nakadi-client 0.2.0.1 → 0.3.0.0

raw patch · 8 files changed

+67/−16 lines, 8 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• nakadi-client.cabal +1/−1
• src/Network/Nakadi/Config.hs +4/−1
• src/Network/Nakadi/Internal/Http.hs +5/−4
• src/Network/Nakadi/Internal/Types/Config.hs +1/−1
• src/Network/Nakadi/Internal/Types/Service.hs +2/−2
• tests/Network/Nakadi/EventTypes/Test.hs +44/−1
• tests/Network/Nakadi/Tests/Common.hs +10/−1
• tests/Tests.hs +0/−5

@@ -3,7 +3,7 @@ -- see: https://github.com/sol/hpack  name:           nakadi-client-version:        0.2.0.1+version:        0.3.0.0 synopsis:       Client library for the Nakadi Event Broker description:    This package implements a client library for interacting with the Nakadi event broker system developed by Zalando. category:       Network

@@ -63,7 +63,10 @@  -- | Install a callback in the provided configuration to use in case -- of deserialization failures when consuming events.-setDeserializationFailureCallback :: (ByteString -> IO ()) -> Config -> Config+setDeserializationFailureCallback ::+  (ByteString -> Text -> IO ())+  -> Config+  -> Config setDeserializationFailureCallback cb = L.deserializationFailureCallback .~ Just cb  -- | Install a callback in the provided configuration which is used

@@ -67,11 +67,12 @@   -> ConduitM ByteString a m () -- ^ Conduit deserializing bytestrings                                 -- into custom values conduitDecode Config { .. } = awaitForever $ \a ->-  case decodeStrict a of-    Just v  -> yield v-    Nothing -> liftIO $ callback a+  case eitherDecodeStrict a of+    Right v  -> yield v+    Left err -> liftIO $ callback a (Text.pack err) -    where callback = fromMaybe (const (return ())) _deserializationFailureCallback+    where callback = fromMaybe dummyCallback _deserializationFailureCallback+          dummyCallback _ _ = return ()  httpBuildRequest ::   (MonadIO m, MonadCatch m)

@@ -28,7 +28,7 @@   , _requestModifier                :: Request -> IO Request   , _manager                        :: Manager   , _consumeParameters              :: ConsumeParameters-  , _deserializationFailureCallback :: Maybe (ByteString -> IO ())+  , _deserializationFailureCallback :: Maybe (ByteString -> Text -> IO ())   , _streamConnectCallback          :: Maybe StreamConnectCallback   , _logFunc                        :: Maybe LogFunc   , _retryPolicy                    :: RetryPolicyM IO

@@ -224,7 +224,7 @@ data Metadata = Metadata   { _eid        :: Text -- ^ Event ID   , _occurredAt :: Timestamp -- ^ Occurred-At timestamp-  , _parentEids :: [Text] -- ^ Event IDs of the Events which triggered this event+  , _parentEids :: Maybe [Text] -- ^ Event IDs of the Events which triggered this event   , _partition  :: Maybe Text -- ^ Partition on which this Event is stored   } deriving (Eq, Show, Generic) @@ -740,7 +740,7 @@   , _occurredAt :: Timestamp   , _receivedAt :: Timestamp   , _version    :: SchemaVersion-  , _parentEids :: [Text]+  , _parentEids :: Maybe [Text]   } deriving (Eq, Show, Generic)  deriveJSON nakadiJsonOptions ''MetadataEnriched

