nakadi-client 0.4.1.0 → 0.5.0.0

raw patch · 61 files changed

+1963/−1324 lines, 61 filesdep +exceptionsdep +fast-loggerdep +lifted-async

Dependencies added: exceptions, fast-logger, lifted-async, monad-control, stm-chans, stm-conduit, transformers-base, unliftio, unliftio-core

Files

AUTHORS.md +1/−1
README.md +7/−8
nakadi-client.cabal +31/−6
src/Network/Nakadi.hs +5/−2
src/Network/Nakadi/Base.hs +8/−0
src/Network/Nakadi/Config.hs +95/−67
src/Network/Nakadi/EventTypes.hs +14/−35
src/Network/Nakadi/EventTypes/CursorDistances.hs +10/−38
src/Network/Nakadi/EventTypes/CursorsLag.hs +13/−39
src/Network/Nakadi/EventTypes/EventType.hs +25/−60
src/Network/Nakadi/EventTypes/Events.hs +57/−69
src/Network/Nakadi/EventTypes/Partitions.hs +16/−42
src/Network/Nakadi/EventTypes/Schemas.hs +10/−39
src/Network/Nakadi/EventTypes/ShiftedCursors.hs +13/−40
src/Network/Nakadi/HttpBackendIO.hs +18/−0
src/Network/Nakadi/Internal/Base.hs +15/−0
src/Network/Nakadi/Internal/Config.hs +12/−1
src/Network/Nakadi/Internal/Http.hs +158/−122
src/Network/Nakadi/Internal/HttpBackendIO.hs +71/−0
src/Network/Nakadi/Internal/Lenses.hs +5/−8
src/Network/Nakadi/Internal/Prelude.hs +3/−2
src/Network/Nakadi/Internal/Retry.hs +30/−20
src/Network/Nakadi/Internal/Types.hs +206/−21
src/Network/Nakadi/Internal/Types/Base.hs +62/−0
src/Network/Nakadi/Internal/Types/Config.hs +31/−29
src/Network/Nakadi/Internal/Types/Exceptions.hs +3/−3
src/Network/Nakadi/Internal/Types/Logger.hs +5/−2
src/Network/Nakadi/Internal/Types/Service.hs +4/−2
src/Network/Nakadi/Internal/Types/Subscription.hs +0/−28
src/Network/Nakadi/Internal/Util.hs +16/−10
src/Network/Nakadi/Prelude.hs +24/−0
src/Network/Nakadi/Registry.hs +6/−16
src/Network/Nakadi/Subscriptions.hs +41/−101
src/Network/Nakadi/Subscriptions/Cursors.hs +22/−65
src/Network/Nakadi/Subscriptions/Events.hs +52/−86
src/Network/Nakadi/Subscriptions/Stats.hs +9/−36
src/Network/Nakadi/Subscriptions/Subscription.hs +9/−37
src/Network/Nakadi/Types.hs +6/−7
src/Network/Nakadi/Types/Config.hs +3/−2
src/Network/Nakadi/Types/Logger.hs +1/−0
src/Network/Nakadi/Types/Service.hs +1/−1
src/Network/Nakadi/Types/Subscription.hs +0/−17
tests/Network/Nakadi/Config/Test.hs +38/−24
tests/Network/Nakadi/Connection/Test.hs +14/−10
tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs +20/−19
tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs +21/−19
tests/Network/Nakadi/EventTypes/Test.hs +110/−87
tests/Network/Nakadi/Examples/Echo/Echo.hs +42/−0
tests/Network/Nakadi/Examples/Echo/Test.hs +84/−0
tests/Network/Nakadi/Examples/ListEventTypes/Test.hs +23/−0
tests/Network/Nakadi/Examples/Subscription/Process.hs +17/−0
tests/Network/Nakadi/Examples/Subscription/Test.hs +74/−0
tests/Network/Nakadi/Examples/Test.hs +15/−0
tests/Network/Nakadi/Internal/Http/Test.hs +67/−34
tests/Network/Nakadi/Internal/Retry/Test.hs +8/−17
tests/Network/Nakadi/MonadicAPI/Test.hs +79/−0
tests/Network/Nakadi/Registry/Test.hs +5/−4
tests/Network/Nakadi/Subscriptions/Processing/Test.hs +137/−0
tests/Network/Nakadi/Subscriptions/Test.hs +32/−27
tests/Network/Nakadi/Tests/Common.hs +45/−15
tests/Tests.hs +14/−6

AUTHORS.md view

@@ -1,6 +1,6 @@ This package contains contributions by the following persons: -* Moritz Schulte [@mtesseract](https://github.com/mtesseract)+* Moritz Clasmeier [@mtesseract](https://github.com/mtesseract) * Viktor Kocherga [@vitold](https://github.com/vitold) * Amr Hassan [@amrhassan](https://github.com/amrhassan) * Eric Torreborre [@etorreborre](https://github.com/etorreborre)

README.md view

@@ -49,15 +49,14 @@  ### Example -Example using the Subscription API:+Example code showing how to dump a subscription:  ```haskell-import qualified Network.Nakadi as Nakadi+dumpSubscription :: (MonadLogger m, MonadNakadi IO m) => Nakadi.SubscriptionId -> m ()+dumpSubscription subscriptionId =+  Nakadi.subscriptionProcess Nothing subscriptionId processBatch -processSubscription :: Nakadi.SubscriptionId -> IO ()-processSubscription subscriptionId = do-  runResourceT $ do-    (connection, source) <- Nakadi.subscriptionSource config Nothing subscriptionId-    Nakadi.runSubscription config connection $-      source .| iterMC processEvent .| Nakadi.subscriptionSink+  where processBatch :: MonadLogger m => Nakadi.SubscriptionEventStreamBatch Value -> m ()+        processBatch batch =+          logInfoN (tshow batch) ```

nakadi-client.cabal view

@@ -2,18 +2,18 @@ -- -- see: https://github.com/sol/hpack ----- hash: 0504d364a6102858824fdbf0a280623f9e9ba3e039e98308fe115e6822289122+-- hash: 8b96a97aa794dc6045432c9cfa1920cfc1ba2845cbf30a06d599896df3f2b6c1  name:           nakadi-client-version:        0.4.1.0+version:        0.5.0.0 synopsis:       Client library for the Nakadi Event Broker description:    This package implements a client library for interacting with the Nakadi event broker system developed by Zalando. category:       Network homepage:       https://github.com/mtesseract/nakadi-haskell#readme bug-reports:    https://github.com/mtesseract/nakadi-haskell/issues-author:         Moritz Schulte+author:         Moritz Clasmeier maintainer:     mtesseract@silverratio.net-copyright:      (c) 2017 Moritz Schulte+copyright:      (c) 2017, 2018 Moritz Clasmeier license:        BSD3 license-file:   LICENSE build-type:     Simple@@ -45,6 +45,7 @@     , conduit-combinators     , conduit-extra     , containers+    , exceptions     , hashable     , http-client     , http-client-tls@@ -52,6 +53,7 @@     , http-types     , iso8601-time     , lens+    , monad-control     , monad-logger     , mtl     , resourcet@@ -63,6 +65,8 @@     , text     , time     , transformers+    , transformers-base+    , unliftio-core     , unordered-containers     , uuid     , vector@@ -72,6 +76,8 @@     ghc-options: -Wall -fno-warn-type-defaults   exposed-modules:       Network.Nakadi+      Network.Nakadi.Prelude+      Network.Nakadi.Base       Network.Nakadi.Config       Network.Nakadi.EventTypes       Network.Nakadi.EventTypes.CursorDistances@@ -92,25 +98,27 @@       Network.Nakadi.Types.Logger       Network.Nakadi.Types.Problem       Network.Nakadi.Types.Service-      Network.Nakadi.Types.Subscription       Network.Nakadi.Registry       Network.Nakadi.Lenses       Network.Nakadi.Internal.Http       Network.Nakadi.Internal.Retry+      Network.Nakadi.HttpBackendIO   other-modules:+      Network.Nakadi.Internal.Base       Network.Nakadi.Internal.Config       Network.Nakadi.Internal.Conversions+      Network.Nakadi.Internal.HttpBackendIO       Network.Nakadi.Internal.Json       Network.Nakadi.Internal.Lenses       Network.Nakadi.Internal.Prelude       Network.Nakadi.Internal.TH       Network.Nakadi.Internal.Types+      Network.Nakadi.Internal.Types.Base       Network.Nakadi.Internal.Types.Config       Network.Nakadi.Internal.Types.Exceptions       Network.Nakadi.Internal.Types.Logger       Network.Nakadi.Internal.Types.Problem       Network.Nakadi.Internal.Types.Service-      Network.Nakadi.Internal.Types.Subscription       Network.Nakadi.Internal.Types.Util       Network.Nakadi.Internal.Util       Paths_nakadi_client@@ -134,6 +142,8 @@     , conduit-combinators     , conduit-extra     , containers+    , exceptions+    , fast-logger     , hashable     , http-client     , http-client-tls@@ -142,6 +152,8 @@     , iso8601-time     , lens     , lens-aeson+    , lifted-async+    , monad-control     , monad-logger     , mtl     , nakadi-client@@ -153,12 +165,17 @@     , scientific     , split     , stm+    , stm-chans+    , stm-conduit     , tasty     , tasty-hunit     , template-haskell     , text     , time     , transformers+    , transformers-base+    , unliftio+    , unliftio-core     , unordered-containers     , uuid     , vector@@ -174,11 +191,19 @@       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

src/Network/Nakadi.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi Description : Nakadi Client Library-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Clasmeier 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -16,11 +16,14 @@   , module Network.Nakadi.Subscriptions   , module Network.Nakadi.Config   , module Network.Nakadi.Registry+  , module Network.Nakadi.HttpBackendIO+  , module Network.Nakadi.Base   ) where +import           Network.Nakadi.Base import           Network.Nakadi.Config import           Network.Nakadi.EventTypes-import           Network.Nakadi.Internal.Lenses+import           Network.Nakadi.HttpBackendIO import           Network.Nakadi.Registry import           Network.Nakadi.Subscriptions import           Network.Nakadi.Types

+ src/Network/Nakadi/Base.hs view

@@ -0,0 +1,8 @@+module Network.Nakadi.Base+  ( NakadiBaseT(..)+  , MonadNakadiBase(..)+  , runNakadiBaseT+  , runNakadiWithBase+  ) where++import           Network.Nakadi.Internal.Base

src/Network/Nakadi/Config.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Config Description : Nakadi Client Configuration-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Clasmeier 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -11,18 +11,37 @@ client configurations. -} -module Network.Nakadi.Config where+module Network.Nakadi.Config+  ( newConfig+  , newConfigIO+  , newConfigWithDedicatedManager+  , setHttpManager+  , setRequestModifier+  , setDeserializationFailureCallback+  , setStreamConnectCallback+  , setHttpErrorCallback+  , setLogFunc+  , setRetryPolicy+  , setMaxUncommittedEvents+  , setBatchLimit+  , setStreamLimit+  , setBatchFlushTimeout+  , setStreamTimeout+  , setStreamKeepAliveLimit+  , setFlowId+  , defaultConsumeParameters+  ) where  import           Network.Nakadi.Internal.Prelude  import           Control.Lens import           Control.Retry-import           Network.HTTP.Client             (Manager, ManagerSettings,-                                                  responseClose, responseOpen)-import           Network.HTTP.Client.TLS         (newTlsManagerWith,-                                                  tlsManagerSettings)-import           Network.HTTP.Simple             (httpLbs)-import qualified Network.Nakadi.Internal.Lenses  as L+import           Network.HTTP.Client                   (Manager,+                                                        ManagerSettings)+import           Network.HTTP.Client.TLS               (newTlsManagerWith)+import           Network.Nakadi.Internal.Config+import           Network.Nakadi.Internal.HttpBackendIO+import qualified Network.Nakadi.Internal.Lenses        as L import           Network.Nakadi.Internal.Types  -- | Default retry policy.@@ -31,61 +50,75 @@  -- | Producs a new configuration, with mandatory HTTP manager, default -- consumption parameters and HTTP request template.-newConfig' ::-  (MonadIO m, MonadThrow m)-  => Manager           -- ^ Manager Settings-  -> ConsumeParameters -- ^ Consumption Parameters+newConfig+  :: Monad b+  => HttpBackend b   -> Request           -- ^ Request Template-  -> m Config          -- ^ Resulting Configuration-newConfig' manager consumeParameters request =-  return Config { _consumeParameters              = consumeParameters-                , _manager                        = manager-                , _requestTemplate                = request-                , _requestModifier                = return-                , _deserializationFailureCallback = Nothing-                , _streamConnectCallback          = Nothing-                , _logFunc                        = Nothing-                , _retryPolicy                    = defaultRetryPolicy-                , _http                           = defaultHttpBackend-                , _httpErrorCallback              = Nothing-                }+  -> Config b          -- ^ Resulting Configuration+newConfig httpBackend request =+  Config { _consumeParameters              = Nothing+         , _manager                        = Nothing+         , _requestTemplate                = request+         , _requestModifier                = pure+         , _deserializationFailureCallback = Nothing+         , _streamConnectCallback          = Nothing+         , _logFunc                        = Nothing+         , _retryPolicy                    = defaultRetryPolicy+         , _http                           = httpBackend+         , _httpErrorCallback              = Nothing+         , _flowId                         = Nothing+         } --- | Default 'HttpBackend' doing IO using http-client.-defaultHttpBackend :: HttpBackend-defaultHttpBackend =-  HttpBackend { _httpLbs                        = httpLbs-              , _responseOpen                   = responseOpen-              , _responseClose                  = responseClose }+-- | Producs a new configuration, with mandatory HTTP manager, default+-- consumption parameters and HTTP request template.+newConfigIO+  :: Request           -- ^ Request Template+  -> ConfigIO          -- ^ Resulting Configuration+newConfigIO = newConfig httpBackendIO  -- | Produce a new configuration, with optional HTTP manager settings -- and mandatory HTTP request template.-newConfig ::-  (MonadIO m, MonadThrow m)-  => Maybe ManagerSettings -- ^ Optional 'ManagerSettings'-  -> Request               -- ^ Request template for Nakadi requests-  -> m Config              -- ^ Resulting Configuration-newConfig mngrSettings request = do-  manager <- newTlsManagerWith (fromMaybe tlsManagerSettings mngrSettings)-  newConfig' manager defaultConsumeParameters request+newConfigWithDedicatedManager ::+  (MonadIO b, MonadMask b, MonadIO m)+  => ManagerSettings -- ^ Optional 'ManagerSettings'+  -> Request         -- ^ Request template for Nakadi requests+  -> m (Config b)    -- ^ Resulting Configuration+newConfigWithDedicatedManager mngrSettings request = do+  manager <- newTlsManagerWith mngrSettings+  pure $ newConfig httpBackendIO request & setHttpManager manager +-- | Install an HTTP Manager in the provided configuration. If not+-- set, HTTP requests will use a global default manager.+setHttpManager+  :: Manager+  -> Config m+  -> Config m+setHttpManager mngr = L.manager .~ Just mngr+ -- | Install a request modifier in the provided configuration. This -- can be used for e.g. including access tokens in HTTP requests to -- Nakadi.-setRequestModifier :: (Request -> IO Request) -> Config -> Config+setRequestModifier ::+  (Request -> m Request)+  -> Config m+  -> Config m setRequestModifier = (L.requestModifier .~)  -- | Install a callback in the provided configuration to use in case -- of deserialization failures when consuming events. setDeserializationFailureCallback ::-  (ByteString -> Text -> IO ())-  -> Config-  -> Config+  (ByteString -> Text -> m ())+  -> Config m+  -> Config m setDeserializationFailureCallback cb = L.deserializationFailureCallback .~ Just cb  -- | Install a callback in the provided configuration which is used -- after having successfully established a streaming Nakadi -- connection.-setStreamConnectCallback :: StreamConnectCallback  -> Config -> Config+setStreamConnectCallback ::+  StreamConnectCallback m+  -> Config m+  -> Config m setStreamConnectCallback cb = L.streamConnectCallback .~ Just cb  -- | Install a callback in the provided configuration which is called@@ -93,33 +126,32 @@ -- conditions by e.g. logging errors or updating metrics. Note that -- this callback is called synchronously, thus blocking in this -- callback delays potential retry attempts.-setHttpErrorCallback :: HttpErrorCallback -> Config -> Config+setHttpErrorCallback ::+  HttpErrorCallback m+  -> Config m+  -> Config m setHttpErrorCallback cb = L.httpErrorCallback .~ Just cb  -- | Install a logger callback in the provided configuration.-setLogFunc :: LogFunc -> Config -> Config+setLogFunc ::+  LogFunc m+  -> Config m+  -> Config m setLogFunc logFunc = L.logFunc .~ Just logFunc  -- | Set a custom retry policy in the provided configuration.-setRetryPolicy :: RetryPolicyM IO -> Config -> Config+setRetryPolicy ::+  RetryPolicyM IO+  -> Config m+  -> Config m setRetryPolicy = (L.retryPolicy .~) --- | Set a custom HTTP Backend in the provided configuration. Can be--- used for testing.-setHttpBackend :: HttpBackend -> Config -> Config-setHttpBackend = (L.http .~)---- | Default parameters for event consumption.-defaultConsumeParameters :: ConsumeParameters-defaultConsumeParameters = ConsumeParameters-  { _maxUncommittedEvents = Nothing-  , _batchLimit           = Nothing-  , _streamLimit          = Nothing-  , _batchFlushTimeout    = Nothing-  , _streamTimeout        = Nothing-  , _streamKeepAliveLimit = Nothing-  , _flowId               = Nothing-  }+-- | Set flow ID in the provided configuration.+setFlowId+  :: FlowId+  -> Config m+  -> Config m+setFlowId flowId = L.flowId .~ Just flowId  -- | Set maximum number of uncommitted events in the provided value of -- consumption parameters.@@ -147,7 +179,3 @@ -- parameters. setStreamKeepAliveLimit :: Int32 -> ConsumeParameters -> ConsumeParameters setStreamKeepAliveLimit n params = params & L.streamKeepAliveLimit .~ Just n---- | Set flow ID in the provided value of value of consumption parameters.-setFlowId :: Text -> ConsumeParameters -> ConsumeParameters-setFlowId flowId = L.flowId .~ Just flowId

src/Network/Nakadi/EventTypes.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.EventTypes Description : Implementation of Nakadi EventTypes API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -22,14 +22,11 @@   , module Network.Nakadi.EventTypes.ShiftedCursors   , module Network.Nakadi.EventTypes.Schemas   , eventTypesList-  , eventTypesListR   , eventTypeCreate-  , eventTypeCreateR   ) where  import           Network.Nakadi.Internal.Prelude -import           Control.Lens import           Network.Nakadi.EventTypes.CursorDistances import           Network.Nakadi.EventTypes.CursorsLag import           Network.Nakadi.EventTypes.Events@@ -39,46 +36,28 @@ import           Network.Nakadi.EventTypes.ShiftedCursors import           Network.Nakadi.Internal.Http -import qualified Network.Nakadi.Internal.Lenses            as L- path :: ByteString path = "/event-types"  -- | @GET@ to @\/event-types@. Retrieves a list of all registered -- event types. eventTypesList ::-  MonadNakadi m-  => Config        -- ^ Configuration-  -> m [EventType] -- ^ Registered Event Types-eventTypesList config =-  httpJsonBody config status200 []-  (setRequestMethod "GET" . setRequestPath path)---- | @GET@ to @\/event-types@. Retrieves a list of all registered--- event types, using the configuration contained in the environment.-eventTypesListR ::-  MonadNakadiEnv r m+  MonadNakadi b m   => m [EventType] -- ^ Registered Event Types-eventTypesListR = do-  config <- asks (view L.nakadiConfig)-  eventTypesList config+eventTypesList = do+  config <- nakadiAsk+  httpJsonBody status200 [] $+    (setRequestMethod "GET"+     . includeFlowId config+     . setRequestPath path)  -- | @POST@ to @\/event-types@. Creates a new event type. eventTypeCreate ::-  MonadNakadi m-  => Config    -- ^ Configuration-  -> EventType -- ^ Event Type to create-  -> m ()-eventTypeCreate config eventType =-  httpJsonNoBody config status201 []-  (setRequestMethod "POST" . setRequestPath path . setRequestBodyJSON eventType)---- | @POST@ to @\/event-types@. Creates a new event type. Uses the--- configuration from the environment.-eventTypeCreateR ::-  MonadNakadiEnv r m+  MonadNakadi b m   => EventType -- ^ Event Type to create   -> m ()-eventTypeCreateR eventType = do-  config <- asks (view L.nakadiConfig)-  eventTypeCreate config eventType+eventTypeCreate eventType =+  httpJsonNoBody status201 []+  (setRequestMethod "POST"+   . setRequestPath path+   . setRequestBodyJSON eventType)

