diff --git a/nakadi-client.cabal b/nakadi-client.cabal --- a/nakadi-client.cabal +++ b/nakadi-client.cabal @@ -1,15 +1,15 @@ --- This file has been generated from package.yaml by hpack version 0.20.0. +-- This file has been generated from package.yaml by hpack version 0.28.2. -- -- see: https://github.com/sol/hpack -- --- hash: 2b8a934a5db5c34ab56daa63679bb7c182fecea6e53b4ee05adc77af5c992d09 +-- hash: 9f130201690462832c511918ca52fc51528dc8ccdcee4b68306cb8ef1ed189c2 name: nakadi-client -version: 0.5.0.3 +version: 0.5.1.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 -homepage: https://github.com/mtesseract/nakadi-haskell#readme +homepage: http://nakadi-client.haskell.silverratio.net bug-reports: https://github.com/mtesseract/nakadi-haskell/issues author: Moritz Clasmeier maintainer: mtesseract@silverratio.net @@ -18,7 +18,6 @@ license-file: LICENSE build-type: Simple cabal-version: >= 1.10 - extra-source-files: .gitignore AUTHORS.md @@ -33,47 +32,6 @@ default: False library - hs-source-dirs: - src - default-extensions: NoImplicitPrelude OverloadedStrings DuplicateRecordFields - build-depends: - aeson - , aeson-casing - , base >=4.7 && <5 - , bytestring - , conduit >=1.2.13 && <1.3.0 - , conduit-combinators >=1.1.2 && <1.3.0 - , conduit-extra >=1.2.3.2 && <1.3.0.0 - , containers - , exceptions - , hashable - , http-client - , http-client-tls - , http-conduit >=2.2.4 && <2.3.0 - , http-types - , iso8601-time - , lens - , monad-control - , monad-logger - , mtl - , resourcet - , retry - , safe-exceptions - , scientific - , split - , template-haskell - , text - , time - , transformers - , transformers-base - , unliftio-core - , unordered-containers - , uuid - , vector - if flag(devel) - ghc-options: -Wall -fno-warn-type-defaults -Werror - else - ghc-options: -Wall -fno-warn-type-defaults exposed-modules: Network.Nakadi Network.Nakadi.Prelude @@ -119,14 +77,79 @@ Network.Nakadi.Internal.Types.Logger Network.Nakadi.Internal.Types.Problem Network.Nakadi.Internal.Types.Service + Network.Nakadi.Internal.Types.Subscriptions Network.Nakadi.Internal.Types.Util Network.Nakadi.Internal.Util + Network.Nakadi.Types.Subscriptions Paths_nakadi_client + hs-source-dirs: + src + default-extensions: NoImplicitPrelude OverloadedStrings DuplicateRecordFields + build-depends: + aeson + , aeson-casing + , async-timer >=0.2.0.0 && <0.3 + , base >=4.7 && <5 + , bytestring + , conduit >=1.3.0 && <1.4.0 + , conduit-extra >=1.3.0 && <1.4.0 + , containers + , exceptions + , hashable + , http-client + , http-client-tls + , http-conduit >=2.3.0 && <2.4 + , http-types + , iso8601-time + , lens + , monad-control + , monad-logger + , mtl + , resourcet + , retry + , safe-exceptions >=0.1.7.0 && <0.2 + , scientific + , split + , stm + , template-haskell + , text + , time + , transformers + , transformers-base + , unliftio >=0.2.4.0 && <0.3 + , unliftio-core + , unordered-containers + , uuid + , vector + if flag(devel) + ghc-options: -Wall -fno-warn-type-defaults -Werror + else + ghc-options: -Wall -fno-warn-type-defaults default-language: Haskell2010 test-suite nakadi-client-test-suite type: exitcode-stdio-1.0 main-is: Tests.hs + other-modules: + Network.Nakadi.Config.Test + Network.Nakadi.Connection.Test + Network.Nakadi.EventTypes.CursorsLag.Test + Network.Nakadi.EventTypes.ShiftedCursors.Test + Network.Nakadi.EventTypes.Test + Network.Nakadi.Examples.ListEventTypes.Test + Network.Nakadi.Examples.Subscription.Process + Network.Nakadi.Examples.Subscription.Test + Network.Nakadi.Examples.Test + Network.Nakadi.Internal.Http.Test + Network.Nakadi.Internal.Retry.Test + Network.Nakadi.Internal.Test + Network.Nakadi.Internal.Types.Test + Network.Nakadi.MonadicAPI.Test + Network.Nakadi.Registry.Test + Network.Nakadi.Subscriptions.Processing.Test + Network.Nakadi.Subscriptions.Test + Network.Nakadi.Tests.Common + Paths_nakadi_client hs-source-dirs: tests default-extensions: NoImplicitPrelude OverloadedStrings DuplicateRecordFields @@ -135,12 +158,12 @@ aeson , aeson-casing , async + , async-timer >=0.2.0.0 && <0.3 , base , bytestring - , classy-prelude - , conduit >=1.2.13 && <1.3.0 - , conduit-combinators >=1.1.2 && <1.3.0 - , conduit-extra >=1.2.3.2 && <1.3.0.0 + , classy-prelude >=1.4.0 && <1.5.0 + , conduit >=1.3.0 && <1.4.0 + , conduit-extra >=1.3.0 && <1.4.0 , containers , exceptions , fast-logger @@ -160,13 +183,13 @@ , random , resourcet , retry - , safe-exceptions + , safe-exceptions >=0.1.7.0 && <0.2 , say , scientific , split , stm , stm-chans - , stm-conduit + , stm-conduit >=4.0.0 && <4.1.0 , tasty , tasty-hunit , template-haskell @@ -185,26 +208,4 @@ ghc-options: -Wall -fno-warn-type-defaults -Werror else ghc-options: -Wall -fno-warn-type-defaults - other-modules: - Network.Nakadi.Config.Test - Network.Nakadi.Connection.Test - Network.Nakadi.EventTypes.CursorsLag.Test - Network.Nakadi.EventTypes.ShiftedCursors.Test - Network.Nakadi.EventTypes.Test - Network.Nakadi.Examples.Echo.Echo - Network.Nakadi.Examples.Echo.Test - Network.Nakadi.Examples.ListEventTypes.Test - Network.Nakadi.Examples.Subscription.Process - Network.Nakadi.Examples.Subscription.Test - Network.Nakadi.Examples.Test - Network.Nakadi.Internal.Http.Test - Network.Nakadi.Internal.Retry.Test - Network.Nakadi.Internal.Test - Network.Nakadi.Internal.Types.Test - Network.Nakadi.MonadicAPI.Test - Network.Nakadi.Registry.Test - Network.Nakadi.Subscriptions.Processing.Test - Network.Nakadi.Subscriptions.Test - Network.Nakadi.Tests.Common - Paths_nakadi_client default-language: Haskell2010 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 @@ -29,6 +29,7 @@ , setStreamTimeout , setStreamKeepAliveLimit , setFlowId + , setCommitStrategy , defaultConsumeParameters ) where @@ -48,6 +49,10 @@ defaultRetryPolicy :: MonadIO m => RetryPolicyM m defaultRetryPolicy = fullJitterBackoff 2 <> limitRetries 5 +-- | Default commit strategy. +defaultCommitStrategy :: CommitStrategy +defaultCommitStrategy = CommitSync + -- | Producs a new configuration, with mandatory HTTP manager, default -- consumption parameters and HTTP request template. newConfig @@ -67,6 +72,7 @@ , _http = httpBackend , _httpErrorCallback = Nothing , _flowId = Nothing + , _commitStrategy = defaultCommitStrategy } -- | Producs a new configuration, with mandatory HTTP manager, default @@ -152,6 +158,13 @@ -> Config m -> Config m setFlowId flowId = L.flowId .~ Just flowId + +-- | Set flow ID in the provided configuration. +setCommitStrategy + :: CommitStrategy + -> Config m + -> Config m +setCommitStrategy = (L.commitStrategy .~) -- | Set maximum number of uncommitted events in the provided value of -- consumption parameters. diff --git a/src/Network/Nakadi/EventTypes/Events.hs b/src/Network/Nakadi/EventTypes/Events.hs --- a/src/Network/Nakadi/EventTypes/Events.hs +++ b/src/Network/Nakadi/EventTypes/Events.hs @@ -19,19 +19,12 @@ {-# LANGUAGE TypeFamilies #-} module Network.Nakadi.EventTypes.Events - ( eventsProcessConduit - , eventsProcess - , eventsPublish + ( eventsPublish ) where import Network.Nakadi.Internal.Prelude -import Conduit import Data.Aeson -import qualified Data.ByteString.Lazy as ByteString.Lazy -import Data.Void -import Network.HTTP.Client (responseBody) -import Network.Nakadi.Internal.Config import Network.Nakadi.Internal.Http path :: EventTypeName -> ByteString @@ -39,49 +32,6 @@ "/event-types/" <> encodeUtf8 (unEventTypeName eventTypeName) <> "/events" - -eventsProcess - :: ( MonadNakadi b m - , FromJSON a ) - => Maybe ConsumeParameters - -> EventTypeName - -> Maybe [Cursor] - -> (EventStreamBatch a -> m r) - -> m r -eventsProcess maybeConsumeParameters eventTypeName maybeCursors processor = - eventsProcess maybeConsumeParameters eventTypeName maybeCursors processor - -eventsProcessConduit - :: ( MonadNakadi b m - , MonadMask m - , FromJSON a ) - => Maybe ConsumeParameters - -> EventTypeName - -> Maybe [Cursor] - -> ConduitM (EventStreamBatch a) Void m r - -> m r -eventsProcessConduit maybeConsumeParameters eventTypeName maybeCursors consumer = do - config <- nakadiAsk - let consumeParams = fromMaybe defaultConsumeParameters maybeConsumeParameters - queryParams = buildConsumeQueryParameters consumeParams - httpJsonBodyStream ok200 [ (status429, errorTooManyRequests) - , (status429, errorEventTypeNotFound) ] - (setRequestPath (path eventTypeName) - . includeFlowId config - . setRequestQueryParameters queryParams - . addCursors) $ - handler config - - where addCursors = case maybeCursors of - Just cursors -> let cursors' = ByteString.Lazy.toStrict (encode cursors) - in addRequestHeader "X-Nakadi-Cursors" cursors' - Nothing -> identity - - handler config response = - responseBody response - .| linesUnboundedAsciiC - .| conduitDecode config - $$ consumer -- | @POST@ to @\/event-types\/NAME\/events@. Publishes a batch of -- events for the specified event type. diff --git a/src/Network/Nakadi/Internal/Lenses.hs b/src/Network/Nakadi/Internal/Lenses.hs --- a/src/Network/Nakadi/Internal/Lenses.hs +++ b/src/Network/Nakadi/Internal/Lenses.hs @@ -35,7 +35,6 @@ makeNakadiLenses ''Config makeNakadiLenses ''HttpBackend makeNakadiLenses ''Cursor -makeNakadiLenses ''EventStreamBatch makeNakadiLenses ''DataChangeEvent makeNakadiLenses ''DataChangeEventEnriched makeNakadiLenses ''SubscriptionEventStreamBatch @@ -54,6 +53,7 @@ makeNakadiLenses ''SubscriptionEventTypeStatsResult makeNakadiLenses ''ConsumeParameters makeNakadiLenses ''SubscriptionCursorCommit +makeNakadiLenses ''SubscriptionCursor makeNakadiLenses ''CursorCommit makeNakadiLenses ''SubscriptionsListResponse makeNakadiLenses ''Subscription diff --git a/src/Network/Nakadi/Internal/Types.hs b/src/Network/Nakadi/Internal/Types.hs --- a/src/Network/Nakadi/Internal/Types.hs +++ b/src/Network/Nakadi/Internal/Types.hs @@ -32,6 +32,7 @@ , module Network.Nakadi.Internal.Types.Service , module Network.Nakadi.Internal.Types.Util , module Network.Nakadi.Internal.Types.Base + , module Network.Nakadi.Internal.Types.Subscriptions , HasNakadiConfig(..) , MonadNakadi(..) , MonadNakadiIO @@ -44,14 +45,14 @@ import Control.Monad.IO.Unlift import Control.Monad.Logger import Control.Monad.State.Class -import qualified Control.Monad.State.Lazy as State.Lazy -import qualified Control.Monad.State.Strict as State.Strict +import qualified Control.Monad.State.Lazy as State.Lazy +import qualified Control.Monad.State.Strict as State.Strict import Control.Monad.Trans.Class import Control.Monad.Trans.Control -import Control.Monad.Trans.Reader (ReaderT (..)) +import Control.Monad.Trans.Reader (ReaderT (..)) import Control.Monad.Trans.Resource -import qualified Control.Monad.Writer.Lazy as Writer.Lazy -import qualified Control.Monad.Writer.Strict as Writer.Strict +import qualified Control.Monad.Writer.Lazy as Writer.Lazy +import qualified Control.Monad.Writer.Strict as Writer.Strict import Network.Nakadi.Internal.Prelude import Network.Nakadi.Internal.Types.Base import Network.Nakadi.Internal.Types.Config @@ -59,6 +60,7 @@ import Network.Nakadi.Internal.Types.Logger import Network.Nakadi.Internal.Types.Problem import Network.Nakadi.Internal.Types.Service +import Network.Nakadi.Internal.Types.Subscriptions import Network.Nakadi.Internal.Types.Util -- * Define Typeclasses 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 @@ -18,11 +18,12 @@ import Conduit import Control.Retry -import qualified Data.ByteString.Lazy as LB +import qualified Data.ByteString.Lazy as LB import Network.HTTP.Client import Network.Nakadi.Internal.Prelude import Network.Nakadi.Internal.Types.Logger import Network.Nakadi.Internal.Types.Service +import Network.Nakadi.Internal.Types.Subscriptions -- | Config @@ -46,6 +47,7 @@ , _http :: HttpBackend m , _httpErrorCallback :: Maybe (HttpErrorCallback m) , _flowId :: Maybe FlowId + , _commitStrategy :: CommitStrategy } -> Config m -- | ConsumeParameters 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 @@ -727,15 +727,6 @@ deriveJSON nakadiJsonOptions ''EventMetadataEnriched --- | EventStreamBatch - -data EventStreamBatch a = EventStreamBatch - { _cursor :: Cursor -- ^ Cursor for this batch - , _events :: Maybe (Vector a) -- ^ Events in this batch - } deriving (Show, Generic) - -deriveJSON nakadiJsonOptions ''EventStreamBatch - -- | SubscriptionEventStreamBatch data SubscriptionEventStreamBatch a = SubscriptionEventStreamBatch diff --git a/src/Network/Nakadi/Internal/Types/Subscriptions.hs b/src/Network/Nakadi/Internal/Types/Subscriptions.hs new file mode 100644 --- /dev/null +++ b/src/Network/Nakadi/Internal/Types/Subscriptions.hs @@ -0,0 +1,40 @@ +{-| +Module : Network.Nakadi.Types.Subscriptions +Description : Nakadi Service Types +Copyright : (c) Moritz Clasmeier 2018 +License : BSD3 +Maintainer : mtesseract@silverratio.net +Stability : experimental +Portability : POSIX + +This module defines types, which are related to consumption of +subscriptions, which are not modelled after the Nakadi API. +-} + +module Network.Nakadi.Internal.Types.Subscriptions where + +import Data.Int + + +-- | This type encodes the supported strategies for subscription +-- cursor committing. +data CommitStrategy + = CommitSync -- ^ This strategy synchronously commits every + -- cursor. + | CommitAsync CommitBufferingStrategy -- ^ This strategy sends cursors to be + -- committed to a dedicated thread + -- responsible for committing them. Cursors + -- are commited one by one, without special + -- buffering logic. + +-- | This type encodes the supported buffering strategies for +-- asynchronous subscription cursor committing. +data CommitBufferingStrategy + = CommitNoBuffer -- ^ No buffering at all. + | CommitTimeBuffer Int32 -- ^ Buffer for the specified duration, + -- given in milliseconds. + | CommitSmartBuffer -- ^ Buffer for a fixed duration, but + -- committing cursors immediately if the + -- number of events processed since the + -- last commit crosses a threshold derived + -- from @maxUncommittedEvents@. diff --git a/src/Network/Nakadi/Internal/Util.hs b/src/Network/Nakadi/Internal/Util.hs --- a/src/Network/Nakadi/Internal/Util.hs +++ b/src/Network/Nakadi/Internal/Util.hs @@ -22,11 +22,9 @@ import Network.Nakadi.Internal.Prelude +import Conduit hiding (throwM) import Data.Aeson -import Data.ByteString.Builder import qualified Data.ByteString.Lazy as ByteString.Lazy -import Data.Conduit -import Data.Conduit.Combinators hiding (decodeUtf8, map) import qualified Data.Text as Text import Network.HTTP.Simple @@ -37,7 +35,7 @@ => ConduitM () ByteString b () -> b ByteString.Lazy.ByteString conduitDrainToLazyByteString conduit = - toLazyByteString <$> (conduit $$ sinkBuilder) + runConduit $ conduit .| sinkLazy decodeThrow :: (FromJSON a, MonadThrow m) diff --git a/src/Network/Nakadi/Subscriptions/Events.hs b/src/Network/Nakadi/Subscriptions/Events.hs --- a/src/Network/Nakadi/Subscriptions/Events.hs +++ b/src/Network/Nakadi/Subscriptions/Events.hs @@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Subscriptions.Events Description : Implementation of Nakadi Subscription Events API -Copyright : (c) Moritz Schulte 2017, 2018 +Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental @@ -13,6 +13,8 @@ {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} @@ -25,38 +27,62 @@ import Network.Nakadi.Internal.Prelude import Conduit hiding (throwM) +import qualified Control.Concurrent.Async.Timer as Timer +import Control.Concurrent.STM (TBQueue, TVar, + atomically, modifyTVar, + newTBQueue, newTVar, + readTBQueue, readTVar, + retry, swapTVar, + writeTBQueue) +import Control.Lens +import Control.Monad.Logger import Data.Aeson -import Data.Void +import qualified Data.Conduit.List as Conduit (mapM_) +import Data.HashMap.Strict (HashMap) +import qualified Data.HashMap.Strict as HashMap +import qualified Data.Vector as Vector import Network.HTTP.Client (responseBody) import Network.HTTP.Simple import Network.HTTP.Types - import Network.Nakadi.Internal.Config import Network.Nakadi.Internal.Conversions import Network.Nakadi.Internal.Http import qualified Network.Nakadi.Internal.Lenses as L import Network.Nakadi.Subscriptions.Cursors +import UnliftIO.Async +-- | Consumes the specified subscription using the commit strategy +-- contained in the configuration. Each consumed batch of subscription +-- events is provided to the provided batch processor action. If this +-- action throws an exception, subscription consumption will terminate. subscriptionProcess :: ( MonadNakadi b m + , MonadUnliftIO m , MonadMask m , FromJSON a ) - => Maybe ConsumeParameters - -> SubscriptionId - -> (SubscriptionEventStreamBatch a -> m ()) + => Maybe ConsumeParameters -- ^ 'ConsumeParameters' + -- to use + -> SubscriptionId -- ^ Subscription to consume + -> (SubscriptionEventStreamBatch a -> m ()) -- ^ Batch processor action -> m () subscriptionProcess maybeConsumeParameters subscriptionId processor = subscriptionProcessConduit maybeConsumeParameters subscriptionId conduit where conduit = iterMC processor +-- | Conduit based interface for subscription consumption. Consumes +-- the specified subscription using the commit strategy contained in +-- the configuration. Each consumed event batch is then processed by +-- the provided conduit. If an exception is thrown inside the conduit, +-- subscription consumption will terminate. subscriptionProcessConduit :: ( MonadNakadi b m + , MonadUnliftIO m , MonadMask m , FromJSON a - , L.HasNakadiSubscriptionCursor c ) - => Maybe ConsumeParameters - -> SubscriptionId - -> ConduitM (SubscriptionEventStreamBatch a) c m () + , batch ~ SubscriptionEventStreamBatch a ) + => Maybe ConsumeParameters -- ^ 'ConsumeParameters' to use + -> SubscriptionId -- ^ Subscription to consume + -> ConduitM batch batch m () -- ^ Conduit processor. -> m () subscriptionProcessConduit maybeConsumeParameters subscriptionId processor = do config <- nakadiAsk @@ -66,38 +92,245 @@ (includeFlowId config . setRequestPath path . setRequestQueryParameters queryParams) $ - handler config - - - where buildSubscriptionEventStream response = - case listToMaybe (getResponseHeader "X-Nakadi-StreamId" response) of - Just streamId -> - pure SubscriptionEventStream - { _streamId = StreamId (decodeUtf8 streamId) - , _subscriptionId = subscriptionId } - Nothing -> - throwM StreamIdMissing + subscriptionProcessHandler consumeParams subscriptionId processor - path = "/subscriptions/" + where path = "/subscriptions/" <> subscriptionIdToByteString subscriptionId <> "/events" - handler config response = do - eventStream <- buildSubscriptionEventStream response - runConduit $ - responseBody response - .| linesUnboundedAsciiC - .| conduitDecode config - .| processor - .| subscriptionSink eventStream +-- | Derive a 'SubscriptionEventStream' from the provided +-- 'SubscriptionId' and Nakadi streaming response. +buildSubscriptionEventStream + :: MonadThrow m + => SubscriptionId + -> Response a + -> m SubscriptionEventStream +buildSubscriptionEventStream subscriptionId response = + case listToMaybe (getResponseHeader "X-Nakadi-StreamId" response) of + Just streamId -> + pure SubscriptionEventStream + { _streamId = StreamId (decodeUtf8 streamId) + , _subscriptionId = subscriptionId } + Nothing -> + throwM StreamIdMissing +-- | This function processes a subscription, taking care of applying +-- the configured committing strategy. +subscriptionProcessHandler + :: ( MonadNakadi b m + , MonadUnliftIO m + , MonadMask m + , FromJSON a + , batch ~ (SubscriptionEventStreamBatch a) ) + => ConsumeParameters + -> SubscriptionId -- ^ Subscription ID required for committing. + -> ConduitM batch batch m () -- ^ User provided Conduit for stream. + -> Response (ConduitM () ByteString m ()) -- ^ Streaming response from Nakadi + -> m () +subscriptionProcessHandler consumeParams subscriptionId processor response = do + config <- nakadiAsk + eventStream <- buildSubscriptionEventStream subscriptionId response + let producer = responseBody response + .| linesUnboundedAsciiC + .| conduitDecode config + .| processor + case config^.L.commitStrategy of + CommitSync -> + -- Synchronous case: Simply use a Conduit sink that commits + -- every cursor. + runConduit $ producer .| subscriptionSink eventStream + CommitAsync bufferingStrategy -> do + -- Asynchronous case: Create a new queue and spawn a cursor + -- committer thread depending on the configured commit buffering + -- method. Then execute the provided Conduit processor with a + -- sink that sends cursor information to the queue. The cursor + -- committer thread reads from this queue and processes the + -- cursors. + queue <- liftIO . atomically $ newTBQueue 1024 + withAsync (subscriptionCommitter bufferingStrategy consumeParams eventStream queue) $ + \ asyncHandle -> do + link asyncHandle + runConduit $ producer .| Conduit.mapM_ (sendToQueue queue) + where sendToQueue queue batch = liftIO . atomically $ do + let cursor = batch^.L.cursor + events = fromMaybe Vector.empty (batch^.L.events) + nEvents = length events + writeTBQueue queue (nEvents, cursor) + -- | Sink which can be used as sink for Conduits processing -- subscriptions events. This sink takes care of committing events. It -- can consume any values which contain Subscription Cursors. -subscriptionSink :: - (MonadNakadi b m, L.HasNakadiSubscriptionCursor a) +subscriptionSink + :: (MonadIO m, MonadNakadi b m) => SubscriptionEventStream - -> ConduitM a Void m () -subscriptionSink eventStream = - awaitForever $ lift . subscriptionCursorCommit eventStream . (: []) + -> ConduitM (SubscriptionEventStreamBatch a) void m () +subscriptionSink eventStream = do + config <- lift nakadiAsk + awaitForever $ \ batch -> lift $ do + let cursor = batch^.L.cursor + catchAny (commitOneCursor eventStream cursor) $ \ exn -> nakadiLiftBase $ + case config^.L.logFunc of + Just logFunc -> logFunc "nakadi-client" LevelWarn $ toLogStr $ + "Failed to synchronously commit cursor: " <> tshow exn + Nothing -> + pure () + +type CursorsMap = HashMap (EventTypeName, PartitionName) (Int, SubscriptionCursor) + +emptyCursorsMap :: CursorsMap +emptyCursorsMap = HashMap.empty + +cursorKey :: SubscriptionCursor -> (EventTypeName, PartitionName) +cursorKey cursor = (cursor^.L.eventType, cursor^.L.partition) + +-- | Implementation for the 'CommitNoBuffer' strategy: We simply read +-- every cursor and commit it in order. +unbufferedCommitLoop + :: (MonadNakadi b m, MonadIO m) + => SubscriptionEventStream + -> TBQueue (Int, SubscriptionCursor) + -> m () +unbufferedCommitLoop eventStream queue = do + config <- nakadiAsk + forever $ do + (_nEvents, cursor) <- liftIO . atomically . readTBQueue $ queue + catchAny (subscriptionCursorCommit eventStream [cursor]) $ \ exn -> do + nakadiLiftBase $ + case config^.L.logFunc of + Just logFunc -> logFunc "nakadi-client" LevelWarn $ toLogStr $ + "Failed to commit cursor " <> tshow cursor <> ": " <> tshow exn + Nothing -> + pure () + +cursorBufferSize :: ConsumeParameters -> Int +cursorBufferSize consumeParams = + case consumeParams^.L.maxUncommittedEvents of + Nothing -> 1 + Just n -> n + & fromIntegral + & (* safetyFactor) + & round + + where safetyFactor = 0.5 + +-- | Main function for the cursor committer thread. Logic depends on +-- the provided buffering strategy. +subscriptionCommitter + :: ( MonadNakadi b m + , MonadUnliftIO m + , MonadMask m ) + => CommitBufferingStrategy + -> ConsumeParameters + -> SubscriptionEventStream + -> TBQueue (Int, SubscriptionCursor) + -> m () + +-- | Implementation for the 'CommitNoBuffer' strategy: We simply read +-- every cursor and commit it in order. +subscriptionCommitter CommitNoBuffer _consumeParams eventStream queue = + unbufferedCommitLoop eventStream queue + +-- | Implementation of the 'CommitTimeBuffer' strategy: We use an +-- async timer for committing cursors at specified intervals. +subscriptionCommitter (CommitTimeBuffer millis) _consumeParams eventStream queue = do + let timerConf = Timer.defaultConf + & Timer.setInitDelay (fromIntegral millis) + & Timer.setInterval (fromIntegral millis) + cursorsMap <- liftIO . atomically $ newTVar emptyCursorsMap + withAsync (cursorConsumer cursorsMap) $ \ asyncCursorConsumer -> do + link asyncCursorConsumer + Timer.withAsyncTimer timerConf $ \ timer -> forever $ do + Timer.wait timer + commitAllCursors eventStream cursorsMap + + where -- | The cursorsConsumer drains the cursors queue and adds each + -- cursor to the provided cursorsMap. + cursorConsumer cursorsMap = forever . liftIO . atomically $ do + (_, cursor) <- readTBQueue queue + modifyTVar cursorsMap (HashMap.insert (cursorKey cursor) (0, cursor)) + +-- | Implementation of the 'CommitSmartBuffer' strategy: We use an +-- async timer for committing cursors at specified intervals, but if +-- the number of uncommitted events reaches some threshold before the +-- next scheduled commit, a commit is being done right away and the +-- timer is resetted. +subscriptionCommitter CommitSmartBuffer consumeParams eventStream queue = do + let millisDefault = 1000 + nMaxEvents = cursorBufferSize consumeParams + timerConf = Timer.defaultConf + & Timer.setInitDelay (fromIntegral millisDefault) + & Timer.setInterval (fromIntegral millisDefault) + if nMaxEvents > 1 + then do cursorsMap <- liftIO . atomically $ newTVar emptyCursorsMap + withAsync (cursorConsumer cursorsMap) $ \ asyncCursorConsumer -> do + link asyncCursorConsumer + Timer.withAsyncTimer timerConf $ cursorCommitter cursorsMap nMaxEvents + else unbufferedCommitLoop eventStream queue + + where -- | The cursorsConsumer drains the cursors queue and adds + -- each cursor to the provided cursorsMap. + cursorConsumer cursorsMap = forever . liftIO . atomically $ do + (nEvents, cursor) <- readTBQueue queue + modifyTVar cursorsMap $ + HashMap.insertWith updateCursor (cursorKey cursor) (nEvents, cursor) + + -- | Adds the old number of events to the new entry in the + -- cursors map. + updateCursor cursorNew _cursorOld @ (nEventsOld, _) = + cursorNew & _1 %~ (+ nEventsOld) + + -- | Committer loop. + cursorCommitter cursorsMap nMaxEvents timer = forever $ do + race (Timer.wait timer) (maxEventsReached cursorsMap nMaxEvents) >>= \ case + Left _ -> + -- Timer has elapsed, simply commit all currently + -- buffered cursors. + commitAllCursors eventStream cursorsMap + Right _ -> do + -- Events processed since last cursor commit have + -- crosses configured threshold for at least one + -- partition. Commit cursors on all such partitions. + Timer.reset timer + commitAllCursors eventStream cursorsMap + + -- | Returns list of cursors that should be committed + -- considering the number of events processed on the + -- respective partition since the last commit. Blocks until at + -- least one such cursor is found. + maxEventsReached cursorsMap nMaxEvents = liftIO . atomically $ do + cursorsList <- HashMap.toList <$> readTVar cursorsMap + let cursorsCommit = filter (shouldBeCommitted nMaxEvents) cursorsList + if null cursorsCommit + then retry + else pure () + + -- | Returns True if the provided cursor should be committed. + shouldBeCommitted nMaxEvents cursor = cursor^._2._1 >= nMaxEvents + +-- | This function commits all cursors in the provided cursorsMap. +commitAllCursors + :: (MonadNakadi b m, MonadIO m) + => SubscriptionEventStream + -> TVar CursorsMap + -> m () +commitAllCursors eventStream cursorsMap = do + cursors <- liftIO . atomically $ swapTVar cursorsMap emptyCursorsMap + forM_ cursors $ \ (_nEvents, cursor) -> commitOneCursor eventStream cursor + +-- | This function takes care of committing a single cursor. Exceptions will be +-- catched and logged, but the failure will NOT be propagated. This means that +-- Nakadi itself is in control of disconnecting us. +commitOneCursor + :: (MonadIO m, MonadNakadi b m) + => SubscriptionEventStream + -> SubscriptionCursor + -> m () +commitOneCursor eventStream cursor = do + config <- nakadiAsk + catchAny (subscriptionCursorCommit eventStream [cursor]) $ \ exn -> nakadiLiftBase $ + case config^.L.logFunc of + Just logFunc -> logFunc "nakadi-client" LevelWarn $ toLogStr $ + "Failed to commit cursor " <> tshow cursor <> ": " <> tshow exn + Nothing -> + pure () diff --git a/src/Network/Nakadi/Types.hs b/src/Network/Nakadi/Types.hs --- a/src/Network/Nakadi/Types.hs +++ b/src/Network/Nakadi/Types.hs @@ -16,6 +16,7 @@ , module Network.Nakadi.Types.Logger , module Network.Nakadi.Types.Problem , module Network.Nakadi.Types.Service + , module Network.Nakadi.Types.Subscriptions , MonadNakadiBase(..) , MonadNakadi(..) , HasNakadiConfig(..) @@ -30,3 +31,4 @@ import Network.Nakadi.Types.Logger import Network.Nakadi.Types.Problem import Network.Nakadi.Types.Service +import Network.Nakadi.Types.Subscriptions diff --git a/src/Network/Nakadi/Types/Service.hs b/src/Network/Nakadi/Types/Service.hs --- a/src/Network/Nakadi/Types/Service.hs +++ b/src/Network/Nakadi/Types/Service.hs @@ -27,7 +27,6 @@ , StreamId(..) , SubscriptionEventStream(..) , Timestamp(..) - , EventStreamBatch(..) , SubscriptionEventStreamBatch(..) , EventId(..) , Partition(..) diff --git a/src/Network/Nakadi/Types/Subscriptions.hs b/src/Network/Nakadi/Types/Subscriptions.hs new file mode 100644 --- /dev/null +++ b/src/Network/Nakadi/Types/Subscriptions.hs @@ -0,0 +1,19 @@ +{-| +Module : Network.Nakadi.Types.Subscriptions +Description : Nakadi Service Subscription Types +Copyright : (c) Moritz Clasmeier 2018 +License : BSD3 +Maintainer : mtesseract@silverratio.net +Stability : experimental +Portability : POSIX + +This module exposes types, which are related to consumption of +subscriptions, which are not modelled after the Nakadi API. +-} + +module Network.Nakadi.Types.Subscriptions + ( CommitStrategy(..) + , CommitBufferingStrategy(..) + ) where + +import Network.Nakadi.Internal.Types.Subscriptions diff --git a/tests/Network/Nakadi/Config/Test.hs b/tests/Network/Nakadi/Config/Test.hs --- a/tests/Network/Nakadi/Config/Test.hs +++ b/tests/Network/Nakadi/Config/Test.hs @@ -9,7 +9,7 @@ module Network.Nakadi.Config.Test where -import ClassyPrelude hiding (catch, throwM) +import ClassyPrelude hiding (catch) import Control.Lens import Control.Monad.Catch (MonadThrow (..)) diff --git a/tests/Network/Nakadi/Connection/Test.hs b/tests/Network/Nakadi/Connection/Test.hs --- a/tests/Network/Nakadi/Connection/Test.hs +++ b/tests/Network/Nakadi/Connection/Test.hs @@ -14,6 +14,7 @@ import System.IO.Unsafe import Test.Tasty import Test.Tasty.HUnit +import UnliftIO.Concurrent testConnection :: TestTree testConnection = testGroup "Connection" @@ -75,9 +76,9 @@ testResponseTimeoutSuccess :: Assertion testResponseTimeoutSuccess = do - let timeout = responseTimeoutMicro (5 * 10^6) -- Accept delay of 5s + let rspTimeout = responseTimeoutMicro (5 * 10^6) -- Accept delay of 5s request = testServerRequest { port = testServerResponseTimeoutPort - , responseTimeout = timeout } + , responseTimeout = rspTimeout } conf = newConfigIO request :: ConfigIO withAsync (run testServerResponseTimeoutPort testServerResponseTimeoutApp) $ \_serverHandle -> do events <- runNakadiT conf eventTypesList @@ -86,9 +87,9 @@ testResponseTimeoutFail :: Assertion testResponseTimeoutFail = do res <- try $ do - let timeout = responseTimeoutMicro (3 * 10^6) -- Accept delay of 3s + let rspTimeout = responseTimeoutMicro (3 * 10^6) -- Accept delay of 3s request = testServerRequest { port = testServerResponseTimeoutPort - , responseTimeout = timeout } + , responseTimeout = rspTimeout } conf = newConfigIO request :: ConfigIO withAsync (run testServerResponseTimeoutPort testServerResponseTimeoutApp) $ \_serverHandle -> runNakadiT conf eventTypesList diff --git a/tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs b/tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs --- a/tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs +++ b/tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs @@ -28,7 +28,7 @@ let cursorsMap = Map.fromList $ map (\Partition { .. } -> (_partition, _newestAvailableOffset)) partitions lagMap <- cursorsLag myEventTypeName cursorsMap - recreateEvent myEventTypeName myEventType + recreateEvent myEventType liftIO $ do Map.size cursorsMap @=? Map.size lagMap forM_ (Map.toList lagMap) $ \(_, lag) -> @@ -38,7 +38,7 @@ testCursorsLagN conf n = runApp . runNakadiT conf $ do now <- liftIO getCurrentTime eid <- EventId <$> liftIO genRandomUUID - recreateEvent myEventTypeName myEventType + recreateEvent myEventType partitions <- eventTypePartitions myEventTypeName let cursorsMap = Map.fromList $ map (\Partition { .. } -> (_partition, _newestAvailableOffset)) partitions diff --git a/tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs b/tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs --- a/tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs +++ b/tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs @@ -25,7 +25,7 @@ testShiftedCursorsZero :: Config App -> Assertion testShiftedCursorsZero conf = runApp . runNakadiT conf $ do - recreateEvent myEventTypeName myEventType + recreateEvent myEventType partitions <- eventTypePartitions myEventTypeName let cursors = map extractCursor partitions cursors' <- cursorsShift myEventTypeName cursors 0 @@ -35,7 +35,7 @@ testShiftedCursorsN conf n = runApp . runNakadiT conf $ do now <- liftIO getCurrentTime eid <- EventId <$> liftIO randomIO - recreateEvent myEventTypeName myEventType + recreateEvent myEventType partitions <- eventTypePartitions myEventTypeName let cursors = map extractCursor partitions liftIO $ length cursors > 0 @=? True 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 @@ -1,4 +1,5 @@ {-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-} @@ -11,18 +12,17 @@ import ClassyPrelude hiding (withAsync) -import Conduit hiding - (runResourceT) import Control.Lens import Data.Function ((&)) import Network.Nakadi import Network.Nakadi.EventTypes.CursorsLag.Test import Network.Nakadi.EventTypes.ShiftedCursors.Test import qualified Network.Nakadi.Lenses as L +import qualified Data.Vector as Vector import Network.Nakadi.Tests.Common -import System.IO.Unsafe import Test.Tasty import Test.Tasty.HUnit +import System.IO.Unsafe import UnliftIO.Async testEventTypes :: Config App -> TestTree @@ -102,6 +102,27 @@ let totalDistances = sum distances liftIO $ totalDistances @=? 10 +mySubscription :: Subscription +mySubscription = Subscription + { _id = Nothing + , _owningApplication = "test-suite" + , _eventTypes = [EventTypeName "test.FOO"] + , _consumerGroup = Nothing -- ?? + , _createdAt = Nothing + , _readFrom = Just SubscriptionPositionEnd + , _initialCursors = Nothing + } + +createMySubscription :: (MonadUnliftIO m, MonadNakadi b m) + => m SubscriptionId +createMySubscription = do + newSubscription <- subscriptionCreate mySubscription `catch` + \ case + SubscriptionExistsAlready s -> pure s + exn -> throwIO exn + let (Just subscriptionId) = newSubscription^.L.id + pure subscriptionId + consumeParametersSingle :: ConsumeParameters consumeParametersSingle = defaultConsumeParameters & setBatchLimit 1 @@ -111,8 +132,7 @@ testEventTypePublishData conf = runApp . runNakadiT conf $ do now <- liftIO getCurrentTime eid <- EventId <$> genRandomUUID - eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ()) - eventTypeCreate myEventType + recreateEvent myEventType let event = DataChangeEvent { _payload = Foo "Hello!" , _metadata = EventMetadata { _eid = eid , _occurredAt = Timestamp now @@ -122,18 +142,30 @@ , _dataType = "test.FOO" , _dataOp = DataOpUpdate } - withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do + subscriptionId <- createMySubscription + batchTv <- newTVarIO Nothing + res <- try $ withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do liftIO $ link asyncHandle - (Just batchConsumed) :: Maybe (EventStreamBatch (DataChangeEvent Foo)) <- - eventsProcessConduit (Just consumeParametersSingle) myEventTypeName Nothing headC - liftIO $ isJust (batchConsumed^.L.events) @=? True + subscriptionProcess (Just consumeParametersSingle) subscriptionId (storeBatch batchTv) + liftIO $ (Left TerminateConsumption) @=? res + Just batch <- atomically $ readTVar batchTv + let Just events = batch^.L.events :: Maybe (Vector (DataChangeEvent Foo)) + liftIO $ True @=? (Vector.length events > 0) +storeBatch + :: MonadIO m + => TVar (Maybe (SubscriptionEventStreamBatch a)) + -> SubscriptionEventStreamBatch a + -> m () +storeBatch batchTv batch = do + atomically $ writeTVar batchTv (Just batch) + throwIO TerminateConsumption + testEventTypeParseFlowId :: Config App -> Assertion testEventTypeParseFlowId conf = runApp . runNakadiT conf $ do now <- liftIO getCurrentTime eid <- EventId <$> genRandomUUID - eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ()) - eventTypeCreate myEventType + recreateEvent myEventType let event = DataChangeEvent { _payload = Foo "Hello!" , _metadata = EventMetadata { _eid = eid , _occurredAt = Timestamp now @@ -144,23 +176,21 @@ , _dataOp = DataOpUpdate } expectedFlowId = Just $ FlowId "12345" - withAsync (delayedPublish expectedFlowId [event]) $ \asyncHandle -> do + subscriptionId <- createMySubscription + batchTv <- newTVarIO Nothing + res <- try $ withAsync (delayedPublish expectedFlowId [event]) $ \ asyncHandle -> do liftIO $ link asyncHandle - Just batchConsumed :: Maybe (EventStreamBatch (DataChangeEventEnriched Foo)) <- - eventsProcessConduit (Just consumeParametersSingle) myEventTypeName Nothing headC - let (Just events) = (batchConsumed^.L.events) - liftIO $ case toList events of - [DataChangeEventEnriched _ x _ _] -> - x^.L.flowId @=? expectedFlowId - [] -> assertFailure "Received no events" - _ -> assertFailure "Did not receive a singleton event list" + subscriptionProcess (Just consumeParametersSingle) subscriptionId (storeBatch batchTv) + liftIO $ (Left TerminateConsumption) @=? res + Just batch <- atomically $ readTVar batchTv + let Just (e:_) = toList <$> (batch^.L.events) :: Maybe [DataChangeEventEnriched Foo] + liftIO $ expectedFlowId @=? e^.L.metadata.L.flowId testEventTypeDeserializationFailureException :: Config App -> Assertion testEventTypeDeserializationFailureException conf = runApp . runNakadiT conf $ do now <- liftIO getCurrentTime eid <- EventId <$> genRandomUUID - eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ()) - eventTypeCreate myEventType + recreateEvent myEventType let event = DataChangeEvent { _payload = Foo "Hello!" , _metadata = EventMetadata { _eid = eid , _occurredAt = Timestamp now @@ -170,10 +200,12 @@ , _dataType = "test.FOO" , _dataOp = DataOpUpdate } - res :: Either NakadiException (Maybe (EventStreamBatch WrongFoo)) <- try $ + subscriptionId <- createMySubscription + res :: Either NakadiException () <- try $ withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do liftIO $ link asyncHandle - eventsProcessConduit (Just consumeParametersSingle) myEventTypeName Nothing headC + subscriptionProcess (Just consumeParametersSingle) subscriptionId $ + \ (_batch :: SubscriptionEventStreamBatch ()) -> pure () case res of Left (DeserializationFailure _ _) -> pure () @@ -186,8 +218,7 @@ testEventTypeDeserializationFailure conf' = runApp . runNakadiT conf $ do now <- liftIO getCurrentTime eid <- EventId <$> genRandomUUID - eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ()) - eventTypeCreate myEventType + recreateEvent myEventType let event = DataChangeEvent { _payload = Foo "Hello!" , _metadata = EventMetadata { _eid = eid , _occurredAt = Timestamp now @@ -197,12 +228,13 @@ , _dataType = "test.FOO" , _dataOp = DataOpUpdate } - withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do + subscriptionId <- createMySubscription + res <- try $ withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do liftIO $ link asyncHandle - eventConsumed :: Maybe (EventStreamBatch WrongFoo) <- - eventsProcessConduit (Just consumeParametersSingle) myEventTypeName Nothing headC - liftIO $ isJust eventConsumed @=? True - + subscriptionProcess (Just consumeParametersSingle) subscriptionId $ + \ (_batch :: SubscriptionEventStreamBatch WrongFoo) -> + throwIO TerminateConsumption + liftIO $ Left TerminateConsumption @=? res counter <- atomically $ readTVar deserializationFailureCounter liftIO $ 1 @=? counter diff --git a/tests/Network/Nakadi/Examples/Echo/Echo.hs b/tests/Network/Nakadi/Examples/Echo/Echo.hs deleted file mode 100644 --- a/tests/Network/Nakadi/Examples/Echo/Echo.hs +++ /dev/null @@ -1,42 +0,0 @@ -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE GADTs #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeFamilies #-} - -module Network.Nakadi.Examples.Echo.Echo (runEcho) where - -import ClassyPrelude -import Conduit -import Control.Concurrent.Async.Lifted (link, waitEither_) -import Control.Lens -import Data.Aeson -import Data.Conduit.TQueue -import Network.Nakadi -import qualified Network.Nakadi.Lenses as L - --- Example program which consumes the events for the event type --- "test-event" and republishes them unchanged under the event type --- "test-event-copy". - -runEcho :: EventTypeName -> EventTypeName -> NakadiT IO IO () -runEcho eventNameInput eventNameOutput = - runResourceT $ do - channel :: TBQueue (Vector Value) <- atomically $ newTBQueue 1024 - consumer <- async $ consumeEvents eventNameInput channel - publisher <- async $ publishEvents eventNameOutput channel - link consumer - link publisher - waitEither_ consumer publisher - - where consumeEvents eventName channel = - eventsProcessConduit (Just consumeParameters) eventName Nothing $ - concatMapC (view L.events) - .| mapC (fmap (toJSON :: DataChangeEvent Value -> Value)) - .| sinkTBQueue channel - - publishEvents eventName channel = - sourceTBQueue channel - .| mapC toList - $$ mapM_C (eventsPublish eventName) - - consumeParameters = defaultConsumeParameters & L.batchFlushTimeout .~ Just 1 diff --git a/tests/Network/Nakadi/Examples/Echo/Test.hs b/tests/Network/Nakadi/Examples/Echo/Test.hs deleted file mode 100644 --- a/tests/Network/Nakadi/Examples/Echo/Test.hs +++ /dev/null @@ -1,84 +0,0 @@ -{-# LANGUAGE DuplicateRecordFields #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeApplications #-} - -module Network.Nakadi.Examples.Echo.Test (testEcho) where - -import ClassyPrelude -import Conduit -import Control.Concurrent.Async.Lifted (wait) -import Control.Lens -import qualified Data.Vector as Vector -import qualified Network.Nakadi as Nakadi -import Network.Nakadi.Examples.Echo.Echo -import qualified Network.Nakadi.Lenses as L -import Network.Nakadi.Tests.Common -import Test.Tasty.HUnit - --- Example program which consumes the events for the event type --- "test-event" and republishes them unchanged under the event type --- "test-event-copy". - -genEvent :: MonadIO m => m (Nakadi.DataChangeEvent Foo) -genEvent = do - now <- liftIO getCurrentTime - eid <- Nakadi.EventId <$> genRandomUUID - let event = Nakadi.DataChangeEvent - { Nakadi._payload = Foo "Hello!" - , Nakadi._metadata = Nakadi.EventMetadata - { Nakadi._eid = eid - , Nakadi._occurredAt = Nakadi.Timestamp now - , Nakadi._parentEids = Nothing - , Nakadi._partition = Nothing - } - , Nakadi._dataType = "test.FOO" - , Nakadi._dataOp = Nakadi.DataOpUpdate - } - pure event - -genEvents :: MonadIO m => m (Vector (Nakadi.DataChangeEvent Foo)) -genEvents = - Vector.fromList <$> sequence (replicate 10 genEvent) - -publishEvents :: Nakadi.MonadNakadi IO m - => Vector (Nakadi.DataChangeEvent Foo) - -> Nakadi.EventTypeName -> m () -publishEvents events eventName = do - Nakadi.eventsPublish eventName (Vector.toList events) - -consumerMain - :: ( Nakadi.MonadNakadi b m - , MonadIO m - , MonadBaseControl IO m - , MonadMask m) - => Nakadi.EventTypeName - -> Int - -> m (Vector Foo) -consumerMain eventName maxSize = runResourceT $ do - let consumeParameters = Nakadi.defaultConsumeParameters - & L.batchFlushTimeout .~ Just 1 - & L.streamLimit .~ Just (fromIntegral maxSize) - Nakadi.eventsProcessConduit (Just consumeParameters) eventName Nothing $ - concatMapC (view L.events) - .| concatC - .| mapC (id :: Nakadi.DataChangeEventEnriched Foo -> Nakadi.DataChangeEventEnriched Foo) - .| mapC (view L.payload) - .| sinkVector - -testEcho :: Nakadi.Config IO -> Assertion -testEcho config = Nakadi.runNakadiT config $ do - recreateEvent myEventTypeName myEventType - recreateEvent myEventTypeNameCopy myEventTypeCopy - events <- genEvents - let eventsPayloads = map (view L.payload) events - withAsync (runEcho myEventTypeName myEventTypeNameCopy) $ \ _echoHandle -> - withAsync (consumerMain myEventTypeNameCopy (length events)) $ \ consumerHandle -> do - threadDelay (10^6) -- Give it some time to connect - publishEvents events myEventTypeName - eventsConsumed <- wait consumerHandle - liftIO $ eventsPayloads @=? eventsConsumed - - where myEventTypeNameCopy = Nakadi.EventTypeName "test.FOO-copy" - myEventTypeCopy = myEventType & L.name .~ myEventTypeNameCopy diff --git a/tests/Network/Nakadi/Examples/Subscription/Process.hs b/tests/Network/Nakadi/Examples/Subscription/Process.hs --- a/tests/Network/Nakadi/Examples/Subscription/Process.hs +++ b/tests/Network/Nakadi/Examples/Subscription/Process.hs @@ -7,8 +7,10 @@ import qualified Network.Nakadi as Nakadi import Network.Nakadi (MonadNakadi) import Control.Monad.Logger +import Control.Monad.Catch (MonadMask) -dumpSubscription :: (MonadLogger m, MonadNakadi IO m, MonadMask m) => Nakadi.SubscriptionId -> m () +dumpSubscription :: (MonadLogger m, MonadNakadi IO m, MonadUnliftIO m, MonadMask m) + => Nakadi.SubscriptionId -> m () dumpSubscription subscriptionId = Nakadi.subscriptionProcess Nothing subscriptionId processBatch diff --git a/tests/Network/Nakadi/Examples/Subscription/Test.hs b/tests/Network/Nakadi/Examples/Subscription/Test.hs --- a/tests/Network/Nakadi/Examples/Subscription/Test.hs +++ b/tests/Network/Nakadi/Examples/Subscription/Test.hs @@ -15,6 +15,7 @@ import qualified Network.Nakadi.Lenses as L import Network.Nakadi.Tests.Common import Test.Tasty.HUnit +import UnliftIO.Concurrent genEvent :: MonadIO m => m (Nakadi.DataChangeEvent Foo) genEvent = do @@ -51,7 +52,7 @@ liftIO $ assertFailure "Subscription Consumption has logged no received batches" where before = do - recreateEvent myEventTypeName myEventType + recreateEvent myEventType subscription <- Nakadi.subscriptionCreate Nakadi.Subscription { _id = Nothing , _owningApplication = "test-suite" diff --git a/tests/Network/Nakadi/Examples/Test.hs b/tests/Network/Nakadi/Examples/Test.hs --- a/tests/Network/Nakadi/Examples/Test.hs +++ b/tests/Network/Nakadi/Examples/Test.hs @@ -4,7 +4,6 @@ import ClassyPrelude import qualified Network.Nakadi as Nakadi -import Network.Nakadi.Examples.Echo.Test import Network.Nakadi.Examples.Subscription.Test import Test.Tasty import Test.Tasty.HUnit @@ -12,4 +11,4 @@ testExamples :: Nakadi.Config IO -> TestTree testExamples conf = testGroup "Examples" [ testCase "Subscription Consumption" (testConsumption conf) - , testCase "Echo" (testEcho conf) ] + ] diff --git a/tests/Network/Nakadi/Internal/Retry/Test.hs b/tests/Network/Nakadi/Internal/Retry/Test.hs --- a/tests/Network/Nakadi/Internal/Retry/Test.hs +++ b/tests/Network/Nakadi/Internal/Retry/Test.hs @@ -40,7 +40,7 @@ return n if current >= successNum then return responseSuccess - else throwM (HttpExceptionRequest defaultRequest (StatusCodeException responseFailure mempty)) + else throwIO (HttpExceptionRequest defaultRequest (StatusCodeException responseFailure mempty)) where responseTemplate = Response { responseStatus = status200 diff --git a/tests/Network/Nakadi/Subscriptions/Processing/Test.hs b/tests/Network/Nakadi/Subscriptions/Processing/Test.hs --- a/tests/Network/Nakadi/Subscriptions/Processing/Test.hs +++ b/tests/Network/Nakadi/Subscriptions/Processing/Test.hs @@ -2,16 +2,15 @@ {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE ScopedTypeVariables #-} - --- This test tests the following the high-level subscription consumption. +-- This test tests the high-level subscription consumption. module Network.Nakadi.Subscriptions.Processing.Test where import ClassyPrelude import Control.Lens +import Control.Monad.Logger import Data.Aeson -import Data.Conduit import Data.Maybe (fromJust) import Network.Nakadi import qualified Network.Nakadi.Lenses as L @@ -20,80 +19,39 @@ import Test.Tasty.HUnit testSubscriptionsProcessing :: Config App -> TestTree -testSubscriptionsProcessing conf = testGroup "Processing" - [ testCase "SubscriptionProcessing" (testSubscriptionHighLevelProcessing conf) - ] +testSubscriptionsProcessing confTemplate = + let mkConf commitStrategy = confTemplate + & setCommitStrategy commitStrategy + in testGroup "Processing" + [ testCase "SubscriptionProcessing/async/TimeBuffer" $ + testSubscriptionHighLevelProcessing (mkConf (CommitAsync (CommitTimeBuffer 200))) + , testCase "SubscriptionProcessing/sync" $ + testSubscriptionHighLevelProcessing (mkConf CommitSync) + , testCase "SubscriptionProcessing/async/NoBuffer" $ + testSubscriptionHighLevelProcessing (mkConf (CommitAsync CommitNoBuffer)) + , testCase "SubscriptionProcessing/async/SmartBuffer" $ + testSubscriptionHighLevelProcessing (mkConf (CommitAsync CommitSmartBuffer)) + ] data ConsumptionDone = ConsumptionDone deriving (Show, Typeable) - instance Exception ConsumptionDone testSubscriptionHighLevelProcessing :: Config App -> Assertion -testSubscriptionHighLevelProcessing conf = runApp . runNakadiT conf $ do - counter <- newIORef 0 - events <- sequence $ - replicate nEvents genMyDataChangeEvent :: NakadiT App App [DataChangeEvent Foo] - publishAndConsume events counter `catch` \ (_exn :: ConsumptionDone) -> pure () - eventsRead <- readIORef counter - liftIO $ nEvents @=? eventsRead - - where before :: MonadNakadi App m => m SubscriptionId - before = do - recreateEvent myEventTypeName myEventType - subscription <- subscriptionCreate Subscription - { _id = Nothing - , _owningApplication = "test-suite" - , _eventTypes = [myEventTypeName] - , _consumerGroup = Nothing -- ?? - , _createdAt = Nothing - , _readFrom = Just SubscriptionPositionEnd - , _initialCursors = Nothing - } - pure . fromJust $ subscription^.L.id - - after :: MonadNakadi App m => SubscriptionId -> m () - after subscriptionId = do - subscriptionDelete subscriptionId - eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ()) - - nEvents :: Int - nEvents = 100 - - publishAndConsume :: (ToJSON a, FromJSON a) - => [DataChangeEvent a] - -> IORef Int - -> NakadiT App App () - publishAndConsume events counter = - bracket before after $ \ subscriptionId -> do - let n = length events - void . async $ delayedPublish Nothing events - subscriptionProcess (Just consumeParameters) subscriptionId $ - \ (batch :: SubscriptionEventStreamBatch (DataChangeEvent Foo)) -> do - -- Make sure that we can use Nakadi from within the high - -- level processing callback: - void $ registryPartitionStrategies - let eventsReceived = fromMaybe mempty (batch^.L.events) - modifyIORef counter (+ (length eventsReceived)) - eventsRead <- readIORef counter - when (n == eventsRead) $ throwM ConsumptionDone - - consumeParameters = defaultConsumeParameters - & setBatchFlushTimeout 1 - & setBatchLimit 1 - -testSubscriptionConduitProcessing :: Config App -> Assertion -testSubscriptionConduitProcessing conf = runApp . runNakadiT conf $ do - counter <- newIORef 0 - events <- sequence $ - replicate nEvents genMyDataChangeEvent :: NakadiT App App [DataChangeEvent Foo] - publishAndConsume events counter `catch` \ (_exn :: ConsumptionDone) -> pure () - eventsRead <- readIORef counter - liftIO $ nEvents @=? eventsRead +testSubscriptionHighLevelProcessing conf = runApp $ do + logger <- askLoggerIO + let logFunc src lev str = liftIO $ logger defaultLoc src lev str + runNakadiT (conf & setLogFunc logFunc) $ do + counter <- newIORef 0 + events <- sequence $ + map genMyDataChangeEventIdx [1..nEvents] :: NakadiT App App [DataChangeEvent Foo] + publishAndConsume events counter `catch` \ (_exn :: ConsumptionDone) -> pure () + eventsRead <- readIORef counter + liftIO $ nEvents @=? eventsRead - where before :: MonadNakadi App m => m SubscriptionId + where before :: (MonadUnliftIO m, MonadNakadi App m) => m SubscriptionId before = do - recreateEvent myEventTypeName myEventType + recreateEvent myEventType subscription <- subscriptionCreate Subscription { _id = Nothing , _owningApplication = "test-suite" @@ -105,33 +63,34 @@ } pure . fromJust $ subscription^.L.id - after :: MonadNakadi App m => SubscriptionId -> m () + after :: (MonadUnliftIO m, MonadNakadi App m) => SubscriptionId -> m () after subscriptionId = do subscriptionDelete subscriptionId eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ()) nEvents :: Int - nEvents = 100 + nEvents = 10000 - publishAndConsume :: (ToJSON a, FromJSON a) + publishAndConsume :: (ToJSON a, FromJSON a, Show a) => [DataChangeEvent a] -> IORef Int -> NakadiT App App () publishAndConsume events counter = bracket before after $ \ subscriptionId -> do let n = length events - void . async $ delayedPublish Nothing events - subscriptionProcessConduit (Just consumeParameters) subscriptionId $ do - awaitForever $ \ (batch :: SubscriptionEventStreamBatch (DataChangeEvent Foo)) -> do - -- Make sure that we can use Nakadi via lifting directly - -- from within the Conduit. - void $ lift registryPartitionStrategies - let eventsReceived = fromMaybe mempty (batch^.L.events) - modifyIORef counter (+ (length eventsReceived)) - eventsRead <- readIORef counter - when (n == eventsRead) $ throwM ConsumptionDone - yield batch + publisherHandle <- async $ do + delayedPublish Nothing events + liftIO $ linkAsync publisherHandle + forever $ do + subscriptionProcess (Just consumeParameters) subscriptionId $ + \ (batch :: SubscriptionEventStreamBatch (DataChangeEvent Foo)) -> do + let eventsReceived = fromMaybe mempty (batch^.L.events) + modifyIORef counter (+ (length eventsReceived)) + eventsRead <- readIORef counter + when (n <= eventsRead) $ throwIO ConsumptionDone + putStrLn $ "Subscription Processing terminated, will restart." consumeParameters = defaultConsumeParameters & setBatchFlushTimeout 1 - & setBatchLimit 1 + & setMaxUncommittedEvents 5000 + & setBatchLimit 10 diff --git a/tests/Network/Nakadi/Subscriptions/Test.hs b/tests/Network/Nakadi/Subscriptions/Test.hs --- a/tests/Network/Nakadi/Subscriptions/Test.hs +++ b/tests/Network/Nakadi/Subscriptions/Test.hs @@ -30,7 +30,7 @@ testSubscriptionsList conf = runApp . runNakadiT conf $ do -- Cleanup deleteSubscriptionsByAppPrefix prefix - recreateEvent myEventTypeName myEventType + recreateEvent myEventType -- Create new Subscriptions maybeSubscriptionIds <- forM [1..n] $ \i -> do let owningApp = ApplicationName (prefix <> tshow i) @@ -74,7 +74,7 @@ testSubscriptionsCreateDelete :: Config App -> Assertion testSubscriptionsCreateDelete conf = runApp . runNakadiT conf $ do - recreateEvent myEventTypeName myEventType + recreateEvent myEventType subscription <- subscriptionCreate (mySubscription Nothing) liftIO $ True @=? isJust (subscription^.L.id) let (Just subscriptionId) = subscription^.L.id @@ -83,7 +83,7 @@ testSubscriptionsDoubleDeleteFailure :: Config App -> Assertion testSubscriptionsDoubleDeleteFailure conf = runApp . runNakadiT conf $ do - recreateEvent myEventTypeName myEventType + recreateEvent myEventType subscription <- subscriptionCreate (mySubscription Nothing) liftIO $ True @=? isJust (subscription^.L.id) let (Just subscriptionId) = subscription^.L.id 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 @@ -8,17 +8,26 @@ import ClassyPrelude +import Control.Exception.Safe (MonadThrow, throwM) import Control.Lens +import Control.Monad.Logger import Data.Aeson -import Data.UUID (UUID) +import Data.List.Split (chunksOf) +import qualified Data.Text as Text +import Data.UUID (UUID) import Network.Nakadi -import qualified Network.Nakadi.Lenses as L +import qualified Network.Nakadi.Lenses as L import System.Random +import UnliftIO.Concurrent -type App = ReaderT () IO +data TerminateConsumption = TerminateConsumption deriving (Eq, Show, Typeable) +instance Exception TerminateConsumption + +type App = LoggingT (ReaderT () IO) + runApp :: App a -> IO a -runApp = flip runReaderT () +runApp = flip runReaderT () . runStdoutLoggingT data Foo = Foo { fortune :: Text } deriving (Show, Eq, Generic) @@ -57,7 +66,7 @@ ignoreExnNotFound :: MonadThrow m => a -> NakadiException -> m a ignoreExnNotFound a (EventTypeNotFound _) = return a -ignoreExnNotFound _ exn = throw exn +ignoreExnNotFound _ exn = throwM exn extractCursor :: Partition -> Cursor extractCursor Partition { ..} = @@ -92,21 +101,42 @@ , _dataOp = DataOpUpdate } +genMyDataChangeEventIdx :: MonadIO m => Int -> m (DataChangeEvent Foo) +genMyDataChangeEventIdx idx = do + eid <- genRandomUUID + now <- liftIO getCurrentTime + pure DataChangeEvent + { _payload = Foo ("Hello " ++ Text.pack (show idx)) + , _metadata = EventMetadata { _eid = EventId eid + , _occurredAt = Timestamp now + , _parentEids = Nothing + , _partition = Nothing + } + , _dataType = "test.FOO" + , _dataOp = DataOpUpdate + } + genRandomUUID :: MonadIO m => m UUID genRandomUUID = liftIO randomIO -recreateEvent :: (MonadCatch m, MonadNakadi b m) => EventTypeName -> EventType -> m () -recreateEvent eventTypeName eventType = do +recreateEvent :: (MonadUnliftIO m, MonadNakadi b m) => EventType -> m () +recreateEvent eventType = do + let eventTypeName = eventType^.L.name subscriptionIds <- subscriptionsList Nothing (Just [eventTypeName]) <&> catMaybes . map (view L.id) mapM_ subscriptionDelete subscriptionIds eventTypeDelete eventTypeName `catch` (ignoreExnNotFound ()) eventTypeCreate eventType -delayedPublish :: (MonadNakadi b m, MonadIO m, ToJSON a) => Maybe FlowId -> [a] -> m () +delayedPublish + :: (MonadNakadi b m, MonadIO m, ToJSON a) + => Maybe FlowId + -> [a] + -> m () delayedPublish maybeFlowId events = do liftIO $ threadDelay (10^6) let flowId = fromMaybe (FlowId "shalom") maybeFlowId config <- nakadiAsk <&> setFlowId flowId + -- Publish events in batches. runNakadiT config $ - eventsPublish myEventTypeName events + forM_ (chunksOf 100 events) (eventsPublish myEventTypeName) diff --git a/tests/Tests.hs b/tests/Tests.hs --- a/tests/Tests.hs +++ b/tests/Tests.hs @@ -14,7 +14,6 @@ import Network.Nakadi.Tests.Common import System.Environment import System.Exit -import System.IO (hFlush) import Test.Tasty main :: IO ()