@@ -19,6 +19,7 @@ import           Network.Nakadi.EventTypes.ShiftedCursors.Test import qualified Network.Nakadi.Lenses                         as L import           Network.Nakadi.Tests.Common+import           System.IO.Unsafe import           Test.Tasty import           Test.Tasty.HUnit @@ -31,6 +32,7 @@   , testCase "EventTypeCursorDistances0" (testEventTypeCursorDistances0 conf)   , testCase "EventTypeCursorDistances10" (testEventTypeCursorDistances10 conf)   , testCase "EventTypePublishData" (testEventTypePublishData conf)+  , testCase "EventTypeDeserializationFailure" (testEventTypeDeserializationFailure conf)   , testEventTypesShiftedCursors conf   , testEventTypesCursorsLag conf   ]@@ -108,7 +110,11 @@   eventTypeDelete conf myEventTypeName `catch` (ignoreExnNotFound ())   eventTypeCreate conf myEventType   let event = DataChangeEvent { _payload = Foo "Hello!"-                              , _metadata = Metadata eid (Timestamp now) [] Nothing+                              , _metadata = Metadata { _eid = eid+                                                     , _occurredAt = Timestamp now+                                                     , _parentEids = Nothing+                                                     , _partition = Nothing+                                                     }                               , _dataType = "test.FOO"                               , _dataOp = DataOpUpdate                               }@@ -122,3 +128,40 @@   where delayedPublish events = do           threadDelay (10^6)           eventPublish conf myEventTypeName Nothing events++testEventTypeDeserializationFailure :: Config -> Assertion+testEventTypeDeserializationFailure conf' = do+  now <- getCurrentTime+  eid <- tshow <$> genRandomUUID+  eventTypeDelete conf myEventTypeName `catch` (ignoreExnNotFound ())+  eventTypeCreate conf myEventType+  let event = DataChangeEvent { _payload = Foo "Hello!"+                              , _metadata = Metadata { _eid = eid+                                                     , _occurredAt = Timestamp now+                                                     , _parentEids = Nothing+                                                     , _partition = Nothing+                                                     }+                              , _dataType = "test.FOO"+                              , _dataOp = DataOpUpdate+                              }+  withAsync (delayedPublish [event]) $ \asyncHandle -> do+    link asyncHandle+    eventConsumed :: Maybe (EventStreamBatch WrongFoo) <- runResourceT $ do+      source <- eventSource conf (Just consumeParametersSingle) myEventTypeName Nothing+      runConduit $ source .| headC+    isJust eventConsumed @=? True++  counter <- atomically $ readTVar deserializationFailureCounter+  1 @=? counter++  where delayedPublish events = do+          threadDelay (10^6)+          eventPublish conf myEventTypeName Nothing events++        conf = conf'+               & setDeserializationFailureCallback (deserializationFailureCb deserializationFailureCounter)++        deserializationFailureCb counter _ errMsg = do+          atomically $ modifyTVar counter (+ 1)++        deserializationFailureCounter = unsafePerformIO $ newTVarIO 0

@@ -17,6 +17,11 @@ deriving instance FromJSON Foo deriving instance ToJSON Foo +data WrongFoo = WrongFoo { fortune :: Int } deriving (Show, Eq, Generic)++deriving instance FromJSON WrongFoo+deriving instance ToJSON WrongFoo+ myEventTypeName :: EventTypeName myEventTypeName = "test.FOO" @@ -54,7 +59,11 @@ myDataChangeEvent :: Text -> UTCTime -> DataChangeEvent Foo myDataChangeEvent eid now =  DataChangeEvent   { _payload = Foo "Hello!"-  , _metadata = Metadata eid (Timestamp now) [] Nothing+  , _metadata = Metadata { _eid        = eid+                         , _occurredAt = Timestamp now+                         , _parentEids = Nothing+                         , _partition  = Nothing+                         }   , _dataType = "test.FOO"   , _dataOp = DataOpUpdate   }

@@ -16,14 +16,9 @@   request <- parseRequest nakadiEndpoint   newConfig Nothing request -deserializationFailureCB :: ByteString -> IO ()-deserializationFailureCB event = do-  error $ "Failed to deserialize: " <> show event- main :: IO () main = do   conf <- createConfig-          <&> setDeserializationFailureCallback deserializationFailureCB   defaultMain (tests conf)  tests :: Config -> TestTree