nakadi-client 0.5.0.3 → 0.5.1.0

raw patch · 28 files changed

+596/−449 lines, 28 filesdep +async-timerdep −conduit-combinatorsdep ~classy-preludedep ~conduitdep ~conduit-extraPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: async-timer

Dependencies removed: conduit-combinators

Dependency ranges changed: classy-prelude, conduit, conduit-extra, http-conduit, safe-exceptions, stm-conduit, unliftio

API changes (from Hackage documentation)

Files

• nakadi-client.cabal +75/−74
• src/Network/Nakadi/Config.hs +13/−0
• src/Network/Nakadi/EventTypes/Events.hs +1/−51
• src/Network/Nakadi/Internal/Lenses.hs +1/−1
• src/Network/Nakadi/Internal/Types.hs +7/−5
• src/Network/Nakadi/Internal/Types/Config.hs +3/−1
• src/Network/Nakadi/Internal/Types/Service.hs +0/−9
• src/Network/Nakadi/Internal/Types/Subscriptions.hs +40/−0
• src/Network/Nakadi/Internal/Util.hs +2/−4
• src/Network/Nakadi/Subscriptions/Events.hs +268/−35
• src/Network/Nakadi/Types.hs +2/−0
• src/Network/Nakadi/Types/Service.hs +0/−1
• src/Network/Nakadi/Types/Subscriptions.hs +19/−0
• tests/Network/Nakadi/Config/Test.hs +1/−1
• tests/Network/Nakadi/Connection/Test.hs +5/−4
• tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs +2/−2
• tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs +2/−2
• tests/Network/Nakadi/EventTypes/Test.hs +63/−31
• tests/Network/Nakadi/Examples/Echo/Echo.hs +0/−42
• tests/Network/Nakadi/Examples/Echo/Test.hs +0/−84
• tests/Network/Nakadi/Examples/Subscription/Process.hs +3/−1
• tests/Network/Nakadi/Examples/Subscription/Test.hs +2/−1
• tests/Network/Nakadi/Examples/Test.hs +1/−2
• tests/Network/Nakadi/Internal/Retry/Test.hs +1/−1
• tests/Network/Nakadi/Subscriptions/Processing/Test.hs +43/−84
• tests/Network/Nakadi/Subscriptions/Test.hs +3/−3
• tests/Network/Nakadi/Tests/Common.hs +39/−9
• tests/Tests.hs +0/−1

@@ -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

@@ -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.

@@ -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.

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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@.

@@ -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)

@@ -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 ()

@@ -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

@@ -27,7 +27,6 @@   , StreamId(..)   , SubscriptionEventStream(..)   , Timestamp(..)-  , EventStreamBatch(..)   , SubscriptionEventStreamBatch(..)   , EventId(..)   , Partition(..)

@@ -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

@@ -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 (..))

@@ -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

@@ -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

@@ -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

@@ -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 

@@ -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

@@ -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

@@ -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 

@@ -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"

@@ -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) ]+  ]

@@ -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

@@ -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

@@ -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

@@ -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)

@@ -14,7 +14,6 @@ import           Network.Nakadi.Tests.Common import           System.Environment import           System.Exit-import           System.IO                         (hFlush) import           Test.Tasty  main :: IO ()