src/Network/Nakadi/EventTypes/CursorDistances.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.EventTypes.CursorDistances Description : Implementation of Nakadi CursorDistances API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -17,9 +17,7 @@  module Network.Nakadi.EventTypes.CursorDistances   ( cursorsDistance'-  , cursorsDistanceR'   , cursorDistance-  , cursorDistanceR   )  where  import           Network.Nakadi.Internal.Prelude@@ -36,53 +34,27 @@  -- | Query for distance between cursors. Low level call. cursorsDistance' ::-  MonadNakadi m-  => Config                   -- ^ Configuration-  -> EventTypeName            -- ^ Event Type+  MonadNakadi b m+  => EventTypeName            -- ^ Event Type   -> [CursorDistanceQuery]    -- ^ List of cursor-distance-queries   -> m [CursorDistanceResult] -- ^ List of cursor-distance-results-cursorsDistance' config eventTypeName cursorDistanceQuery =-  httpJsonBody config ok200 []+cursorsDistance' eventTypeName cursorDistanceQuery =+  httpJsonBody ok200 []   (setRequestMethod "POST"    . setRequestPath (path eventTypeName)    . setRequestBodyJSON cursorDistanceQuery) --- | Query for distance between cursors. Low level call, retrieving--- configuration from environment.-cursorsDistanceR' ::-  MonadNakadiEnv r m-  => EventTypeName            -- ^ Event Type-  -> [CursorDistanceQuery]    -- ^ List of cursor-distance-queries-  -> m [CursorDistanceResult] -- ^ List of cursor-distance-results-cursorsDistanceR' eventTypeName cursorDistanceQuery = do-  config <- asks (view L.nakadiConfig)-  cursorsDistance' config eventTypeName cursorDistanceQuery- -- | Given two cursors, compute the distance between these cursors. cursorDistance ::-  MonadNakadi m-  => Config        -- ^ Configuration-  -> EventTypeName -- ^ Event Type+  MonadNakadi b m+  => EventTypeName -- ^ Event Type   -> Cursor        -- ^ First cursor   -> Cursor        -- ^ Second cursor   -> m Int64       -- ^ Resulting difference between first and second                    -- cursor-cursorDistance config eventTypeName cursor cursor' =+cursorDistance eventTypeName cursor cursor' =   let cursorDistanceQuery = CursorDistanceQuery { _initialCursor = cursor                                                 , _finalCursor   = cursor' }-  in cursorsDistance' config eventTypeName [cursorDistanceQuery] >>= \case+  in cursorsDistance' eventTypeName [cursorDistanceQuery] >>= \case     distanceResult:_ -> return $ distanceResult^.L.distance-    _                -> throwIO CursorDistanceNoResult---- | Given two cursors, compute the distance between these cursors,--- retrieving configuration from environment.-cursorDistanceR ::-  MonadNakadiEnv r m-  => EventTypeName -- ^ Event Type-  -> Cursor        -- ^ First cursor-  -> Cursor        -- ^ Second cursor-  -> m Int64       -- ^ Resulting difference between first and second-                   -- cursor-cursorDistanceR eventTypeName cursor cursor' = do-  config <- asks (view L.nakadiConfig)-  cursorDistance config eventTypeName cursor cursor'+    _                -> throwM CursorDistanceNoResult

src/Network/Nakadi/EventTypes/CursorsLag.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.EventTypes.CursorsLag Description : Implementation of Nakadi CursorsLag API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -16,9 +16,7 @@  module Network.Nakadi.EventTypes.CursorsLag   ( cursorsLag'-  , cursorsLagR'   , cursorsLag-  , cursorsLagR   ) where  import           Network.Nakadi.Internal.Prelude@@ -39,54 +37,30 @@ -- | @POST@ to @\/event-types\/EVENT-TYPE\/cursors-lag@. Low level -- interface. cursorsLag' ::-  MonadNakadi m-  => Config        -- ^ Configuration-  -> EventTypeName -- ^ Event Type-  -> [Cursor]      -- ^ Cursors for which to compute the lag for-  -> m [Partition] -- ^ Resulting partition information containing-                   -- information about unconsumed events.-cursorsLag' config eventTypeName cursors =-  httpJsonBody config ok200 []-  (setRequestMethod "POST"-   . setRequestPath (path eventTypeName)-   . setRequestBodyJSON cursors)---- | @POST@ to @\/event-types\/EVENT-TYPE\/cursors-lag@. Low level--- interface, retrieving configuration from environment.-cursorsLagR' ::-  MonadNakadiEnv r m+  MonadNakadi b m   => EventTypeName -- ^ Event Type   -> [Cursor]      -- ^ Cursors for which to compute the lag for   -> m [Partition] -- ^ Resulting partition information containing                    -- information about unconsumed events.-cursorsLagR' eventTypeName cursors = do-  config <- asks (view L.nakadiConfig)-  cursorsLag' config eventTypeName cursors+cursorsLag' eventTypeName cursors = do+  config <- nakadiAsk+  httpJsonBody ok200 [] $+    (setRequestMethod "POST"+     . includeFlowId config+     . setRequestPath (path eventTypeName)+     . setRequestBodyJSON cursors)  -- | @POST@ to @\/event-types\/EVENT-TYPE\/cursors-lag@. High level -- interface. cursorsLag ::-  MonadNakadi m-  => Config                         -- ^ Configuration-  -> EventTypeName                  -- ^ Event Type+  MonadNakadi b m+  => EventTypeName                  -- ^ Event Type   -> Map PartitionName CursorOffset -- ^ Cursor offsets associated to                                     -- partitions.   -> m (Map PartitionName Int64)    -- ^ Cursor lags associated to partitions.-cursorsLag config eventTypeName cursorsMap = do-  partitionStats <- cursorsLag' config eventTypeName cursors+cursorsLag eventTypeName cursorsMap = do+  partitionStats <- cursorsLag' eventTypeName cursors   return $ partitionStats & map ((view L.partition &&& view L.unconsumedEvents) >>> sequenceSnd)                           & catMaybes                           & Map.fromList   where cursors = map (uncurry Cursor) (Map.toList cursorsMap)---- | @POST@ to @\/event-types\/EVENT-TYPE\/cursors-lag@. High level--- interface, retrieving configuration from environment.-cursorsLagR ::-  MonadNakadiEnv r m-  => EventTypeName                  -- ^ Event Type-  -> Map PartitionName CursorOffset -- ^ Cursor offsets associated to-                                    -- partitions.-  -> m (Map PartitionName Int64)    -- ^ Cursor lags associated to partitions.-cursorsLagR eventTypeName cursorsMap = do-  config <- asks (view L.nakadiConfig)-  cursorsLag config eventTypeName cursorsMap

src/Network/Nakadi/EventTypes/EventType.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.EventTypes.EventType Description : Implementation of Nakadi EventTypes API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -15,89 +15,54 @@  module Network.Nakadi.EventTypes.EventType   ( eventTypeGet-  , eventTypeGetR   , eventTypeUpdate-  , eventTypeUpdateR   , eventTypeDelete-  , eventTypeDeleteR   ) where  import           Network.Nakadi.Internal.Prelude -import           Control.Lens import           Network.Nakadi.Internal.Http-import qualified Network.Nakadi.Internal.Lenses  as L  path :: EventTypeName -> ByteString path eventTypeName = "/event-types/" <> encodeUtf8 (unEventTypeName eventTypeName)  -- | Retrieves an 'EventType' by its 'EventTypeName'. @GET@ to -- @\/event-types\/EVENT-TYPE@.-eventTypeGet ::-  MonadNakadi m-  => Config        -- ^ Configuration-  -> EventTypeName -- ^ Name of Event Type-  -> m EventType   -- ^ Event Type information-eventTypeGet config eventTypeName =-  httpJsonBody config ok200 [(status404, errorEventTypeNotFound)]-  (setRequestMethod "GET"-   . setRequestPath (path eventTypeName))---- | Retrieves an 'EventType' by its 'EventTypeName', using the--- configuration found in the environment. @GET@ to--- @\/event-types\/EVENT-TYPE@.-eventTypeGetR ::-  MonadNakadiEnv r m+eventTypeGet+  :: MonadNakadi b m   => EventTypeName -- ^ Name of Event Type   -> m EventType   -- ^ Event Type information-eventTypeGetR eventTypeName = do-  config <- asks (view L.nakadiConfig)-  eventTypeGet config eventTypeName+eventTypeGet eventTypeName = do+  config <- nakadiAsk+  httpJsonBody ok200 [(status404, errorEventTypeNotFound)] $+    (setRequestMethod "GET"+     . includeFlowId config+     . setRequestPath (path eventTypeName))  -- | Updates an event type given its 'EventTypeName' and its new -- 'EventType' description. @PUT@ to @\/event-types\/EVENT-TYPE@. eventTypeUpdate ::-  MonadNakadi m-  => Config        -- ^ Configuration-  -> EventTypeName -- ^ Name of Event Type-  -> EventType     -- ^ Event Type Settings-  -> m ()-eventTypeUpdate config eventTypeName eventType =-  httpJsonNoBody config ok200 []-  (setRequestMethod "PUT"-   . setRequestPath (path eventTypeName)-   . setRequestBodyJSON eventType)---- | Updates an event type given its 'EventTypeName' and its new--- 'EventType' description, using the configuration found in the--- environment. @PUT@ to @\/event-types\/EVENT-TYPE@.-eventTypeUpdateR ::-  MonadNakadiEnv r m+  MonadNakadi b m   => EventTypeName -- ^ Name of Event Type   -> EventType     -- ^ Event Type Settings   -> m ()-eventTypeUpdateR eventTypeName eventType = do-  config <- asks (view L.nakadiConfig)-  eventTypeUpdate config eventTypeName eventType+eventTypeUpdate eventTypeName eventType = do+  config <- nakadiAsk+  httpJsonNoBody ok200 [] $+    (setRequestMethod "PUT"+     . includeFlowId config+     . setRequestPath (path eventTypeName)+     . setRequestBodyJSON eventType)  -- | Deletes an event type given its 'EventTypeName'. @DELETE@ to -- @\/event-types\/EVENT-TYPE@.-eventTypeDelete ::-  MonadNakadi m-  => Config        -- ^ Configuration-  -> EventTypeName -- ^ Name of Event Type-  -> m ()-eventTypeDelete config eventTypeName =-  httpJsonNoBody config ok200 [(status404, errorEventTypeNotFound)]-  (setRequestMethod "DELETE" . setRequestPath (path eventTypeName))---- | Deletes an event type given its 'EventTypeName', using the--- configuration found in the environment. @DELETE@ to--- @\/event-types\/EVENT-TYPE@.-eventTypeDeleteR ::-  MonadNakadiEnv r m+eventTypeDelete+  :: MonadNakadi b m   => EventTypeName -- ^ Name of Event Type   -> m ()-eventTypeDeleteR eventTypeName = do-  config <- asks (view L.nakadiConfig)-  eventTypeDelete config eventTypeName+eventTypeDelete eventTypeName = do+  config <- nakadiAsk+  httpJsonNoBody ok200 [(status404, errorEventTypeNotFound)] $+    (setRequestMethod "DELETE"+     . includeFlowId config+     . setRequestPath (path eventTypeName))

src/Network/Nakadi/EventTypes/Events.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.EventTypes.Events Description : Implementation of Nakadi Events API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -16,23 +16,23 @@ {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE ScopedTypeVariables   #-} {-# LANGUAGE TupleSections         #-}+{-# LANGUAGE TypeFamilies          #-}  module Network.Nakadi.EventTypes.Events-  ( eventSource-  , eventSourceR-  , eventPublish-  , eventPublishR+  ( eventsProcessConduit+  , eventsProcess+  , eventsPublish   ) where  import           Network.Nakadi.Internal.Prelude  import           Conduit-import           Control.Lens 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-import qualified Network.Nakadi.Internal.Lenses  as L  path :: EventTypeName -> ByteString path eventTypeName =@@ -40,74 +40,62 @@   <> encodeUtf8 (unEventTypeName eventTypeName)   <> "/events" --- | @GET@ to @\/event-types\/NAME\/events@. Returns Conduit source--- for event batch consumption.-eventSource ::-  (MonadNakadi m, MonadResource m, MonadBaseControl IO m, MonadMask m, FromJSON a)-  => Config                  -- ^ Configuration parameter-  -> Maybe ConsumeParameters -- ^ Optional parameters for event consumption-  -> EventTypeName           -- ^ Name of the event type to consume-  -> Maybe [Cursor]          -- ^ Optional list of cursors; by default-                             -- consumption begins with the most-                             -- recent event-  -> m (ConduitM () (EventStreamBatch a)-        m ())                -- ^ Returns a Conduit source.-eventSource config maybeParams eventTypeName maybeCursors = do-  let consumeParams = fromMaybe (config^.L.consumeParameters) maybeParams-      queryParams   = buildSubscriptionConsumeQueryParameters consumeParams-  runReaderC () . snd <$>-    httpJsonBodyStream config ok200 (const (Right ())) [ (status429, errorTooManyRequests)-                                                       , (status429, errorEventTypeNotFound) ]+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-     . case maybeCursors of-         Just cursors -> let cursors' = ByteString.Lazy.toStrict (encode cursors)-                         in addRequestHeader "X-Nakadi-Cursors" cursors'-         Nothing      -> identity)+     . addCursors) $+    handler config --- | @GET@ to @\/event-types\/NAME\/events@. Returns Conduit source--- for event batch consumption. Retrieves configuration from--- environment.-eventSourceR ::-  (MonadNakadiEnv r m, MonadResource m, MonadBaseControl IO m, MonadMask m, FromJSON a)-  => Maybe ConsumeParameters -- ^ Optional parameters for event consumption-  -> EventTypeName           -- ^ Name of the event type to consume-  -> Maybe [Cursor]          -- ^ Optional list of cursors; by default-                             -- consumption begins with the most-                             -- recent event-  -> m (ConduitM () (EventStreamBatch a)-        m ())                -- ^ Returns a Conduit source.-eventSourceR maybeParams eventType maybeCursors = do-  config <- asks (view L.nakadiConfig)-  eventSource config maybeParams eventType maybeCursors+  where addCursors = case maybeCursors of+          Just cursors -> let cursors' = ByteString.Lazy.toStrict (encode cursors)+                          in addRequestHeader "X-Nakadi-Cursors" cursors'+          Nothing      -> identity --- | @POST@ to @\/event-types\/NAME\/events@. Publishes a batch of--- events for the specified event type.-eventPublish ::-  (MonadNakadi m, ToJSON a)-  => Config-  -> EventTypeName-  -> Maybe FlowId-  -> [a]-  -> m ()-eventPublish config eventTypeName maybeFlowId eventBatch =-  httpJsonNoBody config status200-  [ (Status 207 "Multi-Status", errorBatchPartiallySubmitted)-  , (status422, errorBatchNotSubmitted) ]-  (setRequestMethod "POST"-   . setRequestPath (path eventTypeName)-   . maybe identity (addRequestHeader "X-Flow-Id" . encodeUtf8 . unFlowId) maybeFlowId-   . setRequestBodyJSON eventBatch)+        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. Uses the configuration from--- the environment.-eventPublishR ::-  (MonadNakadiEnv r m, ToJSON a)+-- events for the specified event type.+eventsPublish+  :: (MonadNakadi b m, ToJSON a)   => EventTypeName-  -> Maybe FlowId   -> [a]   -> m ()-eventPublishR eventTypeName maybeFlowId eventBatch = do-  config <- asks (view L.nakadiConfig)-  eventPublish config eventTypeName maybeFlowId eventBatch+eventsPublish eventTypeName eventBatch = do+  config <- nakadiAsk+  httpJsonNoBody status200+    [ (Status 207 "Multi-Status", errorBatchPartiallySubmitted)+    , (status422, errorBatchNotSubmitted) ] $+    (setRequestMethod "POST"+     . includeFlowId config+     . setRequestPath (path eventTypeName)+     . setRequestBodyJSON eventBatch)

src/Network/Nakadi/EventTypes/Partitions.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.EventTypes.Partitions Description : Implementation of Nakadi Partitions API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Clasmeier 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -16,16 +16,11 @@  module Network.Nakadi.EventTypes.Partitions   ( eventTypePartitions-  , eventTypePartitionsR   , eventTypePartition-  , eventTypePartitionR   ) where -import           Network.Nakadi.Internal.Prelude--import           Control.Lens import           Network.Nakadi.Internal.Http-import qualified Network.Nakadi.Internal.Lenses  as L+import           Network.Nakadi.Internal.Prelude  path :: EventTypeName -> Maybe PartitionName -> ByteString path eventTypeName maybePartitionName =@@ -38,46 +33,25 @@  -- | @GET@ to @\/event-types\/EVENT-TYPE\/partitions@. Retrieves -- information about all partitions.-eventTypePartitions ::-  MonadNakadi m-  => Config        -- ^ Configuration-  -> EventTypeName -- ^ Name of Event Type-  -> m [Partition] -- ^ Partition Information-eventTypePartitions config eventTypeName =-  httpJsonBody config ok200 [(status404, errorEventTypeNotFound)]-  (setRequestPath (path eventTypeName Nothing))---- | @GET@ to @\/event-types\/EVENT-TYPE\/partitions@. Retrieves--- information about all partitions, using the configuration contained--- in the environment.-eventTypePartitionsR ::-  MonadNakadiEnv r m+eventTypePartitions+  :: MonadNakadi b m   => EventTypeName -- ^ Name of Event Type   -> m [Partition] -- ^ Partition Information-eventTypePartitionsR eventTypeName = do-  config <- asks (view L.nakadiConfig)-  eventTypePartitions config eventTypeName+eventTypePartitions eventTypeName = do+  config <- nakadiAsk+  httpJsonBody ok200 [(status404, errorEventTypeNotFound)] $+    (includeFlowId config+     . setRequestPath (path eventTypeName Nothing))  -- | @GET@ to @\/event-types\/EVENT-TYPE\/partitions\/PARTITION@. -- Retrieves information about a single partition.-eventTypePartition ::-  MonadNakadi m-  => Config        -- ^ Configuration-  -> EventTypeName -- ^ Name of Event Type-  -> PartitionName -- ^ Name of Partition to look up-  -> m Partition   -- ^ Partition Information-eventTypePartition config eventTypeName partitionName =-  httpJsonBody config ok200 []-  (setRequestPath (path eventTypeName (Just partitionName)))---- | @GET@ to @\/event-types\/EVENT-TYPE\/partitions\/PARTITION@.--- Retrieves information about a single partition, using the--- configuration contained in the environment.-eventTypePartitionR ::-  MonadNakadiEnv r m+eventTypePartition+  :: MonadNakadi b m   => EventTypeName -- ^ Name of Event Type   -> PartitionName -- ^ Name of Partition to look up   -> m Partition   -- ^ Partition Information-eventTypePartitionR eventTypeName partitionName = do-  config <- asks (view L.nakadiConfig)-  eventTypePartition config eventTypeName partitionName+eventTypePartition eventTypeName partitionName = do+  config <- nakadiAsk+  httpJsonBody ok200 [] $+    (includeFlowId config+     . setRequestPath (path eventTypeName (Just partitionName)))

src/Network/Nakadi/EventTypes/Schemas.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.EventTypes.Schemas Description : Implementation of Nakadi Schemas API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -18,16 +18,12 @@  module Network.Nakadi.EventTypes.Schemas   ( eventTypeSchemasGet-  , eventTypeSchemasGetR   , eventTypeSchema-  , eventTypeSchemaR   ) where  import           Network.Nakadi.Internal.Prelude -import           Control.Lens import           Network.Nakadi.Internal.Http-import qualified Network.Nakadi.Internal.Lenses  as L  path :: EventTypeName -> Maybe SchemaVersion -> ByteString path eventTypeName maybeSchemaVersion =@@ -41,14 +37,13 @@ -- | Retrieves schemas for the given 'EventTypeName' using low-level -- paging interface. @GET@ to @\/event-types\/NAME\/schemas@. eventTypeSchemasGet ::-  MonadNakadi m-  => Config        -- ^ Configuration-  -> EventTypeName -- ^ Name of Event Type+  MonadNakadi b m+  => EventTypeName -- ^ Name of Event Type   -> Maybe Offset   -> Maybe Limit   -> m EventTypeSchemasResponse-eventTypeSchemasGet config eventTypeName offset limit =-  httpJsonBody config ok200 []+eventTypeSchemasGet eventTypeName offset limit =+  httpJsonBody ok200 []   (setRequestMethod "GET"    . setRequestPath (path eventTypeName Nothing)    . setRequestQueryParameters [ ("offset", offset')@@ -58,39 +53,15 @@         defaultOffset =  0         defaultLimit  = 20 --- | @GET@ to @\/event-types\/NAME\/schemas@. Uses the configuration--- contained in the environment.-eventTypeSchemasGetR ::-  MonadNakadiEnv r m-  => EventTypeName -- ^ Name of Event Type-  -> Maybe Offset-  -> Maybe Limit-  -> m EventTypeSchemasResponse-eventTypeSchemasGetR eventTypeName offset limit = do-  config <- asks (view L.nakadiConfig)-  eventTypeSchemasGet config eventTypeName offset limit- -- | Look up the schema of an event type given its 'EventTypeName' and -- 'SchemaVersion'. @GET@ to -- @\/event-types\/EVENT-TYPE\/schemas\/SCHEMA@. eventTypeSchema ::-  MonadNakadi m-  => Config-  -> EventTypeName-  -> SchemaVersion-  -> m EventTypeSchema-eventTypeSchema config eventTypeName schemaVersion =-  httpJsonBody config ok200 []-  (setRequestMethod "GET" . setRequestPath (path eventTypeName (Just schemaVersion)))---- | Look up the schema of an event type given its 'EventTypeName' and--- 'SchemaVersion', using the configuration found in the environment.--- @GET@ to @\/event-types\/EVENT-TYPE\/schemas\/SCHEMA@.-eventTypeSchemaR ::-  MonadNakadiEnv r m+  MonadNakadi b m   => EventTypeName   -> SchemaVersion   -> m EventTypeSchema-eventTypeSchemaR eventTypeName schemaVersion = do-  config <- asks (view L.nakadiConfig)-  eventTypeSchema config eventTypeName schemaVersion+eventTypeSchema eventTypeName schemaVersion =+  httpJsonBody ok200 []+  (setRequestMethod "GET"+   . setRequestPath (path eventTypeName (Just schemaVersion)))

src/Network/Nakadi/EventTypes/ShiftedCursors.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.EventTypes.ShiftedCursors Description : Implementation of Nakadi ShiftedCursors API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -17,16 +17,12 @@  module Network.Nakadi.EventTypes.ShiftedCursors   ( cursorsShift'-  , cursorsShiftR'   , cursorsShift-  , cursorsShiftR   ) where  import           Network.Nakadi.Internal.Prelude -import           Control.Lens import           Network.Nakadi.Internal.Http-import qualified Network.Nakadi.Internal.Lenses  as L  path :: EventTypeName -> ByteString path eventTypeName =@@ -37,53 +33,30 @@ -- | @POST@ to @\/event-types\/EVENT-TYPE\/shifted-cursors@. Low level -- interface. cursorsShift' ::-  MonadNakadi m-  => Config          -- ^ Configuration-  -> EventTypeName   -- ^ Event Type-  -> [ShiftedCursor] -- ^ Cursors with Shift Distances-  -> m [Cursor]      -- ^ Resulting Cursors-cursorsShift' config eventTypeName cursors =-  httpJsonBody config ok200 []-  (setRequestMethod "POST"-   . setRequestPath (path eventTypeName)-   . setRequestBodyJSON cursors)---- | @POST@ to @\/event-types\/EVENT-TYPE\/shifted-cursors@. Low level--- interface. Retrieves the configuration from the environment.-cursorsShiftR' ::-  MonadNakadiEnv r m+  MonadNakadi b m   => EventTypeName   -- ^ Event Type   -> [ShiftedCursor] -- ^ Cursors with Shift Distances   -> m [Cursor]      -- ^ Resulting Cursors-cursorsShiftR' eventTypeName cursors = do-  config <- asks (view L.nakadiConfig)-  cursorsShift' config eventTypeName cursors+cursorsShift' eventTypeName cursors = do+  config <- nakadiAsk+  httpJsonBody ok200 [] $+    (setRequestMethod "POST"+     . includeFlowId config+     . setRequestPath (path eventTypeName)+     . setRequestBodyJSON cursors)  -- | @POST@ to @\/event-types\/EVENT-TYPE\/shifted-cursors@. High -- level interface. cursorsShift ::-  MonadNakadi m-  => Config        -- ^ Configuration-  -> EventTypeName -- ^ Event Type+  MonadNakadi b m+  => EventTypeName -- ^ Event Type   -> [Cursor]      -- ^ Cursors to shift   -> Int64         -- ^ Shift Distance   -> m [Cursor]    -- ^ Resulting Cursors-cursorsShift config eventTypeName cursors n =-  cursorsShift' config eventTypeName (map makeShiftCursor cursors)+cursorsShift eventTypeName cursors n = do+  cursorsShift' eventTypeName (map makeShiftCursor cursors)    where makeShiftCursor Cursor { .. } =           ShiftedCursor { _partition = _partition                         , _offset    = _offset                         , _shift     = n }---- | @POST@ to @\/event-types\/EVENT-TYPE\/shifted-cursors@. High--- level interface. Retrieves the configuration from the environment.-cursorsShiftR ::-  MonadNakadiEnv r m-  => EventTypeName -- ^ Event Type-  -> [Cursor]      -- ^ Cursors to shift-  -> Int64         -- ^ Shift Distance-  -> m [Cursor]    -- ^ Resulting Cursors-cursorsShiftR eventTypeName cursors n = do-  config <- asks (view L.nakadiConfig)-  cursorsShift config eventTypeName cursors n

+ src/Network/Nakadi/HttpBackendIO.hs view

@@ -0,0 +1,18 @@+{-|+Module      : Network.Nakadi.EventHttpBackendIO+Description : Exports IO based HTTP Backend+Copyright   : (c) Moritz Clasmeier 2018+License     : BSD3+Maintainer  : mtesseract@silverratio.net+Stability   : experimental+Portability : POSIX++This module exports the standard IO based Nakadi HTTP Backend to make+it part of the public API.+-}++module Network.Nakadi.HttpBackendIO+  ( httpBackendIO+  ) where++import           Network.Nakadi.Internal.HttpBackendIO

+ src/Network/Nakadi/Internal/Base.hs view

@@ -0,0 +1,15 @@+module Network.Nakadi.Internal.Base+  ( module Network.Nakadi.Internal.Types.Base+  , runNakadiWithBase+  ) where++import           Network.Nakadi.Internal.Prelude+import           Network.Nakadi.Internal.Types+import           Network.Nakadi.Internal.Types.Base++-- | This is a convenience function, which does `runNakadiT` and+-- `runNakadiBaseT` at once.+runNakadiWithBase+  :: Config b -> NakadiT b (NakadiBaseT m) a -> m a+runNakadiWithBase config =+  runNakadiBaseT . runNakadiT config

src/Network/Nakadi/Internal/Config.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Internal.Config Description : Nakadi Client Configuration (Internal)-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -32,3 +32,14 @@ buildSubscriptionConsumeQueryParameters params@ConsumeParameters { .. } =   buildConsumeQueryParameters params   ++ catMaybes [("max_uncommitted_events",) . encodeUtf8 . tshow <$> _maxUncommittedEvents]++-- | Default parameters for event consumption.+defaultConsumeParameters :: ConsumeParameters+defaultConsumeParameters = ConsumeParameters+  { _maxUncommittedEvents = Nothing+  , _batchLimit           = Nothing+  , _streamLimit          = Nothing+  , _batchFlushTimeout    = Nothing+  , _streamTimeout        = Nothing+  , _streamKeepAliveLimit = Nothing+  }

src/Network/Nakadi/Internal/Http.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Internal.Http Description : Nakadi Client HTTP (Internal)-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -11,10 +11,12 @@ -}  {-# LANGUAGE FlexibleContexts      #-}+{-# LANGUAGE GADTs                 #-} {-# LANGUAGE LambdaCase            #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RecordWildCards       #-} {-# LANGUAGE ScopedTypeVariables   #-}+{-# LANGUAGE TypeFamilies          #-}  module Network.Nakadi.Internal.Http   ( module Network.HTTP.Simple@@ -23,6 +25,9 @@   , httpJsonBody   , httpJsonNoBody   , httpJsonBodyStream+  , httpBuildRequest+  , conduitDecode+  , includeFlowId   , errorClientNotAuthenticated   , errorUnprocessableEntity   , errorAccessForbidden@@ -40,212 +45,243 @@  import           Network.Nakadi.Internal.Prelude -import           Conduit+import           Conduit                         hiding (throwM) import           Control.Arrow import           Control.Lens import           Control.Monad                   (void)-import           Control.Monad.Reader-import           Control.Monad.Trans.Resource import           Data.Aeson import qualified Data.ByteString.Lazy            as ByteString.Lazy-import qualified Data.Conduit.Binary             as Conduit+import qualified Data.ByteString.Lazy            as LB import qualified Data.Text                       as Text import           Network.HTTP.Client             (BodyReader,                                                   HttpException (..),                                                   HttpExceptionContent (..),-                                                  checkResponse, responseStatus)-import           Network.HTTP.Client.Conduit     (bodyReaderSource)-import           Network.HTTP.Simple+                                                  Manager, checkResponse,+                                                  responseBody, responseStatus)+import           Network.HTTP.Simple             hiding (Proxy) import           Network.HTTP.Types import           Network.HTTP.Types.Status import qualified Network.Nakadi.Internal.Lenses  as L-import           Network.Nakadi.Internal.Retry import           Network.Nakadi.Internal.Types import           Network.Nakadi.Internal.Util -conduitDecode ::-  (MonadIO m, FromJSON a)-  => Config -- ^ Configuration, containing the-            -- deserialization-failure-callback.+-- | If no deserializationFailureCallback is set in the provided+-- configuration (which is the default), a+-- DeserializationFailureCallback exception will be thrown. Otherwise,+-- simply run the callback.+conduitDecode+  :: forall a b m+   . (FromJSON a, MonadNakadi b m)+  => Config b -- ^ Configuration, containing the+              -- deserialization-failure-callback.   -> ConduitM ByteString a m () -- ^ Conduit deserializing bytestrings                                 -- into custom values-conduitDecode Config { .. } = awaitForever $ \a ->-  case eitherDecodeStrict a of-    Right v  -> yield v-    Left err -> liftIO $ callback a (Text.pack err)--    where callback = fromMaybe dummyCallback _deserializationFailureCallback-          dummyCallback _ _ = return ()+conduitDecode config = awaitForever $ \ a -> case eitherDecodeStrict' a of+  Right v   -> yield v+  Left  err -> lift . nakadiLiftBase $ callback a (Text.pack err)+ where+  callback bs err = case config ^. L.deserializationFailureCallback of+    Nothing -> throwM $ DeserializationFailure bs err+    Just cb -> cb bs err  -- | Throw 'HttpException' exception on server errors (5xx). checkNakadiResponse :: Request -> Response BodyReader -> IO () checkNakadiResponse request response =-  when (statusCode (responseStatus response) `div` 100 == 5) $-  throwIO $ HttpExceptionRequest request (StatusCodeException (void response) mempty)+  when (statusCode (responseStatus response) `div` 100 == 5)+    $ throwIO+    $ HttpExceptionRequest request (StatusCodeException (void response) mempty) -httpBuildRequest ::-  (MonadIO m, MonadCatch m)-  => Config -- ^ Configuration, contains the impure request modifier-  -> (Request -> Request) -- ^ Pure request modifier+httpBuildRequest+  :: MonadNakadi b m+  => (Request -> Request) -- ^ Pure request modifier   -> m Request -- ^ Resulting request to execute-httpBuildRequest Config { .. } requestDef =-  let manager = _manager-      request = requestDef _requestTemplate-                   & setRequestManager manager-                   & (\req -> req { checkResponse = checkNakadiResponse })-  in modifyRequest _requestModifier request-+httpBuildRequest requestDef = do+  config <- nakadiAsk+  let request = requestDef (config ^. L.requestTemplate)+        & \req -> req { checkResponse = checkNakadiResponse }+  modifyRequest (config ^. L.requestModifier) request  -- | Modify the Request based on a user function in the configuration.-modifyRequest :: (MonadIO m, MonadCatch m) => (Request -> IO Request) -> Request -> m Request-modifyRequest rm request =-  tryAny (liftIO (rm request)) >>= \case-    Right modifiedRequest -> return modifiedRequest-    Left  exn             -> throwIO $ RequestModificationException exn+modifyRequest+  :: MonadNakadi b m => (Request -> b Request) -> Request -> m Request+modifyRequest rm request = tryAny (nakadiLiftBase (rm request)) >>= \case+  Right modifiedRequest -> return modifiedRequest+  Left  exn             -> throwIO $ RequestModificationException exn  -- | Executes an HTTP request using the provided configuration and a -- pure request modifier.-httpExecRequest ::-  (MonadIO m, MonadMask m)-  => Config-  -> (Request -> Request)+httpExecRequest+  :: MonadNakadi b m+  => (Request -> Request)   -> m (Response ByteString.Lazy.ByteString)-httpExecRequest config requestDef = do-  req <- httpBuildRequest config requestDef-  retryAction config req (liftIO . (config^.L.http.L.httpLbs))+httpExecRequest requestDef = do+  config <- nakadiAsk+  req    <- httpBuildRequest requestDef+  nakadiLiftBase $ nakadiHttpLbs config req (config^.L.manager) +nakadiHttpLbs :: Config b+              -> Request+              -> Maybe Manager+              -> b (Response LB.ByteString)+nakadiHttpLbs config req maybeMngr =+  (config^.L.http.L.httpLbs) config req maybeMngr+ -- | Executes an HTTP request using the provided configuration and a -- pure request modifier. Returns the HTTP response and separately the -- response status.-httpExecRequestWithStatus ::-  (MonadIO m, MonadMask m)-  => Config -- ^ Configuration-  -> (Request -> Request) -- ^ Pure request modifier+httpExecRequestWithStatus+  :: MonadNakadi b m+  => (Request -> Request) -- ^ Pure request modifier   -> m (Response ByteString.Lazy.ByteString, Status)-httpExecRequestWithStatus config requestDef =-  (identity &&& getResponseStatus) <$> httpExecRequest config requestDef+httpExecRequestWithStatus requestDef =+  (identity &&& getResponseStatus) <$> httpExecRequest requestDef -httpJsonBody ::-  (MonadMask m, MonadIO m, FromJSON a)-  => Config-  -> Status+httpJsonBody+  :: (MonadNakadi b m, FromJSON a)+  => Status   -> [(Status, ByteString.Lazy.ByteString -> m NakadiException)]   -> (Request -> Request)   -> m a-httpJsonBody config successStatus exceptionMap requestDef = do-  (response, status) <- httpExecRequestWithStatus config requestDef+httpJsonBody successStatus exceptionMap requestDef = do+  (response, status) <- httpExecRequestWithStatus requestDef   if status == successStatus     then decodeThrow (getResponseBody response)     else case lookup status exceptionMap' of-           Just mkExn -> mkExn (getResponseBody response) >>= throwIO-           Nothing    -> throwIO (UnexpectedResponse (void response))-+      Just mkExn -> mkExn (getResponseBody response) >>= throwM+      Nothing    -> throwM (UnexpectedResponse (void response))   where exceptionMap' = exceptionMap ++ defaultExceptionMap -httpJsonNoBody ::-  (MonadMask m, MonadIO m)-  => Config-  -> Status+httpJsonNoBody+  :: MonadNakadi b m+  => Status   -> [(Status, ByteString.Lazy.ByteString -> m NakadiException)]   -> (Request -> Request)   -> m ()-httpJsonNoBody config successStatus exceptionMap requestDef = do-  (response, status) <- httpExecRequestWithStatus config requestDef-  unless (status == successStatus) $-    case lookup status exceptionMap' of-      Just mkExn -> mkExn (getResponseBody response) >>= throwIO-      Nothing    -> throwIO (UnexpectedResponse (void response))-+httpJsonNoBody successStatus exceptionMap requestDef = do+  (response, status) <- httpExecRequestWithStatus requestDef+  unless (status == successStatus) $ case lookup status exceptionMap' of+    Just mkExn -> mkExn (getResponseBody response) >>= throwIO+    Nothing    -> throwIO (UnexpectedResponse (void response))   where exceptionMap' = exceptionMap ++ defaultExceptionMap -httpJsonBodyStream ::-  (MonadMask m, MonadIO m, FromJSON a, MonadBaseControl IO m, MonadResource m)-  => Config-  -> Status-  -> (Response () -> Either Text b)+nakadiHttpResponseOpen :: Config b+                       -> Request+                       -> Maybe Manager+                       -> b (Response (ConduitM () ByteString b ()))+nakadiHttpResponseOpen config req maybeMngr =+  (config^.L.http.L.httpResponseOpen) config req maybeMngr++nakadiHttpResponseClose :: Config b -> Response () -> b ()+nakadiHttpResponseClose config rsp =+  (config^.L.http.L.httpResponseClose) rsp++httpJsonBodyStream+  :: forall b m r+   . (MonadNakadi b m, MonadMask m)+  => Status   -> [(Status, ByteString.Lazy.ByteString -> m NakadiException)]   -> (Request -> Request)-  -> m (b, ConduitM () a (ReaderT r m) ())-httpJsonBodyStream config successStatus f exceptionMap requestDef = do-  request <- httpBuildRequest config requestDef-  let manager           = config^.L.manager-      responseOpen      = config^.L.http.L.responseOpen-      responseClose     = config^.L.http.L.responseClose-      responseOpenRetry = retryAction config request (flip responseOpen manager)-  (_, response) <- allocate responseOpenRetry responseClose-  let response_  = void response-      bodySource = bodyReaderSource (getResponseBody response)-      status     = getResponseStatus response-  if status == successStatus-    then do connectCallback (config^.L.logFunc) response_-            let conduit = bodySource-                          .| Conduit.lines-                          .| conduitDecode config-            case f response_ of-              Left errMsg -> throwString (Text.unpack errMsg)-              Right b     -> return (b, readerC (const conduit))-    else case lookup status exceptionMap' of-           Just mkExn -> conduitDrainToLazyByteString bodySource >>= mkExn >>= throwIO-           Nothing    -> throwIO (UnexpectedResponse response_)+  -> (Response (ConduitM () ByteString m ()) -> m r)+  -> m r+httpJsonBodyStream successStatus exceptionMap requestDef handler = do+  config  <- nakadiAsk+  request <- httpBuildRequest requestDef+  bracket (nakadiLiftBase $ nakadiHttpResponseOpen config request (config^.L.manager))+          (nakadiLiftBase . nakadiHttpResponseClose config . void)+   $ \response -> wrappedHandler config response -  where exceptionMap' = exceptionMap ++ defaultExceptionMap+  where wrappedHandler :: Config b -> Response (ConduitM () ByteString b ()) -> m r+        wrappedHandler config response = do+          let response_      = void response+              status         = responseStatus response_+              bodySource     = responseBody responseLifted+              responseLifted = fmap (transPipe nakadiLiftBase) response+          if status == successStatus+            then do connectCallback config response_+                    handler responseLifted+            else case lookup status exceptionMap' of+                   Just mkExn ->+                     conduitDrainToLazyByteString bodySource >>= mkExn >>= throwM+                   Nothing -> throwM (UnexpectedResponse response_) -        connectCallback = case config^.L.streamConnectCallback of-          Just cb -> \logFunc response -> liftIO $ cb logFunc response-          Nothing -> \_ _ -> return ()+        exceptionMap' = exceptionMap ++ defaultExceptionMap -setRequestQueryParameters ::-  [(ByteString, ByteString)]-  -> Request-  -> Request+        -- connectCallback :: s -> t -> m ()+        connectCallback config response =+          nakadiLiftBase $ case config^.L.streamConnectCallback of+                         Just cb -> cb response+                         Nothing -> pure ()++setRequestQueryParameters :: [(ByteString, ByteString)] -> Request -> Request setRequestQueryParameters parameters = setRequestQueryString parameters'   where parameters' = map (fmap Just) parameters -defaultExceptionMap ::-  MonadThrow m-  => [(Status, ByteString.Lazy.ByteString -> m NakadiException)]+includeFlowId+  :: Config b+  -> Request+  -> Request+includeFlowId config =+  case config^.L.flowId of+    Just flowId -> setRequestHeader "X-Flow-Id" [encodeUtf8 (unFlowId flowId)]+    Nothing     -> identity++defaultExceptionMap+  :: MonadThrow m => [(Status, ByteString.Lazy.ByteString -> m NakadiException)] defaultExceptionMap =   [ (status401, errorClientNotAuthenticated)   , (status403, errorAccessForbidden)   , (status400, errorBadRequest)   , (status422, errorUnprocessableEntity)-  , (status409, errorConflict) ]+  , (status409, errorConflict)+  ] -errorClientNotAuthenticated :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException+errorClientNotAuthenticated+  :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorClientNotAuthenticated s = ClientNotAuthenticated <$> decodeThrow s  errorConflict :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorConflict s = Conflict <$> decodeThrow s -errorUnprocessableEntity :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException+errorUnprocessableEntity+  :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorUnprocessableEntity s = UnprocessableEntity <$> decodeThrow s -errorAccessForbidden :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException+errorAccessForbidden+  :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorAccessForbidden s = AccessForbidden <$> decodeThrow s -errorTooManyRequests :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException+errorTooManyRequests+  :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorTooManyRequests s = TooManyRequests <$> decodeThrow s -errorBadRequest :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException+errorBadRequest+  :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorBadRequest s = BadRequest <$> decodeThrow s -errorSubscriptionNotFound :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException+errorSubscriptionNotFound+  :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorSubscriptionNotFound s = SubscriptionNotFound <$> decodeThrow s -errorCursorAlreadyCommitted :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException+errorCursorAlreadyCommitted+  :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorCursorAlreadyCommitted s = CursorAlreadyCommitted <$> decodeThrow s -errorCursorResetInProgress :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException+errorCursorResetInProgress+  :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorCursorResetInProgress s = CursorResetInProgress <$> decodeThrow s -errorEventTypeNotFound :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException+errorEventTypeNotFound+  :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorEventTypeNotFound s = EventTypeNotFound <$> decodeThrow s -errorSubscriptionExistsAlready :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException+errorSubscriptionExistsAlready+  :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorSubscriptionExistsAlready s = SubscriptionExistsAlready <$> decodeThrow s -errorBatchPartiallySubmitted :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException+errorBatchPartiallySubmitted+  :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorBatchPartiallySubmitted s = BatchPartiallySubmitted <$> decodeThrow s -errorBatchNotSubmitted :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException+errorBatchNotSubmitted+  :: MonadThrow m => ByteString.Lazy.ByteString -> m NakadiException errorBatchNotSubmitted s = BatchNotSubmitted <$> decodeThrow s

+ src/Network/Nakadi/Internal/HttpBackendIO.hs view

@@ -0,0 +1,71 @@+{-|+Module      : Network.Nakadi.EventHttpBackendIO+Description : Implements IO based HTTP Backend+Copyright   : (c) Moritz Clasmeier 2018+License     : BSD3+Maintainer  : mtesseract@silverratio.net+Stability   : experimental+Portability : POSIX++This module implements the standard IO based Nakadi HTTP Backend.+Useful, in case you just want to overwrite a subset of implementations+of the standard backend.+-}++module Network.Nakadi.Internal.HttpBackendIO where++import           Control.Lens                         ((<&>))+import qualified Data.ByteString.Lazy                 as LB+import           Data.Conduit                         (ConduitM, transPipe)+import           Network.HTTP.Client                  (Manager, Request,+                                                       Response)+import qualified Network.HTTP.Client                  as HTTP (httpLbs,+                                                               responseClose,+                                                               responseOpen)+import           Network.HTTP.Client.Conduit          (bodyReaderSource)+import           Network.HTTP.Client.TLS              (getGlobalManager)+import           Network.Nakadi.Internal.Prelude+import           Network.Nakadi.Internal.Retry+import           Network.Nakadi.Internal.Types.Config++getHttpManager+  :: MonadIO m+  => Maybe Manager+  -> m Manager+getHttpManager Nothing        = liftIO getGlobalManager+getHttpManager (Just manager) = pure manager++httpBackendIO+  :: (MonadMask b, MonadIO b)+  => HttpBackend b+httpBackendIO = HttpBackend+  { _httpLbs = httpLbs+  , _httpResponseOpen = responseOpen+  , _httpResponseClose = responseClose+  }++responseOpen+  :: MonadIO b+  => Config b+  -> Request+  -> Maybe Manager+  -> b (Response (ConduitM () ByteString b ()))+responseOpen _config req maybeMngr = do+  mngr <- getHttpManager maybeMngr+  liftIO $ HTTP.responseOpen req mngr <&> fmap (transPipe liftIO . bodyReaderSource)++responseClose+  :: MonadIO b+  => Response ()+  -> b ()+responseClose = liftIO . HTTP.responseClose++httpLbs+  :: (MonadMask b, MonadIO b)+  => Config b+  -> Request+  -> Maybe Manager+  -> b (Response LB.ByteString)+httpLbs config req maybeMngr = do+  mngr <- getHttpManager maybeMngr+  retryAction config req (\r -> liftIO $ HTTP.httpLbs r mngr)

src/Network/Nakadi/Internal/Lenses.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Lenses Description : Nakadi Client Library Lenses (Internal)-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -24,22 +24,20 @@ import           Network.Nakadi.Internal.Prelude  import           Control.Lens-import           Data.Text                                  (Text)+import           Data.Text                             (Text) import           Data.Time.Clock-import           Data.UUID                                  (UUID)+import           Data.UUID                             (UUID) import           Network.Nakadi.Internal.TH  import           Network.Nakadi.Internal.Types.Config import           Network.Nakadi.Internal.Types.Service-import           Network.Nakadi.Internal.Types.Subscription -class HasNakadiConfig s a where-  nakadiConfig :: Lens' s a- makeNakadiLenses ''Config makeNakadiLenses ''HttpBackend makeNakadiLenses ''Cursor makeNakadiLenses ''EventStreamBatch+makeNakadiLenses ''DataChangeEvent+makeNakadiLenses ''DataChangeEventEnriched makeNakadiLenses ''SubscriptionEventStreamBatch makeNakadiLenses ''EventMetadata makeNakadiLenses ''EventMetadataEnriched@@ -48,7 +46,6 @@ makeNakadiLenses ''CursorDistanceResult makeNakadiLenses ''Timestamp makeNakadiLenses ''SubscriptionEventStream-makeNakadiLenses ''SubscriptionEventStreamContext makeNakadiLenses ''EventTypeSchema makeNakadiLenses ''EventType makeNakadiLenses ''EventTypeSchemasResponse

src/Network/Nakadi/Internal/Prelude.hs view

@@ -1,4 +1,4 @@- {-|+{-| Module      : Network.Nakadi.Prelude Description : Nakadi Client Prelude (Internal) Copyright   : (c) Moritz Schulte 2017@@ -28,6 +28,7 @@   , undefined   , error   , MonadIO+  , liftIO   , Request   , Response   ) where@@ -45,7 +46,7 @@ import qualified Data.Text                  as Text import           Data.Text.Encoding import           Network.HTTP.Client        (Request, Response)-import           Prelude                    hiding (id, undefined, error)+import           Prelude                    hiding (error, id, undefined) import qualified Prelude  tshow :: Show a => a -> Text

src/Network/Nakadi/Internal/Retry.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Internal.Retry Description : Nakadi Client Retry Mechanism-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -10,8 +10,8 @@ This module provides the basic retry mechanism via the retry package. -} -{-# LANGUAGE LambdaCase          #-}-{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase       #-}  module Network.Nakadi.Internal.Retry   ( retryAction@@ -20,46 +20,56 @@ import           Network.Nakadi.Internal.Prelude  import           Control.Lens-import           Control.Monad.IO.Class import           Control.Retry import           Network.HTTP.Client import           Network.HTTP.Types.Status-import qualified Network.Nakadi.Internal.Lenses  as L-import           Network.Nakadi.Internal.Types+import qualified Network.Nakadi.Internal.Lenses       as L+import           Network.Nakadi.Internal.Types.Config  -- | Invokes the HTTP Error Callback set in the configuration for the -- provided 'Request', 'HttpException' and 'RetryStatus'. If no -- callback is set, this is no-op.-invokeHttpErrorCallback :: MonadIO m => Config -> Request -> HttpException -> RetryStatus -> m ()-invokeHttpErrorCallback config req exn retryStatus = liftIO $+invokeHttpErrorCallback ::+  (MonadIO b)+  => Config b+  -> Request+  -> HttpException+  -> RetryStatus+  -> b ()+invokeHttpErrorCallback config req exn retryStatus =   case config^.L.httpErrorCallback of     Just cb -> do-      finalFailure <- applyPolicy (config^.L.retryPolicy) retryStatus >>= \case-        Just _  -> pure False-        Nothing -> pure True+      finalFailure <- isFinalFailure       cb req exn retryStatus finalFailure     Nothing -> pure () +  where isFinalFailure = do+          let policy = liftIORetryPolicy (config^.L.retryPolicy)+          applyPolicy policy retryStatus >>= \case+            Just _  -> pure False+            Nothing -> pure True+++liftIORetryPolicy :: MonadIO b => RetryPolicyM IO -> RetryPolicyM b+liftIORetryPolicy rp = RetryPolicyM $ liftIO . getRetryPolicyM rp+ -- | Try to execute the provided IO action using the provided retry -- policy. If executing the IO action raises specific exceptions of -- type 'HttpException', the action will be potentially retried -- (depending on the retry policy). retryAction ::-  (MonadIO m, MonadMask m)-  => Config+  (MonadIO b, MonadMask b)+  => Config b   -> Request-  -> (Request -> m a)-  -> m a+  -> (Request -> b a)+  -> b a retryAction config req ma =-  let policy = config^.L.retryPolicy-      nakadiRetryPolicy = RetryPolicyM $ \retryStatus ->-        liftIO (getRetryPolicyM policy retryStatus)-  in recovering nakadiRetryPolicy [handlerHttp] (const (ma req))+  let policy = liftIORetryPolicy (config^.L.retryPolicy)+  in recovering policy [handlerHttp] (\_retryStatus -> ma req)    where handlerHttp retryStatus = Handler $ \exn -> do           invokeHttpErrorCallback config req exn retryStatus           pure $ shouldRetry exn-         shouldRetry (HttpExceptionRequest _ exceptionContent) =           case exceptionContent of             StatusCodeException response _ ->

src/Network/Nakadi/Internal/Types.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Internal.Types Description : Nakadi Client Types (Internal)-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Clasmeier 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -10,9 +10,19 @@ Exports all types for internal usage. -} -{-# LANGUAGE ConstraintKinds  #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE RankNTypes       #-}+{-# LANGUAGE ConstraintKinds            #-}+{-# LANGUAGE DefaultSignatures          #-}+{-# LANGUAGE DeriveFunctor              #-}+{-# LANGUAGE FlexibleContexts           #-}+{-# LANGUAGE FlexibleInstances          #-}+{-# LANGUAGE FunctionalDependencies     #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses      #-}+{-# LANGUAGE RankNTypes                 #-}+{-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE ScopedTypeVariables        #-}+{-# LANGUAGE TypeFamilies               #-}+{-# LANGUAGE UndecidableInstances       #-}  module Network.Nakadi.Internal.Types   ( module Network.Nakadi.Internal.Types.Config@@ -20,33 +30,208 @@   , module Network.Nakadi.Internal.Types.Logger   , module Network.Nakadi.Internal.Types.Problem   , module Network.Nakadi.Internal.Types.Service-  , module Network.Nakadi.Internal.Types.Subscription   , module Network.Nakadi.Internal.Types.Util-  , MonadNakadi-  , MonadNakadiEnv-  , HasNakadiConfig+  , module Network.Nakadi.Internal.Types.Base+  , HasNakadiConfig(..)+  , MonadNakadi(..)+  , MonadNakadiIO+  , NakadiT(..)+  , runNakadiT   ) where -import           Network.Nakadi.Internal.Lenses+import           Control.Monad.Base+import           Control.Monad.Catch+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           Control.Monad.Trans.Class+import           Control.Monad.Trans.Control+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           Network.Nakadi.Internal.Prelude+import           Network.Nakadi.Internal.Types.Base import           Network.Nakadi.Internal.Types.Config import           Network.Nakadi.Internal.Types.Exceptions import           Network.Nakadi.Internal.Types.Logger import           Network.Nakadi.Internal.Types.Problem import           Network.Nakadi.Internal.Types.Service-import           Network.Nakadi.Internal.Types.Subscription import           Network.Nakadi.Internal.Types.Util +-- * Define Typeclasses -import           Control.Exception.Safe-import           Control.Monad.IO.Class-import           Control.Monad.Reader.Class+class HasNakadiConfig b r | r -> b where+  nakadiConfig :: r -> Config b --- | Type constraint synonym for encapsulating the monad constraints--- required by most funtions in this package.-type MonadNakadi m = (MonadIO m, MonadCatch m, MonadThrow m, MonadMask m)+-- | The `MonadNakadi` typeclass is implemented by monads in which+-- Nakadi can be called. The first parameter (`b`) denotes the `base+-- monad`. This is the monad in which the core actions are run. This+-- includes executing (non-streaming) HTTP requests and running+-- user-provided callbacks. The typeclass provides methods for+-- * retrieving the Nakadi configuration+-- * locally changing the Nakadi configuration+-- * extracting specific Nakadi configuration values+-- * lifting actions from the+-- The `MonadNakadi` typeclass is modelled closely after `MonadReader`.+class (MonadNakadiBase b m, MonadThrow b, MonadMask b, MonadThrow m, MonadCatch m)+   => MonadNakadi b m | m -> b where+  nakadiAsk :: m (Config b)+  default nakadiAsk :: (MonadNakadi b n, MonadTrans t, m ~ t n) => m (Config b)+  nakadiAsk = lift nakadiAsk --- | Type constraint synonym for encapsulating the monad constraints--- required by most funtions in this package. Reader Monad version,--- expects a 'Config' to be available in the current reader--- environment.-type MonadNakadiEnv r m = (MonadNakadi m, MonadReader r m, HasNakadiConfig r Config)+-- * Define Types++type MonadNakadiIO = MonadNakadi IO++-- | The `NakadiT` type is just a specialized `ReaderT` monad.+newtype NakadiT b m a = NakadiT { _runNakadiT :: Config b -> m a }++-- * Provide Typeclass Implementaions++-- ** Implement typeclass instances for `NakadiT`.++-- | `Functor` for `NakadiT`.+instance Functor m => Functor (NakadiT b m) where+  fmap f (NakadiT n) = NakadiT (\c -> fmap f (n c))++-- | `Applicative` for `NakadiT`.+instance (Applicative m) => Applicative (NakadiT b m) where+  pure a  = NakadiT $ \_conf -> pure a+  {-# INLINE pure #-}+  f <*> v = NakadiT $ \ c -> _runNakadiT f c <*> _runNakadiT v c+  {-# INLINE (<*>) #-}+  u *> v = NakadiT $ \ c -> _runNakadiT u c *> _runNakadiT v c+  {-# INLINE (*>) #-}+  u <* v = NakadiT $ \ c -> _runNakadiT u c <* _runNakadiT v c+  {-# INLINE (<*) #-}++-- | 'Monad'+instance (Monad m) => Monad (NakadiT b m) where+  return   = lift . return+  m >>= k  = NakadiT $ \ c -> do+    a <- _runNakadiT m c+    _runNakadiT (k a) c+  {-# INLINE (>>=) #-}+  (>>) = (*>)+  {-# INLINE (>>) #-}+  fail msg = lift (fail msg)+  {-# INLINE fail #-}++-- | 'MonadTrans'+instance MonadTrans (NakadiT b) where+    lift a = NakadiT (const a)+    {-# INLINE lift #-}++-- | 'MonadThrow'+instance (Monad b, MonadThrow m) => MonadThrow (NakadiT b m) where+  throwM e = lift $ Control.Monad.Catch.throwM e++-- | 'MonadCatch'+instance (Monad b, MonadCatch m) => MonadCatch (NakadiT b m) where+  catch (NakadiT b) h =+    NakadiT $ \ c -> b c `Control.Monad.Catch.catch` \e -> _runNakadiT (h e) c++-- | 'MonadMask'+instance (Monad b, MonadMask m) => MonadMask (NakadiT b m) where+  mask a = NakadiT $ \e -> mask $ \u -> _runNakadiT (a $ q u) e+    where q :: (m a -> m a) -> NakadiT e m a -> NakadiT e m a+          q u (NakadiT b) = NakadiT (u . b)+  uninterruptibleMask a =+    NakadiT $ \e -> uninterruptibleMask $ \u -> _runNakadiT (a $ q u) e+      where q :: (m a -> m a) -> NakadiT e m a -> NakadiT e m a+            q u (NakadiT b) = NakadiT (u . b)++-- | 'MonadIO'+instance (Monad b, MonadIO m) => MonadIO (NakadiT b m) where+  liftIO = lift . liftIO++-- | 'MonadBase'+instance (Monad m, MonadBase b' m) => MonadBase b' (NakadiT b m) where+  liftBase = liftBaseDefault++-- | 'MonadReader'+instance (Monad b, MonadReader r m) => MonadReader r (NakadiT b m) where+  ask = lift ask+  local = mapNakadiT . local++-- | 'MonadLogger'+instance MonadLogger m => MonadLogger (NakadiT b m)++-- | 'MonadState'+instance (Monad b, MonadState s m) => MonadState s (NakadiT b m) where+  get = lift get+  put = lift . put++-- | 'MonadUnliftIO'+instance (Monad b, MonadUnliftIO m) => MonadUnliftIO (NakadiT b m) where+  {-# INLINE askUnliftIO #-}+  askUnliftIO =+    NakadiT $ \r ->+    withUnliftIO $ \u ->+    return (UnliftIO (unliftIO u . runNakadiT r))++-- | 'MonadTransControl'+instance MonadTransControl (NakadiT b) where+  type StT (NakadiT b) a = a+  liftWith f = NakadiT $ \r -> f $ \t -> _runNakadiT t r+  restoreT = NakadiT . const+  {-# INLINABLE liftWith #-}+  {-# INLINABLE restoreT #-}++-- | 'MonadBaseControl'+instance MonadBaseControl b' m => MonadBaseControl b' (NakadiT b m) where+  type StM (NakadiT b m) a = ComposeSt (NakadiT b) m a+  liftBaseWith = defaultLiftBaseWith+  restoreM     = defaultRestoreM++-- | 'MonadNakadiBase'+instance {-# OVERLAPPABLE #-} MonadNakadiBase b m => MonadNakadiBase b (NakadiT b m)++-- ** Implementations 'MonadNakadi' typeclass for transformers.++-- | 'ReaderT'+instance ( MonadMask b+         , MonadCatch m+         , MonadNakadiBase b (ReaderT r m)+         , HasNakadiConfig b r )+      => MonadNakadi b (ReaderT r m) where+  nakadiAsk = asks nakadiConfig++-- | 'NakadiT'+instance ( MonadCatch m+         , MonadMask b+         , MonadNakadiBase b (NakadiT b m) )+      => MonadNakadi b (NakadiT b m) where+  nakadiAsk = NakadiT return++-- | 'WriterT' (lazy)+instance (MonadNakadi b m, Monoid w) => MonadNakadi b (Writer.Lazy.WriterT w m)++-- | 'WriterT' (strict)+instance (MonadNakadi b m, Monoid w) => MonadNakadi b (Writer.Strict.WriterT w m)++-- | 'StateT' (strict)+instance (MonadNakadi b m) => MonadNakadi b (State.Strict.StateT s m)++-- | 'StateT' (lazy)+instance (MonadNakadi b m) => MonadNakadi b (State.Lazy.StateT s m)++-- | 'LoggingT'+instance (MonadNakadi b m) => MonadNakadi b (LoggingT m)++-- | 'NoLoggingT'+instance (MonadNakadi b m) => MonadNakadi b (NoLoggingT m)++-- | 'ResourceT'.+instance (MonadNakadi b m) => MonadNakadi b (ResourceT m)++-- * Convenience Functions++runNakadiT :: Config b -> NakadiT b m a -> m a+runNakadiT = flip _runNakadiT++mapNakadiT :: (m a -> m a) -> NakadiT b m a -> NakadiT b m a+mapNakadiT f n = NakadiT $ \ c -> f (_runNakadiT n c)

+ src/Network/Nakadi/Internal/Types/Base.hs view

@@ -0,0 +1,62 @@+{-# LANGUAGE DefaultSignatures          #-}+{-# LANGUAGE DeriveFunctor              #-}+{-# LANGUAGE FlexibleContexts           #-}+{-# LANGUAGE FlexibleInstances          #-}+{-# LANGUAGE GADTs                      #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses      #-}+{-# LANGUAGE TypeFamilies               #-}+{-# LANGUAGE UndecidableInstances       #-}++module Network.Nakadi.Internal.Types.Base where++import           Control.Monad.Base+import           Control.Monad.Logger+import           Control.Monad.Reader+import           Control.Monad.State+import qualified Control.Monad.State.Lazy        as State.Lazy+import qualified Control.Monad.State.Strict      as State.Strict+import           Control.Monad.Trans.Resource+import           Control.Monad.Writer+import qualified Control.Monad.Writer.Lazy       as Writer.Lazy+import qualified Control.Monad.Writer.Strict     as Writer.Strict+import           Network.Nakadi.Internal.Prelude++class (Monad b, Monad m) => MonadNakadiBase b m where+  nakadiLiftBase :: b a -> m a+  default nakadiLiftBase :: (MonadNakadiBase b n, MonadTrans t, m ~ t n) => b a -> m a+  nakadiLiftBase = lift . nakadiLiftBase++instance {-# OVERLAPPING #-} MonadNakadiBase IO IO where+  nakadiLiftBase = identity++instance {-# OVERLAPPING #-} Monad m => MonadNakadiBase (ReaderT r m) (ReaderT r m) where+  nakadiLiftBase = identity++instance {-# OVERLAPPING #-} Monad m => MonadNakadiBase (LoggingT (ReaderT r m)) (LoggingT (ReaderT r m)) where+  nakadiLiftBase = identity++instance {-# OVERLAPPING #-} Monad m => MonadNakadiBase (NakadiBaseT m) (NakadiBaseT m) where+  nakadiLiftBase = identity++instance {-# OVERLAPPABLE #-} MonadNakadiBase b m => MonadNakadiBase b (ReaderT r m)+instance {-# OVERLAPPABLE #-} (MonadNakadiBase b m, Monoid w) => MonadNakadiBase b (Writer.Strict.WriterT w m)+instance {-# OVERLAPPABLE #-} (MonadNakadiBase b m, Monoid w) => MonadNakadiBase b (Writer.Lazy.WriterT w m)+instance {-# OVERLAPPABLE #-} MonadNakadiBase b m => MonadNakadiBase b (LoggingT m)+instance {-# OVERLAPPABLE #-} MonadNakadiBase b m => MonadNakadiBase b (NoLoggingT m)+instance {-# OVERLAPPABLE #-} MonadNakadiBase b m => MonadNakadiBase b (ResourceT m)+instance {-# OVERLAPPABLE #-} (MonadNakadiBase b m) => MonadNakadiBase b (State.Strict.StateT s m)+instance {-# OVERLAPPABLE #-} (MonadNakadiBase b m) => MonadNakadiBase b (State.Lazy.StateT s m)++newtype NakadiBaseT m a = NakadiBaseT+  { runNakadiBaseT :: m a+  } deriving ( Functor, Applicative, Monad, MonadIO+             , MonadThrow, MonadCatch, MonadMask+             , MonadReader r, MonadWriter w, MonadState s+             , MonadLogger)++instance MonadTrans NakadiBaseT where+    lift = NakadiBaseT++instance (MonadBase b m) => MonadBase b (NakadiBaseT m) where+  liftBase = liftBaseDefault

src/Network/Nakadi/Internal/Types/Config.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Internal.Types.Config Description : Nakadi Client Configuration Types (Internal)-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -10,47 +10,43 @@ Internal configuration specific types. -} +{-# LANGUAGE GADTs      #-}+{-# LANGUAGE RankNTypes #-} {-# LANGUAGE StrictData #-}  module Network.Nakadi.Internal.Types.Config where -import           Network.Nakadi.Internal.Prelude-+import           Conduit import           Control.Retry+import qualified Data.ByteString.Lazy                  as LB import           Network.HTTP.Client--import qualified Data.ByteString.Lazy            as LB (ByteString)-import           Network.Nakadi.Types.Logger+import           Network.Nakadi.Internal.Prelude+import           Network.Nakadi.Internal.Types.Logger+import           Network.Nakadi.Internal.Types.Service  -- | Config -type StreamConnectCallback = Maybe LogFunc -> Response () -> IO ()+type StreamConnectCallback m = Response () -> m ()  -- | Type synonym for user-provided callbacks which are used for HTTP -- Errror propagation.-type HttpErrorCallback = Request -> HttpException -> RetryStatus -> Bool -> IO ()+type HttpErrorCallback m = Request -> HttpException -> RetryStatus -> Bool -> m () -data Config = Config-  { _requestTemplate                :: Request-  , _requestModifier                :: Request -> IO Request-  , _manager                        :: Manager-  , _consumeParameters              :: ConsumeParameters-  , _deserializationFailureCallback :: Maybe (ByteString -> Text -> IO ())-  , _streamConnectCallback          :: Maybe StreamConnectCallback-  , _logFunc                        :: Maybe LogFunc-  , _retryPolicy                    :: RetryPolicyM IO-  , _http                           :: HttpBackend-  , _httpErrorCallback              :: Maybe HttpErrorCallback-  }+type ConfigIO = Config IO --- | Type encapsulating the HTTP backend functions used by this--- package. By default the corresponding functions from the--- http-client package are used. Useful, for e.g., testing.-data HttpBackend = HttpBackend-  { _httpLbs       :: Request -> IO (Response LB.ByteString)-  , _responseOpen  :: Request -> Manager -> IO (Response BodyReader)-  , _responseClose :: Response BodyReader -> IO ()-  }+data Config m where+  Config :: { _requestTemplate                :: Request+            , _requestModifier                :: Request -> m Request+            , _manager                        :: Maybe Manager+            , _consumeParameters              :: Maybe ConsumeParameters+            , _deserializationFailureCallback :: Maybe (ByteString -> Text -> m ())+            , _streamConnectCallback          :: Maybe (StreamConnectCallback m)+            , _logFunc                        :: Maybe (LogFunc m)+            , _retryPolicy                    :: RetryPolicyM IO+            , _http                           :: HttpBackend m+            , _httpErrorCallback              :: Maybe (HttpErrorCallback m)+            , _flowId                         :: Maybe FlowId+            } -> Config m  -- | ConsumeParameters @@ -61,5 +57,11 @@   , _batchFlushTimeout    :: Maybe Int32   , _streamTimeout        :: Maybe Int32   , _streamKeepAliveLimit :: Maybe Int32-  , _flowId               :: Maybe Text   } deriving (Show, Eq, Ord)+++data HttpBackend b = HttpBackend+  { _httpLbs           :: Config b -> Request -> Maybe Manager -> b (Response LB.ByteString)+  , _httpResponseOpen  :: Config b -> Request -> Maybe Manager -> b (Response (ConduitM () ByteString b ()))+  , _httpResponseClose :: Response () -> b ()+  }

src/Network/Nakadi/Internal/Types/Exceptions.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Internal.Types.Exceptions Description : Nakadi Client Exceptions (Internal)-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Clasmeier 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -16,7 +16,6 @@  import           Network.Nakadi.Internal.Prelude -import qualified Data.ByteString.Lazy            as ByteString.Lazy import           Network.Nakadi.Types.Problem import           Network.Nakadi.Types.Service @@ -27,7 +26,7 @@                      | AccessForbidden Problem                      | UnprocessableEntity Problem                      | Conflict Problem-                     | DeserializationFailure ByteString.Lazy.ByteString+                     | DeserializationFailure ByteString Text                      | UnexpectedResponse (Response ())                      | NotFound Problem                      | TooManyRequests Problem@@ -39,6 +38,7 @@                      | SubscriptionExistsAlready Subscription                      | RequestModificationException SomeException                      | CursorDistanceNoResult+                     | StreamIdMissing                      deriving (Show, Typeable)  instance Exception NakadiException

src/Network/Nakadi/Internal/Types/Logger.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Internal.Types.Logger Description : Nakadi Client Logger Types (Internal)-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -17,4 +17,7 @@  -- | Type of a logger callback provided to nakadi-client for logging -- purposes.-type LogFunc = Loc -> LogSource -> LogLevel -> LogStr -> IO ()+type LogFunc m = LogSource -> LogLevel -> LogStr -> m ()++-- | 'LogFunc' specialized to IO.+type LogFuncIO = LogFunc IO

src/Network/Nakadi/Internal/Types/Service.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Internal.Types.Service Description : Nakadi Client Service Types (Internal)-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -213,6 +213,9 @@   }   deriving (Show, Eq, Ord, Generic) +instance IsString FlowId where+  fromString = FlowId . Text.pack+ instance ToJSON FlowId where   toJSON = String . unFlowId @@ -779,7 +782,6 @@ deriveJSON nakadiJsonOptions ''DataChangeEvent  -- | A DataChangeEvent enriched by Nakadi- data DataChangeEventEnriched a = DataChangeEventEnriched   { _payload  :: a -- Cannot be named '_data', as this this would                    -- cause the lense 'data' to be created, which is a

− src/Network/Nakadi/Internal/Types/Subscription.hs

@@ -1,28 +0,0 @@-{-|-Module      : Network.Nakadi.Internal.Types.Subscription-Description : Nakadi Client Subscription Types (Internal)-Copyright   : (c) Moritz Schulte 2017-License     : BSD3-Maintainer  : mtesseract@silverratio.net-Stability   : experimental-Portability : POSIX--Internal Subscription specific types, which are not part of the Nakadi-Service API but custom to this package.--}--{-# LANGUAGE StrictData #-}--module Network.Nakadi.Internal.Types.Subscription where--import           Network.Nakadi.Internal.Types.Config-import           Network.Nakadi.Internal.Types.Service---- | This context is required in the environment for running a--- subscription. It is managed by the library, not by the user.--data SubscriptionEventStreamContext = SubscriptionEventStreamContext-  { _streamId       :: StreamId-  , _subscriptionId :: SubscriptionId-  , _ctxConfig      :: Config-  }

src/Network/Nakadi/Internal/Util.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Internal.Util Description : Nakadi Client Utilities (Internal)-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Clasmeier 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -10,7 +10,8 @@ Internal utility functions. -} -{-# LANGUAGE TupleSections #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TupleSections    #-}  module Network.Nakadi.Internal.Util   ( conduitDrainToLazyByteString@@ -26,20 +27,25 @@ 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 -import           Network.Nakadi.Types+import           Network.Nakadi.Internal.Types -conduitDrainToLazyByteString :: Monad m-                             => ConduitM () ByteString m ()-                             -> m ByteString.Lazy.ByteString+conduitDrainToLazyByteString ::+  Monad b+  => ConduitM () ByteString b ()+  -> b ByteString.Lazy.ByteString conduitDrainToLazyByteString conduit =   toLazyByteString <$> (conduit $$ sinkBuilder) -decodeThrow :: (FromJSON a, MonadThrow m) => ByteString.Lazy.ByteString -> m a-decodeThrow s = case decode s of-  Just a  -> return a-  Nothing -> throwIO (DeserializationFailure s)+decodeThrow+  :: (FromJSON a, MonadThrow m)+  => ByteString.Lazy.ByteString+  -> m a+decodeThrow s = case eitherDecode' s of+  Right a  -> pure a+  Left err -> throwM (DeserializationFailure (ByteString.Lazy.toStrict s) (Text.pack err))  sequenceSnd :: Functor f => (a, f b) -> f (a, b) sequenceSnd (a, fb) = (a,) <$> fb

+ src/Network/Nakadi/Prelude.hs view

@@ -0,0 +1,24 @@+{-|+Module      : Network.Nakadi.Prelude+Description : Nakadi Client Library+Copyright   : (c) Moritz Clasmeier 2018+License     : BSD3+Maintainer  : mtesseract@silverratio.net+Stability   : experimental+Portability : POSIX++Convenience exports that should be fine to import unqualified.+-}++module Network.Nakadi.Prelude+  ( MonadNakadi(..)+  , MonadNakadiBase(..)+  , HasNakadiConfig(..)+  , NakadiT+  , NakadiBaseT+  , NakadiException+  , runNakadiT+  , runNakadiBaseT+  ) where++import           Network.Nakadi

src/Network/Nakadi/Registry.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Registry Description : Implementation of Nakadi Registry API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -10,32 +10,22 @@ This module implements the @\/registry@ API. -} -{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}  module Network.Nakadi.Registry   ( registryPartitionStrategies-  , registryPartitionStrategiesR   ) where -import           Control.Lens import           Network.Nakadi.Internal.Http import           Network.Nakadi.Internal.Prelude -import qualified Network.Nakadi.Internal.Lenses  as L- path :: ByteString path = "/registry"  -- | Retrieves supported partitioning strategies from Nakadi. @GET@ to -- @\/registry\/partition-strategies@.-registryPartitionStrategies :: MonadNakadi m => Config -> m [PartitionStrategy]-registryPartitionStrategies config =-  httpJsonBody config status200 []+registryPartitionStrategies :: MonadNakadi r m => m [PartitionStrategy]+registryPartitionStrategies = httpJsonBody+  status200+  []   (setRequestMethod "GET" . setRequestPath (path <> "/partition-strategies"))---- | Retrieves supported partitioning strategies from Nakadi,--- obtaining configuration from environment.-registryPartitionStrategiesR :: MonadNakadiEnv r m => m [PartitionStrategy]-registryPartitionStrategiesR = do-  config <- asks (view L.nakadiConfig)-  registryPartitionStrategies config

src/Network/Nakadi/Subscriptions.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Subscriptions.Stats Description : Implementation of Nakadi Subscription API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -20,15 +20,10 @@   , module Network.Nakadi.Subscriptions.Stats   , module Network.Nakadi.Subscriptions.Subscription   , subscriptionCreate'-  , subscriptionCreateR'   , subscriptionCreate-  , subscriptionCreateR   , subscriptionsList'-  , subscriptionsListR'   , subscriptionsSource-  , subscriptionsSourceR   , subscriptionsList-  , subscriptionsListR   ) where  import           Network.Nakadi.Internal.Prelude@@ -50,72 +45,39 @@  -- | @POST@ to @\/subscriptions@. Creates a new subscription. Low -- level interface.-subscriptionCreate' :: MonadNakadi m-                    => Config-                    -> Subscription-                    -> m Subscription-subscriptionCreate' config subscription =-  httpJsonBody config status201 [(ok200, errorSubscriptionExistsAlready)]-  (setRequestMethod "POST" . setRequestPath path . setRequestBodyJSON subscription)---- | @POST@ to @\/subscriptions@. Creates a new subscription. Low--- level interface. Retrieves configuration from the environment.-subscriptionCreateR' ::-  MonadNakadiEnv r m+subscriptionCreate' ::+  MonadNakadi b m   => Subscription   -> m Subscription-subscriptionCreateR' subscription = do-  config <- asks (view L.nakadiConfig)-  subscriptionCreate config subscription+subscriptionCreate' subscription =+  httpJsonBody status201 [(ok200, errorSubscriptionExistsAlready)]+  (setRequestMethod "POST"+   . setRequestPath path+   . setRequestBodyJSON subscription)  -- | @POST@ to @\/subscriptions@. Creates a new subscription. Does not -- fail if the requested subscription does already exist.-subscriptionCreate :: MonadNakadi m-                   => Config-                   -> Subscription-                   -> m Subscription-subscriptionCreate config subscription =-  Safe.catchJust exceptionPredicate (subscriptionCreate' config subscription) return+subscriptionCreate ::+  (MonadNakadi b m, MonadCatch m)+  => Subscription+  -> m Subscription+subscriptionCreate subscription = do+  Safe.catchJust exceptionPredicate (subscriptionCreate' subscription) return    where exceptionPredicate (SubscriptionExistsAlready s) = Just s         exceptionPredicate _                             = Nothing --- | @POST@ to @\/subscriptions@. Creates a new subscription. Does not--- fail if the requested subscription does already exist. Retrieves--- configuration from the environment.-subscriptionCreateR ::-  MonadNakadiEnv r m-  => Subscription-  -> m Subscription-subscriptionCreateR subscription = do-  config <- asks (view L.nakadiConfig)-  subscriptionCreate config subscription- -- | @GET@ to @\/subscriptions@. Internal low-level interface.-subscriptionsGet :: MonadNakadi m-                 => Config-                 -> [(ByteString, ByteString)]-                 -> m SubscriptionsListResponse-subscriptionsGet config queryParameters =-  httpJsonBody config ok200 []+subscriptionsGet ::+  MonadNakadi b m+  => [(ByteString, ByteString)]+  -> m SubscriptionsListResponse+subscriptionsGet queryParameters =+  httpJsonBody ok200 []   (setRequestMethod "GET"    . setRequestPath path    . setRequestQueryParameters queryParameters) --- | @GET@ to @\/subscriptions@. Retrieves all subscriptions matching--- the provided filter criteria. Low-level interface using pagination.-subscriptionsList' :: MonadNakadi m-                   => Config-                   -> Maybe ApplicationName-                   -> Maybe [EventTypeName]-                   -> Maybe Limit-                   -> Maybe Offset-                   -> m SubscriptionsListResponse-subscriptionsList' config maybeOwningApp maybeEventTypeNames maybeLimit maybeOffset =-  subscriptionsGet config queryParameters-  where queryParameters =-          buildQueryParameters maybeOwningApp maybeEventTypeNames maybeLimit maybeOffset- buildQueryParameters :: Maybe ApplicationName                      -> Maybe [EventTypeName]                      -> Maybe Limit@@ -131,31 +93,30 @@        Nothing -> []  -- | @GET@ to @\/subscriptions@. Retrieves all subscriptions matching--- the provided filter criteria. Uses configuration contained in the--- environment.-subscriptionsListR' ::-  (MonadNakadiEnv r m, MonadMask m)+-- the provided filter criteria. Low-level interface using pagination.+subscriptionsList' ::+  (MonadNakadi b m)   => Maybe ApplicationName   -> Maybe [EventTypeName]   -> Maybe Limit   -> Maybe Offset   -> m SubscriptionsListResponse-subscriptionsListR' owningApp eventTypeNames maybeLimit maybeOffset = do-  config <- asks (view L.nakadiConfig)-  subscriptionsList' config owningApp eventTypeNames maybeLimit maybeOffset+subscriptionsList' maybeOwningApp maybeEventTypeNames maybeLimit maybeOffset = do+  subscriptionsGet queryParameters+  where queryParameters =+          buildQueryParameters maybeOwningApp maybeEventTypeNames maybeLimit maybeOffset  -- | @GET@ to @\/subscriptions@. Retrieves all subscriptions matching -- the provided filter criteria. High-level Conduit interface.-subscriptionsSource :: (MonadNakadi m, MonadMask m)-                    => Config-                    -> Maybe ApplicationName+subscriptionsSource :: ( MonadNakadi b m)+                    => Maybe ApplicationName                     -> Maybe [EventTypeName]-                    -> Source m [Subscription]-subscriptionsSource config maybeOwningApp maybeEventTypeNames =-  nextPage initialQueryParameters+                    -> m (ConduitM () [Subscription] m ())+subscriptionsSource maybeOwningApp maybeEventTypeNames =+  pure $ nextPage initialQueryParameters    where nextPage queryParameters = do-          resp <- lift $ subscriptionsGet config queryParameters+          resp <- lift $ subscriptionsGet queryParameters           yield (resp^.L.items)           let maybeNextPath = Text.unpack . (view L.href) <$> (resp^.L.links.L.next)           case maybeNextPath >>= extractQueryParametersFromPath  of@@ -168,33 +129,12 @@           buildQueryParameters maybeOwningApp maybeEventTypeNames Nothing Nothing  -- | @GET@ to @\/subscriptions@. Retrieves all subscriptions matching--- the provided filter criteria. High-level Conduit interface,--- obtaining the configuration from the environment.-subscriptionsSourceR :: (MonadNakadiEnv r m, MonadMask m)-                     => Maybe ApplicationName-                     -> Maybe [EventTypeName]-                     -> Source m [Subscription]-subscriptionsSourceR maybeOwningApp maybeEventTypeNames = do-  config <- asks (view L.nakadiConfig)-  subscriptionsSource config maybeOwningApp maybeEventTypeNames---- | @GET@ to @\/subscriptions@. Retrieves all subscriptions matching -- the provided filter criteria. High-level list interface.-subscriptionsList :: (MonadNakadi m, MonadMask m)-                  => Config-                  -> Maybe ApplicationName-                  -> Maybe [EventTypeName]-                  -> m [Subscription]-subscriptionsList config maybeOwningApp maybeEventTypeNames = runConduit $-  subscriptionsSource config maybeOwningApp maybeEventTypeNames .| concatC .| sinkList---- | @GET@ to @\/subscriptions@. Retrieves all subscriptions matching--- the provided filter criteria. High-level Conduit interface,--- obtaining the configuration from the environment.-subscriptionsListR :: (MonadNakadiEnv r m, MonadMask m)-                   => Maybe ApplicationName-                   -> Maybe [EventTypeName]-                   -> m [Subscription]-subscriptionsListR maybeOwningApp maybeEventTypeNames = do-  config <- asks (view L.nakadiConfig)-  subscriptionsList config maybeOwningApp maybeEventTypeNames+subscriptionsList ::+  MonadNakadi b m+  => Maybe ApplicationName+  -> Maybe [EventTypeName]+  -> m [Subscription]+subscriptionsList maybeOwningApp maybeEventTypeNames = do+  source <- subscriptionsSource maybeOwningApp maybeEventTypeNames+  runConduit $ source .| concatC .| sinkList

src/Network/Nakadi/Subscriptions/Cursors.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Subscriptions.Cursors Description : Implementation of Nakadi Subscription Cursors API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -20,12 +20,9 @@  module Network.Nakadi.Subscriptions.Cursors   ( subscriptionCursorCommit'-  , subscriptionCursorCommitR'-  , subscriptionCommit+  , subscriptionCursorCommit   , subscriptionCursors-  , subscriptionCursorsR   , subscriptionCursorsReset-  , subscriptionCursorsResetR   ) where  import           Network.Nakadi.Internal.Prelude@@ -34,7 +31,6 @@  import qualified Control.Exception.Safe              as Safe import           Control.Lens-import           Control.Monad.Reader import qualified Data.HashMap.Lazy                   as HashMap import           Network.Nakadi.Internal.Conversions import           Network.Nakadi.Internal.Http@@ -50,43 +46,29 @@ -- | @POST@ to @\/subscriptions\/SUBSCRIPTION-ID\/cursors@. Commits -- cursors using low level interface. subscriptionCursorCommit' ::-  MonadNakadi m-  => Config                   -- ^ Configuration-  -> SubscriptionId           -- ^ Subsciption ID+  MonadNakadi b m+  => SubscriptionId           -- ^ Subsciption ID   -> StreamId                 -- ^ Stream ID   -> SubscriptionCursorCommit -- ^ Subscription Cursor to commit   -> m ()-subscriptionCursorCommit' config subscriptionId streamId cursors =-  httpJsonNoBody config status204 [(ok200, errorCursorAlreadyCommitted)]+subscriptionCursorCommit' subscriptionId streamId cursors =+  httpJsonNoBody status204 [(ok200, errorCursorAlreadyCommitted)]   (setRequestMethod "POST"    . addRequestHeader "X-Nakadi-StreamId" (encodeUtf8 (unStreamId streamId))    . setRequestBodyJSON cursors    . setRequestPath (path subscriptionId)) --- | @POST@ to @\/subscriptions\/SUBSCRIPTION-ID\/cursors@. Commits--- cursors using low level interface. Uses the configuration contained--- in the environment.-subscriptionCursorCommitR' ::-  MonadNakadiEnv r m-  => SubscriptionId           -- ^ Subsciption ID-  -> StreamId                 -- ^ Stream ID-  -> SubscriptionCursorCommit -- ^ Subscription Cursor to commit-  -> m ()-subscriptionCursorCommitR' subscriptionId streamId cursors = do-  config <- asks (view L.nakadiConfig)-  subscriptionCursorCommit' config subscriptionId streamId cursors- -- | @POST@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. Commits -- cursors using high level interface.-subscriptionCommit ::-  (MonadNakadi m, MonadCatch m, HasNakadiSubscriptionCursor a)-  => [a] -- ^ Values containing Subscription Cursors to commit-  -> ReaderT SubscriptionEventStreamContext m ()-subscriptionCommit as = do-  SubscriptionEventStreamContext { .. } <- ask+subscriptionCursorCommit ::+  (MonadNakadi b m, MonadCatch m, HasNakadiSubscriptionCursor a)+  => SubscriptionEventStream+  -> [a] -- ^ Values containing Subscription Cursors to commit+  -> m ()+subscriptionCursorCommit SubscriptionEventStream { .. } as  = do   Safe.catchJust     exceptionPredicate-    (subscriptionCursorCommit' _ctxConfig _subscriptionId _streamId cursorsCommit)+    (subscriptionCursorCommit' _subscriptionId _streamId cursorsCommit)     (const (return ()))    where exceptionPredicate = \case@@ -99,48 +81,23 @@ -- | @GET@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. Retrieves -- subscriptions cursors. subscriptionCursors ::-  MonadNakadi m-  => Config                 -- ^ Configuration-  -> SubscriptionId         -- ^ Subscription ID-  -> m [SubscriptionCursor] -- ^ Subscription Cursors for the specified Subscription-subscriptionCursors config subscriptionId =-  httpJsonBody config ok200 []-  (setRequestMethod "GET" . setRequestPath (path subscriptionId))---- | @GET@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. Retrieves--- subscriptions cursors, using the configuration from the--- environment.-subscriptionCursorsR ::-  MonadNakadiEnv r m+  MonadNakadi b m   => SubscriptionId         -- ^ Subscription ID   -> m [SubscriptionCursor] -- ^ Subscription Cursors for the specified Subscription-subscriptionCursorsR subscriptionId = do-  config <- asks (view L.nakadiConfig)-  subscriptionCursors config subscriptionId+subscriptionCursors subscriptionId =+  httpJsonBody ok200 []+  (setRequestMethod "GET" . setRequestPath (path subscriptionId))  -- | @PATCH@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. Resets -- subscriptions cursors. subscriptionCursorsReset ::-  MonadNakadi m-  => Config                           -- ^ Configuration-  -> SubscriptionId                   -- ^ Subscription ID+  MonadNakadi b m+  => SubscriptionId                   -- ^ Subscription ID   -> [SubscriptionCursorWithoutToken] -- ^ Subscription Cursors to reset   -> m ()-subscriptionCursorsReset config subscriptionId cursors =-  httpJsonNoBody config status204 [ (status404, errorSubscriptionNotFound)-                                 , (status409, errorCursorResetInProgress) ]+subscriptionCursorsReset subscriptionId cursors =+  httpJsonNoBody status204 [ (status404, errorSubscriptionNotFound)+                           , (status409, errorCursorResetInProgress) ]   (setRequestMethod "PATCH"    . setRequestPath (path subscriptionId)    . setRequestBodyJSON (Object (HashMap.fromList [("items", toJSON cursors)])))---- | @PATCH@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. Resets--- subscriptions cursors, using the configuration from the--- environment.-subscriptionCursorsResetR ::-  MonadNakadiEnv r m-  => SubscriptionId                   -- ^ Subscription ID-  -> [SubscriptionCursorWithoutToken] -- ^ Subscription Cursors to reset-  -> m ()-subscriptionCursorsResetR subscriptionId cursors = do-  config <- asks (view L.nakadiConfig)-  subscriptionCursorsReset config subscriptionId cursors

src/Network/Nakadi/Subscriptions/Events.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Subscriptions.Events Description : Implementation of Nakadi Subscription Events API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -15,123 +15,89 @@ {-# LANGUAGE FlexibleContexts      #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RecordWildCards       #-}+{-# LANGUAGE ScopedTypeVariables   #-}  module Network.Nakadi.Subscriptions.Events-  ( subscriptionSource-  , subscriptionSourceR-  , runSubscription-  , runSubscriptionR-  , subscriptionSink+  ( subscriptionProcessConduit+  , subscriptionProcess   ) where  import           Network.Nakadi.Internal.Prelude -import           Conduit-import           Control.Lens-import           Control.Monad.Reader+import           Conduit                              hiding (throwM) import           Data.Aeson+import           Data.Void+import           Network.HTTP.Client                  (responseBody) import           Network.HTTP.Simple import           Network.HTTP.Types -import           Data.Void import           Network.Nakadi.Internal.Config import           Network.Nakadi.Internal.Conversions import           Network.Nakadi.Internal.Http-import           Network.Nakadi.Internal.Lenses       (HasNakadiSubscriptionCursor) import qualified Network.Nakadi.Internal.Lenses       as L import           Network.Nakadi.Subscriptions.Cursors --- | GET to @\/subscriptions\/SUBSCRIPTION\/events@. Creates a Conduit--- Source producing events from a Subscription's event stream.-subscriptionSource ::-  (MonadNakadi m, MonadResource m, MonadBaseControl IO m, MonadMask m, FromJSON a)-  => Config                  -- ^ Configuration-  -> Maybe ConsumeParameters -- ^ Optional Consumption Parameters-  -> SubscriptionId          -- ^ Subscription ID-  -> m ( SubscriptionEventStream-       , ConduitM ()-         (SubscriptionEventStreamBatch a)-         (ReaderT SubscriptionEventStreamContext m)-         () )                -- ^ Returns a Pair consisting of subscription-                             -- connection information ('SubscriptionEventStream')-                             -- and a Conduit source.-subscriptionSource config maybeParams subscriptionId = do-  let consumeParams = fromMaybe (config^.L.consumeParameters) maybeParams-      queryParams = buildSubscriptionConsumeQueryParameters consumeParams+subscriptionProcess+  :: ( MonadNakadi b m+     , MonadMask m+     , FromJSON a )+  => Maybe ConsumeParameters+  -> SubscriptionId+  -> (SubscriptionEventStreamBatch a -> m ())+  -> m ()+subscriptionProcess maybeConsumeParameters subscriptionId processor =+  subscriptionProcessConduit maybeConsumeParameters subscriptionId conduit+  where conduit = iterMC processor -      addFlowId     = case consumeParams^.L.flowId of-                        Just flowId -> setRequestHeader "X-Flow-Id" [encodeUtf8 flowId]-                        Nothing     -> identity-  httpJsonBodyStream config ok200 buildSubscriptionEventStream-    [(status404, errorSubscriptionNotFound)]-    (setRequestPath path . addFlowId . setRequestQueryParameters queryParams)+subscriptionProcessConduit+  :: ( MonadNakadi b m+     , MonadMask m+     , FromJSON a+     , L.HasNakadiSubscriptionCursor c )+  => Maybe ConsumeParameters+  -> SubscriptionId+  -> ConduitM (SubscriptionEventStreamBatch a) c m ()+  -> m ()+subscriptionProcessConduit maybeConsumeParameters subscriptionId processor = do+  config <- nakadiAsk+  let consumeParams = fromMaybe defaultConsumeParameters maybeConsumeParameters+      queryParams   = buildSubscriptionConsumeQueryParameters consumeParams+  httpJsonBodyStream ok200 [(status404, errorSubscriptionNotFound)]+    (includeFlowId config+     . setRequestPath path+     . setRequestQueryParameters queryParams) $+    handler config +   where buildSubscriptionEventStream response =           case listToMaybe (getResponseHeader "X-Nakadi-StreamId" response) of             Just streamId ->-              Right SubscriptionEventStream+              pure SubscriptionEventStream               { _streamId       = StreamId (decodeUtf8 streamId)               , _subscriptionId = subscriptionId }             Nothing ->-              Left "X-Nakadi-StreamId not found"+              throwM StreamIdMissing          path = "/subscriptions/"                <> subscriptionIdToByteString subscriptionId                <> "/events" --- | GET to @\/subscriptions\/SUBSCRIPTION\/events@. Creates a Conduit--- Source producing events from a Subscription's event stream. Uses--- the configuration from the environment.-subscriptionSourceR ::-  (MonadNakadiEnv r m, MonadResource m, MonadBaseControl IO m, MonadMask m, FromJSON a)-  => Maybe ConsumeParameters -- ^ Optional Consumption Parameters-  -> SubscriptionId          -- ^ Subscription ID-  -> m ( SubscriptionEventStream-       , ConduitM ()-         (SubscriptionEventStreamBatch a)-         (ReaderT SubscriptionEventStreamContext m)-         () )                -- ^ Returns a Pair consisting of subscription-                             -- connection information ('SubscriptionEventStream')-                             -- and a Conduit source.-subscriptionSourceR maybeParams subscriptionId = do-  config <- asks (view L.nakadiConfig)-  subscriptionSource config maybeParams subscriptionId+        handler config response = do+          eventStream <- buildSubscriptionEventStream response+          runConduit $+            responseBody response+            .| linesUnboundedAsciiC+            .| conduitDecode config+            .| processor+            .| subscriptionSink eventStream --- | Run a Subscription processing Conduit.-runSubscription ::-  (Monad m, MonadBaseControl IO m, MonadResource m)-  => Config                  -- ^ Configuration-  -> SubscriptionEventStream -- ^ Connection information for the Subscription-  -> ConduitM ()-              Void-              (ReaderT SubscriptionEventStreamContext m)-              r              -- ^ Subscription Conduit to run-  -> m r                     -- ^ Result of the Conduit-runSubscription config SubscriptionEventStream { .. } =-  let subscriptionStreamContext = SubscriptionEventStreamContext-                                  { _streamId       = _streamId-                                  , _subscriptionId = _subscriptionId-                                  , _ctxConfig      = config }-  in runConduit . runReaderC subscriptionStreamContext --- | Run a Subscription processing Conduit. Uses the configuration--- contained in the environment.-runSubscriptionR ::-  (Monad m, MonadBaseControl IO m, MonadResource m, MonadReader r m, L.HasNakadiConfig r Config)-  => SubscriptionEventStream -- ^ Connection information for the Subscription-  -> ConduitM ()-              Void-              (ReaderT SubscriptionEventStreamContext m)-              s              -- ^ Subscription Conduit to run-  -> m s                     -- ^ Result of the Conduit-runSubscriptionR subscriptionEventStream conduit = do-  config <- asks (view L.nakadiConfig)-  runSubscription config subscriptionEventStream conduit- -- | 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 m, HasNakadiSubscriptionCursor a )-  => ConduitM a Void (ReaderT SubscriptionEventStreamContext m) ()-subscriptionSink = awaitForever $ lift . subscriptionCommit . (: [])+  (MonadNakadi b m, L.HasNakadiSubscriptionCursor a)+  => SubscriptionEventStream+  -> ConduitM a Void m ()+subscriptionSink eventStream =+  awaitForever $ lift . subscriptionCursorCommit eventStream . (: [])

src/Network/Nakadi/Subscriptions/Stats.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Subscriptions.Stats Description : Implementation of Nakadi Subscription Stats API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -17,9 +17,7 @@  module Network.Nakadi.Subscriptions.Stats   ( subscriptionStats'-  , subscriptionStatsR'   , subscriptionStats-  , subscriptionStatsR   ) where  import           Network.Nakadi.Internal.Prelude@@ -39,45 +37,20 @@ -- | @GET@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. Low level -- interface for Subscriptions Statistics retrieval. subscriptionStats' ::-  MonadNakadi m-  => Config                             -- ^ Configuration-  -> SubscriptionId                     -- ^ Subscription ID-  -> m SubscriptionEventTypeStatsResult -- ^ Subscription Statistics-subscriptionStats' config subscriptionId =-  httpJsonBody config ok200 [(status404, errorSubscriptionNotFound)]-  (setRequestMethod "GET" . setRequestPath (path subscriptionId))---- | @GET@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. Low level--- interface for Subscriptions Statistics retrieval. Obtains--- configuration from environment.-subscriptionStatsR' ::-  MonadNakadiEnv r m+  MonadNakadi b m   => SubscriptionId                     -- ^ Subscription ID   -> m SubscriptionEventTypeStatsResult -- ^ Subscription Statistics-subscriptionStatsR' subscriptionId = do-  config <- asks (view L.nakadiConfig)-  subscriptionStats' config subscriptionId+subscriptionStats' subscriptionId =+  httpJsonBody ok200 [(status404, errorSubscriptionNotFound)]+  (setRequestMethod "GET" . setRequestPath (path subscriptionId))  -- | @GET@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. High level -- interface for Subscription Statistics retrieval. subscriptionStats ::-  MonadNakadi m-  => Config                                -- ^  Configuration-  -> SubscriptionId                        -- ^ Subscription ID-  -> m (Map EventTypeName [PartitionStat]) -- ^ Subscription-                                           -- Statistics as a 'Map'.-subscriptionStats config subscriptionId = do-  items <- subscriptionStats' config subscriptionId <&> view L.items-  return . Map.fromList . map (\SubscriptionEventTypeStats { .. } -> (_eventType, _partitions)) $ items---- | @GET@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. High level--- interface for Subscription Statistics retrieval, obtains--- configuration from environment..-subscriptionStatsR ::-  MonadNakadiEnv r m+  MonadNakadi b m   => SubscriptionId                        -- ^ Subscription ID   -> m (Map EventTypeName [PartitionStat]) -- ^ Subscription                                            -- Statistics as a 'Map'.-subscriptionStatsR subscriptionId = do-  config <- asks (view L.nakadiConfig)-  subscriptionStats config subscriptionId+subscriptionStats subscriptionId = do+  items <- subscriptionStats' subscriptionId <&> view L.items+  return . Map.fromList . map (\SubscriptionEventTypeStats { .. } -> (_eventType, _partitions)) $ items

src/Network/Nakadi/Subscriptions/Subscription.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Subscriptions.Stats Description : Implementation of Nakadi Subscription API-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -15,17 +15,13 @@  module Network.Nakadi.Subscriptions.Subscription   ( subscriptionGet-  , subscriptionGetR   , subscriptionDelete-  , subscriptionDeleteR   ) where  import           Network.Nakadi.Internal.Prelude -import           Control.Lens import           Network.Nakadi.Internal.Conversions import           Network.Nakadi.Internal.Http-import qualified Network.Nakadi.Internal.Lenses      as L  path :: SubscriptionId -> ByteString path subscriptionId =@@ -35,43 +31,19 @@ -- | @GET@ to @\/subscriptions\/SUBSCRIPTION@. Looks up subscription -- information for a subscription ID. subscriptionGet ::-  MonadNakadi m-  => Config         -- ^ Configuration-  -> SubscriptionId -- ^ Subscription ID-  -> m Subscription -- ^ Resulting Subscription Information-subscriptionGet config subscriptionId =-  httpJsonBody config ok200 [(status404, errorSubscriptionNotFound)]-  (setRequestMethod "GET" . setRequestPath (path subscriptionId))---- | @GET@ to @\/subscriptions\/SUBSCRIPTION@. Looks up subscription--- information for a subscription ID. Uses configuration from the--- environment.-subscriptionGetR ::-  MonadNakadiEnv r m+  MonadNakadi b m   => SubscriptionId -- ^ Subscription ID   -> m Subscription -- ^ Resulting Subscription Information-subscriptionGetR subscriptionId = do-  config <- asks (view L.nakadiConfig)-  subscriptionGet config subscriptionId+subscriptionGet subscriptionId =+  httpJsonBody ok200 [(status404, errorSubscriptionNotFound)]+  (setRequestMethod "GET" . setRequestPath (path subscriptionId))  -- | @DELETE@ to @\/subscriptions\/SUBSCRIPTION@. Deletes a -- subscription by subscription ID. subscriptionDelete ::-  MonadNakadi m-  => Config         -- ^ Configuration-  -> SubscriptionId -- ^ ID of the Subcription to delete-  -> m ()-subscriptionDelete config subscriptionId =-  httpJsonNoBody config status204 [(status404, errorSubscriptionNotFound)]-  (setRequestMethod "DELETE" . setRequestPath (path subscriptionId))---- | @DELETE@ to @\/subscriptions\/SUBSCRIPTION@. Deletes a--- subscription by subscription ID. Uses configuration contained in--- the environment.-subscriptionDeleteR ::-  MonadNakadiEnv r m+  MonadNakadi b m   => SubscriptionId -- ^ ID of the Subcription to delete   -> m ()-subscriptionDeleteR subscriptionId = do-  config <- asks (view L.nakadiConfig)-  subscriptionDelete config subscriptionId+subscriptionDelete subscriptionId =+  httpJsonNoBody status204 [(status404, errorSubscriptionNotFound)]+  (setRequestMethod "DELETE" . setRequestPath (path subscriptionId))

src/Network/Nakadi/Types.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Types Description : Nakadi API Types-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Clasmeier 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -16,18 +16,17 @@   , module Network.Nakadi.Types.Logger   , module Network.Nakadi.Types.Problem   , module Network.Nakadi.Types.Service-  , module Network.Nakadi.Types.Subscription-  , MonadNakadi-  , MonadNakadiEnv+  , MonadNakadiBase(..)+  , MonadNakadi(..)   , HasNakadiConfig(..)+  , NakadiT+  , runNakadiT   ) where -import           Network.Nakadi.Internal.Types     (MonadNakadi, MonadNakadiEnv)+import           Network.Nakadi.Internal.Types -import           Network.Nakadi.Internal.Lenses import           Network.Nakadi.Types.Config import           Network.Nakadi.Types.Exceptions import           Network.Nakadi.Types.Logger import           Network.Nakadi.Types.Problem import           Network.Nakadi.Types.Service-import           Network.Nakadi.Types.Subscription

src/Network/Nakadi/Types/Config.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Types.Config Description : Nakadi Config Types-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental@@ -12,9 +12,10 @@  module Network.Nakadi.Types.Config   ( Config-  , HttpBackend(..)+  , ConfigIO   , ConsumeParameters   , StreamConnectCallback+  , HttpBackend(..)   ) where  import           Network.Nakadi.Internal.Types.Config

src/Network/Nakadi/Types/Logger.hs view

@@ -12,6 +12,7 @@  module Network.Nakadi.Types.Logger   ( LogFunc+  , LogFuncIO   ) where  import           Network.Nakadi.Internal.Types.Logger

src/Network/Nakadi/Types/Service.hs view

@@ -1,7 +1,7 @@ {-| Module      : Network.Nakadi.Types.Service Description : Nakadi Service Types-Copyright   : (c) Moritz Schulte 2017+Copyright   : (c) Moritz Schulte 2017, 2018 License     : BSD3 Maintainer  : mtesseract@silverratio.net Stability   : experimental

− src/Network/Nakadi/Types/Subscription.hs

@@ -1,17 +0,0 @@-{-|-Module      : Network.Nakadi.Types.Subscription-Description : Nakadi Subscription Types-Copyright   : (c) Moritz Schulte 2017-License     : BSD3-Maintainer  : mtesseract@silverratio.net-Stability   : experimental-Portability : POSIX--This module provides the Nakadi Subscription Types.--}--module Network.Nakadi.Types.Subscription-  ( SubscriptionEventStreamContext(..)-  ) where--import           Network.Nakadi.Internal.Types.Subscription

tests/Network/Nakadi/Config/Test.hs view

@@ -1,11 +1,25 @@+{-# LANGUAGE DeriveFunctor              #-}+{-# LANGUAGE FlexibleContexts           #-}+{-# LANGUAGE FlexibleInstances          #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses      #-}+{-# LANGUAGE RankNTypes                 #-}+{-# LANGUAGE ScopedTypeVariables        #-}+{-# LANGUAGE UndecidableInstances       #-}+ module Network.Nakadi.Config.Test where -import           ClassyPrelude-import           Control.Lens         ((<&>))-import           Control.Retry-import qualified Data.ByteString.Lazy as LB+import           ClassyPrelude               hiding (catch, throwM)++import           Control.Lens+import           Control.Monad.Catch         (MonadThrow (..))+import qualified Data.ByteString.Lazy        as LB+import           Data.Conduit                (ConduitM, transPipe) import           Network.HTTP.Client+import           Network.HTTP.Client.Conduit (bodyReaderSource)+import           Network.HTTP.Client.TLS     (getGlobalManager) import           Network.Nakadi+import           Network.Nakadi.Tests.Common import           System.IO.Unsafe import           Test.Tasty import           Test.Tasty.HUnit@@ -18,31 +32,31 @@ requestsExecuted :: TVar [Request] requestsExecuted = unsafePerformIO . newTVarIO $ [] -dummyHttpLbs :: Request -> IO (Response LB.ByteString)-dummyHttpLbs req = do-  atomically $ modifyTVar requestsExecuted (req :)-  throwIO (HttpExceptionRequest req ResponseTimeout)--dummyResponseOpen :: Request -> Manager -> IO (Response (IO ByteString))-dummyResponseOpen req _manager = do+mockHttpBackendLbs :: Config b -> Request -> Maybe Manager -> App (Response LB.ByteString)+mockHttpBackendLbs _conf req _mngr = do   atomically $ modifyTVar requestsExecuted (req :)-  throwIO (HttpExceptionRequest req ResponseTimeout)+  throwM (HttpExceptionRequest req ResponseTimeout) -dummyResponseClose :: Response BodyReader -> IO ()-dummyResponseClose _response = return ()+mockHttpBackendResponseOpen :: Config b -> Request -> Maybe Manager -> App (Response (ConduitM i ByteString App ()))+mockHttpBackendResponseOpen _config req _maybeMngr = do+  mngr <- liftIO getGlobalManager+  liftIO $ responseOpen req mngr <&> fmap (transPipe liftIO . bodyReaderSource) -dummyHttpBackend :: HttpBackend-dummyHttpBackend = HttpBackend dummyHttpLbs dummyResponseOpen dummyResponseClose+mockHttpBackendResponseClose :: Response a -> App ()+mockHttpBackendResponseClose = liftIO . responseClose  testCustomHttpBackend :: Assertion-testCustomHttpBackend = do-  let trivialRetryPolicy = limitRetries 0-  conf <- newConfig Nothing defaultRequest-          <&> setHttpBackend dummyHttpBackend-          <&> setRetryPolicy trivialRetryPolicy-  res0 <- try $ registryPartitionStrategies conf -- This should use httpLbs-  case res0 of+testCustomHttpBackend = runApp $ do+  res0 <- try $ runNakadiT mockConfig $+    registryPartitionStrategies -- This uses httpLbs.+  liftIO $ case res0 of     Left (HttpExceptionRequest _ ResponseTimeout) -> return ()     _ -> assertFailure "Expected ResponseTimeout exception from dummy HttpBackend"   requests <- atomically . readTVar $ requestsExecuted-  1 @=? length requests+  liftIO $ 1 @=? length requests++  where mockConfig = newConfig mockHttpBackend defaultRequest+        mockHttpBackend = HttpBackend+                          { _httpLbs           = mockHttpBackendLbs+                          , _httpResponseOpen  = mockHttpBackendResponseOpen+                          , _httpResponseClose = mockHttpBackendResponseClose }

tests/Network/Nakadi/Connection/Test.hs view

@@ -1,3 +1,5 @@+{-# LANGUAGE ScopedTypeVariables #-}+ module Network.Nakadi.Connection.Test where  import           ClassyPrelude@@ -66,28 +68,30 @@  testSimpleRetry :: Assertion testSimpleRetry = do-  conf <- newConfig Nothing testServerRequest { port = testServerRetryPort }-  withAsync (run testServerRetryPort (testServerRetryApp 2)) $ \_serverHandle -> do-    events <- eventTypesList conf+  let conf = newConfigIO testServerRequest { port = testServerRetryPort } :: ConfigIO+  withAsync (run testServerRetryPort (testServerRetryApp 1)) $ \_serverHandle -> do+    events <- runNakadiT conf eventTypesList     [] @=? events  testResponseTimeoutSuccess :: Assertion testResponseTimeoutSuccess = do   let timeout = responseTimeoutMicro (5 * 10^6) -- Accept delay of 5s-  conf <- newConfig Nothing testServerRequest { port = testServerResponseTimeoutPort-                                              , responseTimeout = timeout }+      request = testServerRequest { port = testServerResponseTimeoutPort+                                  , responseTimeout = timeout }+      conf    = newConfigIO request :: ConfigIO   withAsync (run testServerResponseTimeoutPort testServerResponseTimeoutApp) $ \_serverHandle -> do-    events <- eventTypesList conf+    events <- runNakadiT conf eventTypesList     [] @=? events  testResponseTimeoutFail :: Assertion testResponseTimeoutFail = do   res <- try $ do     let timeout = responseTimeoutMicro (3 * 10^6) -- Accept delay of 3s-    conf <- newConfig Nothing testServerRequest { port = testServerResponseTimeoutPort-                                                , responseTimeout = timeout }-    withAsync (run testServerResponseTimeoutPort testServerResponseTimeoutApp) $ \_serverHandle -> do-      eventTypesList conf+        request = testServerRequest { port = testServerResponseTimeoutPort+                                    , responseTimeout = timeout }+        conf    = newConfigIO request :: ConfigIO+    withAsync (run testServerResponseTimeoutPort testServerResponseTimeoutApp) $ \_serverHandle ->+      runNakadiT conf eventTypesList   case res of     Left (HttpExceptionRequest _request ResponseTimeout) -> return ()     _ -> assertFailure "Expected HttpExceptionRequest with content ResponseTimeout"

tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs view

@@ -16,34 +16,35 @@ import           Test.Tasty import           Test.Tasty.HUnit -testEventTypesCursorsLag :: Config -> TestTree+testEventTypesCursorsLag :: Config App -> TestTree testEventTypesCursorsLag conf = testGroup "CursorsLag"   [ testCase "CursorsLagZero" (testCursorsLagZero conf)   , testCase "CursorsLag10" (testCursorsLagN conf 10)   ] -testCursorsLagZero :: Config -> Assertion-testCursorsLagZero conf = do-  recreateEvent conf myEventTypeName myEventType-  partitions <- eventTypePartitions conf myEventTypeName+testCursorsLagZero :: Config App -> Assertion+testCursorsLagZero conf = runApp . runNakadiT conf $ do+  partitions <- eventTypePartitions myEventTypeName   let cursorsMap = Map.fromList $         map (\Partition { .. } -> (_partition, _newestAvailableOffset)) partitions-  lagMap <- cursorsLag conf myEventTypeName cursorsMap-  Map.size cursorsMap @=? Map.size lagMap-  forM_ (Map.toList lagMap) $ \(_, lag) ->-    lag @=? 0+  lagMap <- cursorsLag myEventTypeName cursorsMap+  recreateEvent myEventTypeName myEventType+  liftIO $ do+    Map.size cursorsMap @=? Map.size lagMap+    forM_ (Map.toList lagMap) $ \(_, lag) ->+      liftIO $ lag @=? 0 -testCursorsLagN :: Config -> Int64 -> Assertion-testCursorsLagN conf n = do-  now <- getCurrentTime-  eid <- EventId <$> genRandomUUID-  recreateEvent conf myEventTypeName myEventType-  partitions <- eventTypePartitions conf myEventTypeName+testCursorsLagN :: Config App -> Int64 -> Assertion+testCursorsLagN conf n = runApp . runNakadiT conf $ do+  now <- liftIO getCurrentTime+  eid <- EventId <$> liftIO genRandomUUID+  recreateEvent myEventTypeName myEventType+  partitions <- eventTypePartitions myEventTypeName   let cursorsMap = Map.fromList $         map (\Partition { .. } -> (_partition, _newestAvailableOffset)) partitions   forM_ [1..n] $ \_ ->-    eventPublish conf myEventTypeName Nothing [myDataChangeEvent eid now]-  lagMap <- cursorsLag conf myEventTypeName cursorsMap-  Map.size cursorsMap @=? Map.size lagMap+    eventsPublish myEventTypeName [myDataChangeEvent eid now]+  lagMap <- cursorsLag myEventTypeName cursorsMap+  liftIO $ Map.size cursorsMap @=? Map.size lagMap   let lag = sum $ map snd (Map.toList lagMap)-  n @=? lag+  liftIO $ n @=? lag

tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs view

@@ -10,37 +10,39 @@  import           ClassyPrelude +import           Data.UUID                   () import           Network.Nakadi import           Network.Nakadi.Tests.Common+import           System.Random import           Test.Tasty import           Test.Tasty.HUnit -testEventTypesShiftedCursors :: Config -> TestTree+testEventTypesShiftedCursors :: Config App -> TestTree testEventTypesShiftedCursors conf = testGroup "ShiftedCursors"   [ testCase "ShiftedCursorsZero" (testShiftedCursorsZero conf)   , testCase "ShiftedCursorsN" (testShiftedCursorsN conf 10)   ] -testShiftedCursorsZero :: Config -> Assertion-testShiftedCursorsZero conf = do-  recreateEvent conf myEventTypeName myEventType-  partitions <- eventTypePartitions conf myEventTypeName+testShiftedCursorsZero :: Config App -> Assertion+testShiftedCursorsZero conf = runApp . runNakadiT conf $ do+  recreateEvent myEventTypeName myEventType+  partitions <- eventTypePartitions myEventTypeName   let cursors = map extractCursor partitions-  cursors' <- cursorsShift conf myEventTypeName cursors 0-  cursors @=? cursors'+  cursors' <- cursorsShift myEventTypeName cursors 0+  liftIO $ cursors @=? cursors' -testShiftedCursorsN :: Config -> Int64 -> Assertion-testShiftedCursorsN conf n = do-  now <- getCurrentTime-  eid <- EventId <$> genRandomUUID-  recreateEvent conf myEventTypeName myEventType-  partitions <- eventTypePartitions conf myEventTypeName+testShiftedCursorsN :: Config App -> Int64 -> Assertion+testShiftedCursorsN conf n = runApp . runNakadiT conf $ do+  now <- liftIO getCurrentTime+  eid <- EventId <$> liftIO randomIO+  recreateEvent myEventTypeName myEventType+  partitions <- eventTypePartitions myEventTypeName   let cursors = map extractCursor partitions-  length cursors > 0 @=? True+  liftIO $ length cursors > 0 @=? True   forM_ [1..n] $ \_ -> do-    eventPublish conf myEventTypeName Nothing [myDataChangeEvent eid now]-  cursors' <- cursorsShift conf myEventTypeName cursors n-  length cursors' @=? length cursors+    eventsPublish myEventTypeName [myDataChangeEvent eid now]+  cursors' <- cursorsShift myEventTypeName cursors n+  liftIO $ length cursors' @=? length cursors   forM_ (zip cursors cursors') $ \(c, c') -> do-    distance <- cursorDistance conf myEventTypeName c c'-    distance @=? n+    distance <- cursorDistance myEventTypeName c c'+    liftIO $ distance @=? n

tests/Network/Nakadi/EventTypes/Test.hs view

@@ -8,10 +8,11 @@  module Network.Nakadi.EventTypes.Test where -import           ClassyPrelude+import           ClassyPrelude                                 hiding+                                                                (withAsync) -import           Conduit-import           Control.Concurrent.Async                      (link)+import           Conduit                                       hiding+                                                                (runResourceT) import           Control.Lens import           Data.Function                                 ((&)) import           Network.Nakadi@@ -22,8 +23,9 @@ import           System.IO.Unsafe import           Test.Tasty import           Test.Tasty.HUnit+import           UnliftIO.Async -testEventTypes :: Config -> TestTree+testEventTypes :: Config App -> TestTree testEventTypes conf = testGroup "EventTypes"   [ testCase "EventTypesPrepare" (testEventTypesPrepare conf)   , testCase "EventTypesGet" (testEventTypesGet conf)@@ -33,83 +35,84 @@   , testCase "EventTypeCursorDistances10" (testEventTypeCursorDistances10 conf)   , testCase "EventTypePublishData" (testEventTypePublishData conf)   , testCase "EventTypeParseFlowId" (testEventTypeParseFlowId conf)+  , testCase "EventTypeDeserializationFailureException" (testEventTypeDeserializationFailureException conf)   , testCase "EventTypeDeserializationFailure" (testEventTypeDeserializationFailure conf)   , testEventTypesShiftedCursors conf   , testEventTypesCursorsLag conf   ] -testEventTypesPrepare :: Config -> Assertion-testEventTypesPrepare conf = do-  subscriptions <- subscriptionsList conf Nothing Nothing+testEventTypesPrepare :: Config App -> Assertion+testEventTypesPrepare conf = runApp . runNakadiT conf $ do+  subscriptions <- subscriptionsList Nothing Nothing   let subscriptionIds = catMaybes . map (view L.id) $ subscriptions-  forM_ subscriptionIds (subscriptionDelete conf)+  forM_ subscriptionIds subscriptionDelete -testEventTypesGet :: Config -> Assertion-testEventTypesGet conf =-  void $ eventTypesList conf+testEventTypesGet :: Config App -> Assertion+testEventTypesGet conf = runApp . runNakadiT conf $+  void eventTypesList -testEventTypesDeleteCreateGet :: Config -> Assertion-testEventTypesDeleteCreateGet conf = do-  eventTypeDelete conf myEventTypeName `catch` (ignoreExnNotFound ())-  eventTypeCreate conf myEventType-  myEventTypes <- filterMyEvent <$> eventTypesList conf-  length myEventTypes @=? 1-  eventTypeDelete conf myEventTypeName-  myEventTypes' <- filterMyEvent <$> eventTypesList conf-  length myEventTypes' @=? 0+testEventTypesDeleteCreateGet :: Config App -> Assertion+testEventTypesDeleteCreateGet conf = runApp . runNakadiT conf $ do+  eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+  eventTypeCreate myEventType+  myEventTypes <- filterMyEvent <$> eventTypesList+  liftIO $ length myEventTypes @=? 1+  eventTypeDelete myEventTypeName+  myEventTypes' <- filterMyEvent <$> eventTypesList+  liftIO $ length myEventTypes' @=? 0    where filterMyEvent = filter ((myEventTypeName ==) . (view L.name)) -testEventTypePartitionsGet :: Config -> Assertion-testEventTypePartitionsGet conf = do-  eventTypeDelete conf myEventTypeName `catch` (ignoreExnNotFound ())-  eventTypeCreate conf myEventType-  void $ eventTypePartitions conf myEventTypeName+testEventTypePartitionsGet :: Config App -> Assertion+testEventTypePartitionsGet conf = runApp . runNakadiT conf $ do+  eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+  eventTypeCreate myEventType+  void $ eventTypePartitions myEventTypeName -testEventTypeCursorDistances0 :: Config -> Assertion-testEventTypeCursorDistances0 conf = do-  eventTypeDelete conf myEventTypeName `catch` (ignoreExnNotFound ())-  eventTypeCreate conf myEventType-  partitions <- eventTypePartitions conf myEventTypeName+testEventTypeCursorDistances0 :: Config App -> Assertion+testEventTypeCursorDistances0 conf = runApp . runNakadiT conf $ do+  eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+  eventTypeCreate myEventType+  partitions <- eventTypePartitions myEventTypeName   let cursors = map extractCursor partitions   forM_ cursors $ \cursor -> do-    distance <- cursorDistance conf myEventTypeName cursor cursor-    distance @=? 0+    distance <- cursorDistance myEventTypeName cursor cursor+    liftIO $ distance @=? 0 -testEventTypeCursorDistances10 :: Config -> Assertion-testEventTypeCursorDistances10 conf = do-  eventTypeDelete conf myEventTypeName `catch` (ignoreExnNotFound ())-  eventTypeCreate conf myEventType-  partitions <- eventTypePartitions conf myEventTypeName+testEventTypeCursorDistances10 :: Config App -> Assertion+testEventTypeCursorDistances10 conf = runApp . runNakadiT conf $ do+  eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+  eventTypeCreate myEventType+  partitions <- eventTypePartitions myEventTypeName   let cursors = map extractCursor partitions    forM_ [1..10] $ \_ -> do-    now <- getCurrentTime+    now <- liftIO getCurrentTime     eid <- EventId <$> genRandomUUID-    eventPublish conf myEventTypeName Nothing [myDataChangeEvent eid now]+    eventsPublish myEventTypeName [myDataChangeEvent eid now]    cursorPairs <- forM cursors $ \cursor@Cursor { .. } -> do-    part <- eventTypePartition conf myEventTypeName _partition+    part <- eventTypePartition myEventTypeName _partition     let cursor' = extractCursor part     return (cursor, cursor')    distances <- forM cursorPairs $ \(c, c') -> do-    cursorDistance conf myEventTypeName c c'+    cursorDistance myEventTypeName c c'    let totalDistances = sum distances-  totalDistances @=? 10+  liftIO $ totalDistances @=? 10  consumeParametersSingle :: ConsumeParameters consumeParametersSingle = defaultConsumeParameters                           & setBatchLimit 1                           & setBatchFlushTimeout 1 -testEventTypePublishData :: Config -> Assertion-testEventTypePublishData conf = do-  now <- getCurrentTime+testEventTypePublishData :: Config App -> Assertion+testEventTypePublishData conf = runApp . runNakadiT conf $ do+  now <- liftIO getCurrentTime   eid <- EventId <$> genRandomUUID-  eventTypeDelete conf myEventTypeName `catch` (ignoreExnNotFound ())-  eventTypeCreate conf myEventType+  eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+  eventTypeCreate myEventType   let event = DataChangeEvent { _payload = Foo "Hello!"                               , _metadata = EventMetadata { _eid = eid                                                           , _occurredAt = Timestamp now@@ -119,19 +122,18 @@                               , _dataType = "test.FOO"                               , _dataOp = DataOpUpdate                               }-  withAsync (delayedPublish conf Nothing [event]) $ \asyncHandle -> do-    link asyncHandle-    eventConsumed :: Maybe (EventStreamBatch Foo) <- runResourceT $ do-      source <- eventSource conf (Just consumeParametersSingle) myEventTypeName Nothing-      runConduit $ source .| headC-    isJust eventConsumed @=? True+  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 -testEventTypeParseFlowId :: Config -> Assertion-testEventTypeParseFlowId conf = do-  now <- getCurrentTime+testEventTypeParseFlowId :: Config App -> Assertion+testEventTypeParseFlowId conf = runApp . runNakadiT conf $ do+  now <- liftIO getCurrentTime   eid <- EventId <$> genRandomUUID-  eventTypeDelete conf myEventTypeName `catch` (ignoreExnNotFound ())-  eventTypeCreate conf myEventType+  eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+  eventTypeCreate myEventType   let event = DataChangeEvent { _payload = Foo "Hello!"                               , _metadata = EventMetadata { _eid = eid                                                           , _occurredAt = Timestamp now@@ -142,28 +144,50 @@                               , _dataOp = DataOpUpdate                               }       expectedFlowId = Just $ FlowId "12345"-  withAsync (delayedPublish conf expectedFlowId [event]) $ \asyncHandle -> do-    link asyncHandle-    eventConsumed :: Maybe (EventStreamBatch (DataChangeEventEnriched Foo)) <- runResourceT $ do-      source <- eventSource conf (Just consumeParametersSingle) myEventTypeName Nothing-      runConduit $ source .| headC-    isJust eventConsumed @=? True-    let events = eventConsumed >>= (\batch -> batch^.L.events)-    isJust events @=? True+  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" -    case events of-      Nothing -> assertFailure "Received no events"-      Just v -> case toList v of-        [DataChangeEventEnriched _ x _ _] ->-          x^.L.flowId @=? expectedFlowId-        _ -> assertFailure "Received not a singleton event list"+testEventTypeDeserializationFailureException :: Config App -> Assertion+testEventTypeDeserializationFailureException conf = runApp . runNakadiT conf $ do+  now <- liftIO getCurrentTime+  eid <- EventId <$> genRandomUUID+  eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+  eventTypeCreate myEventType+  let event = DataChangeEvent { _payload = Foo "Hello!"+                              , _metadata = EventMetadata { _eid = eid+                                                          , _occurredAt = Timestamp now+                                                          , _parentEids = Nothing+                                                          , _partition  = Nothing+                                                          }+                              , _dataType = "test.FOO"+                              , _dataOp = DataOpUpdate+                              }+  res :: Either NakadiException (Maybe (EventStreamBatch WrongFoo)) <- try $+    withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do+    liftIO $ link asyncHandle+    eventsProcessConduit (Just consumeParametersSingle) myEventTypeName Nothing headC+  case res of+    Left (DeserializationFailure _ _) ->+      pure ()+    Left exn ->+      liftIO $ assertFailure $ "Unexpected exception: " <> show exn+    Right events ->+      liftIO $ assertFailure $ "Unexpected success: " <> show events -testEventTypeDeserializationFailure :: Config -> Assertion-testEventTypeDeserializationFailure conf' = do-  now <- getCurrentTime+testEventTypeDeserializationFailure :: Config App -> Assertion+testEventTypeDeserializationFailure conf' = runApp . runNakadiT conf $ do+  now <- liftIO getCurrentTime   eid <- EventId <$> genRandomUUID-  eventTypeDelete conf myEventTypeName `catch` (ignoreExnNotFound ())-  eventTypeCreate conf myEventType+  eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+  eventTypeCreate myEventType   let event = DataChangeEvent { _payload = Foo "Hello!"                               , _metadata = EventMetadata { _eid = eid                                                           , _occurredAt = Timestamp now@@ -173,20 +197,19 @@                               , _dataType = "test.FOO"                               , _dataOp = DataOpUpdate                               }-  withAsync (delayedPublish conf Nothing [event]) $ \asyncHandle -> do-    link asyncHandle-    eventConsumed :: Maybe (EventStreamBatch WrongFoo) <- runResourceT $ do-      source <- eventSource conf (Just consumeParametersSingle) myEventTypeName Nothing-      runConduit $ source .| headC-    isJust eventConsumed @=? True+  withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do+    liftIO $ link asyncHandle+    eventConsumed :: Maybe (EventStreamBatch WrongFoo) <-+      eventsProcessConduit (Just consumeParametersSingle) myEventTypeName Nothing headC+    liftIO $ isJust eventConsumed @=? True    counter <- atomically $ readTVar deserializationFailureCounter-  1 @=? counter+  liftIO $ 1 @=? counter    where conf = conf'-               & setDeserializationFailureCallback (deserializationFailureCb deserializationFailureCounter)+               & setDeserializationFailureCallback deserializationFailureCb -        deserializationFailureCb counter _ _ =-          atomically $ modifyTVar counter (+ 1)+        deserializationFailureCb _ _ =+          atomically $ modifyTVar deserializationFailureCounter (+ 1)          deserializationFailureCounter = unsafePerformIO $ newTVarIO 0

+ tests/Network/Nakadi/Examples/Echo/Echo.hs view

@@ -0,0 +1,42 @@+{-# 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

+ tests/Network/Nakadi/Examples/Echo/Test.hs view

@@ -0,0 +1,84 @@+{-# 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

+ tests/Network/Nakadi/Examples/ListEventTypes/Test.hs view

@@ -0,0 +1,23 @@+{-# LANGUAGE DuplicateRecordFields #-}+{-# LANGUAGE FlexibleContexts      #-}+{-# LANGUAGE ScopedTypeVariables   #-}++module Network.Nakadi.Examples.ListEventTypes.Test+  ( testListEventTypes+  ) where++import           ClassyPrelude+import           Control.Lens+import           Control.Monad.Logger+import qualified Network.Nakadi        as Nakadi+import qualified Network.Nakadi.Lenses as L++testListEventTypes :: Nakadi.ConfigIO -> Bool -> IO ()+testListEventTypes config withLogging =+  runStdoutLoggingT . filterLogger (\ _ _ -> withLogging) $ dumpEventTypes config++dumpEventTypes :: Nakadi.ConfigIO -> LoggingT IO ()+dumpEventTypes config = Nakadi.runNakadiT config $ do+  eventTypes <- Nakadi.eventTypesList+  forM_ eventTypes $ \ eventType ->+    logInfoN $ tshow (eventType^.L.name)

+ tests/Network/Nakadi/Examples/Subscription/Process.hs view

@@ -0,0 +1,17 @@+{-# LANGUAGE FlexibleContexts    #-}++module Network.Nakadi.Examples.Subscription.Process (dumpSubscription) where++import           ClassyPrelude+import           Data.Aeson+import qualified Network.Nakadi as Nakadi+import Network.Nakadi (MonadNakadi)+import Control.Monad.Logger++dumpSubscription :: (MonadLogger m, MonadNakadi IO m, MonadMask m) => Nakadi.SubscriptionId -> m ()+dumpSubscription subscriptionId =+  Nakadi.subscriptionProcess Nothing subscriptionId processBatch++  where processBatch :: MonadLogger m => Nakadi.SubscriptionEventStreamBatch Value -> m ()+        processBatch batch =+          logInfoN (tshow batch)

+ tests/Network/Nakadi/Examples/Subscription/Test.hs view

@@ -0,0 +1,74 @@+{-# LANGUAGE DuplicateRecordFields #-}+{-# LANGUAGE FlexibleContexts      #-}+{-# LANGUAGE ScopedTypeVariables   #-}++module Network.Nakadi.Examples.Subscription.Test+  ( testConsumption+  ) where++import           ClassyPrelude+import           Control.Lens+import           Control.Monad.Logger+import           Data.Maybe                                   (fromJust)+import qualified Network.Nakadi                               as Nakadi+import           Network.Nakadi.Examples.Subscription.Process+import qualified Network.Nakadi.Lenses                        as L+import           Network.Nakadi.Tests.Common+import           Test.Tasty.HUnit++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 [Nakadi.DataChangeEvent Foo]+genEvents =+  sequence (replicate 10 genEvent)++testConsumption :: Nakadi.Config IO -> Assertion+testConsumption config = Nakadi.runNakadiT config $ do+  nakadiLogRef <- liftIO $ newIORef []+  bracket before after $ \ subscriptionId -> do+    _ <- flip runLoggingT (logger nakadiLogRef) $ do+      events <- genEvents+      void . async $ delayedPublish Nothing events+      withAsync (dumpSubscription subscriptionId) $ \ _dumpHandle ->+        threadDelay (2 * 10^6) -- Give Nakadi some time to transmit the published events+    nakadiLog <- liftIO $ readIORef nakadiLogRef+    when (length nakadiLog == 0) $+      liftIO $ assertFailure "Subscription Consumption has logged no received batches"++  where before = do+          recreateEvent myEventTypeName myEventType+          subscription <- Nakadi.subscriptionCreate Nakadi.Subscription+            { _id = Nothing+            , _owningApplication = "test-suite"+            , _eventTypes = [myEventTypeName]+            , _consumerGroup = Nothing -- ??+            , _createdAt = Nothing+            , _readFrom = Just Nakadi.SubscriptionPositionEnd+            , _initialCursors = Nothing+            }+          pure . fromJust $ subscription^.L.id++        after subscriptionId = do+          Nakadi.subscriptionDelete subscriptionId+          Nakadi.eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())++        -- nEvents = 100++        logger nakadiLog loc logSource logLevel logStr = do+          let str = defaultLogStr loc logSource logLevel logStr+          liftIO $ modifyIORef nakadiLog (str :)

+ tests/Network/Nakadi/Examples/Test.hs view

@@ -0,0 +1,15 @@+{-# LANGUAGE FlexibleContexts #-}++module Network.Nakadi.Examples.Test where++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++testExamples :: Nakadi.Config IO -> TestTree+testExamples conf = testGroup "Examples"+  [ testCase "Subscription Consumption" (testConsumption conf)+  , testCase "Echo" (testEcho conf) ]

tests/Network/Nakadi/Internal/Http/Test.hs view

@@ -1,58 +1,91 @@-{-# LANGUAGE OverloadedStrings     #-}-{-# LANGUAGE ScopedTypeVariables   #-}-{-# LANGUAGE RecordWildCards       #-}+{-# LANGUAGE GADTs               #-}+{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE RecordWildCards     #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies        #-}  module Network.Nakadi.Internal.Http.Test   ( testHttp   ) where +import           ClassyPrelude+import           Control.Arrow+import           Control.Lens+import qualified Data.ByteString.Lazy         as LB import           Network.HTTP.Client import           Network.HTTP.Types+import           Network.Nakadi+import           Network.Nakadi.Internal.Http+import qualified Network.Nakadi.Lenses        as L import           Test.Tasty import           Test.Tasty.HUnit-import           Control.Monad-import           Control.Monad.Reader-import           ClassyPrelude-import           Network.Nakadi.Internal.Http-import           Network.HTTP.Client.Internal (CookieJar (..), Request(..), Response (..), ResponseClose (..))-import           Network.Nakadi-import           Conduit  testHttp :: TestTree testHttp = testGroup "Http"   [ testCase "HttpRequestModifier" testHttpRequestModifier+  , testCase "FlowIdInclusion" testFlowIdInclusion+  , testCase "FlowIdMissing" testFlowIdMissing+  , testCase "FlowIdInclusionHttp" testFlowIdInclusionHttp+  , testCase "FlowIdMissingHttp" testFlowIdMissingHttp   ] -resp :: Response (IO ByteString)-resp = Response-    { responseStatus = status200-    , responseVersion = http11-    , responseHeaders = []-    , responseBody = pure ""-    , responseCookieJar = CJ []-    , responseClose' = ResponseClose (pure ())-    }+extractFlowId :: Request -> Maybe FlowId+extractFlowId =+  requestHeaders+  >>> filter (\ (key, _) -> key == "X-Flow-Id")+  >>> listToMaybe+  >>> fmap (FlowId . decodeUtf8 . snd)  headers :: RequestHeaders headers = [("test-header", "header-value")] -dummyResponseOpen :: Request -> Manager -> IO (Response (IO ByteString))-dummyResponseOpen Request { .. } _ = do-  requestHeaders @=? headers-  pure resp--dummyHttpBackend :: HttpBackend-dummyHttpBackend = defaultHttpBackend {-    _responseOpen = dummyResponseOpen-}- dummyRequestModifier :: Request -> IO Request dummyRequestModifier request = pure (request { requestHeaders = headers })  testHttpRequestModifier :: Assertion testHttpRequestModifier = do-    conf <- newConfig Nothing defaultRequest-    let config = conf {_http = dummyHttpBackend, _requestModifier = dummyRequestModifier }-    (_, source) <- runResourceT $ httpJsonBodyStream config ok200 (const (Right ())) [] id-    (_ :: Maybe Text) <- runResourceT $ runReaderT (runConduit $ source .| headC) ()-    return ()+  let conf = newConfigIO defaultRequest+             & setRequestModifier dummyRequestModifier+  request <- runNakadiT conf $ httpBuildRequest id+  requestHeaders request @=? headers++testFlowIdInclusion :: Assertion+testFlowIdInclusion = do+  let flowId = FlowId "shalom"+      config = newConfigIO defaultRequest & setFlowId flowId+  Just flowId @=? (defaultRequest & (includeFlowId config >>> extractFlowId))++testFlowIdMissing :: Assertion+testFlowIdMissing = do+  let config = newConfigIO defaultRequest+  Nothing @=? (defaultRequest & (includeFlowId config >>> extractFlowId))++mockHttpLbs+  :: b ~ IO+  => TVar (Maybe Request)+  -> Config b+  -> Request+  -> Maybe Manager+  -> b (Response LB.ByteString)+mockHttpLbs tv _config request  _manager = do+  liftIO . atomically $ writeTVar tv (Just request)+  throwString "Mock"++testFlowIdInclusionHttp :: Assertion+testFlowIdInclusionHttp = do+  tv <- atomically $ newTVar Nothing+  let flowId      = FlowId "shalom"+      httpBackend = httpBackendIO & L.httpLbs .~ (mockHttpLbs tv)+      config      = newConfig httpBackend defaultRequest & setFlowId flowId+  Left (StringException _ _) <- try $ runNakadiT config eventTypesList+  Just requestExecuted <- liftIO . atomically $ readTVar tv+  Just flowId @=? extractFlowId requestExecuted++testFlowIdMissingHttp :: Assertion+testFlowIdMissingHttp = do+  tv <- atomically $ newTVar Nothing+  let httpBackend = httpBackendIO & L.httpLbs .~ (mockHttpLbs tv)+      config      = newConfig httpBackend defaultRequest+  Left (StringException _ _) <- try $ runNakadiT config eventTypesList+  Just requestExecuted <- liftIO . atomically $ readTVar tv+  Nothing @=? extractFlowId requestExecuted

tests/Network/Nakadi/Internal/Retry/Test.hs view

@@ -6,9 +6,9 @@   ) where  import           ClassyPrelude-import           Control.Lens import           Control.Retry import qualified Data.ByteString.Lazy          as LB+import           Data.Function                 ((&)) import           Network.HTTP.Client import           Network.HTTP.Client.Internal  (CookieJar (..), Request (..),                                                 Response (..),@@ -66,30 +66,21 @@      then True @=? finalFailure      else False @=? finalFailure -mockHttpBackend :: Int -> IO HttpBackend-mockHttpBackend numFailures = do-  responder <- prepareMockResponse numFailures-  return HttpBackend-    { _httpLbs       = mockHttpLbs responder-    , _responseOpen  = responseOpen-    , _responseClose = responseClose-    }--    where mockHttpLbs responder _request = responder- -- | Tests that the callback is called exactly numFailures times -- before the request succeeds — depending on the retry policy. testHttpErrorCallbackN :: Int -> Assertion testHttpErrorCallbackN numFailures = do-  httpBackend <- mockHttpBackend numFailures   counter <- newTVarIO 0-  conf <- newConfig Nothing defaultRequest-          <&> setHttpErrorCallback (httpErrorCallback counter)-          <&> setRetryPolicy (fullJitterBackoff 2 ++ limitRetries maxRetries)+  responder <- prepareMockResponse numFailures+  let conf = newConfigIO defaultRequest+             & setHttpErrorCallback (httpErrorCallback counter)+             & setRetryPolicy (fullJitterBackoff 2 ++ limitRetries maxRetries)   _response :: Either HttpException (Response LB.ByteString) <- try $-    retryAction conf defaultRequest (_httpLbs httpBackend)+    retryAction conf defaultRequest (mockHttpLbs responder)   current <- readTVarIO counter   numFailures @=? current++  where mockHttpLbs responder _request = responder  testHttpErrorCallback0 :: Assertion testHttpErrorCallback0 = testHttpErrorCallbackN 0

+ tests/Network/Nakadi/MonadicAPI/Test.hs view

@@ -0,0 +1,79 @@+{-# LANGUAGE FlexibleContexts      #-}+{-# LANGUAGE FlexibleInstances     #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeSynonymInstances  #-}++module Network.Nakadi.MonadicAPI.Test where++import           ClassyPrelude+import           Control.Lens+import           Control.Monad.Logger+import           Control.Monad.State.Strict+import           Control.Monad.Trans.Resource+import           Control.Retry+import           Network.HTTP.Client+import qualified Network.Nakadi               as Nakadi+import           Network.Nakadi.Base+import qualified Network.Nakadi.Lenses        as L+import           Network.Nakadi.Tests.Common+import           Test.Tasty+import           Test.Tasty.HUnit++httpErrorCallback :: Request -> HttpException -> RetryStatus -> Bool -> App ()+httpErrorCallback _req exn _retryStatus _isFinalFailure =+  putStrLn $ "nakadi-client triggered exception: " <> tshow exn++testMonadicAPI :: Nakadi.Config App -> TestTree+testMonadicAPI conf = testGroup "Monadic API"+  [ testCase "Simple Monadic API" (testSimpleMonadicAPI conf')+  , testCase "Simple NakadiBaseT" (testSimpleNakadiBaseT confLifted)+  , testCase "NakadiBaseT with non-trivial stack" (testNonTrivNakadiBaseT confLifted)+  , testCase "MonadNakadi instance for ReaderT" (testReaderT conf)+  ]++  where conf' = Nakadi.setHttpErrorCallback httpErrorCallback conf+        confLifted = Nakadi.newConfig Nakadi.httpBackendIO (conf^.L.requestTemplate)++testSimpleMonadicAPI :: Nakadi.Config App -> Assertion+testSimpleMonadicAPI conf = runApp . Nakadi.runNakadiT conf $ do+  -- Tests Nakadi call within NakadiT monad transformer+  _events <- Nakadi.eventTypesList+  -- But we can also call Nakadi from within monad transformers on top of NakadiT:+  void $ flip runStateT () $ do+    _events <- Nakadi.eventTypesList+    pure ()+  -- Required for consuming via subscription API is the ability to+  -- call Nakadi from within ResourceT:+  runResourceT $ do+    _partitionStrategies <- Nakadi.registryPartitionStrategies+    pure ()+  -- We can retrieve the configuration using nakadiAsk:+  _conf <- Nakadi.nakadiAsk+  pure ()++testSimpleNakadiBaseT :: Nakadi.Config (NakadiBaseT App) -> Assertion+testSimpleNakadiBaseT conf = runApp . Nakadi.runNakadiWithBase conf $ do+  _events <- Nakadi.eventTypesList+  pure ()++-- | This tests the verbose but more general method of seperating+-- runNakadiBaseT from runNakadiT, changing the monad transformer+-- stack inbetween (by adding NoLoggingT).+testNonTrivNakadiBaseT :: Nakadi.Config (NakadiBaseT App) -> Assertion+testNonTrivNakadiBaseT conf =+  runApp . Nakadi.runNakadiBaseT . runNoLoggingT . Nakadi.runNakadiT conf $ do+  _events <- Nakadi.eventTypesList+  pure ()+++data Env = Env { nakadiConfiguration :: Nakadi.Config App }++instance Nakadi.HasNakadiConfig App Env where+  nakadiConfig = nakadiConfiguration++-- | This test uses the 'MonadNakadi' instance for 'ReaderT' with a+-- 'HasNakadiConfig' constraint.+testReaderT :: Nakadi.Config App -> Assertion+testReaderT conf = runApp . flip runReaderT (Env conf) $ do+  _events <- Nakadi.eventTypesList+  pure ()

tests/Network/Nakadi/Registry/Test.hs view

@@ -11,14 +11,15 @@ import           ClassyPrelude  import           Network.Nakadi+import           Network.Nakadi.Tests.Common import           Test.Tasty import           Test.Tasty.HUnit -testRegistry :: Config -> TestTree+testRegistry :: Config App -> TestTree testRegistry conf = testGroup "Registry"   [ testCase "PartitionStrategies" (testPartitionStrategies conf)   ] -testPartitionStrategies :: Config -> Assertion-testPartitionStrategies conf = do-  void $ registryPartitionStrategies conf+testPartitionStrategies :: Config App -> Assertion+testPartitionStrategies conf = runApp . runNakadiT conf $ do+  void $ registryPartitionStrategies

+ tests/Network/Nakadi/Subscriptions/Processing/Test.hs view

@@ -0,0 +1,137 @@+{-# LANGUAGE FlexibleContexts      #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables   #-}+++-- This test tests the following the high-level subscription consumption.++module Network.Nakadi.Subscriptions.Processing.Test where++import           ClassyPrelude++import           Control.Lens+import           Data.Aeson+import           Data.Conduit+import           Data.Maybe                  (fromJust)+import           Network.Nakadi+import qualified Network.Nakadi.Lenses       as L+import           Network.Nakadi.Tests.Common+import           Test.Tasty+import           Test.Tasty.HUnit++testSubscriptionsProcessing :: Config App -> TestTree+testSubscriptionsProcessing conf = testGroup "Processing"+  [ testCase "SubscriptionProcessing" (testSubscriptionHighLevelProcessing conf)+  ]++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++  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+          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++        consumeParameters = defaultConsumeParameters+                            & setBatchFlushTimeout 1+                            & setBatchLimit 1

tests/Network/Nakadi/Subscriptions/Test.hs view

@@ -16,42 +16,45 @@ import           Network.Nakadi.Tests.Common import           Test.Tasty import           Test.Tasty.HUnit+import Network.Nakadi.Subscriptions.Processing.Test -testSubscriptions :: Config -> TestTree+testSubscriptions :: Config App -> TestTree testSubscriptions conf = testGroup "Subscriptions"-  [ testCase "SubscriptionsList" (testSubscriptionsList conf)+  [ testSubscriptionsProcessing conf+  , testCase "SubscriptionsList" (testSubscriptionsList conf)   , testCase "SubscriptionsCreateDelete" (testSubscriptionsCreateDelete conf)   , testCase "SubscriptionDoubleDeleteFailure" (testSubscriptionsDoubleDeleteFailure conf)   ] -testSubscriptionsList :: Config -> Assertion-testSubscriptionsList conf = do+testSubscriptionsList :: Config App -> Assertion+testSubscriptionsList conf = runApp . runNakadiT conf $ do   -- Cleanup-  deleteSubscriptionsByAppPrefix conf prefix+  deleteSubscriptionsByAppPrefix prefix+  recreateEvent myEventTypeName myEventType   -- Create new Subscriptions   maybeSubscriptionIds <- forM [1..n] $ \i -> do     let owningApp = ApplicationName (prefix <> tshow i)-    subscription <- subscriptionCreate conf (mySubscription (Just owningApp))+    subscription <- subscriptionCreate (mySubscription (Just owningApp))     return (subscription^.L.id)   let subscriptionIds = catMaybes maybeSubscriptionIds-  n @=? length subscriptionIds+  liftIO $ n @=? length subscriptionIds   -- Retrieve list of all Subscriptions-  subscriptions' <- subscriptionsList conf Nothing Nothing+  subscriptions' <- subscriptionsList Nothing Nothing   -- Filter for the subscriptions we have created above   let subscriptionsFiltered = filter (subscriptionAppHasPrefix prefix) subscriptions'       subscriptionIdsFiltered = catMaybes . map (view L.id) $ subscriptionsFiltered-  n @=? length subscriptionIdsFiltered-  sort subscriptionIds @=? sort subscriptionIdsFiltered-+  liftIO $ n @=? length subscriptionIdsFiltered+  liftIO $ sort subscriptionIds @=? sort subscriptionIdsFiltered+     where n = 100         prefix = "test-suite-list-" -deleteSubscriptionsByAppPrefix :: Config -> Text -> IO ()-deleteSubscriptionsByAppPrefix conf prefix = do-  subscriptions <- subscriptionsList conf Nothing Nothing+deleteSubscriptionsByAppPrefix :: MonadNakadi b m => Text -> m ()+deleteSubscriptionsByAppPrefix prefix = do+  subscriptions <- subscriptionsList Nothing Nothing   let subscriptionsFiltered = filter (subscriptionAppHasPrefix prefix) subscriptions       subscriptionIdsFiltered = catMaybes . map (view L.id) $ subscriptionsFiltered-  forM_ subscriptionIdsFiltered (subscriptionDelete conf)+  forM_ subscriptionIdsFiltered subscriptionDelete  subscriptionAppHasPrefix :: Text -> Subscription -> Bool subscriptionAppHasPrefix prefix subscription =@@ -69,22 +72,24 @@   , _initialCursors = Nothing   } -testSubscriptionsCreateDelete :: Config -> Assertion-testSubscriptionsCreateDelete conf = do-  subscription <- subscriptionCreate conf (mySubscription Nothing)-  True @=? isJust (subscription^.L.id)+testSubscriptionsCreateDelete :: Config App -> Assertion+testSubscriptionsCreateDelete conf = runApp . runNakadiT conf $ do+  recreateEvent myEventTypeName myEventType+  subscription <- subscriptionCreate (mySubscription Nothing)+  liftIO $ True @=? isJust (subscription^.L.id)   let (Just subscriptionId) = subscription^.L.id-  subscriptionDelete conf subscriptionId+  subscriptionDelete subscriptionId   return () -testSubscriptionsDoubleDeleteFailure :: Config -> Assertion-testSubscriptionsDoubleDeleteFailure conf = do-  subscription <- subscriptionCreate conf (mySubscription Nothing)-  True @=? isJust (subscription^.L.id)+testSubscriptionsDoubleDeleteFailure :: Config App -> Assertion+testSubscriptionsDoubleDeleteFailure conf = runApp . runNakadiT conf $ do+  recreateEvent myEventTypeName myEventType+  subscription <- subscriptionCreate (mySubscription Nothing)+  liftIO $ True @=? isJust (subscription^.L.id)   let (Just subscriptionId) = subscription^.L.id-  subscriptionDelete conf subscriptionId-  res <- try (subscriptionDelete conf subscriptionId)+  subscriptionDelete subscriptionId+  res <- try (subscriptionDelete subscriptionId)   case res of     Left (SubscriptionNotFound _) -> return ()-    _ -> assertFailure "Expected SubscriptionNotFound exception"+    _ -> liftIO $ assertFailure "Expected SubscriptionNotFound exception"   return ()

tests/Network/Nakadi/Tests/Common.hs view

@@ -1,17 +1,25 @@-{-# LANGUAGE DeriveAnyClass     #-}-{-# LANGUAGE DeriveGeneric      #-}-{-# LANGUAGE RecordWildCards    #-}-{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE DeriveAnyClass        #-}+{-# LANGUAGE DeriveGeneric         #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RecordWildCards       #-}+{-# LANGUAGE StandaloneDeriving    #-}  module Network.Nakadi.Tests.Common where  import           ClassyPrelude +import           Control.Lens import           Data.Aeson-import           Data.UUID      (UUID)+import           Data.UUID             (UUID) import           Network.Nakadi+import qualified Network.Nakadi.Lenses as L import           System.Random +type App = ReaderT () IO++runApp :: App a -> IO a+runApp = flip runReaderT ()+ data Foo = Foo { fortune :: Text } deriving (Show, Eq, Generic)  deriving instance FromJSON Foo@@ -68,15 +76,37 @@   , _dataOp = DataOpUpdate   } -genRandomUUID :: IO UUID-genRandomUUID = randomIO -recreateEvent :: Config -> EventTypeName -> EventType -> IO ()-recreateEvent conf eventTypeName eventType = do-  eventTypeDelete conf eventTypeName `catch` (ignoreExnNotFound ())-  eventTypeCreate conf eventType+genMyDataChangeEvent :: MonadIO m => m (DataChangeEvent Foo)+genMyDataChangeEvent = do+  eid <- genRandomUUID+  now <- liftIO getCurrentTime+  pure DataChangeEvent+    { _payload = Foo "Hello!"+    , _metadata = EventMetadata { _eid        = EventId eid+                                , _occurredAt = Timestamp now+                                , _parentEids = Nothing+                                , _partition  = Nothing+                                }+    , _dataType = "test.FOO"+    , _dataOp = DataOpUpdate+    } -delayedPublish :: (ToJSON a) => Config -> Maybe FlowId -> [a] -> IO ()-delayedPublish conf flowId events  = do-  threadDelay (10^6)-  eventPublish conf myEventTypeName flowId events+genRandomUUID :: MonadIO m => m UUID+genRandomUUID = liftIO randomIO++recreateEvent :: (MonadCatch m, MonadNakadi b m) => EventTypeName -> EventType -> m ()+recreateEvent eventTypeName eventType = do+  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 maybeFlowId events  = do+  liftIO $ threadDelay (10^6)+  let flowId = fromMaybe (FlowId "shalom") maybeFlowId+  config <- nakadiAsk <&> setFlowId flowId+  runNakadiT config $+    eventsPublish myEventTypeName events

tests/Tests.hs view

@@ -1,3 +1,4 @@+{-# LANGUAGE ScopedTypeVariables #-} import           ClassyPrelude  import           Network.HTTP.Client@@ -5,9 +6,12 @@ import           Network.Nakadi.Config.Test import           Network.Nakadi.Connection.Test import           Network.Nakadi.EventTypes.Test+import           Network.Nakadi.Examples.Test import           Network.Nakadi.Internal.Test+import           Network.Nakadi.MonadicAPI.Test import           Network.Nakadi.Registry.Test import           Network.Nakadi.Subscriptions.Test+import           Network.Nakadi.Tests.Common import           System.Environment import           System.Exit import           System.IO                         (hFlush)@@ -48,8 +52,10 @@       let nakadiEndpointT = pack nakadiEndpoint       case parseRequest nakadiEndpoint of         Just request -> do-          conf <- newConfig Nothing request-          return ("nakadi-client Test Suite", unitTests ++ integrationTests conf)+          let conf   = newConfig httpBackendIO request+              confIO = newConfigIO request+          return ("nakadi-client Test Suite",+                  integrationTests conf confIO ++ unitTests)         Nothing -> do           hPut stderr . encodeUtf8 $             "Failed to parse Nakadi URL in TEST_NAKADI_ENDPOINT (" <> nakadiEndpointT <> ")"@@ -64,9 +70,11 @@   , testConnection   ] -integrationTests :: Config -> [TestTree]-integrationTests conf =-  [ testEventTypes conf+integrationTests :: Config App -> ConfigIO -> [TestTree]+integrationTests conf confIO =+  [ testSubscriptions conf+  , testExamples confIO+  , testEventTypes conf   , testRegistry conf-  , testSubscriptions conf+  , testMonadicAPI conf   ]