diff --git a/nakadi-client.cabal b/nakadi-client.cabal --- a/nakadi-client.cabal +++ b/nakadi-client.cabal @@ -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 diff --git a/src/Network/Nakadi/Config.hs b/src/Network/Nakadi/Config.hs --- a/src/Network/Nakadi/Config.hs +++ b/src/Network/Nakadi/Config.hs @@ -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 diff --git a/src/Network/Nakadi/Internal/Http.hs b/src/Network/Nakadi/Internal/Http.hs --- a/src/Network/Nakadi/Internal/Http.hs +++ b/src/Network/Nakadi/Internal/Http.hs @@ -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) diff --git a/src/Network/Nakadi/Internal/Types/Config.hs b/src/Network/Nakadi/Internal/Types/Config.hs --- a/src/Network/Nakadi/Internal/Types/Config.hs +++ b/src/Network/Nakadi/Internal/Types/Config.hs @@ -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 diff --git a/src/Network/Nakadi/Internal/Types/Service.hs b/src/Network/Nakadi/Internal/Types/Service.hs --- a/src/Network/Nakadi/Internal/Types/Service.hs +++ b/src/Network/Nakadi/Internal/Types/Service.hs @@ -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 diff --git a/tests/Network/Nakadi/EventTypes/Test.hs b/tests/Network/Nakadi/EventTypes/Test.hs --- a/tests/Network/Nakadi/EventTypes/Test.hs +++ b/tests/Network/Nakadi/EventTypes/Test.hs @@ -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 diff --git a/tests/Network/Nakadi/Tests/Common.hs b/tests/Network/Nakadi/Tests/Common.hs --- a/tests/Network/Nakadi/Tests/Common.hs +++ b/tests/Network/Nakadi/Tests/Common.hs @@ -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 } diff --git a/tests/Tests.hs b/tests/Tests.hs --- a/tests/Tests.hs +++ b/tests/Tests.hs @@ -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