nakadi-client 0.5.1.0 → 0.6.0.0
raw patch · 69 files changed
+2375/−1292 lines, 69 filesdep +aeson-qqdep −lifted-asyncdep ~asyncdep ~basedep ~resourcet
Dependencies added: aeson-qq
Dependencies removed: lifted-async
Dependency ranges changed: async, base, resourcet
Files
- AUTHORS.md +2/−1
- CHANGES.md +21/−0
- LICENSE +1/−1
- README.md +8/−23
- nakadi-client.cabal +22/−6
- src/Network/Nakadi/Config.hs +57/−102
- src/Network/Nakadi/EventTypes.hs +11/−16
- src/Network/Nakadi/EventTypes/CursorDistances.hs +1/−1
- src/Network/Nakadi/EventTypes/CursorsLag.hs +2/−2
- src/Network/Nakadi/EventTypes/EventType.hs +21/−17
- src/Network/Nakadi/EventTypes/Events.hs +13/−19
- src/Network/Nakadi/EventTypes/Partitions.hs +14/−12
- src/Network/Nakadi/EventTypes/Schemas.hs +1/−1
- src/Network/Nakadi/EventTypes/ShiftedCursors.hs +3/−3
- src/Network/Nakadi/Internal/Committer.hs +61/−0
- src/Network/Nakadi/Internal/Committer/NoBuffer.hs +30/−0
- src/Network/Nakadi/Internal/Committer/Shared.hs +82/−0
- src/Network/Nakadi/Internal/Committer/SmartBuffer.hs +133/−0
- src/Network/Nakadi/Internal/Committer/TimeBuffer.hs +53/−0
- src/Network/Nakadi/Internal/Config.hs +59/−22
- src/Network/Nakadi/Internal/Conversions.hs +1/−1
- src/Network/Nakadi/Internal/GlobalConfig.hs +14/−0
- src/Network/Nakadi/Internal/Http.hs +80/−104
- src/Network/Nakadi/Internal/Json.hs +1/−1
- src/Network/Nakadi/Internal/Lenses.hs +15/−5
- src/Network/Nakadi/Internal/Prelude.hs +15/−9
- src/Network/Nakadi/Internal/Retry.hs +1/−1
- src/Network/Nakadi/Internal/TH.hs +4/−7
- src/Network/Nakadi/Internal/Types.hs +36/−12
- src/Network/Nakadi/Internal/Types/Committer.hs +12/−0
- src/Network/Nakadi/Internal/Types/Config.hs +31/−28
- src/Network/Nakadi/Internal/Types/Exceptions.hs +2/−0
- src/Network/Nakadi/Internal/Types/Logger.hs +1/−1
- src/Network/Nakadi/Internal/Types/Problem.hs +1/−1
- src/Network/Nakadi/Internal/Types/Service.hs +202/−31
- src/Network/Nakadi/Internal/Types/Util.hs +16/−11
- src/Network/Nakadi/Internal/Types/Worker.hs +25/−0
- src/Network/Nakadi/Internal/Unsafe/IO.hs +52/−0
- src/Network/Nakadi/Internal/Util.hs +19/−18
- src/Network/Nakadi/Internal/Worker.hs +199/−0
- src/Network/Nakadi/Lenses.hs +1/−1
- src/Network/Nakadi/Registry.hs +1/−1
- src/Network/Nakadi/Subscriptions.hs +112/−73
- src/Network/Nakadi/Subscriptions/Cursors.hs +42/−43
- src/Network/Nakadi/Subscriptions/Events.hs +121/−260
- src/Network/Nakadi/Subscriptions/Stats.hs +36/−20
- src/Network/Nakadi/Subscriptions/Subscription.hs +14/−13
- src/Network/Nakadi/Types/Config.hs +1/−2
- src/Network/Nakadi/Types/Exceptions.hs +1/−1
- src/Network/Nakadi/Types/Logger.hs +1/−1
- src/Network/Nakadi/Types/Problem.hs +1/−6
- src/Network/Nakadi/Types/Service.hs +8/−3
- src/Network/Nakadi/Unsafe/IO.hs +26/−0
- tests/Network/Nakadi/Config/Test.hs +21/−21
- tests/Network/Nakadi/EventTypes/BusinessEvents/Test.hs +135/−0
- tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs +0/−3
- tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs +1/−5
- tests/Network/Nakadi/EventTypes/Test.hs +116/−123
- tests/Network/Nakadi/Examples/Subscription/Process.hs +18/−13
- tests/Network/Nakadi/Examples/Subscription/Test.hs +38/−40
- tests/Network/Nakadi/Internal/Http/Test.hs +2/−3
- tests/Network/Nakadi/Internal/Types/Test.hs +80/−13
- tests/Network/Nakadi/MonadicAPI/Test.hs +55/−26
- tests/Network/Nakadi/Registry/Test.hs +3/−9
- tests/Network/Nakadi/Subscriptions/Processing/Test.hs +58/−61
- tests/Network/Nakadi/Subscriptions/Stats/Test.hs +72/−0
- tests/Network/Nakadi/Subscriptions/Test.hs +31/−39
- tests/Network/Nakadi/Tests/Common.hs +56/−54
- tests/Tests.hs +2/−2
AUTHORS.md view
@@ -4,5 +4,6 @@ * Viktor Kocherga [@vitold](https://github.com/vitold) * Amr Hassan [@amrhassan](https://github.com/amrhassan) * Eric Torreborre [@etorreborre](https://github.com/etorreborre)+* Dmitrii Dolgov [@erthalion](https://github.com/erthalion) -Thanks to everybody for code, testing, discussions, bug reports, etc.!+Thank you all!
+ CHANGES.md view
@@ -0,0 +1,21 @@+# Nakadi-Client Change Log++## v0.6.0.0++* Nakadi *business events* are supported.++* Support for concurrent workers consuming a subscription. This feature is activated using the function `setWorkerThreads`.++* The modelling of the subscription API has been simplified by by differentiating between `SubscriptionRequest` (before creation) and `Subscription` (after creation).++* An experimental API for creating a Conduit event source for a given subscription is included, allowing simulating the old low-level API using the Subscription API.++* A new function `withSubscription` is included, providing an interface for creating a subscription and automatically passing its subscription ID to some user-provided action.++* A new function `withTemporarySubscription` is included, which is very similar to `withSubscription`, but with the crucicial difference that the subscription will be automatically deleted after the user-provided action has terminated.++* Support for the new `show_time_lag` field when retrieving subscription statistics has been added.++* An experimental `MonadNakadi` instance for the `IO` monad using a global Nakadi configuration has been added. The new module `Network.Nakadi.Unsafe.IO` exposes functionality for accessing this global Nakadi configuration. The new instance for `IO` allows e.g. the evaluation of Nakadi calls interactively in GHCi without the need to run any monad transformers.++* A new convenience function `newConfigFromEnv` is exposed, which allows creating a Nakadi configuration with the Nakadi service URL being derived automatically from the environment variable `NAKADI_URL`.
LICENSE view
@@ -1,6 +1,6 @@ Most of this code is BSD3-licensed with the following copyright notice: -Copyright Moritz Schulte (c) 2017+Copyright Moritz Clasmeier (c) 2017, 2018 All rights reserved.
README.md view
@@ -13,39 +13,24 @@ `nakadi-client` provides: - Docker based test suite testing against the official Nakadi [docker- image](https://github.com/zalando/nakadi#running-a-server) (in- progress).--- A rather direct translation of Nakadi's REST API to Haskell. Thus,- if you are familiar with Nakadi's REST API, the API exposed by- `nakadi-client` will feel very familiar.--- Where suitable, `nakadi-client` provides *additional* higher-level- interfaces.+ image](https://github.com/zalando/nakadi#running-a-server). - A type-safe API for interacting with Nakadi. For example, the name of an event type has type `EventTypeName`, not `Text` or something- generic.+ generic. Correct types for values like `CursorOffset` are provided+ (which must be treated as opaque strings). - Integrated and configurable retry mechanism. - Conduit based interfaces for streaming events. -- Convenient Subscription API interface (`subscriptionSource` &- `runSubscription`), which frees the user from any manual bookkeeping- of the Subscription Stream ID necessary for commiting cursors.--- Mechanism for registering callbacks for logging and token injection.+- Support for temporary subscriptions. -- Correct types for values like `CursorOffset` (which must be treated- as opaque strings).+- Convenient Subscription API interface (`subscriptionProcess` &+ `subscriptionProcessConduit`), which frees the user from any manual+ bookkeeping. -- Basically each API function is exposed in two versions: One which- requires the caller to pass in a Nakadi configuration value- containing the information about how to connect to Nakadi and one- which is suffixed with `R` (think: Reader monad), which expects to- find the Nakadi configuration in the environment provided by a- reader monad in your application's monad stack.+- Mechanism for registering callbacks for logging and token injection. ### Example
nakadi-client.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 9f130201690462832c511918ca52fc51528dc8ccdcee4b68306cb8ef1ed189c2+-- hash: 61b5d898775025659f82db56b3249871d299f52224409c2b8eb75d1d4adc90ca name: nakadi-client-version: 0.5.1.0+version: 0.6.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@@ -21,6 +21,7 @@ extra-source-files: .gitignore AUTHORS.md+ CHANGES.md README.md source-repository head@@ -61,10 +62,17 @@ Network.Nakadi.Internal.Http Network.Nakadi.Internal.Retry Network.Nakadi.HttpBackendIO+ Network.Nakadi.Unsafe.IO other-modules: Network.Nakadi.Internal.Base+ Network.Nakadi.Internal.Committer+ Network.Nakadi.Internal.Committer.NoBuffer+ Network.Nakadi.Internal.Committer.Shared+ Network.Nakadi.Internal.Committer.SmartBuffer+ Network.Nakadi.Internal.Committer.TimeBuffer Network.Nakadi.Internal.Config Network.Nakadi.Internal.Conversions+ Network.Nakadi.Internal.GlobalConfig Network.Nakadi.Internal.HttpBackendIO Network.Nakadi.Internal.Json Network.Nakadi.Internal.Lenses@@ -72,6 +80,7 @@ Network.Nakadi.Internal.TH Network.Nakadi.Internal.Types Network.Nakadi.Internal.Types.Base+ Network.Nakadi.Internal.Types.Committer Network.Nakadi.Internal.Types.Config Network.Nakadi.Internal.Types.Exceptions Network.Nakadi.Internal.Types.Logger@@ -79,7 +88,10 @@ Network.Nakadi.Internal.Types.Service Network.Nakadi.Internal.Types.Subscriptions Network.Nakadi.Internal.Types.Util+ Network.Nakadi.Internal.Types.Worker+ Network.Nakadi.Internal.Unsafe.IO Network.Nakadi.Internal.Util+ Network.Nakadi.Internal.Worker Network.Nakadi.Types.Subscriptions Paths_nakadi_client hs-source-dirs:@@ -88,6 +100,7 @@ build-depends: aeson , aeson-casing+ , async >=2.2.1 && <2.3.0 , async-timer >=0.2.0.0 && <0.3 , base >=4.7 && <5 , bytestring@@ -105,12 +118,13 @@ , monad-control , monad-logger , mtl- , resourcet+ , resourcet >=1.2.0 && <1.3 , retry , safe-exceptions >=0.1.7.0 && <0.2 , scientific , split , stm+ , stm-chans , template-haskell , text , time@@ -133,6 +147,7 @@ other-modules: Network.Nakadi.Config.Test Network.Nakadi.Connection.Test+ Network.Nakadi.EventTypes.BusinessEvents.Test Network.Nakadi.EventTypes.CursorsLag.Test Network.Nakadi.EventTypes.ShiftedCursors.Test Network.Nakadi.EventTypes.Test@@ -147,6 +162,7 @@ Network.Nakadi.MonadicAPI.Test Network.Nakadi.Registry.Test Network.Nakadi.Subscriptions.Processing.Test+ Network.Nakadi.Subscriptions.Stats.Test Network.Nakadi.Subscriptions.Test Network.Nakadi.Tests.Common Paths_nakadi_client@@ -157,9 +173,10 @@ build-depends: aeson , aeson-casing+ , aeson-qq >=0.8.2 && <0.9 , async , async-timer >=0.2.0.0 && <0.3- , base+ , base >=4.7 && <5 , bytestring , classy-prelude >=1.4.0 && <1.5.0 , conduit >=1.3.0 && <1.4.0@@ -175,13 +192,12 @@ , iso8601-time , lens , lens-aeson- , lifted-async , monad-control , monad-logger , mtl , nakadi-client , random- , resourcet+ , resourcet >=1.2.0 && <1.3 , retry , safe-exceptions >=0.1.7.0 && <0.2 , say
src/Network/Nakadi/Config.hs view
@@ -15,6 +15,7 @@ ( newConfig , newConfigIO , newConfigWithDedicatedManager+ , newConfigFromEnv , setHttpManager , setRequestModifier , setDeserializationFailureCallback@@ -22,6 +23,7 @@ , setHttpErrorCallback , setLogFunc , setRetryPolicy+ , setWorkerThreads , setMaxUncommittedEvents , setBatchLimit , setStreamLimit@@ -30,50 +32,23 @@ , setStreamKeepAliveLimit , setFlowId , setCommitStrategy- , defaultConsumeParameters- ) where+ , setShowTimeLag+ )+where import Network.Nakadi.Internal.Prelude import Control.Lens import Control.Retry-import Network.HTTP.Client (Manager,- ManagerSettings)-import Network.HTTP.Client.TLS (newTlsManagerWith)-import Network.Nakadi.Internal.Config+import Network.HTTP.Client ( Manager+ , ManagerSettings+ )+import Network.HTTP.Client.TLS ( newTlsManagerWith ) import Network.Nakadi.Internal.HttpBackendIO-import qualified Network.Nakadi.Internal.Lenses as L+import qualified Network.Nakadi.Internal.Lenses+ as L import Network.Nakadi.Internal.Types---- | Default retry policy.-defaultRetryPolicy :: MonadIO m => RetryPolicyM m-defaultRetryPolicy = fullJitterBackoff 2 <> limitRetries 5---- | Default commit strategy.-defaultCommitStrategy :: CommitStrategy-defaultCommitStrategy = CommitSync---- | Producs a new configuration, with mandatory HTTP manager, default--- consumption parameters and HTTP request template.-newConfig- :: Monad b- => HttpBackend b- -> Request -- ^ Request Template- -> 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- , _commitStrategy = defaultCommitStrategy- }+import Network.Nakadi.Internal.Config -- | Producs a new configuration, with mandatory HTTP manager, default -- consumption parameters and HTTP request template.@@ -84,8 +59,8 @@ -- | Produce a new configuration, with optional HTTP manager settings -- and mandatory HTTP request template.-newConfigWithDedicatedManager ::- (MonadIO b, MonadMask b, MonadIO m)+newConfigWithDedicatedManager+ :: (MonadIO b, MonadMask b, MonadIO m) => ManagerSettings -- ^ Optional 'ManagerSettings' -> Request -- ^ Request template for Nakadi requests -> m (Config b) -- ^ Resulting Configuration@@ -95,100 +70,80 @@ -- | 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+setHttpManager :: Manager -> Config m -> Config m+setHttpManager = (L.manager ?~) -- | 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 -> m Request)- -> Config m- -> Config m+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 -> m ())- -> Config m- -> Config m-setDeserializationFailureCallback cb = L.deserializationFailureCallback .~ Just cb+setDeserializationFailureCallback :: (ByteString -> Text -> m ()) -> Config m -> Config m+setDeserializationFailureCallback = (L.deserializationFailureCallback ?~) -- | Install a callback in the provided configuration which is used -- after having successfully established a streaming Nakadi -- connection.-setStreamConnectCallback ::- StreamConnectCallback m- -> Config m- -> Config m-setStreamConnectCallback cb = L.streamConnectCallback .~ Just cb+setStreamConnectCallback :: StreamConnectCallback m -> Config m -> Config m+setStreamConnectCallback = (L.streamConnectCallback ?~) -- | Install a callback in the provided configuration which is called -- on HTTP 5xx errors. This allows the user to act on such error -- 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 m- -> Config m- -> Config m-setHttpErrorCallback cb = L.httpErrorCallback .~ Just cb+setHttpErrorCallback :: HttpErrorCallback m -> Config m -> Config m+setHttpErrorCallback = (L.httpErrorCallback ?~) -- | Install a logger callback in the provided configuration.-setLogFunc ::- LogFunc m- -> Config m- -> Config m-setLogFunc logFunc = L.logFunc .~ Just logFunc+setLogFunc :: LogFunc m -> Config m -> Config m+setLogFunc = (L.logFunc ?~) -- | Set a custom retry policy in the provided configuration.-setRetryPolicy ::- RetryPolicyM IO- -> Config m- -> Config m+setRetryPolicy :: RetryPolicyM IO -> Config m -> Config m setRetryPolicy = (L.retryPolicy .~) -- | Set flow ID in the provided configuration.-setFlowId- :: FlowId- -> Config m- -> Config m-setFlowId flowId = L.flowId .~ Just flowId+setFlowId :: FlowId -> Config m -> Config m+setFlowId = (L.flowId ?~) -- | Set flow ID in the provided configuration.-setCommitStrategy- :: CommitStrategy- -> Config m- -> Config m+setCommitStrategy :: CommitStrategy -> Config m -> Config m setCommitStrategy = (L.commitStrategy .~) --- | Set maximum number of uncommitted events in the provided value of--- consumption parameters.-setMaxUncommittedEvents :: Int32 -> ConsumeParameters -> ConsumeParameters-setMaxUncommittedEvents n params = params & L.maxUncommittedEvents .~ Just n+-- | Set number of worker threads that should be spawned on+-- subscription consumption. The (per event-type) partitions of the+-- subscription to be consumed will then be mapped onto this finite+-- set of workers.+setWorkerThreads :: Int -> Config m -> Config m+setWorkerThreads n = (L.worker . L.nThreads .~ n) --- | Set batch limit in the provided value of consumption parameters.-setBatchLimit :: Int32 -> ConsumeParameters -> ConsumeParameters-setBatchLimit n params = params & L.batchLimit .~ Just n+-- | Set maximum number of uncommitted events.+setMaxUncommittedEvents :: Int32 -> Config m -> Config m+setMaxUncommittedEvents = (L.maxUncommittedEvents ?~) --- | Set stream limit in the provided value of consumption parameters.-setStreamLimit :: Int32 -> ConsumeParameters -> ConsumeParameters-setStreamLimit n params = params & L.streamLimit .~ Just n+-- | Set batch limit.+setBatchLimit :: Int32 -> Config m -> Config m+setBatchLimit = (L.batchLimit ?~) --- | Set batch-flush-timeout limit in the provided value of--- consumption parameters.-setBatchFlushTimeout :: Int32 -> ConsumeParameters -> ConsumeParameters-setBatchFlushTimeout n params = params & L.batchFlushTimeout .~ Just n+-- | Set stream limit.+setStreamLimit :: Int32 -> Config m -> Config m+setStreamLimit = (L.streamLimit ?~) --- | Set stream timeout in the provided value of consumption parameters.-setStreamTimeout :: Int32 -> ConsumeParameters -> ConsumeParameters-setStreamTimeout n params = params & L.streamTimeout .~ Just n+-- | Set batch-flush-timeout limit.+setBatchFlushTimeout :: Int32 -> Config m -> Config m+setBatchFlushTimeout = (L.batchFlushTimeout ?~) --- | Set stream-keep-alive-limit in the provided value of consumption--- parameters.-setStreamKeepAliveLimit :: Int32 -> ConsumeParameters -> ConsumeParameters-setStreamKeepAliveLimit n params = params & L.streamKeepAliveLimit .~ Just n+-- | Set stream timeout.+setStreamTimeout :: Int32 -> Config m -> Config m+setStreamTimeout = (L.streamTimeout ?~)++-- | Set stream-keep-alive-limit.+setStreamKeepAliveLimit :: Int32 -> Config m -> Config m+setStreamKeepAliveLimit = (L.streamKeepAliveLimit ?~)++setShowTimeLag :: Bool -> Config m -> Config m+setShowTimeLag flag = L.subscriptionStats ?~ SubscriptionStatsConf {_showTimeLag = flag}
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -23,7 +23,8 @@ , module Network.Nakadi.EventTypes.Schemas , eventTypesList , eventTypeCreate- ) where+ )+where import Network.Nakadi.Internal.Prelude @@ -41,23 +42,17 @@ -- | @GET@ to @\/event-types@. Retrieves a list of all registered -- event types.-eventTypesList ::- MonadNakadi b m- => m [EventType] -- ^ Registered Event Types+eventTypesList :: MonadNakadi b m => m [EventType] -- ^ Registered Event Types eventTypesList = do config <- nakadiAsk- httpJsonBody status200 [] $- (setRequestMethod "GET"- . includeFlowId config- . setRequestPath path)+ httpJsonBody status200 [] (setRequestMethod "GET" . includeFlowId config . setRequestPath path) -- | @POST@ to @\/event-types@. Creates a new event type.-eventTypeCreate ::- MonadNakadi b m+eventTypeCreate+ :: MonadNakadi b m => EventType -- ^ Event Type to create -> m ()-eventTypeCreate eventType =- httpJsonNoBody status201 []- (setRequestMethod "POST"- . setRequestPath path- . setRequestBodyJSON 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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -44,7 +44,7 @@ -- information about unconsumed events. cursorsLag' eventTypeName cursors = do config <- nakadiAsk- httpJsonBody ok200 [] $+ httpJsonBody ok200 [] (setRequestMethod "POST" . includeFlowId config . setRequestPath (path eventTypeName)
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -17,7 +17,8 @@ ( eventTypeGet , eventTypeUpdate , eventTypeDelete- ) where+ )+where import Network.Nakadi.Internal.Prelude @@ -34,25 +35,28 @@ -> m EventType -- ^ Event Type information eventTypeGet eventTypeName = do config <- nakadiAsk- httpJsonBody ok200 [(status404, errorEventTypeNotFound)] $- (setRequestMethod "GET"- . includeFlowId config- . setRequestPath (path eventTypeName))+ 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 b m+eventTypeUpdate+ :: MonadNakadi b m => EventTypeName -- ^ Name of Event Type -> EventType -- ^ Event Type Settings -> m () eventTypeUpdate eventTypeName eventType = do config <- nakadiAsk- httpJsonNoBody ok200 [] $- (setRequestMethod "PUT"- . includeFlowId config- . setRequestPath (path eventTypeName)- . setRequestBodyJSON eventType)+ httpJsonNoBody+ ok200+ []+ ( setRequestMethod "PUT"+ . includeFlowId config+ . setRequestPath (path eventTypeName)+ . setRequestBodyJSON eventType+ ) -- | Deletes an event type given its 'EventTypeName'. @DELETE@ to -- @\/event-types\/EVENT-TYPE@.@@ -62,7 +66,7 @@ -> m () eventTypeDelete eventTypeName = do config <- nakadiAsk- httpJsonNoBody ok200 [(status404, errorEventTypeNotFound)] $- (setRequestMethod "DELETE"- . includeFlowId config- . setRequestPath (path eventTypeName))+ 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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -15,12 +15,12 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TupleSections #-} {-# LANGUAGE TypeFamilies #-} module Network.Nakadi.EventTypes.Events ( eventsPublish- ) where+ )+where import Network.Nakadi.Internal.Prelude @@ -28,24 +28,18 @@ import Network.Nakadi.Internal.Http path :: EventTypeName -> ByteString-path eventTypeName =- "/event-types/"- <> encodeUtf8 (unEventTypeName eventTypeName)- <> "/events"+path eventTypeName = "/event-types/" <> encodeUtf8 (unEventTypeName eventTypeName) <> "/events" -- | @POST@ to @\/event-types\/NAME\/events@. Publishes a batch of -- events for the specified event type.-eventsPublish- :: (MonadNakadi b m, ToJSON a)- => EventTypeName- -> [a]- -> m ()+eventsPublish :: (MonadNakadi b m, ToJSON a) => EventTypeName -> [a] -> m () eventsPublish eventTypeName eventBatch = do config <- nakadiAsk- httpJsonNoBody status200- [ (Status 207 "Multi-Status", errorBatchPartiallySubmitted)- , (status422, errorBatchNotSubmitted) ] $- (setRequestMethod "POST"- . includeFlowId config- . setRequestPath (path eventTypeName)- . setRequestBodyJSON eventBatch)+ 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
@@ -17,7 +17,8 @@ module Network.Nakadi.EventTypes.Partitions ( eventTypePartitions , eventTypePartition- ) where+ )+where import Network.Nakadi.Internal.Http import Network.Nakadi.Internal.Prelude@@ -25,11 +26,12 @@ path :: EventTypeName -> Maybe PartitionName -> ByteString path eventTypeName maybePartitionName = "/event-types/"- <> encodeUtf8 (unEventTypeName eventTypeName)- <> "/partitions"- <> (case maybePartitionName of- Just partitionName -> "/" <> encodeUtf8 (unPartitionName partitionName)- Nothing -> "")+ <> encodeUtf8 (unEventTypeName eventTypeName)+ <> "/partitions"+ <> (case maybePartitionName of+ Just partitionName -> "/" <> encodeUtf8 (unPartitionName partitionName)+ Nothing -> ""+ ) -- | @GET@ to @\/event-types\/EVENT-TYPE\/partitions@. Retrieves -- information about all partitions.@@ -39,9 +41,9 @@ -> m [Partition] -- ^ Partition Information eventTypePartitions eventTypeName = do config <- nakadiAsk- httpJsonBody ok200 [(status404, errorEventTypeNotFound)] $- (includeFlowId config- . setRequestPath (path eventTypeName Nothing))+ httpJsonBody ok200+ [(status404, errorEventTypeNotFound)]+ (includeFlowId config . setRequestPath (path eventTypeName Nothing)) -- | @GET@ to @\/event-types\/EVENT-TYPE\/partitions\/PARTITION@. -- Retrieves information about a single partition.@@ -52,6 +54,6 @@ -> m Partition -- ^ Partition Information eventTypePartition eventTypeName partitionName = do config <- nakadiAsk- httpJsonBody ok200 [] $- (includeFlowId config- . setRequestPath (path eventTypeName (Just partitionName)))+ 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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -39,7 +39,7 @@ -> m [Cursor] -- ^ Resulting Cursors cursorsShift' eventTypeName cursors = do config <- nakadiAsk- httpJsonBody ok200 [] $+ httpJsonBody ok200 [] (setRequestMethod "POST" . includeFlowId config . setRequestPath (path eventTypeName)@@ -53,7 +53,7 @@ -> [Cursor] -- ^ Cursors to shift -> Int64 -- ^ Shift Distance -> m [Cursor] -- ^ Resulting Cursors-cursorsShift eventTypeName cursors n = do+cursorsShift eventTypeName cursors n = cursorsShift' eventTypeName (map makeShiftCursor cursors) where makeShiftCursor Cursor { .. } =
+ src/Network/Nakadi/Internal/Committer.hs view
@@ -0,0 +1,61 @@+{-|+Module : Network.Nakadi.Internal.Committer+Description : Implementation of Cursor Committing Strategies+Copyright : (c) Moritz Clasmeier 2017, 2018+License : BSD3+Maintainer : mtesseract@silverratio.net+Stability : experimental+Portability : POSIX++This internal module implements cursor committing strategies to be+used by the subscription API.+-}++{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Network.Nakadi.Internal.Committer where++import Network.Nakadi.Internal.Prelude++import Conduit+import Control.Lens+import UnliftIO.STM++import qualified Network.Nakadi.Internal.Lenses+ as L+import Network.Nakadi.Internal.Types++import Network.Nakadi.Internal.Committer.NoBuffer+import Network.Nakadi.Internal.Committer.Shared+import Network.Nakadi.Internal.Committer.SmartBuffer+import Network.Nakadi.Internal.Committer.TimeBuffer++-- | Main function for the cursor committer thread. Logic depends on+-- the provided buffering strategy. This function dispatches to the+-- actual commit strategy implementations:+--+-- * 'committerNoBuffer' defined in @Network.Nakadi.Internal.Committer.NoBuffer@,+-- * 'committerTimeBuffer' defined in @Network.Nakadi.Internal.Committer.TimeBuffer@ and+-- * 'committerSmartBuffer' defined in @Network.Nakadi.Internal.Committer.SmartBuffer@.+subscriptionCommitter+ :: forall b m+ . (MonadNakadi b m, MonadUnliftIO m, MonadMask m)+ => CommitBufferingStrategy+ -> SubscriptionEventStream+ -> TBQueue (Int, SubscriptionCursor)+ -> m ()+subscriptionCommitter CommitNoBuffer = committerNoBuffer+subscriptionCommitter (CommitTimeBuffer millis) = committerTimeBuffer millis+subscriptionCommitter CommitSmartBuffer = committerSmartBuffer++-- | Sink which can be used as sink for Conduits processing+-- subscription batches. This sink takes care of committing+-- all cursors synchronously, one after the other.+subscriptionSink+ :: (MonadIO m, MonadNakadi b m)+ => SubscriptionEventStream+ -> ConduitM (SubscriptionEventStreamBatch a) void m ()+subscriptionSink eventStream =+ awaitForever $ \batch -> lift $ commitOneCursor eventStream (batch ^. L.cursor)
+ src/Network/Nakadi/Internal/Committer/NoBuffer.hs view
@@ -0,0 +1,30 @@+{-|+Module : Network.Nakadi.Internal.Committer.NoBuffer+Description : Implementation unbuffered Cursor Committing Strategy+Copyright : (c) Moritz Clasmeier 2018+License : BSD3+Maintainer : mtesseract@silverratio.net+Stability : experimental+Portability : POSIX+-}++{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}++module Network.Nakadi.Internal.Committer.NoBuffer where++import Network.Nakadi.Internal.Prelude++import Network.Nakadi.Internal.Committer.Shared+import Network.Nakadi.Internal.Types++import UnliftIO.STM++-- | Implementation for the 'CommitNoBuffer' strategy: We simply read+-- every cursor and commit it in order.+committerNoBuffer+ :: (MonadNakadi b m, MonadUnliftIO m, MonadMask m)+ => SubscriptionEventStream+ -> TBQueue (Int, SubscriptionCursor)+ -> m ()+committerNoBuffer = unbufferedCommitLoop
@@ -0,0 +1,82 @@+{-|+Module : Network.Nakadi.Internal.Committer.Shared+Description : Shared code for Committing Strategies+Copyright : (c) Moritz Clasmeier 2018+License : BSD3+Maintainer : mtesseract@silverratio.net+Stability : experimental+Portability : POSIX+-}++{-# LANGUAGE MultiParamTypeClasses #-}++module Network.Nakadi.Internal.Committer.Shared where++import Network.Nakadi.Internal.Prelude++import Control.Lens+import Control.Monad.Logger+import qualified Data.HashMap.Strict as HashMap++import qualified Network.Nakadi.Internal.Lenses+ as L+import Network.Nakadi.Internal.Types+import Network.Nakadi.Subscriptions.Cursors++import UnliftIO.STM++-- | This function commits all cursors in the provided map of staged+-- cursors.+commitAllCursors+ :: (MonadNakadi b m, MonadIO m)+ => (a -> SubscriptionCursor)+ -> SubscriptionEventStream+ -> TVar (StagedCursorsMap a)+ -> m ()+commitAllCursors extractCursor eventStream stagedCursorsTv = do+ stagedCursors <- liftIO . atomically $ swapTVar stagedCursorsTv HashMap.empty+ let cursors = map extractCursor $ HashMap.elems stagedCursors+ forM_ cursors (commitOneCursor eventStream)++-- | This function takes care of committing a single cursor. Exceptions will be+-- catched and logged, but the failure will NOT be propagated. This means that+-- Nakadi itself is in control of disconnecting us.+commitOneCursor+ :: (MonadIO m, MonadNakadi b m) => SubscriptionEventStream -> SubscriptionCursor -> m ()+commitOneCursor eventStream cursor = do+ config <- nakadiAsk+ catchAny (subscriptionCursorCommit eventStream [cursor])+ $ \exn -> nakadiLiftBase $ case config ^. L.logFunc of+ Just logFunc ->+ logFunc "nakadi-client" LevelWarn+ $ toLogStr+ $ "Failed to commit cursor "+ <> tshow cursor+ <> ": "+ <> tshow exn+ Nothing -> pure ()++-- | Naive cursor commit loop: We simply read every cursor and commit+-- it in order.+unbufferedCommitLoop+ :: (MonadNakadi b m, MonadIO m)+ => SubscriptionEventStream+ -> TBQueue (Int, SubscriptionCursor)+ -> m ()+unbufferedCommitLoop eventStream queue = do+ config <- nakadiAsk+ forever $ do+ (_nEvents, cursor) <- liftIO . atomically . readTBQueue $ queue+ catchAny (subscriptionCursorCommit eventStream [cursor])+ $ \exn -> nakadiLiftBase $ case config ^. L.logFunc of+ Just logFunc ->+ logFunc "nakadi-client" LevelWarn+ $ toLogStr+ $ "Failed to commit cursor "+ <> tshow cursor+ <> ": "+ <> tshow exn+ Nothing -> pure ()++cursorKey :: SubscriptionCursor -> (EventTypeName, PartitionName)+cursorKey cursor = (cursor ^. L.eventType, cursor ^. L.partition)
+ src/Network/Nakadi/Internal/Committer/SmartBuffer.hs view
@@ -0,0 +1,133 @@+{-|+Module : Network.Nakadi.Internal.Committer.SmartBuffer+Description : Implementation of SmartBuffer based Cursor Committing Strategy+Copyright : (c) Moritz Clasmeier 2018+License : BSD3+Maintainer : mtesseract@silverratio.net+Stability : experimental+Portability : POSIX+-}++{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Network.Nakadi.Internal.Committer.SmartBuffer where++import Network.Nakadi.Internal.Prelude++import Control.Concurrent.Async.Timer ( Timer )+import qualified Control.Concurrent.Async.Timer+ as Timer+import qualified Data.HashMap.Strict as HashMap++import Control.Concurrent.STM ( retry )+import Control.Lens+import Data.Function ( (&) )++import Network.Nakadi.Internal.Committer.Shared+import qualified Network.Nakadi.Internal.Lenses+ as L+import Network.Nakadi.Internal.Types++import UnliftIO.Async+import UnliftIO.STM++-- | Implementation of the 'CommitSmartBuffer' strategy: We use an+-- async timer for committing cursors at specified intervals, but if+-- the number of uncommitted events reaches the threshold currently+-- specified by @maxUncommittedEvents@ divided by 2 (safety factor)+-- before the next scheduled commit, a commit is being done right away+-- and the timer is resetted.+--+-- The 'StagedCursor's in the 'CommitSmartBuffer' case carry an 'Int',+-- which is used for counting the number of events processed on the+-- respective partition since the last commit.+committerSmartBuffer+ :: forall b m+ . (MonadNakadi b m, MonadUnliftIO m, MonadMask m)+ => SubscriptionEventStream+ -> TBQueue (Int, SubscriptionCursor)+ -> m ()+committerSmartBuffer eventStream queue = do+ nMaxEvents <- cursorBufferSize+ let millisDefault = 1000+ timerConf =+ Timer.defaultConf & Timer.setInitDelay (fromIntegral millisDefault) & Timer.setInterval+ (fromIntegral millisDefault)+ if nMaxEvents > 1+ then do+ cursorsMap <- liftIO . atomically $ newTVar HashMap.empty+ withAsync (cursorConsumer queue cursorsMap) $ \asyncCursorConsumer -> do+ link asyncCursorConsumer+ Timer.withAsyncTimer timerConf $ cursorCommitter eventStream cursorsMap nMaxEvents+ else unbufferedCommitLoop eventStream queue++-- | The cursorsConsumer drains the cursors queue and adds+-- each cursor to the provided cursorsMap.+cursorConsumer+ :: (MonadIO m)+ => TBQueue (Int, SubscriptionCursor)+ -> TVar (StagedCursorsMap SubscriptionCursorWithCounter)+ -> m ()+cursorConsumer queue cursorsMap = forever . liftIO . atomically $ do+ (nEvents, cursor) <- readTBQueue queue+ let key = cursorKey cursor+ stagedCursor = SubscriptionCursorWithCounter cursor nEvents+ modifyTVar cursorsMap $ HashMap.insertWith updateCursor key stagedCursor++-- | Adds the old number of events to the new entry in the+-- cursors map.+updateCursor+ :: SubscriptionCursorWithCounter -> SubscriptionCursorWithCounter -> SubscriptionCursorWithCounter+updateCursor cursorNew cursorOld = cursorNew & L.nEvents %~ (+ (cursorOld ^. L.nEvents))++-- | Committer loop.+cursorCommitter+ :: (MonadNakadi b m, MonadUnliftIO m)+ => SubscriptionEventStream+ -> TVar (StagedCursorsMap SubscriptionCursorWithCounter)+ -> Int+ -> Timer+ -> m ()+cursorCommitter eventStream cursorsMap nMaxEvents timer =+ forever+ $ race (Timer.wait timer) (maxEventsReached cursorsMap nMaxEvents)+ >>= \case+ Left _ ->+ -- Timer has elapsed, simply commit all currently+ -- buffered cursors.+ commitAllCursors (view L.cursor) eventStream cursorsMap+ Right _ -> do+ -- Events processed since last cursor commit have+ -- crosses configured threshold for at least one+ -- partition. Commit cursors on all such partitions.+ Timer.reset timer+ commitAllCursors (view L.cursor) eventStream cursorsMap++-- | Returns list of cursors that should be committed+-- considering the number of events processed on the+-- respective partition since the last commit. Blocks until at+-- least one such cursor is found.+maxEventsReached+ :: MonadIO m => TVar (StagedCursorsMap SubscriptionCursorWithCounter) -> Int -> m ()+maxEventsReached stagedCursorsTv nMaxEvents = liftIO . atomically $ do+ stagedCursors <- readTVar stagedCursorsTv+ let cursorsList = HashMap.elems stagedCursors+ cursorsCommit = filter (shouldBeCommitted nMaxEvents) cursorsList+ if null cursorsCommit then retry else pure ()++-- | Returns True if the provided staged cursor should be committed.+-- It is expected that the provided staged cursor carries an integral+-- enrichment of the same type as @nMaxEvents@.+shouldBeCommitted :: Int -> SubscriptionCursorWithCounter -> Bool+shouldBeCommitted nMaxEvents stagedCursor = stagedCursor ^. L.nEvents >= nMaxEvents++cursorBufferSize :: MonadNakadi b m => m Int+cursorBufferSize = do+ conf <- nakadiAsk+ pure $ case conf ^. L.maxUncommittedEvents of+ Nothing -> 1+ Just n -> n & fromIntegral & (* safetyFactor) & round+ where safetyFactor = 0.5
+ src/Network/Nakadi/Internal/Committer/TimeBuffer.hs view
@@ -0,0 +1,53 @@+{-|+Module : Network.Nakadi.Internal.Committer.TimeBuffer+Description : Implementation of TimeBuffer based Cursor Committing Strategy+Copyright : (c) Moritz Clasmeier 2018+License : BSD3+Maintainer : mtesseract@silverratio.net+Stability : experimental+Portability : POSIX+-}++{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}++module Network.Nakadi.Internal.Committer.TimeBuffer where++import Network.Nakadi.Internal.Prelude++import qualified Control.Concurrent.Async.Timer+ as Timer+import qualified Data.HashMap.Strict as HashMap++import Data.Function ( (&) )+import Network.Nakadi.Internal.Committer.Shared+import Network.Nakadi.Internal.Types++import UnliftIO.Async+import UnliftIO.STM++-- | Implementation of the 'CommitTimeBuffer' strategy: We use an+-- async timer for committing cursors at specified intervals.+--+-- The 'StagedCursor's in the 'CommitTimeBuffer' case carry no+-- additional information, just the subscription cursors.+committerTimeBuffer+ :: (MonadNakadi b m, MonadUnliftIO m, MonadMask m)+ => Int32+ -> SubscriptionEventStream+ -> TBQueue (Int, SubscriptionCursor)+ -> m ()+committerTimeBuffer millis eventStream queue = do+ let timerConf = Timer.defaultConf & Timer.setInitDelay (fromIntegral millis) & Timer.setInterval+ (fromIntegral millis)+ cursorsMap <- liftIO $ newTVarIO (HashMap.empty :: StagedCursorsMap SubscriptionCursor)+ withAsync (cursorConsumer cursorsMap) $ \asyncCursorConsumer -> do+ link asyncCursorConsumer+ Timer.withAsyncTimer timerConf $ \timer -> forever $ do+ Timer.wait timer+ commitAllCursors identity eventStream cursorsMap+ where -- | The cursorsConsumer drains the cursors queue and adds each+ -- cursor to the provided cursorsMap.+ cursorConsumer cursorsMap = forever . liftIO . atomically $ do+ (_, cursor) <- readTBQueue queue+ modifyTVar cursorsMap (HashMap.insert (cursorKey cursor) cursor)
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -16,30 +16,67 @@ module Network.Nakadi.Internal.Config where -import Network.Nakadi.Internal.Prelude+import System.Environment +import Control.Retry+import Network.HTTP.Client ( parseRequest )++import Network.Nakadi.Internal.Prelude+import Network.Nakadi.Internal.HttpBackendIO import Network.Nakadi.Internal.Types -buildConsumeQueryParameters :: ConsumeParameters -> [(ByteString, ByteString)]-buildConsumeQueryParameters ConsumeParameters { .. } = catMaybes- [ ("batch_limit",) . encodeUtf8 . tshow <$> _batchLimit- , ("stream_limit",) . encodeUtf8 . tshow <$> _streamLimit- , ("batch_flush_timeout",) . encodeUtf8 . tshow <$> _batchFlushTimeout- , ("stream_timeout",) . encodeUtf8 . tshow <$> _streamTimeout- , ("stream_keep_alive_limit",) . encodeUtf8 . tshow <$> _streamKeepAliveLimit ]+buildConsumeQueryParameters :: Config m -> [(ByteString, ByteString)]+buildConsumeQueryParameters Config {..} = catMaybes+ [ ("batch_limit", ) . encodeUtf8 . tshow <$> _batchLimit+ , ("stream_limit", ) . encodeUtf8 . tshow <$> _streamLimit+ , ("batch_flush_timeout", ) . encodeUtf8 . tshow <$> _batchFlushTimeout+ , ("stream_timeout", ) . encodeUtf8 . tshow <$> _streamTimeout+ , ("max_uncommitted_events", ) . encodeUtf8 . tshow <$> _maxUncommittedEvents+ , ("stream_keep_alive_limit", ) . encodeUtf8 . tshow <$> _streamKeepAliveLimit+ ] -buildSubscriptionConsumeQueryParameters :: ConsumeParameters -> [(ByteString, ByteString)]-buildSubscriptionConsumeQueryParameters params@ConsumeParameters { .. } =- buildConsumeQueryParameters params- ++ catMaybes [("max_uncommitted_events",) . encodeUtf8 . tshow <$> _maxUncommittedEvents]+newConfigFromEnv :: (MonadIO m, MonadThrow m, MonadMask b, MonadIO b) => m (Config b)+newConfigFromEnv = do+ request <- liftIO (lookupEnv "NAKADI_URL") >>= maybe (throwM NakadiUrlMissing) parseRequest+ pure $ newConfig httpBackendIO request --- | Default parameters for event consumption.-defaultConsumeParameters :: ConsumeParameters-defaultConsumeParameters = ConsumeParameters- { _maxUncommittedEvents = Nothing- , _batchLimit = Nothing- , _streamLimit = Nothing- , _batchFlushTimeout = Nothing- , _streamTimeout = Nothing- , _streamKeepAliveLimit = Nothing+-- | Default retry policy.+defaultRetryPolicy :: MonadIO m => RetryPolicyM m+defaultRetryPolicy = fullJitterBackoff 2 <> limitRetries 5++-- | Default commit strategy.+defaultCommitStrategy :: CommitStrategy+defaultCommitStrategy = CommitSync++-- | Default worker configuration. This specifies single-threaded consumption of subscriptions.+defaultWorkerConfig :: WorkerConfig+defaultWorkerConfig = WorkerConfig {_nThreads = 1}++-- | Producs a new configuration, with mandatory HTTP manager, default+-- consumption parameters and HTTP request template.+newConfig+ :: Monad b+ => HttpBackend b+ -> Request -- ^ Request Template+ -> Config b -- ^ Resulting Configuration+newConfig httpBackend request = Config+ { _manager = Nothing+ , _requestTemplate = request+ , _requestModifier = pure+ , _deserializationFailureCallback = Nothing+ , _streamConnectCallback = Nothing+ , _logFunc = Nothing+ , _retryPolicy = defaultRetryPolicy+ , _http = httpBackend+ , _httpErrorCallback = Nothing+ , _flowId = Nothing+ , _commitStrategy = defaultCommitStrategy+ , _subscriptionStats = Nothing+ , _maxUncommittedEvents = Nothing+ , _batchLimit = Nothing+ , _streamLimit = Nothing+ , _batchFlushTimeout = Nothing+ , _streamTimeout = Nothing+ , _streamKeepAliveLimit = Nothing+ , _worker = defaultWorkerConfig }
src/Network/Nakadi/Internal/Conversions.hs view
@@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Internal.Conversions Description : Nakadi Client Conversions (Internal)-Copyright : (c) Moritz Schulte 2017+Copyright : (c) Moritz Clasmeier 2017 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental
+ src/Network/Nakadi/Internal/GlobalConfig.hs view
@@ -0,0 +1,14 @@+{-# LANGUAGE MultiParamTypeClasses #-}++module Network.Nakadi.Internal.GlobalConfig where++import UnliftIO.STM++import System.IO.Unsafe++import Network.Nakadi.Internal.Types.Config+import Network.Nakadi.Internal.Prelude++{-# NOINLINE globalConfiguration #-}+globalConfiguration :: TVar (Maybe ConfigIO)+globalConfiguration = unsafePerformIO $ newTVarIO 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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -14,7 +14,6 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-} @@ -41,27 +40,32 @@ , errorBatchPartiallySubmitted , errorBatchNotSubmitted , setRequestQueryParameters- ) where+ )+where import Network.Nakadi.Internal.Prelude -import Conduit hiding (throwM)+import Conduit hiding ( throwM ) import Control.Arrow import Control.Lens-import Control.Monad (void)+import Control.Monad ( void ) import Data.Aeson-import qualified Data.ByteString.Lazy as ByteString.Lazy-import qualified Data.ByteString.Lazy as LB-import qualified Data.Text as Text-import Network.HTTP.Client (BodyReader,- HttpException (..),- HttpExceptionContent (..),- Manager, checkResponse,- responseBody, responseStatus)-import Network.HTTP.Simple hiding (Proxy)+import qualified Data.ByteString.Lazy as ByteString.Lazy+import qualified Data.ByteString.Lazy as LB+import qualified Data.Text as Text+import Network.HTTP.Client ( BodyReader+ , HttpException(..)+ , HttpExceptionContent(..)+ , 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 qualified Network.Nakadi.Internal.Lenses+ as L import Network.Nakadi.Internal.Types import Network.Nakadi.Internal.Util @@ -72,24 +76,25 @@ 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+ => ConduitM ByteString a m () -- ^ Conduit deserializing bytestrings -- into custom values-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+conduitDecode = do+ config <- lift nakadiAsk+ awaitForever $ \ a -> case eitherDecodeStrict' a of+ Right v -> yield v+ Left err -> lift . nakadiLiftBase $ callback config a (Text.pack err) + where callback config 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 :: MonadNakadi b m@@ -102,8 +107,7 @@ modifyRequest (config ^. L.requestModifier) request -- | Modify the Request based on a user function in the configuration.-modifyRequest- :: MonadNakadi b m => (Request -> b Request) -> Request -> m Request+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@@ -111,20 +115,14 @@ -- | Executes an HTTP request using the provided configuration and a -- pure request modifier. httpExecRequest- :: MonadNakadi b m- => (Request -> Request)- -> m (Response ByteString.Lazy.ByteString)+ :: MonadNakadi b m => (Request -> Request) -> m (Response ByteString.Lazy.ByteString) httpExecRequest requestDef = do config <- nakadiAsk req <- httpBuildRequest requestDef- nakadiLiftBase $ nakadiHttpLbs config req (config^.L.manager)+ 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+nakadiHttpLbs :: Config b -> Request -> Maybe Manager -> b (Response LB.ByteString)+nakadiHttpLbs config = (config ^. L.http . L.httpLbs) config -- | Executes an HTTP request using the provided configuration and a -- pure request modifier. Returns the HTTP response and separately the@@ -164,16 +162,12 @@ Nothing -> throwIO (UnexpectedResponse (void response)) where exceptionMap' = exceptionMap ++ defaultExceptionMap -nakadiHttpResponseOpen :: Config b- -> Request- -> Maybe Manager- -> b (Response (ConduitM () ByteString b ()))-nakadiHttpResponseOpen config req maybeMngr =- (config^.L.http.L.httpResponseOpen) config req maybeMngr+nakadiHttpResponseOpen+ :: Config b -> Request -> Maybe Manager -> b (Response (ConduitM () ByteString b ()))+nakadiHttpResponseOpen config = (config ^. L.http . L.httpResponseOpen) config nakadiHttpResponseClose :: Config b -> Response () -> b ()-nakadiHttpResponseClose config rsp =- (config^.L.http.L.httpResponseClose) rsp+nakadiHttpResponseClose = view (L.http . L.httpResponseClose) httpJsonBodyStream :: forall b m r@@ -186,47 +180,41 @@ httpJsonBodyStream successStatus exceptionMap requestDef handler = do config <- nakadiAsk request <- httpBuildRequest requestDef- bracket (nakadiLiftBase $ nakadiHttpResponseOpen config request (config^.L.manager))+ bracket (nakadiLiftBase $ nakadiHttpResponseOpen config request (config ^. L.manager)) (nakadiLiftBase . nakadiHttpResponseClose config . void)- $ \response -> wrappedHandler config response-- 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_)+ $ \response -> wrappedHandler config response+ 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_) - exceptionMap' = exceptionMap ++ defaultExceptionMap+ exceptionMap' = exceptionMap ++ defaultExceptionMap - -- connectCallback :: s -> t -> m ()- connectCallback config response =- nakadiLiftBase $ case config^.L.streamConnectCallback of- Just cb -> cb response- Nothing -> pure ()+ -- 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 -includeFlowId- :: Config b- -> Request- -> Request-includeFlowId config =- case config^.L.flowId of- Just flowId -> setRequestHeader "X-Flow-Id" [encodeUtf8 (unFlowId flowId)]- Nothing -> identity+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 :: MonadThrow m => [(Status, ByteString.Lazy.ByteString -> m NakadiException)] defaultExceptionMap = [ (status401, errorClientNotAuthenticated) , (status403, errorAccessForbidden)@@ -235,53 +223,41 @@ , (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/Json.hs view
@@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Internal.Json Description : Nakadi Client Json (Internal)-Copyright : (c) Moritz Schulte 2017+Copyright : (c) Moritz Clasmeier 2017 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -24,23 +24,29 @@ 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.Committer import Network.Nakadi.Internal.Types.Config import Network.Nakadi.Internal.Types.Service+import Network.Nakadi.Internal.Types.Worker makeNakadiLenses ''Config+makeNakadiLenses ''SubscriptionStatsConf makeNakadiLenses ''HttpBackend makeNakadiLenses ''Cursor makeNakadiLenses ''DataChangeEvent makeNakadiLenses ''DataChangeEventEnriched+makeNakadiLenses ''BusinessEvent+makeNakadiLenses ''BusinessEventEnriched makeNakadiLenses ''SubscriptionEventStreamBatch makeNakadiLenses ''EventMetadata makeNakadiLenses ''EventMetadataEnriched makeNakadiLenses ''Partition+makeNakadiLenses ''PartitionStat makeNakadiLenses ''CursorDistanceQuery makeNakadiLenses ''CursorDistanceResult makeNakadiLenses ''Timestamp@@ -50,13 +56,17 @@ makeNakadiLenses ''EventTypeSchemasResponse makeNakadiLenses ''PaginationLink makeNakadiLenses ''PaginationLinks-makeNakadiLenses ''SubscriptionEventTypeStatsResult-makeNakadiLenses ''ConsumeParameters+makeNakadiLenses ''SubscriptionStats makeNakadiLenses ''SubscriptionCursorCommit makeNakadiLenses ''SubscriptionCursor makeNakadiLenses ''CursorCommit makeNakadiLenses ''SubscriptionsListResponse makeNakadiLenses ''Subscription+makeNakadiLenses ''WorkerConfig+makeNakadiLenses ''Worker+makeNakadiLenses ''WorkerRegistry+makeNakadiLenses ''SubscriptionRequest+makeNakadiLenses ''SubscriptionCursorWithCounter instance HasNakadiId StreamId Text where id f (StreamId a) = StreamId <$> f a
src/Network/Nakadi/Internal/Prelude.hs view
@@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Prelude Description : Nakadi Client Prelude (Internal)-Copyright : (c) Moritz Schulte 2017+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -12,6 +12,7 @@ module Network.Nakadi.Internal.Prelude ( module Prelude+ , module Control.Monad.IO.Unlift , module Data.Int , module Data.Monoid , module Data.Maybe@@ -27,26 +28,31 @@ , identity , undefined , error- , MonadIO- , liftIO , Request , Response- ) where+ )+where import Control.Exception.Safe import Control.Monad import Control.Monad.IO.Class+import Control.Monad.IO.Unlift import Control.Monad.Reader.Class import Data.ByteString import Data.Int-import Data.Map (Map)+import Data.Map ( Map ) import Data.Maybe import Data.Monoid-import Data.Text (Text)-import qualified Data.Text as Text+import Data.Text ( Text )+import qualified Data.Text as Text import Data.Text.Encoding-import Network.HTTP.Client (Request, Response)-import Prelude hiding (error, id, undefined)+import Network.HTTP.Client ( Request+ , Response+ )+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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental
src/Network/Nakadi/Internal/TH.hs view
@@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Internal.TH Description : Nakadi Client TemplateHaskell (Internal)-Copyright : (c) Moritz Schulte 2017+Copyright : (c) Moritz Clasmeier 2017 License : BSD2 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -10,8 +10,6 @@ Internal TemplateHaskell specific code. -} -{-# LANGUAGE TemplateHaskell #-}- module Network.Nakadi.Internal.TH where import Control.Lens@@ -35,13 +33,12 @@ -- | Rules for creating nakadi-client lenses. nakadiLensRules :: LensRules-nakadiLensRules =- defaultFieldRules & lensField .~ nakadiLensNamer+nakadiLensRules = defaultFieldRules & lensField .~ nakadiLensNamer -- | Convenience function, copied from the lens package. -- -- License: BSD2 -- Copyright (C) 2012-2016 Edward A. Kmett overHead :: (a -> a) -> [a] -> [a]-overHead _ [] = []-overHead f (x:xs) = f x : xs+overHead _ [] = []+overHead f (x : xs) = f x : xs
src/Network/Nakadi/Internal/Types.hs view
@@ -12,20 +12,20 @@ {-# 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 LambdaCase #-} {-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE CPP #-} module Network.Nakadi.Internal.Types ( module Network.Nakadi.Internal.Types.Config+ , module Network.Nakadi.Internal.Types.Committer , module Network.Nakadi.Internal.Types.Exceptions , module Network.Nakadi.Internal.Types.Logger , module Network.Nakadi.Internal.Types.Problem@@ -33,35 +33,41 @@ , module Network.Nakadi.Internal.Types.Util , module Network.Nakadi.Internal.Types.Base , module Network.Nakadi.Internal.Types.Subscriptions+ , module Network.Nakadi.Internal.Types.Worker , HasNakadiConfig(..) , MonadNakadi(..) , MonadNakadiIO , NakadiT(..) , runNakadiT- ) where+ )+where +import UnliftIO.STM ( readTVarIO ) 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 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 Control.Monad.Trans.Reader ( ReaderT(..) )+import Control.Monad.Trans.Resource hiding (release)+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.Committer 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.GlobalConfig import Network.Nakadi.Internal.Types.Subscriptions import Network.Nakadi.Internal.Types.Util+import Network.Nakadi.Internal.Types.Worker -- * Define Typeclasses @@ -97,7 +103,7 @@ -- | `Functor` for `NakadiT`. instance Functor m => Functor (NakadiT b m) where- fmap f (NakadiT n) = NakadiT (\c -> fmap f (n c))+ fmap f (NakadiT n) = NakadiT (fmap f . n) -- | `Applicative` for `NakadiT`. instance (Applicative m) => Applicative (NakadiT b m) where@@ -145,6 +151,13 @@ 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)+#if MIN_VERSION_exceptions(0, 10, 0)+ generalBracket acquire release use = NakadiT $ \ conf ->+ generalBracket+ (_runNakadiT acquire conf)+ (\resource exitCase -> _runNakadiT (release resource exitCase) conf)+ (\resource -> _runNakadiT (use resource) conf)+#endif -- | 'MonadIO' instance (Monad b, MonadIO m) => MonadIO (NakadiT b m) where@@ -170,6 +183,9 @@ get = lift get put = lift . put +instance (Monad b, MonadResource m) => MonadResource (NakadiT b m) where+ liftResourceT = lift . liftResourceT+ -- | 'MonadUnliftIO' instance (Monad b, MonadUnliftIO m) => MonadUnliftIO (NakadiT b m) where {-# INLINE askUnliftIO #-}@@ -195,6 +211,14 @@ -- | 'MonadNakadiBase' instance {-# OVERLAPPABLE #-} MonadNakadiBase b m => MonadNakadiBase b (NakadiT b m) +-- ** Implementation of 'MonadNakadi' for IO. Useful for testing purposes or scripting.++instance MonadNakadi IO IO where+ nakadiAsk =+ liftIO (readTVarIO globalConfiguration) >>= \case+ Nothing -> throwIO ConfigurationMissing+ Just config -> pure config+ -- ** Implementations 'MonadNakadi' typeclass for transformers. -- | 'ReaderT'@@ -239,4 +263,4 @@ 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)+mapNakadiT f n = NakadiT $ \c -> f (_runNakadiT n c)
+ src/Network/Nakadi/Internal/Types/Committer.hs view
@@ -0,0 +1,12 @@+{-# LANGUAGE StrictData #-}++module Network.Nakadi.Internal.Types.Committer where++import Data.Int+import Data.HashMap.Strict ( HashMap )++import Network.Nakadi.Internal.Types.Service++data SubscriptionCursorWithCounter = SubscriptionCursorWithCounter { _cursor :: SubscriptionCursor, _nEvents :: Int }++type StagedCursorsMap a = HashMap (EventTypeName, PartitionName) a
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -18,7 +18,7 @@ import Conduit import Control.Retry-import qualified Data.ByteString.Lazy as LB+import qualified Data.ByteString.Lazy as LB import Network.HTTP.Client import Network.Nakadi.Internal.Prelude import Network.Nakadi.Internal.Types.Logger@@ -35,35 +35,38 @@ type ConfigIO = Config 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- , _commitStrategy :: CommitStrategy- } -> Config m---- | ConsumeParameters--data ConsumeParameters = ConsumeParameters- { _maxUncommittedEvents :: Maybe Int32- , _batchLimit :: Maybe Int32- , _streamLimit :: Maybe Int32- , _batchFlushTimeout :: Maybe Int32- , _streamTimeout :: Maybe Int32- , _streamKeepAliveLimit :: Maybe Int32- } deriving (Show, Eq, Ord)-+data Config m =+ Config { _requestTemplate :: Request+ , _requestModifier :: Request -> m Request+ , _manager :: Maybe Manager+ , _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+ , _commitStrategy :: CommitStrategy+ , _subscriptionStats :: Maybe SubscriptionStatsConf+ , _maxUncommittedEvents :: Maybe Int32+ , _batchLimit :: Maybe Int32+ , _streamLimit :: Maybe Int32+ , _batchFlushTimeout :: Maybe Int32+ , _streamTimeout :: Maybe Int32+ , _streamKeepAliveLimit :: Maybe Int32+ , _worker :: WorkerConfig+ } 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 ()+ }++data WorkerConfig = WorkerConfig+ { _nThreads :: Int+ }++data SubscriptionStatsConf = SubscriptionStatsConf+ { _showTimeLag :: Bool }
src/Network/Nakadi/Internal/Types/Exceptions.hs view
@@ -39,6 +39,8 @@ | RequestModificationException SomeException | CursorDistanceNoResult | StreamIdMissing+ | NakadiUrlMissing+ | ConfigurationMissing 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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental
src/Network/Nakadi/Internal/Types/Problem.hs view
@@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Internal.Types.Problem Description : Nakadi Client Problem Type (Internal)-Copyright : (c) Moritz Schulte 2017+Copyright : (c) Moritz Clasmeier 2017 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -16,6 +16,7 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE StrictData #-} {-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE RecordWildCards #-} module Network.Nakadi.Internal.Types.Service where @@ -25,12 +26,13 @@ import Data.Aeson.TH import Data.Aeson.Types import Data.Hashable+import qualified Data.HashMap.Strict as HashMap import Data.String-import qualified Data.Text as Text+import qualified Data.Text as Text import Data.Time import Data.Time.ISO8601 import Data.UUID-import Data.Vector (Vector)+import Data.Vector ( Vector ) import GHC.Generics import Network.Nakadi.Internal.Json@@ -279,40 +281,154 @@ deriveJSON nakadiJsonOptions ''CursorDistanceResult --- | Type for subscription positions.+-- | This type models the "read_from" field contained in subscription objects. +data SubscriptionReadFrom = SubscriptionReadFromBegin+ | SubscriptionReadFromEnd+ | SubscriptionReadFromCursors+ deriving (Show, Eq, Ord, Generic, Hashable)++instance ToJSON SubscriptionReadFrom where+ toJSON = \case+ SubscriptionReadFromBegin -> "begin"+ SubscriptionReadFromEnd -> "end"+ SubscriptionReadFromCursors -> "cursors"++instance FromJSON SubscriptionReadFrom where+ parseJSON = \case+ "begin" -> return SubscriptionReadFromBegin+ "end" -> return SubscriptionReadFromEnd+ "cursors" -> return SubscriptionReadFromCursors+ invalid -> typeMismatch "SubscriptionReadFrom" invalid++-- | Type modelling a subscription position.+ data SubscriptionPosition = SubscriptionPositionBegin | SubscriptionPositionEnd- | SubscriptionPositionCursors+ | SubscriptionPositionCursors [SubscriptionCursorWithoutToken] deriving (Show, Eq, Ord, Generic, Hashable) +-- | Internal helper function for converting a 'SubscriptionPosition' into a+-- JSON Object (not a JSON Value). Removes the need for partial pattern matching later.+subscriptionPositionToObject :: SubscriptionPosition -> Object+subscriptionPositionToObject = \case+ SubscriptionPositionBegin ->+ HashMap.fromList [("read_from", toJSON SubscriptionReadFromBegin)]+ SubscriptionPositionEnd ->+ HashMap.fromList [("read_from", toJSON SubscriptionReadFromEnd)]+ SubscriptionPositionCursors cursors ->+ HashMap.fromList [("read_from", toJSON SubscriptionReadFromCursors)+ ,("cursors", toJSON cursors)]+ instance ToJSON SubscriptionPosition where- toJSON pos = case pos of- SubscriptionPositionBegin -> "begin"- SubscriptionPositionEnd -> "end"- SubscriptionPositionCursors -> "cursors"+ toJSON = Object . subscriptionPositionToObject instance FromJSON SubscriptionPosition where- parseJSON pos = case pos of- "begin" -> return SubscriptionPositionBegin- "end" -> return SubscriptionPositionEnd- "cursors" -> return SubscriptionPositionCursors- invalid -> typeMismatch "SubscriptionPosition" invalid+ parseJSON = withObject "SubscriptionPosition" $ \obj -> do+ readFrom <- obj .: "read_from" >>= parseJSON+ case readFrom of+ SubscriptionReadFromBegin ->+ pure SubscriptionPositionBegin+ SubscriptionReadFromEnd ->+ pure SubscriptionPositionEnd+ SubscriptionReadFromCursors ->+ SubscriptionPositionCursors <$> obj .: "cursors" --- | Type for a Subscription.+-- | This type models the value describing the use case of a subscription.+-- In general this is an additional identifier used to differ subscriptions+-- having the same owning application and event types.+newtype ConsumerGroup = ConsumerGroup { unConsumerGroup :: Text }+ deriving (Eq, Ord, Show, Generic, Hashable) +instance IsString ConsumerGroup where+ fromString = ConsumerGroup . Text.pack++instance ToJSON ConsumerGroup where+ toJSON = String . unConsumerGroup++instance FromJSON ConsumerGroup where+ parseJSON = parseString "ConsumerGroup" ConsumerGroup++-- | Type for a Subscription which has already been created.+--+-- When a subscription object is retrieved from Nakadi the following fields+-- are regarded as mandatory:+--+-- * @id@+-- * @owning_application@+-- * @event_types@+-- * @consumer_group@+-- * @created_at@+-- * @read_from@+-- * depending on @read_from@ also @cursors@. data Subscription = Subscription- { _id :: Maybe SubscriptionId- , _owningApplication :: ApplicationName- , _eventTypes :: [EventTypeName]- , _consumerGroup :: Maybe Text- , _createdAt :: Maybe Timestamp- , _readFrom :: Maybe SubscriptionPosition- , _initialCursors :: Maybe [SubscriptionCursorWithoutToken]+ { _id :: SubscriptionId+ , _owningApplication :: ApplicationName+ , _eventTypes :: [EventTypeName]+ , _consumerGroup :: ConsumerGroup+ , _createdAt :: Timestamp+ , _subscriptionPosition :: SubscriptionPosition } deriving (Show, Eq, Ord, Generic, Hashable) -deriveJSON nakadiJsonOptions ''Subscription+instance FromJSON Subscription where+ parseJSON = withObject "Subscription" $ \obj ->+ Subscription <$> obj .: "id"+ <*> obj .: "owning_application"+ <*> obj .: "event_types"+ <*> obj .: "consumer_group"+ <*> obj .: "created_at"+ <*> parseJSON (Object obj) +instance ToJSON Subscription where+ toJSON Subscription {..} =+ let obj = HashMap.fromList [ ("id", toJSON _id)+ , ("owning_application", toJSON _owningApplication)+ , ("event_types", toJSON _eventTypes)+ , ("consumer_group", toJSON _consumerGroup)+ , ("created_at", toJSON _createdAt) ]+ <> subscriptionPositionToObject _subscriptionPosition+ in Object obj++-- | Type for a Subscription which is to be created.+--+-- When a subscription is to be created the following fields+-- are regarded as mandatory in the subscription object:+--+-- * @owning_application@+-- * @event_types@+--+-- The remaining fields are regarded as optional:+--+-- * @consumer_group@+-- * @read_from@+-- * depending on @read_from@ the field @cursors@ might+-- have to be present as well.+data SubscriptionRequest = SubscriptionRequest+ { _owningApplication :: ApplicationName+ , _eventTypes :: [EventTypeName]+ , _consumerGroup :: Maybe ConsumerGroup+ , _subscriptionPosition :: Maybe SubscriptionPosition+ } deriving (Show, Eq, Ord, Generic, Hashable)++instance FromJSON SubscriptionRequest where+ parseJSON = withObject "SubscriptionRequest" $ \obj ->+ SubscriptionRequest <$> obj .: "owning_application"+ <*> obj .: "event_types"+ <*> obj .:? "consumer_group"+ <*> parseJSON (Object obj)++instance ToJSON SubscriptionRequest where+ toJSON SubscriptionRequest {..} =+ let obj = subscriptionPositionToObject (fromMaybe SubscriptionPositionEnd _subscriptionPosition)+ <> HashMap.fromList [ ("owning_application", toJSON _owningApplication)+ , ("event_types", toJSON _eventTypes) ]+ <> case _consumerGroup of+ Just consumerGroup ->+ HashMap.fromList [ ("consumer_group", toJSON consumerGroup) ]+ Nothing ->+ HashMap.empty+ in Object obj+ -- | Type for publishing status. data PublishingStatus = PublishingStatusSubmitted | PublishingStatusFailed@@ -541,10 +657,11 @@ -- | Type for per-partition statistics. data PartitionStat = PartitionStat- { _partition :: PartitionName- , _state :: PartitionState- , _unconsumedEvents :: Int64- , _streamId :: StreamId+ { _partition :: PartitionName+ , _state :: PartitionState+ , _unconsumedEvents :: Maybe Int64+ , _streamId :: Maybe StreamId+ , _consumerLagSeconds :: Maybe Int64 } deriving (Show, Eq, Ord, Generic) deriveJSON nakadiJsonOptions ''PartitionStat@@ -558,15 +675,16 @@ deriveJSON nakadiJsonOptions ''SubscriptionEventTypeStats --- | SubscriptionEventTypeStatsResult+-- | Type modelling per-subscription statistics. Objects of this type are returned by+-- requests to /subscriptions/SUBSCRIPTION-ID/stats. -newtype SubscriptionEventTypeStatsResult = SubscriptionEventTypeStatsResult+newtype SubscriptionStats = SubscriptionStats { _items :: [SubscriptionEventTypeStats] } deriving (Show, Eq, Ord, Generic) -deriveJSON nakadiJsonOptions ''SubscriptionEventTypeStatsResult+deriveJSON nakadiJsonOptions ''SubscriptionStats --- | Type for the category of an 'EventType'.+-- | Type for the category of an 'EventType'. data EventTypeCategory = EventTypeCategoryUndefined | EventTypeCategoryData@@ -783,3 +901,56 @@ } deriving (Eq, Show, Generic) deriveJSON nakadiJsonOptions ''DataChangeEventEnriched++-- | Type modelling a "Business Event". Their JSON encodings are special since the payload+-- object is directly enriched with a @metadata@ field. "Data Change Events" on the other side+-- are JSON-encoded such that the complete event payload is contained in a seperate object field.+--+-- On the Haskell API side we split payload from meta data, which requires us to write custom+-- 'ToJSON' and 'FromJSON' implementations.++data BusinessEvent a = BusinessEvent+ { _payload :: a -- ^ Event Payload.+ , _metadata :: EventMetadata -- ^ Event Metadata.+ } deriving (Eq, Show, Generic)++instance FromJSON a => FromJSON (BusinessEvent a) where+ parseJSON = withObject "BusinessEvent" $ \obj ->+ BusinessEvent <$> parseJSON (Object obj) <*> obj .: "metadata"++instance ToJSON a => ToJSON (BusinessEvent a) where+ toJSON BusinessEvent {..} =+ case toJSON _payload of+ Object o ->+ -- If the JSON encoding of the event payload is an object, extend+ -- the object with metadata.+ Object (o <> metadata)+ _ ->+ -- Otherwise, produce an object that only contains metadata.+ Object metadata+ where metadata = HashMap.fromList [("metadata", toJSON _metadata)]++-- | Type modelling a Nakadi-enriched "Business Event". JSON encoding is basically the same as+-- for the non-enriched Business Events.+data BusinessEventEnriched a = BusinessEventEnriched+ { _payload :: a -- Cannot be named '_data', as this this would+ -- cause the lense 'data' to be created, which is a+ -- reserved keyword.+ , _metadata :: EventMetadataEnriched+ } deriving (Eq, Show, Generic)++instance FromJSON a => FromJSON (BusinessEventEnriched a) where+ parseJSON = withObject "BusinessEventEnriched" $ \obj ->+ BusinessEventEnriched <$> parseJSON (Object obj) <*> obj .: "metadata"++instance ToJSON a => ToJSON (BusinessEventEnriched a) where+ toJSON BusinessEventEnriched {..} =+ case toJSON _payload of+ Object o ->+ -- If the JSON encoding of the event payload is an object, extend+ -- the object with metadata.+ Object (o <> metadata)+ _ ->+ -- Otherwise, produce an object that only contains metadata.+ Object metadata+ where metadata = HashMap.fromList [("metadata", toJSON _metadata)]
src/Network/Nakadi/Internal/Types/Util.hs view
@@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Internal.Types.Util Description : Nakadi Client Utilities for Types (Internal)-Copyright : (c) Moritz Schulte 2017+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -14,9 +14,10 @@ import Data.Aeson import Data.Aeson.Types+import Data.Text ( Text ) import Data.Maybe-import Data.Scientific (toBoundedInteger)-import qualified Data.Text as Text+import Data.Scientific ( toBoundedInteger )+import qualified Data.Text as Text import Data.UUID import Prelude @@ -26,14 +27,18 @@ parseUUID :: String -> (UUID -> a) -> Value -> Parser a parseUUID label constructor val@(String valS) =- case Data.UUID.fromString (Text.unpack valS) of- Just uuid -> return (constructor uuid)- Nothing -> typeMismatch label val+ case Data.UUID.fromString (Text.unpack valS) of+ Just uuid -> return (constructor uuid)+ Nothing -> typeMismatch label val parseUUID label _ val = typeMismatch label val -parseInteger :: (Integral i, Bounded i) => String -> (i -> a) -> Value -> Parser a-parseInteger label constructor val@(Number n) =- case toBoundedInteger n of- Just i -> return $ constructor i- Nothing -> typeMismatch label val+parseInteger+ :: (Integral i, Bounded i) => String -> (i -> a) -> Value -> Parser a+parseInteger label constructor val@(Number n) = case toBoundedInteger n of+ Just i -> return $ constructor i+ Nothing -> typeMismatch label val parseInteger label _ val = typeMismatch label val++parseString :: String -> (Text -> a) -> Value -> Parser a+parseString _label ctor (String s) = pure (ctor s)+parseString label _ val = typeMismatch label val
+ src/Network/Nakadi/Internal/Types/Worker.hs view
@@ -0,0 +1,25 @@+module Network.Nakadi.Internal.Types.Worker where++import Network.Nakadi.Internal.Prelude++import UnliftIO.Async+import UnliftIO.STM++import Data.HashMap.Strict (HashMap)+import Data.List.NonEmpty (NonEmpty)++import Network.Nakadi.Internal.Types.Service++-- | Data type denoting an asynchronous worker.+data Worker a = Worker { _queue :: TBQueue (SubscriptionEventStreamBatch a)+ , _async :: Async ()+ }++-- | Data type containing a non-empty list of worker references.+data WorkerRegistry a =+ WorkerRegistry { _workers :: NonEmpty (Worker a)+ , _partitionIndexMap :: PartitionIndexMap }++-- | Map used for mapping subscription batch cursors to worked+-- indices.+type PartitionIndexMap = HashMap (PartitionName, EventTypeName) Int
+ src/Network/Nakadi/Internal/Unsafe/IO.hs view
@@ -0,0 +1,52 @@+{-|+Module : Network.Nakadi.Unsafe.IO+Description : Support for MonadNakadi IO Instance+Copyright : (c) Moritz Clasmeier 2018+License : BSD3+Maintainer : mtesseract@silverratio.net+Stability : experimental+Portability : POSIX++Nakadi-client provides a 'MonadNakadi' instance for the 'IO' monad+accessing a global mutable configuration.++This module implements the functionality for managing this global+configuration.+-}++{-# LANGUAGE MultiParamTypeClasses #-}++module Network.Nakadi.Internal.Unsafe.IO where++import UnliftIO.STM++import Network.Nakadi.Internal.Types+import Network.Nakadi.Internal.Prelude+import Network.Nakadi.Internal.Config++import Network.Nakadi.Internal.GlobalConfig++-- | Initialize the global configuration used by the 'MonadNakadi' instance+-- for the 'IO' monad via 'newConfigFromEnv'.+--+-- Experimental API.+initializeGlobalConfigurationFromEnv :: (MonadIO m, MonadThrow m) => m ()+initializeGlobalConfigurationFromEnv = do+ config <- newConfigFromEnv+ atomically $ writeTVar globalConfiguration (Just config)++-- | Sets the global configuration used by the 'MonadNakadi' instance+-- for the 'IO' monad to the provided configuration.+--+-- Experimental API.+setGlobalConfiguration :: MonadIO m => ConfigIO -> m ()+setGlobalConfiguration = atomically . writeTVar globalConfiguration . Just++-- | Modifies the global configuration used by the 'MonadNakadi' instance+-- for the 'IO' monad using the provided function. If no global configuration has+-- been set so far (using 'initializeGlobalConfigurationFromEnv' or+-- 'setGlobalConfiguration'), this will throw a 'ConfigurationMissing' exception.+--+-- Experimental API.+modifyGlobalConfiguration :: MonadIO m => (ConfigIO -> ConfigIO) -> m ()+modifyGlobalConfiguration f = atomically $ modifyTVar' globalConfiguration (fmap f)
src/Network/Nakadi/Internal/Util.hs view
@@ -18,38 +18,39 @@ , decodeThrow , sequenceSnd , extractQueryParametersFromPath- ) where+ , encodeStrict+ )+where import Network.Nakadi.Internal.Prelude -import Conduit hiding (throwM)+import Conduit hiding ( throwM ) import Data.Aeson-import qualified Data.ByteString.Lazy as ByteString.Lazy-import qualified Data.Text as Text+import qualified Data.ByteString.Lazy as ByteString.Lazy+import qualified Data.Text as Text import Network.HTTP.Simple import Network.Nakadi.Internal.Types -conduitDrainToLazyByteString ::- Monad b- => ConduitM () ByteString b ()- -> b ByteString.Lazy.ByteString-conduitDrainToLazyByteString conduit =- runConduit $ conduit .| sinkLazy+conduitDrainToLazyByteString+ :: Monad b => ConduitM () ByteString b () -> b ByteString.Lazy.ByteString+conduitDrainToLazyByteString conduit = runConduit $ conduit .| sinkLazy -decodeThrow- :: (FromJSON a, MonadThrow m)- => ByteString.Lazy.ByteString- -> m a+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))+ 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+sequenceSnd (a, fb) = (a, ) <$> fb extractQueryParametersFromPath :: String -> Maybe [(ByteString, ByteString)] extractQueryParametersFromPath path = case parseRequest ("http://localhost" <> path) of- Just req -> Just . catMaybes . map sequenceSnd . getRequestQueryString $ req+ Just req ->+ Just . mapMaybe sequenceSnd . getRequestQueryString $ req Nothing -> Nothing++encodeStrict :: ToJSON a => a -> ByteString+encodeStrict = ByteString.Lazy.toStrict . encode
+ src/Network/Nakadi/Internal/Worker.hs view
@@ -0,0 +1,199 @@+{-|+Module : Network.Nakadi.Internal.Worker+Description : Implementation of Subscription Consumption Workers+Copyright : (c) Moritz Clasmeier 2018+License : BSD3+Maintainer : mtesseract@silverratio.net+Stability : experimental+Portability : POSIX++This internal module implements dispatching of batches to workers.+-}++{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}++module Network.Nakadi.Internal.Worker+ ( WorkerRegistry+ , spawnWorkers+ , workerDispatchSink+ , workersWait+ )+where++import Network.Nakadi.Internal.Prelude++import Conduit+import Control.Lens+import qualified Control.Monad.Trans.Resource as Resource+import Data.Aeson+import qualified Data.HashMap.Strict as HashMap+import Data.List.NonEmpty ( NonEmpty(..) )+import qualified Data.List.NonEmpty as NonEmpty+import qualified Data.Vector as Vector+import Network.Nakadi.EventTypes.Partitions+import Network.Nakadi.Subscriptions.Subscription++import Network.Nakadi.Internal.Committer+import qualified Network.Nakadi.Internal.Lenses+ as L+import Network.Nakadi.Internal.Types+import UnliftIO.Async+import UnliftIO.STM++-- | Spawn a single worker.+spawnWorker+ :: ( MonadNakadi b m+ , MonadResource m+ , MonadUnliftIO m+ , MonadMask m+ , MonadIO m+ , FromJSON a+ , batch ~ (SubscriptionEventStreamBatch a)+ )+ => SubscriptionEventStream+ -> ConduitM batch batch m ()+ -> m (Worker a)+spawnWorker eventStream processor = do+ u <- askUnliftIO+ workerQueue <- atomically $ newTBQueue 1024+ let workerIO = unliftIO u (subscriptionWorker processor eventStream workerQueue)+ (_, workerAsync) <- Resource.allocate (async workerIO) cancel+ pure Worker {_queue = workerQueue, _async = workerAsync}++-- | Spawn multiple workers and return a 'WorkerRegistry'. It is guaranteed that the worker+-- registry contains at least one worker.+spawnWorkers+ :: ( MonadNakadi b m+ , MonadUnliftIO m+ , MonadMask m+ , MonadResource m+ , FromJSON a+ , batch ~ SubscriptionEventStreamBatch a+ )+ => SubscriptionId+ -> SubscriptionEventStream+ -> Int+ -> ConduitM batch batch m ()+ -> m (WorkerRegistry a)+spawnWorkers subscriptionId eventStream nWorkers processor = do+ let workerIndices = fromMaybe (1 :| []) (NonEmpty.nonEmpty [1 .. nWorkers])+ workers <- forM workerIndices (\_idx -> spawnWorker eventStream processor)+ partitionIndexMap <- retrievePartitionIndexMap subscriptionId+ pure WorkerRegistry {_workers = workers, _partitionIndexMap = partitionIndexMap}++-- | This function processes a subscription, taking care of applying+-- the configured committing strategy.+subscriptionWorker+ :: ( MonadNakadi b m+ , MonadUnliftIO m+ , MonadResource m+ , MonadMask m+ , FromJSON a+ , batch ~ (SubscriptionEventStreamBatch a)+ )+ => ConduitM batch batch m () -- ^ User provided Conduit for stream.+ -> SubscriptionEventStream+ -> TBQueue batch -- ^ Streaming response from Nakadi+ -> m ()+subscriptionWorker processor eventStream queue = do+ config <- nakadiAsk+ -- This @producer@ is a Conduit producing subscription batches that have been successfully+ -- processed from the user provided callback @processor@.+ --+ -- What remains to be done with the batches produced by the producer is committing the+ -- corresponding cursors.+ let producer = repeatMC (atomically (readTBQueue queue)) .| processor+ -- For committing we distinguish between two distinct strategies: synchronous and+ -- asynchronous comitting.+ case config ^. L.commitStrategy of+ CommitSync ->+ -- Synchronous case: Simply use a Conduit sink that commits+ -- every cursor.+ --+ -- Run the Conduit, which reads batches from the queue, processes them+ -- and commits their cursors.+ runConduit $ producer .| subscriptionSink eventStream+ CommitAsync bufferingStrategy -> do+ -- Asynchronous case: Create a new queue and spawn a cursor+ -- committer thread depending on the configured commit buffering+ -- method. Then execute the provided Conduit processor with a+ -- sink that sends cursor information to the queue. The cursor+ -- committer thread reads from this queue and processes the+ -- cursors.+ --+ -- Run the Conduit, which reads batches from the queue, processes them+ -- and sends their cursors to the cursor committer thread implementing+ -- the actual cursor committing logic.+ commitQueue <- liftIO . atomically $ newTBQueue asyncCommitQueueBufferSize+ withAsync (subscriptionCommitter bufferingStrategy eventStream commitQueue) $ \asyncHandle ->+ do+ -- This makes sure that if the cursor committing thread dies because+ -- of an exception, this exception will be re-raised in the current thread.+ link asyncHandle+ runConduit $ producer .| mapM_C (sendToQueue commitQueue)+ where+ sendToQueue commitQueue batch = liftIO . atomically $ do+ let cursor = batch ^. L.cursor+ events = fromMaybe Vector.empty (batch ^. L.events)+ nEvents = length events+ writeTBQueue commitQueue (nEvents, cursor)++ asyncCommitQueueBufferSize = 1024++-- | Retrieve the 'PartitionIndexMap' for the given subscription. This map is used for mapping+-- per-event type partitions to worker indices. Given a pair consisting of an event type and+-- a partition ID, the worker index references the worker in the worker registry responsible+-- for processing batches originating from that partition.+retrievePartitionIndexMap :: MonadNakadi b m => SubscriptionId -> m PartitionIndexMap+retrievePartitionIndexMap subscriptionId = do+ eventTypes <- (view L.eventTypes) <$> subscriptionGet subscriptionId+ eventTypesWithPartition <- concat <$> forM eventTypes extractPartitionsForEventType+ pure . HashMap.fromList $ zip eventTypesWithPartition [0 ..]+ where+ extractPartitionsForEventType eventType = do+ partitions <- map (view L.partition) <$> eventTypePartitions eventType+ pure (zip partitions (repeat eventType))++-- | Conduit sink which dispatches a batch to a worker contained in+-- the registry.+workerDispatchSink+ :: (MonadIO m) => WorkerRegistry a -> ConduitM (SubscriptionEventStreamBatch a) Void m ()+workerDispatchSink registry = awaitForever $ \batch -> do+ let partition = batch ^. L.cursor . L.partition+ eventType = batch ^. L.cursor . L.eventType+ worker = pickWorker registry eventType partition+ atomically $ writeTBQueue (_queue worker) batch++-- | Given a 'SubscriptionEventStreamBatch', produce the worker that+-- should handle the batch. The worker is found using the 'PartitionIndexMap'.+pickWorker :: WorkerRegistry a -> EventTypeName -> PartitionName -> Worker a+pickWorker registry eventType partition =+ let workers = registry ^. L.workers+ nWorkers = NonEmpty.length workers+ in case HashMap.lookup (partition, eventType) (registry ^. L.partitionIndexMap) of+ Nothing ->+ -- We failed to find an entry in the PartitionIndexMap for that partition.+ -- This should rarely happen.+ -- It could, for example, happen if the number of partitions of an event+ -- type belonging to the subscription to be consumed was increased after+ -- subscription consumption has started.+ --+ -- In the future, we could improve this fallback mechanism, if there is a+ -- need to do so. At the moment we simply use the first worker in case of+ -- lookup failures.+ NonEmpty.head workers+ Just idx ->+ -- Lookup successful. Truncate the resulting index using modulo in order to+ -- obtain a worker index and return the corresponding worker reference.+ workers NonEmpty.!! (idx `mod` nWorkers)++-- | Block as long no worker has finished. Workers are supposed to run forever, unless+-- cancelled. Therefore, using 'waitAnyCancel' essentially means that if some worker+-- fails due to an uncaught exception, then all other workers are cancelled as well.+workersWait :: MonadIO m => WorkerRegistry a -> m ()+workersWait registry = do+ let workerHandles = map _async $ NonEmpty.toList (registry ^. L.workers)+ void . waitAnyCancel $ workerHandles
src/Network/Nakadi/Lenses.hs view
@@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Lenses Description : Nakadi Client Library Lenses-Copyright : (c) Moritz Schulte 2017+Copyright : (c) Moritz Clasmeier 2017 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -24,117 +24,156 @@ , subscriptionsList' , subscriptionsSource , subscriptionsList- ) where+ , withSubscription+ , withTemporarySubscription+ )+where import Network.Nakadi.Internal.Prelude import Conduit-import qualified Control.Exception.Safe as Safe+import qualified Control.Exception.Safe as Safe import Control.Lens-import qualified Data.Text as Text+import qualified Data.Text as Text import Network.Nakadi.Internal.Http-import qualified Network.Nakadi.Internal.Lenses as L+import qualified Network.Nakadi.Internal.Lenses+ as L import Network.Nakadi.Internal.Util import Network.Nakadi.Subscriptions.Cursors import Network.Nakadi.Subscriptions.Events import Network.Nakadi.Subscriptions.Stats import Network.Nakadi.Subscriptions.Subscription+import qualified Data.Set as Set+import Data.Set ( Set ) path :: ByteString path = "/subscriptions" -- | @POST@ to @\/subscriptions@. Creates a new subscription. Low -- level interface.-subscriptionCreate' ::- MonadNakadi b m- => Subscription- -> m Subscription-subscriptionCreate' subscription =- httpJsonBody status201 [(ok200, errorSubscriptionExistsAlready)]- (setRequestMethod "POST"- . setRequestPath path- . setRequestBodyJSON subscription)+subscriptionCreate' :: MonadNakadi b m => SubscriptionRequest -> m 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 b m, MonadCatch m)- => Subscription- -> m Subscription-subscriptionCreate subscription = do- Safe.catchJust exceptionPredicate (subscriptionCreate' subscription) return-- where exceptionPredicate (SubscriptionExistsAlready s) = Just s- exceptionPredicate _ = Nothing+subscriptionCreate :: (MonadNakadi b m, MonadCatch m) => SubscriptionRequest -> m Subscription+subscriptionCreate subscription = Safe.catchJust exceptionPredicate+ (subscriptionCreate' subscription)+ return+ where+ exceptionPredicate (SubscriptionExistsAlready s) = Just s+ exceptionPredicate _ = Nothing -- | @GET@ to @\/subscriptions@. Internal low-level interface.-subscriptionsGet ::- MonadNakadi b m- => [(ByteString, ByteString)]- -> m SubscriptionsListResponse-subscriptionsGet queryParameters =- httpJsonBody ok200 []- (setRequestMethod "GET"- . setRequestPath path- . setRequestQueryParameters queryParameters)+subscriptionsGet :: MonadNakadi b m => [(ByteString, ByteString)] -> m SubscriptionsListResponse+subscriptionsGet queryParameters = httpJsonBody+ ok200+ []+ (setRequestMethod "GET" . setRequestPath path . setRequestQueryParameters queryParameters) -buildQueryParameters :: Maybe ApplicationName- -> Maybe [EventTypeName]- -> Maybe Limit- -> Maybe Offset- -> [(ByteString, ByteString)]+buildQueryParameters+ :: Maybe ApplicationName+ -> Maybe [EventTypeName]+ -> Maybe Limit+ -> Maybe Offset+ -> [(ByteString, ByteString)] buildQueryParameters maybeOwningApp maybeEventTypeNames maybeLimit maybeOffset =- catMaybes $- [ ("owning_application",) . encodeUtf8 . unApplicationName <$> maybeOwningApp- , ("limit",) . encodeUtf8 . tshow <$> maybeLimit- , ("offset",) . encodeUtf8 . tshow <$> maybeOffset ]- ++ case maybeEventTypeNames of- Just eventTypeNames -> map (Just . ("event_type",) . encodeUtf8 . unEventTypeName) eventTypeNames- Nothing -> []+ catMaybes+ $ [ ("owning_application", ) . encodeUtf8 . unApplicationName <$> maybeOwningApp+ , ("limit", ) . encodeUtf8 . tshow <$> maybeLimit+ , ("offset", ) . encodeUtf8 . tshow <$> maybeOffset+ ]+ ++ case maybeEventTypeNames of+ Just eventTypeNames ->+ map (Just . ("event_type", ) . encodeUtf8 . unEventTypeName) eventTypeNames+ Nothing -> [] -- | @GET@ to @\/subscriptions@. Retrieves all subscriptions matching -- the provided filter criteria. Low-level interface using pagination.-subscriptionsList' ::- (MonadNakadi b m)+subscriptionsList'+ :: (MonadNakadi b m) => Maybe ApplicationName -> Maybe [EventTypeName] -> Maybe Limit -> Maybe Offset -> m SubscriptionsListResponse-subscriptionsList' maybeOwningApp maybeEventTypeNames maybeLimit maybeOffset = do- subscriptionsGet queryParameters- where queryParameters =- buildQueryParameters maybeOwningApp maybeEventTypeNames maybeLimit maybeOffset+subscriptionsList' maybeOwningApp maybeEventTypeNames maybeLimit maybeOffset = 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 b m)- => Maybe ApplicationName- -> Maybe [EventTypeName]- -> m (ConduitM () [Subscription] m ())-subscriptionsSource maybeOwningApp maybeEventTypeNames =- pure $ nextPage initialQueryParameters-- where nextPage queryParameters = do- resp <- lift $ subscriptionsGet queryParameters- yield (resp^.L.items)- let maybeNextPath = Text.unpack . (view L.href) <$> (resp^.L.links.L.next)- case maybeNextPath >>= extractQueryParametersFromPath of- Just nextQueryParameters -> do- nextPage nextQueryParameters- Nothing ->- return ()+subscriptionsSource+ :: (MonadNakadi b m)+ => Maybe ApplicationName+ -> Maybe [EventTypeName]+ -> m (ConduitM () [Subscription] m ())+subscriptionsSource maybeOwningApp maybeEventTypeNames = pure $ nextPage initialQueryParameters+ where+ nextPage queryParameters = do+ resp <- lift $ subscriptionsGet queryParameters+ yield (resp ^. L.items)+ let maybeNextPath = Text.unpack . view L.href <$> (resp ^. L.links . L.next)+ forM_ (maybeNextPath >>= extractQueryParametersFromPath) nextPage - initialQueryParameters =- buildQueryParameters maybeOwningApp maybeEventTypeNames Nothing Nothing+ initialQueryParameters = buildQueryParameters maybeOwningApp maybeEventTypeNames Nothing Nothing -- | @GET@ to @\/subscriptions@. Retrieves all subscriptions matching -- the provided filter criteria. High-level list interface.-subscriptionsList ::- MonadNakadi b m- => Maybe ApplicationName- -> Maybe [EventTypeName]- -> m [Subscription]+subscriptionsList+ :: MonadNakadi b m => Maybe ApplicationName -> Maybe [EventTypeName] -> m [Subscription] subscriptionsList maybeOwningApp maybeEventTypeNames = do source <- subscriptionsSource maybeOwningApp maybeEventTypeNames runConduit $ source .| concatC .| sinkList++-- | Experimental API.+--+-- Creates a new temporary subscription using the specified parameters via `subscriptionCreate`+-- and pass it to the provided action. Note that `bracket` is used to enforce the deletion of+-- the subscription via `subscriptionDelete` when the provided action returns.+--+-- Do NOT use this function if the specified subscription should not be deleted.+withTemporarySubscription+ :: (MonadNakadi b m, MonadMask m)+ => ApplicationName+ -> ConsumerGroup+ -> Set EventTypeName+ -> SubscriptionPosition+ -> (Subscription -> m r)+ -> m r+withTemporarySubscription owningApp consumerGroup eventTypeNames subscriptionPosition = bracket+ (subscriptionCreate subscriptionRequest)+ (subscriptionDelete . view L.id)+ where+ subscriptionRequest = SubscriptionRequest+ { _owningApplication = owningApp+ , _eventTypes = Set.toList eventTypeNames+ , _consumerGroup = Just consumerGroup+ , _subscriptionPosition = Just subscriptionPosition+ }++-- | Experimental API.+--+-- Creates a new subscription using the specified parameters via `subscriptionCreate`+-- and pass it to the provided action.+withSubscription+ :: (MonadNakadi b m, MonadMask m)+ => ApplicationName+ -> ConsumerGroup+ -> Set EventTypeName+ -> SubscriptionPosition+ -> (Subscription -> m r)+ -> m r+withSubscription owningApp consumerGroup eventTypeNames subscriptionPosition f =+ subscriptionCreate subscriptionRequest >>= f+ where+ subscriptionRequest = SubscriptionRequest+ { _owningApplication = owningApp+ , _eventTypes = Set.toList eventTypeNames+ , _consumerGroup = Just consumerGroup+ , _subscriptionPosition = Just subscriptionPosition+ }
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -23,81 +23,80 @@ , subscriptionCursorCommit , subscriptionCursors , subscriptionCursorsReset- ) where+ )+where import Network.Nakadi.Internal.Prelude import Data.Aeson -import qualified Control.Exception.Safe as Safe+import qualified Control.Exception.Safe as Safe import Control.Lens-import qualified Data.HashMap.Lazy as HashMap+import qualified Data.HashMap.Lazy as HashMap 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.Internal.Lenses ( HasNakadiSubscriptionCursor )+import qualified Network.Nakadi.Internal.Lenses+ as L path :: SubscriptionId -> ByteString-path subscriptionId =- "/subscriptions/"- <> subscriptionIdToByteString subscriptionId- <> "/cursors"+path subscriptionId = "/subscriptions/" <> subscriptionIdToByteString subscriptionId <> "/cursors" -- | @POST@ to @\/subscriptions\/SUBSCRIPTION-ID\/cursors@. Commits -- cursors using low level interface.-subscriptionCursorCommit' ::- MonadNakadi b m+subscriptionCursorCommit'+ :: MonadNakadi b m => SubscriptionId -- ^ Subsciption ID -> StreamId -- ^ Stream ID -> SubscriptionCursorCommit -- ^ Subscription Cursor to commit -> m ()-subscriptionCursorCommit' subscriptionId streamId cursors =- httpJsonNoBody status204 [(ok200, errorCursorAlreadyCommitted)]- (setRequestMethod "POST"- . addRequestHeader "X-Nakadi-StreamId" (encodeUtf8 (unStreamId streamId))- . setRequestBodyJSON cursors- . setRequestPath (path subscriptionId))+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\/cursors@. Commits -- cursors using high level interface.-subscriptionCursorCommit ::- (MonadNakadi b m, MonadCatch m, HasNakadiSubscriptionCursor a)+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' _subscriptionId _streamId cursorsCommit)- (const (return ()))-- where exceptionPredicate = \case- CursorAlreadyCommitted _ -> Just ()- _ -> Nothing+subscriptionCursorCommit SubscriptionEventStream {..} as = Safe.catchJust+ exceptionPredicate+ (subscriptionCursorCommit' _subscriptionId _streamId cursorsCommit)+ (const (return ()))+ where+ exceptionPredicate = \case+ CursorAlreadyCommitted _ -> Just ()+ _ -> Nothing - cursors = map (^. L.subscriptionCursor) as- cursorsCommit = SubscriptionCursorCommit cursors+ cursors = map (^. L.subscriptionCursor) as+ cursorsCommit = SubscriptionCursorCommit cursors -- | @GET@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. Retrieves -- subscriptions cursors.-subscriptionCursors ::- MonadNakadi b m+subscriptionCursors+ :: MonadNakadi b m => SubscriptionId -- ^ Subscription ID -> m [SubscriptionCursor] -- ^ Subscription Cursors for the specified Subscription subscriptionCursors subscriptionId =- httpJsonBody ok200 []- (setRequestMethod "GET" . setRequestPath (path subscriptionId))+ httpJsonBody ok200 [] (setRequestMethod "GET" . setRequestPath (path subscriptionId)) -- | @PATCH@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. Resets -- subscriptions cursors.-subscriptionCursorsReset ::- MonadNakadi b m+subscriptionCursorsReset+ :: MonadNakadi b m => SubscriptionId -- ^ Subscription ID -> [SubscriptionCursorWithoutToken] -- ^ Subscription Cursors to reset -> m ()-subscriptionCursorsReset subscriptionId cursors =- httpJsonNoBody status204 [ (status404, errorSubscriptionNotFound)- , (status409, errorCursorResetInProgress) ]- (setRequestMethod "PATCH"- . setRequestPath (path subscriptionId)- . setRequestBodyJSON (Object (HashMap.fromList [("items", toJSON cursors)])))+subscriptionCursorsReset subscriptionId cursors = httpJsonNoBody+ status204+ [(status404, errorSubscriptionNotFound), (status409, errorCursorResetInProgress)]+ (setRequestMethod "PATCH" . setRequestPath (path subscriptionId) . setRequestBodyJSON+ (Object (HashMap.fromList [("items", toJSON cursors)]))+ )
src/Network/Nakadi/Subscriptions/Events.hs view
@@ -16,57 +16,69 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE RecordWildCards #-} module Network.Nakadi.Subscriptions.Events ( subscriptionProcessConduit , subscriptionProcess- ) where+ , subscriptionSource+ , subscriptionSourceEvents+ )+where import Network.Nakadi.Internal.Prelude -import Conduit hiding (throwM)-import qualified Control.Concurrent.Async.Timer as Timer-import Control.Concurrent.STM (TBQueue, TVar,- atomically, modifyTVar,- newTBQueue, newTVar,- readTBQueue, readTVar,- retry, swapTVar,- writeTBQueue)+import Control.Concurrent.STM.TBMQueue+ ( TBMQueue+ , newTBMQueue+ , writeTBMQueue+ , closeTBMQueue+ , readTBMQueue+ )+import UnliftIO.STM ( atomically )++import Conduit hiding ( throwM ) import Control.Lens-import Control.Monad.Logger import Data.Aeson-import qualified Data.Conduit.List as Conduit (mapM_)-import Data.HashMap.Strict (HashMap)-import qualified Data.HashMap.Strict as HashMap-import qualified Data.Vector as Vector-import Network.HTTP.Client (responseBody)+import Control.Monad.Trans.Resource ( allocate )+import Network.HTTP.Client ( responseBody ) import Network.HTTP.Simple import Network.HTTP.Types import Network.Nakadi.Internal.Config import Network.Nakadi.Internal.Conversions import Network.Nakadi.Internal.Http-import qualified Network.Nakadi.Internal.Lenses as L-import Network.Nakadi.Subscriptions.Cursors+import qualified Network.Nakadi.Internal.Lenses+ as L+import Network.Nakadi.Internal.Worker import UnliftIO.Async +-- For dispatching batches to workers, we maintain an integer-indexed+-- (non-empty) list of workers. Thus, we need have a way for mapping a+-- subscription batch, more precisely the cursors belonging to a+-- subscription batch, to some worker index. Mapping a cursor to an+-- Integer is sufficient, as we can simply take the reminder modulo+-- number of workers in our workers list.+--+-- Nakadi subscription cursors contain a partition reference and an+-- event type name — both given as strings. How do we derive an+-- integer from this data? The idea is to built a 'PartitionIndexMap'+-- upfrom, which allows us to establish this mapping. It is expected+-- that this map contains an entry for every valid @(PartitionName,+-- EventTypeName)@ combination for the subscription to be consumed.+-- If, for some reason, we receive cursors not contained in this map,+-- we map it to the zero index.+ -- | Consumes the specified subscription using the commit strategy -- contained in the configuration. Each consumed batch of subscription -- events is provided to the provided batch processor action. If this -- action throws an exception, subscription consumption will terminate. subscriptionProcess- :: ( MonadNakadi b m- , MonadUnliftIO m- , MonadMask m- , FromJSON a )- => Maybe ConsumeParameters -- ^ 'ConsumeParameters'- -- to use- -> SubscriptionId -- ^ Subscription to consume+ :: (MonadNakadi b m, MonadUnliftIO m, MonadMask m, MonadResource m, FromJSON a)+ => SubscriptionId -- ^ Subscription to consume -> (SubscriptionEventStreamBatch a -> m ()) -- ^ Batch processor action -> m ()-subscriptionProcess maybeConsumeParameters subscriptionId processor =- subscriptionProcessConduit maybeConsumeParameters subscriptionId conduit+subscriptionProcess subscriptionId processor = subscriptionProcessConduit subscriptionId conduit where conduit = iterMC processor -- | Conduit based interface for subscription consumption. Consumes@@ -77,260 +89,109 @@ subscriptionProcessConduit :: ( MonadNakadi b m , MonadUnliftIO m+ , MonadResource m , MonadMask m , FromJSON a- , batch ~ SubscriptionEventStreamBatch a )- => Maybe ConsumeParameters -- ^ 'ConsumeParameters' to use- -> SubscriptionId -- ^ Subscription to consume+ , batch ~ SubscriptionEventStreamBatch a+ )+ => SubscriptionId -- ^ Subscription to consume -> ConduitM batch batch m () -- ^ Conduit processor. -> m ()-subscriptionProcessConduit maybeConsumeParameters subscriptionId processor = do+subscriptionProcessConduit subscriptionId processor = do config <- nakadiAsk- let consumeParams = fromMaybe defaultConsumeParameters maybeConsumeParameters- queryParams = buildSubscriptionConsumeQueryParameters consumeParams- httpJsonBodyStream ok200 [(status404, errorSubscriptionNotFound)]- (includeFlowId config- . setRequestPath path- . setRequestQueryParameters queryParams) $- subscriptionProcessHandler consumeParams subscriptionId processor-- where path = "/subscriptions/"- <> subscriptionIdToByteString subscriptionId- <> "/events"+ let queryParams = buildConsumeQueryParameters config+ httpJsonBodyStream+ ok200+ [(status404, errorSubscriptionNotFound)]+ (includeFlowId config . setRequestPath path . setRequestQueryParameters queryParams)+ $ subscriptionProcessHandler subscriptionId processor+ where path = "/subscriptions/" <> subscriptionIdToByteString subscriptionId <> "/events" -- | Derive a 'SubscriptionEventStream' from the provided -- 'SubscriptionId' and Nakadi streaming response. buildSubscriptionEventStream- :: MonadThrow m- => SubscriptionId- -> Response a- -> m SubscriptionEventStream+ :: MonadThrow m => SubscriptionId -> Response a -> m SubscriptionEventStream buildSubscriptionEventStream subscriptionId response = case listToMaybe (getResponseHeader "X-Nakadi-StreamId" response) of- Just streamId ->- pure SubscriptionEventStream+ Just streamId -> pure SubscriptionEventStream { _streamId = StreamId (decodeUtf8 streamId)- , _subscriptionId = subscriptionId }- Nothing ->- throwM StreamIdMissing+ , _subscriptionId = subscriptionId+ }+ Nothing -> throwM StreamIdMissing --- | This function processes a subscription, taking care of applying--- the configured committing strategy.+-- | This function processes a subscription, taking care of+-- dispatching to worker threads and applying the configured+-- committing strategy. subscriptionProcessHandler- :: ( MonadNakadi b m+ :: forall a b m batch+ . ( MonadNakadi b m , MonadUnliftIO m+ , MonadResource m , MonadMask m , FromJSON a- , batch ~ (SubscriptionEventStreamBatch a) )- => ConsumeParameters- -> SubscriptionId -- ^ Subscription ID required for committing.+ , batch ~ (SubscriptionEventStreamBatch a)+ )+ => SubscriptionId -- ^ Subscription ID required for committing. -> ConduitM batch batch m () -- ^ User provided Conduit for stream. -> Response (ConduitM () ByteString m ()) -- ^ Streaming response from Nakadi -> m ()-subscriptionProcessHandler consumeParams subscriptionId processor response = do- config <- nakadiAsk- eventStream <- buildSubscriptionEventStream subscriptionId response- let producer = responseBody response- .| linesUnboundedAsciiC- .| conduitDecode config- .| processor- case config^.L.commitStrategy of- CommitSync ->- -- Synchronous case: Simply use a Conduit sink that commits- -- every cursor.- runConduit $ producer .| subscriptionSink eventStream- CommitAsync bufferingStrategy -> do- -- Asynchronous case: Create a new queue and spawn a cursor- -- committer thread depending on the configured commit buffering- -- method. Then execute the provided Conduit processor with a- -- sink that sends cursor information to the queue. The cursor- -- committer thread reads from this queue and processes the- -- cursors.- queue <- liftIO . atomically $ newTBQueue 1024- withAsync (subscriptionCommitter bufferingStrategy consumeParams eventStream queue) $- \ asyncHandle -> do- link asyncHandle- runConduit $ producer .| Conduit.mapM_ (sendToQueue queue)-- where sendToQueue queue batch = liftIO . atomically $ do- let cursor = batch^.L.cursor- events = fromMaybe Vector.empty (batch^.L.events)- nEvents = length events- writeTBQueue queue (nEvents, cursor)---- | Sink which can be used as sink for Conduits processing--- subscriptions events. This sink takes care of committing events. It--- can consume any values which contain Subscription Cursors.-subscriptionSink- :: (MonadIO m, MonadNakadi b m)- => SubscriptionEventStream- -> ConduitM (SubscriptionEventStreamBatch a) void m ()-subscriptionSink eventStream = do- config <- lift nakadiAsk- awaitForever $ \ batch -> lift $ do- let cursor = batch^.L.cursor- catchAny (commitOneCursor eventStream cursor) $ \ exn -> nakadiLiftBase $- case config^.L.logFunc of- Just logFunc -> logFunc "nakadi-client" LevelWarn $ toLogStr $- "Failed to synchronously commit cursor: " <> tshow exn- Nothing ->- pure ()--type CursorsMap = HashMap (EventTypeName, PartitionName) (Int, SubscriptionCursor)--emptyCursorsMap :: CursorsMap-emptyCursorsMap = HashMap.empty--cursorKey :: SubscriptionCursor -> (EventTypeName, PartitionName)-cursorKey cursor = (cursor^.L.eventType, cursor^.L.partition)---- | Implementation for the 'CommitNoBuffer' strategy: We simply read--- every cursor and commit it in order.-unbufferedCommitLoop- :: (MonadNakadi b m, MonadIO m)- => SubscriptionEventStream- -> TBQueue (Int, SubscriptionCursor)- -> m ()-unbufferedCommitLoop eventStream queue = do+subscriptionProcessHandler subscriptionId processor response = do config <- nakadiAsk- forever $ do- (_nEvents, cursor) <- liftIO . atomically . readTBQueue $ queue- catchAny (subscriptionCursorCommit eventStream [cursor]) $ \ exn -> do- nakadiLiftBase $- case config^.L.logFunc of- Just logFunc -> logFunc "nakadi-client" LevelWarn $ toLogStr $- "Failed to commit cursor " <> tshow cursor <> ": " <> tshow exn- Nothing ->- pure ()--cursorBufferSize :: ConsumeParameters -> Int-cursorBufferSize consumeParams =- case consumeParams^.L.maxUncommittedEvents of- Nothing -> 1- Just n -> n- & fromIntegral- & (* safetyFactor)- & round-- where safetyFactor = 0.5---- | Main function for the cursor committer thread. Logic depends on--- the provided buffering strategy.-subscriptionCommitter- :: ( MonadNakadi b m- , MonadUnliftIO m- , MonadMask m )- => CommitBufferingStrategy- -> ConsumeParameters- -> SubscriptionEventStream- -> TBQueue (Int, SubscriptionCursor)- -> m ()---- | Implementation for the 'CommitNoBuffer' strategy: We simply read--- every cursor and commit it in order.-subscriptionCommitter CommitNoBuffer _consumeParams eventStream queue =- unbufferedCommitLoop eventStream queue---- | Implementation of the 'CommitTimeBuffer' strategy: We use an--- async timer for committing cursors at specified intervals.-subscriptionCommitter (CommitTimeBuffer millis) _consumeParams eventStream queue = do- let timerConf = Timer.defaultConf- & Timer.setInitDelay (fromIntegral millis)- & Timer.setInterval (fromIntegral millis)- cursorsMap <- liftIO . atomically $ newTVar emptyCursorsMap- withAsync (cursorConsumer cursorsMap) $ \ asyncCursorConsumer -> do- link asyncCursorConsumer- Timer.withAsyncTimer timerConf $ \ timer -> forever $ do- Timer.wait timer- commitAllCursors eventStream cursorsMap-- where -- | The cursorsConsumer drains the cursors queue and adds each- -- cursor to the provided cursorsMap.- cursorConsumer cursorsMap = forever . liftIO . atomically $ do- (_, cursor) <- readTBQueue queue- modifyTVar cursorsMap (HashMap.insert (cursorKey cursor) (0, cursor))---- | Implementation of the 'CommitSmartBuffer' strategy: We use an--- async timer for committing cursors at specified intervals, but if--- the number of uncommitted events reaches some threshold before the--- next scheduled commit, a commit is being done right away and the--- timer is resetted.-subscriptionCommitter CommitSmartBuffer consumeParams eventStream queue = do- let millisDefault = 1000- nMaxEvents = cursorBufferSize consumeParams- timerConf = Timer.defaultConf- & Timer.setInitDelay (fromIntegral millisDefault)- & Timer.setInterval (fromIntegral millisDefault)- if nMaxEvents > 1- then do cursorsMap <- liftIO . atomically $ newTVar emptyCursorsMap- withAsync (cursorConsumer cursorsMap) $ \ asyncCursorConsumer -> do- link asyncCursorConsumer- Timer.withAsyncTimer timerConf $ cursorCommitter cursorsMap nMaxEvents- else unbufferedCommitLoop eventStream queue-- where -- | The cursorsConsumer drains the cursors queue and adds- -- each cursor to the provided cursorsMap.- cursorConsumer cursorsMap = forever . liftIO . atomically $ do- (nEvents, cursor) <- readTBQueue queue- modifyTVar cursorsMap $- HashMap.insertWith updateCursor (cursorKey cursor) (nEvents, cursor)-- -- | Adds the old number of events to the new entry in the- -- cursors map.- updateCursor cursorNew _cursorOld @ (nEventsOld, _) =- cursorNew & _1 %~ (+ nEventsOld)-- -- | Committer loop.- cursorCommitter cursorsMap nMaxEvents timer = forever $ do- race (Timer.wait timer) (maxEventsReached cursorsMap nMaxEvents) >>= \ case- Left _ ->- -- Timer has elapsed, simply commit all currently- -- buffered cursors.- commitAllCursors eventStream cursorsMap- Right _ -> do- -- Events processed since last cursor commit have- -- crosses configured threshold for at least one- -- partition. Commit cursors on all such partitions.- Timer.reset timer- commitAllCursors eventStream cursorsMap-- -- | Returns list of cursors that should be committed- -- considering the number of events processed on the- -- respective partition since the last commit. Blocks until at- -- least one such cursor is found.- maxEventsReached cursorsMap nMaxEvents = liftIO . atomically $ do- cursorsList <- HashMap.toList <$> readTVar cursorsMap- let cursorsCommit = filter (shouldBeCommitted nMaxEvents) cursorsList- if null cursorsCommit- then retry- else pure ()+ let nWorkers = config ^. L.worker . L.nThreads+ eventStream <- buildSubscriptionEventStream subscriptionId response+ workerRegistry <- spawnWorkers subscriptionId eventStream nWorkers processor+ race_ (workersWait workerRegistry)+ $ runConduit+ $ responseBody response+ .| linesUnboundedAsciiC+ .| conduitDecode+ .| mapC (identity :: batch -> batch)+ .| workerDispatchSink workerRegistry - -- | Returns True if the provided cursor should be committed.- shouldBeCommitted nMaxEvents cursor = cursor^._2._1 >= nMaxEvents+-- | Experimental API.+--+-- Creates a Conduit source from a subscription ID. The source will produce subscription+-- event stream batches. Note the batches will be asynchronously committed irregardless of+-- any event processing logic. Use this function only if the guarantees provided by Nakadi+-- for event processing and cursor committing are not required or not desired.+subscriptionSource+ :: forall a b m+ . (MonadNakadi b m, MonadUnliftIO m, MonadMask m, MonadResource m, FromJSON a)+ => SubscriptionId -- ^ Subscription to consume.+ -> ConduitM () (SubscriptionEventStreamBatch a) m () -- ^ Conduit source.+subscriptionSource subscriptionId = do+ UnliftIO {..} <- lift askUnliftIO+ streamLimit <- lift nakadiAsk <&> (fmap fromIntegral . view L.streamLimit)+ queue <- atomically $ newTBMQueue queueSize+ (_releaseKey, asyncHandle) <- allocate+ (unliftIO (async (subscriptionConsumer streamLimit queue)))+ cancel+ link asyncHandle+ drain queue+ where+ queueSize = 2048+ drain queue = atomically (readTBMQueue queue) >>= \case+ Just a -> yield a >> drain queue+ Nothing -> pure () --- | This function commits all cursors in the provided cursorsMap.-commitAllCursors- :: (MonadNakadi b m, MonadIO m)- => SubscriptionEventStream- -> TVar CursorsMap- -> m ()-commitAllCursors eventStream cursorsMap = do- cursors <- liftIO . atomically $ swapTVar cursorsMap emptyCursorsMap- forM_ cursors $ \ (_nEvents, cursor) -> commitOneCursor eventStream cursor+ subscriptionConsumer :: Maybe Int -> TBMQueue (SubscriptionEventStreamBatch a) -> m ()+ subscriptionConsumer maybeStreamLimit queue = go `finally` atomically (closeTBMQueue queue)+ where+ go = do+ subscriptionProcess subscriptionId (void . atomically . writeTBMQueue queue)+ -- We only reconnect automatically when no stream limit is set.+ -- This effectively means that we don't try to reach @streamLimit@ events exactly.+ -- We simply regard @streamLimit@ as an upper bound and in case Nakadi disconnects us+ -- earlier or the connection breaks, we produce less than @streamLimit@ events.+ when (isNothing maybeStreamLimit) go --- | This function takes care of committing a single cursor. Exceptions will be--- catched and logged, but the failure will NOT be propagated. This means that--- Nakadi itself is in control of disconnecting us.-commitOneCursor- :: (MonadIO m, MonadNakadi b m)- => SubscriptionEventStream- -> SubscriptionCursor- -> m ()-commitOneCursor eventStream cursor = do- config <- nakadiAsk- catchAny (subscriptionCursorCommit eventStream [cursor]) $ \ exn -> nakadiLiftBase $- case config^.L.logFunc of- Just logFunc -> logFunc "nakadi-client" LevelWarn $ toLogStr $- "Failed to commit cursor " <> tshow cursor <> ": " <> tshow exn- Nothing ->- pure ()+-- | Experimental API.+--+-- Similar to `subscriptionSource`, but the created Conduit source provides events instead+-- of event batches.+subscriptionSourceEvents+ :: (MonadNakadi b m, MonadUnliftIO m, MonadMask m, MonadResource m, FromJSON a)+ => SubscriptionId -- ^ Subscription to consume+ -> ConduitM () a m () -- ^ Conduit processor.+subscriptionSourceEvents subscriptionId =+ subscriptionSource subscriptionId .| concatMapC (\batch -> fromMaybe mempty (batch & _events))
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -18,39 +18,55 @@ module Network.Nakadi.Subscriptions.Stats ( subscriptionStats' , subscriptionStats- ) where+ )+where import Network.Nakadi.Internal.Prelude import Control.Lens-import qualified Data.Map.Strict as Map+import Data.Aeson+import qualified Data.Map.Strict as Map import Network.Nakadi.Internal.Conversions import Network.Nakadi.Internal.Http-import qualified Network.Nakadi.Internal.Lenses as L+import qualified Network.Nakadi.Internal.Lenses+ as L+import Network.Nakadi.Internal.Util path :: SubscriptionId -> ByteString path subscriptionId =- "/subscriptions/"- <> subscriptionIdToByteString subscriptionId- <> "/cursors"+ "/subscriptions/" <> subscriptionIdToByteString subscriptionId <> "/stats" --- | @GET@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. Low level+-- | @GET@ to @\/subscriptions\/SUBSCRIPTION\/stats@. Low level -- interface for Subscriptions Statistics retrieval.-subscriptionStats' ::- MonadNakadi b m- => SubscriptionId -- ^ Subscription ID- -> m SubscriptionEventTypeStatsResult -- ^ Subscription Statistics-subscriptionStats' subscriptionId =- httpJsonBody ok200 [(status404, errorSubscriptionNotFound)]- (setRequestMethod "GET" . setRequestPath (path subscriptionId))+subscriptionStats'+ :: MonadNakadi b m+ => SubscriptionId -- ^ Subscription ID+ -> Bool -- ^ Whether to show time lag.+ -> m SubscriptionStats -- ^ Subscription Statistics+subscriptionStats' subscriptionId showTimeLag = httpJsonBody+ ok200+ [(status404, errorSubscriptionNotFound)]+ ( setRequestMethod "GET"+ . setRequestPath (path subscriptionId)+ . setRequestQueryParameters queryParameters+ )+ where queryParameters = [("show_time_lag", encodeStrict (Bool showTimeLag))] --- | @GET@ to @\/subscriptions\/SUBSCRIPTION\/cursors@. High level+-- | @GET@ to @\/subscriptions\/SUBSCRIPTION\/stats@. High level -- interface for Subscription Statistics retrieval.-subscriptionStats ::- MonadNakadi b m+subscriptionStats+ :: MonadNakadi b m => SubscriptionId -- ^ Subscription ID -> m (Map EventTypeName [PartitionStat]) -- ^ Subscription -- Statistics as a 'Map'. subscriptionStats subscriptionId = do- items <- subscriptionStats' subscriptionId <&> view L.items- return . Map.fromList . map (\SubscriptionEventTypeStats { .. } -> (_eventType, _partitions)) $ items+ subscriptionStatsConf <- nakadiAsk <&> (^. L.subscriptionStats)+ items <- subscriptionStats' subscriptionId (showTimeLag subscriptionStatsConf) <&> view L.items+ return+ . Map.fromList+ . map (\SubscriptionEventTypeStats {..} -> (_eventType, _partitions))+ $ items+ where+ showTimeLag :: Maybe SubscriptionStatsConf -> Bool+ showTimeLag Nothing = False+ showTimeLag (Just conf) = conf ^. L.showTimeLag
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -16,7 +16,8 @@ module Network.Nakadi.Subscriptions.Subscription ( subscriptionGet , subscriptionDelete- ) where+ )+where import Network.Nakadi.Internal.Prelude @@ -24,26 +25,26 @@ import Network.Nakadi.Internal.Http path :: SubscriptionId -> ByteString-path subscriptionId =- "/subscriptions/"- <> subscriptionIdToByteString subscriptionId+path subscriptionId = "/subscriptions/" <> subscriptionIdToByteString subscriptionId -- | @GET@ to @\/subscriptions\/SUBSCRIPTION@. Looks up subscription -- information for a subscription ID.-subscriptionGet ::- MonadNakadi b m+subscriptionGet+ :: MonadNakadi b m => SubscriptionId -- ^ Subscription ID -> m Subscription -- ^ Resulting Subscription Information-subscriptionGet subscriptionId =- httpJsonBody ok200 [(status404, errorSubscriptionNotFound)]+subscriptionGet subscriptionId = httpJsonBody+ ok200+ [(status404, errorSubscriptionNotFound)] (setRequestMethod "GET" . setRequestPath (path subscriptionId)) -- | @DELETE@ to @\/subscriptions\/SUBSCRIPTION@. Deletes a -- subscription by subscription ID.-subscriptionDelete ::- MonadNakadi b m+subscriptionDelete+ :: MonadNakadi b m => SubscriptionId -- ^ ID of the Subcription to delete -> m ()-subscriptionDelete subscriptionId =- httpJsonNoBody status204 [(status404, errorSubscriptionNotFound)]+subscriptionDelete subscriptionId = httpJsonNoBody+ status204+ [(status404, errorSubscriptionNotFound)] (setRequestMethod "DELETE" . setRequestPath (path subscriptionId))
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -13,7 +13,6 @@ module Network.Nakadi.Types.Config ( Config , ConfigIO- , ConsumeParameters , StreamConnectCallback , HttpBackend(..) ) where
src/Network/Nakadi/Types/Exceptions.hs view
@@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Types.Exceptions Description : Nakadi Exception Types-Copyright : (c) Moritz Schulte 2017+Copyright : (c) Moritz Clasmeier 2017 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental
src/Network/Nakadi/Types/Logger.hs view
@@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Types.Logger Description : Nakadi Exception Types-Copyright : (c) Moritz Schulte 2017+Copyright : (c) Moritz Clasmeier 2017 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental
src/Network/Nakadi/Types/Problem.hs view
@@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Types.Problem Description : Nakadi Problem Type-Copyright : (c) Moritz Schulte 2017+Copyright : (c) Moritz Clasmeier 2017 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -9,11 +9,6 @@ This module provides the Nakadi Problem Type. -}--{-# LANGUAGE ApplicativeDo #-}-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE TupleSections #-} module Network.Nakadi.Types.Problem ( Problem(..)
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, 2018+Copyright : (c) Moritz Clasmeier 2017, 2018 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental@@ -35,6 +35,8 @@ , CursorDistanceResult(..) , SubscriptionPosition(..) , Subscription(..)+ , SubscriptionRequest(..)+ , ConsumerGroup(..) , PublishingStatus(..) , Step(..) , BatchItemResponse(..)@@ -54,7 +56,7 @@ , PartitionState(..) , PartitionStat(..) , SubscriptionEventTypeStats(..)- , SubscriptionEventTypeStatsResult(..)+ , SubscriptionStats(..) , EventTypeCategory(..) , PartitionStrategy(..) , EnrichmentStrategy(..)@@ -63,11 +65,14 @@ , EventType(..) , DataChangeEvent(..) , DataChangeEventEnriched(..)+ , BusinessEvent(..)+ , BusinessEventEnriched(..) , DataOp(..) , EventMetadata(..) , EventMetadataEnriched(..) , EventTypeStatistics(..) , EventTypeOptions(..)- ) where+ )+ where import Network.Nakadi.Internal.Types.Service
+ src/Network/Nakadi/Unsafe/IO.hs view
@@ -0,0 +1,26 @@+{-|+Module : Network.Nakadi.Unsafe.IO+Description : Support for MonadNakadi IO Instance+Copyright : (c) Moritz Clasmeier 2018+License : BSD3+Maintainer : mtesseract@silverratio.net+Stability : experimental+Portability : POSIX++This is an experimental API.++Nakadi-client provides a 'MonadNakadi' instance for the 'IO' monad+accessing a global mutable configuration.++This module exports the user-visible API for managing this global+configuration.+-}++module Network.Nakadi.Unsafe.IO+ ( initializeGlobalConfigurationFromEnv+ , setGlobalConfiguration+ , modifyGlobalConfiguration+ )+where++import Network.Nakadi.Internal.Unsafe.IO
tests/Network/Nakadi/Config/Test.hs view
@@ -1,7 +1,5 @@-{-# LANGUAGE DeriveFunctor #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-}@@ -9,15 +7,16 @@ module Network.Nakadi.Config.Test where -import ClassyPrelude hiding (catch)+import ClassyPrelude -import Control.Lens-import Control.Monad.Catch (MonadThrow (..))-import qualified Data.ByteString.Lazy as LB-import Data.Conduit (ConduitM, transPipe)+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.HTTP.Client.Conduit ( bodyReaderSource )+import Network.HTTP.Client.TLS ( getGlobalManager ) import Network.Nakadi import Network.Nakadi.Tests.Common import System.IO.Unsafe@@ -25,8 +24,7 @@ import Test.Tasty.HUnit testConfig :: TestTree-testConfig = testGroup "Config"- [ testCase "Use Custom HttpBackend" testCustomHttpBackend ]+testConfig = testGroup "Config" [testCase "Use Custom HttpBackend" testCustomHttpBackend] {-# NOINLINE requestsExecuted #-} requestsExecuted :: TVar [Request]@@ -37,26 +35,28 @@ atomically $ modifyTVar requestsExecuted (req :) throwM (HttpExceptionRequest req ResponseTimeout) -mockHttpBackendResponseOpen :: Config b -> Request -> Maybe Manager -> App (Response (ConduitM i ByteString App ()))+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)+ response <- liftIO $ responseOpen req mngr+ pure $ fmap (transPipe liftIO . bodyReaderSource) response mockHttpBackendResponseClose :: Response a -> App () mockHttpBackendResponseClose = liftIO . responseClose testCustomHttpBackend :: Assertion testCustomHttpBackend = runApp $ do- res0 <- try $ runNakadiT mockConfig $- registryPartitionStrategies -- This uses httpLbs.+ 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 liftIO $ 1 @=? length requests-- where mockConfig = newConfig mockHttpBackend defaultRequest- mockHttpBackend = HttpBackend- { _httpLbs = mockHttpBackendLbs- , _httpResponseOpen = mockHttpBackendResponseOpen- , _httpResponseClose = mockHttpBackendResponseClose }+ where+ mockConfig = newConfig mockHttpBackend defaultRequest+ mockHttpBackend = HttpBackend+ { _httpLbs = mockHttpBackendLbs+ , _httpResponseOpen = mockHttpBackendResponseOpen+ , _httpResponseClose = mockHttpBackendResponseClose+ }
+ tests/Network/Nakadi/EventTypes/BusinessEvents/Test.hs view
@@ -0,0 +1,135 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Network.Nakadi.EventTypes.BusinessEvents.Test where++import ClassyPrelude+import Data.Function ( (&) )+import Network.Nakadi+import qualified Network.Nakadi.Lenses as L+import Network.Nakadi.Tests.Common+import Test.Tasty+import Test.Tasty.HUnit+import System.Random+import Control.Lens+import qualified Data.Set as Set+import Data.Aeson++import Conduit++data TestEventA = TestEventA+ { aInt :: Int+ , aString :: String+ } deriving (Eq, Generic, Show)++instance FromJSON TestEventA+instance ToJSON TestEventA++data EventSpec a b c = EventSpec+ { eventGenerator :: IO a+ , eventType :: EventType+ , eventPayload :: a -> c+ , eventEnrichedPayload :: b -> c+ }++eventTypeA :: EventType+eventTypeA = EventType+ { _name = "event-type-a"+ , _owningApplication = Just "test-suite"+ , _category = Just EventTypeCategoryBusiness+ , _enrichmentStrategies = Just [EnrichmentStrategyMetadata]+ , _partitionStrategy = Nothing+ , _compatibilityMode = Nothing+ , _schema = EventTypeSchema+ { _version = Nothing+ , _createdAt = Nothing+ , _schemaType = SchemaTypeJson+ , _schema = "{\+ \ \"properties\": \+ \ { \+ \ \"aInt\": {\"type\": \"number\"}, \+ \ \"aString\": {\"type\": \"string\"} \+ \ }, \+ \ \"required\": [\"aInt\", \"aString\"] \+ \}"+ }+ , _partitionKeyFields = Nothing+ , _defaultStatistic = Nothing+ , _options = Nothing+ }++eventSpecA :: EventSpec (BusinessEvent TestEventA) (BusinessEventEnriched TestEventA) TestEventA+eventSpecA = EventSpec genBusinessEventA eventTypeA (view L.payload) (view L.payload)++genBusinessEventA :: IO (BusinessEvent TestEventA)+genBusinessEventA = do+ payload <- TestEventA <$> randomRIO (-100, 100) <*> pure "Hello"+ eid <- EventId <$> randomIO+ timestamp <- Timestamp <$> getCurrentTime+ pure BusinessEvent+ { _payload = payload+ , _metadata = EventMetadata+ { _eid = eid+ , _occurredAt = timestamp+ , _parentEids = Nothing+ , _partition = Nothing+ }+ }++testBusinessEvents :: Config App -> TestTree+testBusinessEvents conf = testEvents conf "BusinessEvents" eventSpecA++testEvents+ :: (FromJSON a, ToJSON a, FromJSON b, ToJSON b, Eq c, Show c)+ => Config App+ -> String+ -> EventSpec a b c+ -> TestTree+testEvents conf label eventSpec = testGroup+ label+ [ testCase "createAndDeleteEvent" (createAndDeleteEvent conf eventSpec)+ , testCase "publishAndConsume" (publishAndConsume conf eventSpec)+ ]++createEventTypeFromSpec :: (MonadUnliftIO m, MonadNakadi base m) => EventSpec a b c -> m ()+createEventTypeFromSpec eventSpec = do+ subscriptionIds <- subscriptionsList Nothing (Just [eventSpec & eventType & _name])+ <&> map (view L.id)+ mapM_ subscriptionDelete subscriptionIds+ eventTypeDelete (eventSpec & eventType & _name) `catch` ignoreExnNotFound ()+ eventTypeCreate (eventType eventSpec)++deleteEventTypeFromSpec :: MonadNakadi base m => EventSpec a b c -> m ()+deleteEventTypeFromSpec eventSpec = eventTypeDelete (eventSpec & eventType & _name)++publishAndConsume+ :: forall a b c+ . (FromJSON a, ToJSON a, FromJSON b, ToJSON b, Eq c, Show c)+ => Config App+ -> EventSpec a b c+ -> Assertion+publishAndConsume conf' eventSpec =+ runApp+ . runNakadiT conf+ $ bracket_ (createEventTypeFromSpec eventSpec) (deleteEventTypeFromSpec eventSpec)+ $ withTemporarySubscription "test-suite"+ "business-event-test"+ (Set.fromList [eventSpec & eventType & _name])+ SubscriptionPositionBegin+ $ \subscription -> do+ events :: [a] <- liftIO $ replicateM 10 (eventSpec & eventGenerator)+ eventsPublish (eventSpec & eventType & _name) events+ consumed :: [b] <-+ runConduitRes $ subscriptionSourceEvents (subscription ^. L.id) .| sinkList+ liftIO+ $ map (eventPayload eventSpec) events+ @=? map (eventEnrichedPayload eventSpec) consumed+ where conf = conf' & setBatchFlushTimeout 1 & setStreamLimit 10++createAndDeleteEvent :: Config App -> EventSpec a b c -> Assertion+createAndDeleteEvent conf eventSpec =+ runApp+ . runNakadiT conf+ $ bracket_ (createEventTypeFromSpec eventSpec) (deleteEventTypeFromSpec eventSpec)+ $ pure ()
tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs view
@@ -1,10 +1,7 @@-{-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE StandaloneDeriving #-} module Network.Nakadi.EventTypes.CursorsLag.Test where
tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs view
@@ -1,10 +1,6 @@-{-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE StandaloneDeriving #-} module Network.Nakadi.EventTypes.ShiftedCursors.Test where @@ -39,7 +35,7 @@ partitions <- eventTypePartitions myEventTypeName let cursors = map extractCursor partitions liftIO $ length cursors > 0 @=? True- forM_ [1..n] $ \_ -> do+ forM_ [1..n] $ \_ -> eventsPublish myEventTypeName [myDataChangeEvent eid now] cursors' <- cursorsShift myEventTypeName cursors n liftIO $ length cursors' @=? length cursors
tests/Network/Nakadi/EventTypes/Test.hs view
@@ -1,41 +1,42 @@-{-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE StandaloneDeriving #-} module Network.Nakadi.EventTypes.Test where -import ClassyPrelude hiding- (withAsync)+import ClassyPrelude hiding ( withAsync ) import Control.Lens-import Data.Function ((&))+import Data.Function ( (&) ) import Network.Nakadi import Network.Nakadi.EventTypes.CursorsLag.Test import Network.Nakadi.EventTypes.ShiftedCursors.Test-import qualified Network.Nakadi.Lenses as L-import qualified Data.Vector as Vector+import Network.Nakadi.EventTypes.BusinessEvents.Test+import qualified Network.Nakadi.Lenses as L+import qualified Data.Vector as Vector import Network.Nakadi.Tests.Common import Test.Tasty import Test.Tasty.HUnit-import System.IO.Unsafe+import System.IO.Unsafe import UnliftIO.Async+import Control.Monad.Trans.Resource testEventTypes :: Config App -> TestTree-testEventTypes conf = testGroup "EventTypes"- [ testCase "EventTypesPrepare" (testEventTypesPrepare conf)- , testCase "EventTypesGet" (testEventTypesGet conf)+testEventTypes conf = testGroup+ "EventTypes"+ [ testBusinessEvents conf+ , testCase "EventTypesPrepare" (testEventTypesPrepare conf)+ , testCase "EventTypesGet" (testEventTypesGet conf) , testCase "EventTypesDeleteCreateAndGet" (testEventTypesDeleteCreateGet conf)- , testCase "EventTypePartitionsGet" (testEventTypePartitionsGet conf)- , testCase "EventTypeCursorDistances0" (testEventTypeCursorDistances0 conf)- , testCase "EventTypeCursorDistances10" (testEventTypeCursorDistances10 conf)- , testCase "EventTypePublishData" (testEventTypePublishData conf)- , testCase "EventTypeParseFlowId" (testEventTypeParseFlowId conf)- , testCase "EventTypeDeserializationFailureException" (testEventTypeDeserializationFailureException conf)+ , testCase "EventTypePartitionsGet" (testEventTypePartitionsGet conf)+ , testCase "EventTypeCursorDistances0" (testEventTypeCursorDistances0 conf)+ , testCase "EventTypeCursorDistances10" (testEventTypeCursorDistances10 conf)+ , testCase "EventTypePublishData" (testEventTypePublishData conf)+ , testCase "EventTypeParseFlowId" (testEventTypeParseFlowId conf)+ , testCase "EventTypeDeserializationFailureException"+ (testEventTypeDeserializationFailureException conf) , testCase "EventTypeDeserializationFailure" (testEventTypeDeserializationFailure conf) , testEventTypesShiftedCursors conf , testEventTypesCursorsLag conf@@ -44,34 +45,32 @@ testEventTypesPrepare :: Config App -> Assertion testEventTypesPrepare conf = runApp . runNakadiT conf $ do subscriptions <- subscriptionsList Nothing Nothing- let subscriptionIds = catMaybes . map (view L.id) $ subscriptions+ let subscriptionIds = map (view L.id) subscriptions forM_ subscriptionIds subscriptionDelete testEventTypesGet :: Config App -> Assertion-testEventTypesGet conf = runApp . runNakadiT conf $- void eventTypesList+testEventTypesGet conf = runApp . runNakadiT conf $ void eventTypesList testEventTypesDeleteCreateGet :: Config App -> Assertion testEventTypesDeleteCreateGet conf = runApp . runNakadiT conf $ do- eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+ 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))+ where filterMyEvent = filter ((myEventTypeName ==) . view L.name) testEventTypePartitionsGet :: Config App -> Assertion testEventTypePartitionsGet conf = runApp . runNakadiT conf $ do- eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+ eventTypeDelete myEventTypeName `catch` ignoreExnNotFound () eventTypeCreate myEventType void $ eventTypePartitions myEventTypeName testEventTypeCursorDistances0 :: Config App -> Assertion testEventTypeCursorDistances0 conf = runApp . runNakadiT conf $ do- eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+ eventTypeDelete myEventTypeName `catch` ignoreExnNotFound () eventTypeCreate myEventType partitions <- eventTypePartitions myEventTypeName let cursors = map extractCursor partitions@@ -81,76 +80,67 @@ testEventTypeCursorDistances10 :: Config App -> Assertion testEventTypeCursorDistances10 conf = runApp . runNakadiT conf $ do- eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+ eventTypeDelete myEventTypeName `catch` ignoreExnNotFound () eventTypeCreate myEventType partitions <- eventTypePartitions myEventTypeName let cursors = map extractCursor partitions - forM_ [1..10] $ \_ -> do+ forM_ [1 .. 10] $ \_ -> do now <- liftIO getCurrentTime eid <- EventId <$> genRandomUUID eventsPublish myEventTypeName [myDataChangeEvent eid now] - cursorPairs <- forM cursors $ \cursor@Cursor { .. } -> do+ cursorPairs <- forM cursors $ \cursor@Cursor {..} -> do part <- eventTypePartition myEventTypeName _partition let cursor' = extractCursor part return (cursor, cursor') - distances <- forM cursorPairs $ \(c, c') -> do- cursorDistance myEventTypeName c c'+ distances <- forM cursorPairs $ uncurry (cursorDistance myEventTypeName) let totalDistances = sum distances liftIO $ totalDistances @=? 10 -mySubscription :: Subscription-mySubscription = Subscription- { _id = Nothing- , _owningApplication = "test-suite"- , _eventTypes = [EventTypeName "test.FOO"]- , _consumerGroup = Nothing -- ??- , _createdAt = Nothing- , _readFrom = Just SubscriptionPositionEnd- , _initialCursors = Nothing+mySubscription :: SubscriptionRequest+mySubscription = SubscriptionRequest+ { _owningApplication = "test-suite"+ , _eventTypes = [EventTypeName "test.FOO"]+ , _consumerGroup = Nothing+ , _subscriptionPosition = Nothing } -createMySubscription :: (MonadUnliftIO m, MonadNakadi b m)- => m SubscriptionId+createMySubscription :: (MonadUnliftIO m, MonadNakadi b m) => m SubscriptionId createMySubscription = do- newSubscription <- subscriptionCreate mySubscription `catch`- \ case- SubscriptionExistsAlready s -> pure s- exn -> throwIO exn- let (Just subscriptionId) = newSubscription^.L.id- pure subscriptionId--consumeParametersSingle :: ConsumeParameters-consumeParametersSingle = defaultConsumeParameters- & setBatchLimit 1- & setBatchFlushTimeout 1+ newSubscription <- subscriptionCreate mySubscription `catch` \case+ SubscriptionExistsAlready s -> pure s+ exn -> throwIO exn+ pure (newSubscription ^. L.id) testEventTypePublishData :: Config App -> Assertion-testEventTypePublishData conf = runApp . runNakadiT conf $ do+testEventTypePublishData conf' = runApp . runNakadiT conf $ do now <- liftIO getCurrentTime eid <- EventId <$> genRandomUUID recreateEvent myEventType- let event = DataChangeEvent { _payload = Foo "Hello!"- , _metadata = EventMetadata { _eid = eid- , _occurredAt = Timestamp now- , _parentEids = Nothing- , _partition = Nothing- }- , _dataType = "test.FOO"- , _dataOp = DataOpUpdate- }+ let event = DataChangeEvent+ { _payload = Foo "Hello!"+ , _metadata = EventMetadata+ { _eid = eid+ , _occurredAt = Timestamp now+ , _parentEids = Nothing+ , _partition = Nothing+ }+ , _dataType = "test.FOO"+ , _dataOp = DataOpUpdate+ } subscriptionId <- createMySubscription- batchTv <- newTVarIO Nothing- res <- try $ withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do+ batchTv <- newTVarIO Nothing+ res <- try $ withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do liftIO $ link asyncHandle- subscriptionProcess (Just consumeParametersSingle) subscriptionId (storeBatch batchTv)+ runResourceT $ subscriptionProcess subscriptionId (storeBatch batchTv) liftIO $ (Left TerminateConsumption) @=? res Just batch <- atomically $ readTVar batchTv- let Just events = batch^.L.events :: Maybe (Vector (DataChangeEvent Foo))+ let Just events = batch ^. L.events :: Maybe (Vector (DataChangeEvent Foo)) liftIO $ True @=? (Vector.length events > 0)+ where conf = conf' & setBatchLimit 1 & setBatchFlushTimeout 1 storeBatch :: MonadIO m@@ -166,82 +156,85 @@ now <- liftIO getCurrentTime eid <- EventId <$> genRandomUUID recreateEvent myEventType- let event = DataChangeEvent { _payload = Foo "Hello!"- , _metadata = EventMetadata { _eid = eid- , _occurredAt = Timestamp now- , _parentEids = Nothing- , _partition = Nothing- }- , _dataType = "test.FOO"- , _dataOp = DataOpUpdate- }+ let event = DataChangeEvent+ { _payload = Foo "Hello!"+ , _metadata = EventMetadata+ { _eid = eid+ , _occurredAt = Timestamp now+ , _parentEids = Nothing+ , _partition = Nothing+ }+ , _dataType = "test.FOO"+ , _dataOp = DataOpUpdate+ } expectedFlowId = Just $ FlowId "12345" subscriptionId <- createMySubscription- batchTv <- newTVarIO Nothing- res <- try $ withAsync (delayedPublish expectedFlowId [event]) $ \ asyncHandle -> do+ batchTv <- newTVarIO Nothing+ res <- try $ withAsync (delayedPublish expectedFlowId [event]) $ \asyncHandle -> do liftIO $ link asyncHandle- subscriptionProcess (Just consumeParametersSingle) subscriptionId (storeBatch batchTv)+ runResourceT $ subscriptionProcess subscriptionId (storeBatch batchTv) liftIO $ (Left TerminateConsumption) @=? res Just batch <- atomically $ readTVar batchTv- let Just (e:_) = toList <$> (batch^.L.events) :: Maybe [DataChangeEventEnriched Foo]- liftIO $ expectedFlowId @=? e^.L.metadata.L.flowId+ let Just (e : _) = toList <$> (batch ^. L.events) :: Maybe [DataChangeEventEnriched Foo]+ liftIO $ expectedFlowId @=? e ^. L.metadata . L.flowId testEventTypeDeserializationFailureException :: Config App -> Assertion-testEventTypeDeserializationFailureException conf = runApp . runNakadiT conf $ do+testEventTypeDeserializationFailureException conf' = runApp . runNakadiT conf $ do now <- liftIO getCurrentTime eid <- EventId <$> genRandomUUID recreateEvent myEventType- let event = DataChangeEvent { _payload = Foo "Hello!"- , _metadata = EventMetadata { _eid = eid- , _occurredAt = Timestamp now- , _parentEids = Nothing- , _partition = Nothing- }- , _dataType = "test.FOO"- , _dataOp = DataOpUpdate- }- subscriptionId <- createMySubscription- res :: Either NakadiException () <- try $- withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do- liftIO $ link asyncHandle- subscriptionProcess (Just consumeParametersSingle) subscriptionId $- \ (_batch :: SubscriptionEventStreamBatch ()) -> pure ()+ let event = DataChangeEvent+ { _payload = Foo "Hello!"+ , _metadata = EventMetadata+ { _eid = eid+ , _occurredAt = Timestamp now+ , _parentEids = Nothing+ , _partition = Nothing+ }+ , _dataType = "test.FOO"+ , _dataOp = DataOpUpdate+ }+ subscriptionId <- createMySubscription+ res :: Either NakadiException () <-+ try $ withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do+ liftIO $ link asyncHandle+ runResourceT+ $ subscriptionProcess subscriptionId+ $ \(_batch :: SubscriptionEventStreamBatch ()) -> pure () case res of- Left (DeserializationFailure _ _) ->- pure ()- Left exn ->- liftIO $ assertFailure $ "Unexpected exception: " <> show exn- Right events ->- liftIO $ assertFailure $ "Unexpected success: " <> show events+ Left (DeserializationFailure _ _) -> pure ()+ Left exn -> liftIO $ assertFailure $ "Unexpected exception: " <> show exn+ Right events -> liftIO $ assertFailure $ "Unexpected success: " <> show events+ where conf = conf' & setBatchLimit 1 & setBatchFlushTimeout 1 testEventTypeDeserializationFailure :: Config App -> Assertion testEventTypeDeserializationFailure conf' = runApp . runNakadiT conf $ do now <- liftIO getCurrentTime eid <- EventId <$> genRandomUUID recreateEvent myEventType- let event = DataChangeEvent { _payload = Foo "Hello!"- , _metadata = EventMetadata { _eid = eid- , _occurredAt = Timestamp now- , _parentEids = Nothing- , _partition = Nothing- }- , _dataType = "test.FOO"- , _dataOp = DataOpUpdate- }+ let event = DataChangeEvent+ { _payload = Foo "Hello!"+ , _metadata = EventMetadata+ { _eid = eid+ , _occurredAt = Timestamp now+ , _parentEids = Nothing+ , _partition = Nothing+ }+ , _dataType = "test.FOO"+ , _dataOp = DataOpUpdate+ } subscriptionId <- createMySubscription- res <- try $ withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do+ res <- try $ withAsync (delayedPublish Nothing [event]) $ \asyncHandle -> do liftIO $ link asyncHandle- subscriptionProcess (Just consumeParametersSingle) subscriptionId $- \ (_batch :: SubscriptionEventStreamBatch WrongFoo) ->- throwIO TerminateConsumption+ runResourceT+ $ subscriptionProcess subscriptionId+ $ \(_batch :: SubscriptionEventStreamBatch WrongFoo) -> throwIO TerminateConsumption liftIO $ Left TerminateConsumption @=? res counter <- atomically $ readTVar deserializationFailureCounter liftIO $ 1 @=? counter-- where conf = conf'- & setDeserializationFailureCallback deserializationFailureCb+ where+ conf = conf' & setDeserializationFailureCallback deserializationFailureCb - deserializationFailureCb _ _ =- atomically $ modifyTVar deserializationFailureCounter (+ 1)+ deserializationFailureCb _ _ = atomically $ modifyTVar deserializationFailureCounter (+ 1) - deserializationFailureCounter = unsafePerformIO $ newTVarIO 0+ deserializationFailureCounter = unsafePerformIO $ newTVarIO 0
tests/Network/Nakadi/Examples/Subscription/Process.hs view
@@ -1,19 +1,24 @@ {-# LANGUAGE FlexibleContexts #-} -module Network.Nakadi.Examples.Subscription.Process (dumpSubscription) where+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-import Control.Monad.Catch (MonadMask)--dumpSubscription :: (MonadLogger m, MonadNakadi IO m, MonadUnliftIO m, MonadMask m)- => Nakadi.SubscriptionId -> m ()-dumpSubscription subscriptionId =- Nakadi.subscriptionProcess Nothing subscriptionId processBatch+import qualified Network.Nakadi as Nakadi+import Network.Nakadi ( MonadNakadi )+import Control.Monad.Logger+import Control.Monad.Catch ( MonadMask )+import Control.Monad.Trans.Resource - where processBatch :: MonadLogger m => Nakadi.SubscriptionEventStreamBatch Value -> m ()- processBatch batch =- logInfoN (tshow batch)+dumpSubscription+ :: (MonadLogger m, MonadNakadi IO m, MonadUnliftIO m, MonadMask m)+ => Nakadi.SubscriptionId+ -> m ()+dumpSubscription subscriptionId = runResourceT+ $ Nakadi.subscriptionProcess subscriptionId processBatch+ where+ processBatch :: MonadLogger m => Nakadi.SubscriptionEventStreamBatch Value -> m ()+ processBatch batch = logInfoN (tshow batch)
tests/Network/Nakadi/Examples/Subscription/Test.hs view
@@ -4,15 +4,15 @@ module Network.Nakadi.Examples.Subscription.Test ( testConsumption- ) where+ )+where import ClassyPrelude import Control.Lens import Control.Monad.Logger-import Data.Maybe (fromJust)-import qualified Network.Nakadi as Nakadi+import qualified Network.Nakadi as Nakadi import Network.Nakadi.Examples.Subscription.Process-import qualified Network.Nakadi.Lenses as L+import qualified Network.Nakadi.Lenses as L import Network.Nakadi.Tests.Common import Test.Tasty.HUnit import UnliftIO.Concurrent@@ -22,54 +22,52 @@ now <- liftIO getCurrentTime eid <- Nakadi.EventId <$> genRandomUUID let event = Nakadi.DataChangeEvent- { Nakadi._payload = Foo "Hello!"+ { Nakadi._payload = Foo "Hello!" , Nakadi._metadata = Nakadi.EventMetadata- { Nakadi._eid = eid- , Nakadi._occurredAt = Nakadi.Timestamp now- , Nakadi._parentEids = Nothing- , Nakadi._partition = Nothing- }+ { Nakadi._eid = eid+ , Nakadi._occurredAt = Nakadi.Timestamp now+ , Nakadi._parentEids = Nothing+ , Nakadi._partition = Nothing+ } , Nakadi._dataType = "test.FOO"- , Nakadi._dataOp = Nakadi.DataOpUpdate+ , Nakadi._dataOp = Nakadi.DataOpUpdate } pure event genEvents :: MonadIO m => m [Nakadi.DataChangeEvent Foo]-genEvents =- sequence (replicate 10 genEvent)+genEvents = replicateM 10 genEvent testConsumption :: Nakadi.Config IO -> Assertion testConsumption config = Nakadi.runNakadiT config $ do nakadiLogRef <- liftIO $ newIORef []- bracket before after $ \ subscriptionId -> do+ 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+ events <- genEvents+ publisherHandle <- async $ delayedPublish Nothing events+ link publisherHandle+ withAsync (dumpSubscription subscriptionId) $ \dumpHandle -> do+ link dumpHandle+ threadDelay (5 * 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 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+ when (null nakadiLog) $ liftIO $ assertFailure+ "Subscription Consumption has logged no received batches"+ where+ before = do+ recreateEvent myEventType+ subscription <- Nakadi.subscriptionCreate Nakadi.SubscriptionRequest+ { _owningApplication = "test-suite"+ , _eventTypes = [myEventTypeName]+ , _consumerGroup = Nothing -- ??+ , _subscriptionPosition = Just Nakadi.SubscriptionPositionEnd+ }+ pure $ subscription ^. L.id - after subscriptionId = do- Nakadi.subscriptionDelete subscriptionId- Nakadi.eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+ after subscriptionId = do+ Nakadi.subscriptionDelete subscriptionId+ Nakadi.eventTypeDelete myEventTypeName `catch` ignoreExnNotFound () - -- nEvents = 100+ -- nEvents = 100 - logger nakadiLog loc logSource logLevel logStr = do- let str = defaultLogStr loc logSource logLevel logStr- liftIO $ modifyIORef nakadiLog (str :)+ logger nakadiLog loc logSource logLevel logStr = do+ let str = defaultLogStr loc logSource logLevel logStr+ liftIO $ modifyIORef nakadiLog (str :)
tests/Network/Nakadi/Internal/Http/Test.hs view
@@ -1,6 +1,5 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-} @@ -75,7 +74,7 @@ testFlowIdInclusionHttp = do tv <- atomically $ newTVar Nothing let flowId = FlowId "shalom"- httpBackend = httpBackendIO & L.httpLbs .~ (mockHttpLbs tv)+ httpBackend = httpBackendIO & L.httpLbs .~ mockHttpLbs tv config = newConfig httpBackend defaultRequest & setFlowId flowId Left (StringException _ _) <- try $ runNakadiT config eventTypesList Just requestExecuted <- liftIO . atomically $ readTVar tv@@ -84,7 +83,7 @@ testFlowIdMissingHttp :: Assertion testFlowIdMissingHttp = do tv <- atomically $ newTVar Nothing- let httpBackend = httpBackendIO & L.httpLbs .~ (mockHttpLbs tv)+ let httpBackend = httpBackendIO & L.httpLbs .~ mockHttpLbs tv config = newConfig httpBackend defaultRequest Left (StringException _ _) <- try $ runNakadiT config eventTypesList Just requestExecuted <- liftIO . atomically $ readTVar tv
tests/Network/Nakadi/Internal/Types/Test.hs view
@@ -1,23 +1,90 @@-{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-} module Network.Nakadi.Internal.Types.Test where -import ClassyPrelude-import Test.Tasty-import Test.Tasty.HUnit-import Data.Aeson-import Network.Nakadi.Types.Service+import qualified Data.ByteString.Lazy as LB+import ClassyPrelude+import Test.Tasty+import Test.Tasty.HUnit+import Data.Aeson+import Network.Nakadi.Types.Service+import Data.Aeson.QQ+import Data.Maybe (fromJust) testTypes :: TestTree-testTypes = testGroup "Types"- [ testService- ]+testTypes = testGroup "Types" [testService] testService :: TestTree-testService = testGroup "Service"- [ testCase "JSON-decoding CursorCommitResults" testDecodeCursorCommitResults+testService = testGroup+ "Service"+ [ testCase "JSON Decode: CursorCommitResults" testDecodeCursorCommitResults+ , testCase "JSON Decode: SubscriptionRequest" testDecodeSubscriptionRequest ] +jsonEqual :: LB.ByteString -> LB.ByteString -> Bool+jsonEqual a b =+ let Just a' = decode a :: Maybe Value+ Just b' = decode b :: Maybe Value+ in a' == b'++testDecodeSubscriptionRequest :: Assertion+testDecodeSubscriptionRequest = do+ let+ req pos = SubscriptionRequest+ { _owningApplication = "test"+ , _eventTypes = ["event1"]+ , _consumerGroup = Nothing+ , _subscriptionPosition = pos+ }+ req'End = encode [aesonQQ|{owning_application: "test", event_types: ["event1"], read_from: "end"}|]+ req'Begin = encode+ [aesonQQ|{owning_application: "test", event_types: ["event1"], read_from: "begin"}|]+ cursors = [] :: [SubscriptionCursorWithoutToken]+ req'Cursors+ = encode [aesonQQ|{owning_application: "test", event_types: ["event1"], read_from: "cursors", cursors: #{cursors}}|]++ assertBool "Failed to serialize SubscriptionRequest with SubscriptionPositionEnd"+ (jsonEqual (encode (req (Just SubscriptionPositionEnd))) req'End)+ assertBool "Failed to deserialize SubscriptionRequest with SubscriptionPositionEnd"+ (decode req'End == Just (req (Just SubscriptionPositionEnd)))++ assertBool "Failed to serialize SubscriptionRequest with SubscriptionPositionBegin"+ (jsonEqual (encode (req (Just SubscriptionPositionBegin))) req'Begin)+ assertBool "Failed to deserialize SubscriptionRequest with SubscriptionPositionBegin"+ (decode req'Begin == Just (req (Just SubscriptionPositionBegin)))++ assertBool "Failed to serialize SubscriptionRequest with SubscriptionPositionCursors"+ (jsonEqual (encode (req (Just (SubscriptionPositionCursors cursors)))) req'Cursors)+ assertBool "Failed to deserialize SubscriptionRequest with SubscriptionPositionCursors"+ (decode req'Cursors == Just (req (Just (SubscriptionPositionCursors cursors))))++ assertBool "Failed to serialize SubscriptionRequest without SubscriptionPosition"+ (jsonEqual (encode (req Nothing)) req'End)+ testDecodeCursorCommitResults :: Assertion-testDecodeCursorCommitResults = assertBool "Failed to decode" $ isJust (decode sampleResponse :: (Maybe CursorCommitResults))- where sampleResponse = "{\"items\":[{\"cursor\":{\"partition\":\"0\",\"offset\":\"001-0001-000000000000007598\",\"event_type\":\"http4s-nakadi.test-event\",\"cursor_token\":\"3bb3a590-ede5-43a9-981e-2bea26347c99\"},\"result\":\"outdated\"}]}"+testDecodeCursorCommitResults = do+ let eventType = "http4s-nakadi.test-event"+ offset = "001-0001-000000000000007598"+ cursorToken = "3bb3a590-ede5-43a9-981e-2bea26347c99"+ partitionName = "0"+ result = CursorCommitResultOutdated+ cursorCommitResultsJson+ = encode [aesonQQ|+ { items: [ { cursor: { partition: #{partitionName}+ , offset: #{offset}+ , event_type: #{eventType}+ , cursor_token: #{cursorToken}+ },+ result: #{result}+ }+ ]+ }+ |]+ cursor = SubscriptionCursor { _partition = partitionName+ , _offset = offset+ , _eventType = eventType+ , _cursorToken = cursorToken}+ cursorCommitResults = CursorCommitResults { _items = [CursorCommitResult { _cursor = cursor , _result = result }] }+ assertBool "Failed to decode CursorsCommitResults" $+ fromJust (decode cursorCommitResultsJson) == cursorCommitResults
tests/Network/Nakadi/MonadicAPI/Test.hs view
@@ -12,11 +12,13 @@ import Control.Monad.Trans.Resource import Control.Retry import Network.HTTP.Client-import qualified Network.Nakadi as Nakadi+import qualified Network.Nakadi as Nakadi import Network.Nakadi.Base-import qualified Network.Nakadi.Lenses as L+import qualified Network.Nakadi.Lenses as L import Network.Nakadi.Tests.Common import Test.Tasty+import System.Environment+import qualified Network.Nakadi.Unsafe.IO as Nakadi import Test.Tasty.HUnit httpErrorCallback :: Request -> HttpException -> RetryStatus -> Bool -> App ()@@ -24,32 +26,38 @@ 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)+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)+ , testCase "MonadNakadi instance for IO" (testIO conf)+ , testCase "MonadNakadi instance for IO, from Env" (testIOFromEnv conf) ]-- where conf' = Nakadi.setHttpErrorCallback httpErrorCallback conf- confLifted = Nakadi.newConfig Nakadi.httpBackendIO (conf^.L.requestTemplate)+ 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+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 ()+ _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@@ -62,8 +70,8 @@ testNonTrivNakadiBaseT :: Nakadi.Config (NakadiBaseT App) -> Assertion testNonTrivNakadiBaseT conf = runApp . Nakadi.runNakadiBaseT . runNoLoggingT . Nakadi.runNakadiT conf $ do- _events <- Nakadi.eventTypesList- pure ()+ _events <- Nakadi.eventTypesList+ pure () data Env = Env { nakadiConfiguration :: Nakadi.Config App }@@ -76,4 +84,25 @@ testReaderT :: Nakadi.Config App -> Assertion testReaderT conf = runApp . flip runReaderT (Env conf) $ do _events <- Nakadi.eventTypesList+ pure ()++-- | Test the 'MonadNakadi' instance for 'IO' global configuration+-- initialized from the environment (environment variable @NAKADI_URL@).+testIOFromEnv :: Nakadi.Config App -> Assertion+testIOFromEnv conf = flip finally (unsetEnv envNakadiUrl) $ do+ setEnv envNakadiUrl url+ Nakadi.initializeGlobalConfigurationFromEnv+ _eventsTypes <- Nakadi.eventTypesList+ pure ()+ where+ url = show (getUri (conf ^. L.requestTemplate))+ envNakadiUrl = "NAKADI_URL"++-- | Test the 'MonadNakadi' instance for 'IO' using explicitely+-- set global configuration.+testIO :: Nakadi.Config App -> Assertion+testIO conf = do+ let confIO = Nakadi.newConfigIO (conf ^. L.requestTemplate)+ Nakadi.setGlobalConfiguration confIO+ _eventsTypes <- Nakadi.eventTypesList pure ()
tests/Network/Nakadi/Registry/Test.hs view
@@ -1,10 +1,6 @@-{-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE StandaloneDeriving #-} module Network.Nakadi.Registry.Test where @@ -16,10 +12,8 @@ import Test.Tasty.HUnit testRegistry :: Config App -> TestTree-testRegistry conf = testGroup "Registry"- [ testCase "PartitionStrategies" (testPartitionStrategies conf)- ]+testRegistry conf =+ testGroup "Registry" [testCase "PartitionStrategies" (testPartitionStrategies conf)] testPartitionStrategies :: Config App -> Assertion-testPartitionStrategies conf = runApp . runNakadiT conf $ do- void $ registryPartitionStrategies+testPartitionStrategies conf = runApp . runNakadiT conf . void $ registryPartitionStrategies
tests/Network/Nakadi/Subscriptions/Processing/Test.hs view
@@ -10,87 +10,84 @@ import Control.Lens import Control.Monad.Logger+import Control.Monad.Trans.Resource import Data.Aeson-import Data.Maybe (fromJust) import Network.Nakadi-import qualified Network.Nakadi.Lenses as L+import qualified Network.Nakadi.Lenses as L import Network.Nakadi.Tests.Common import Test.Tasty import Test.Tasty.HUnit testSubscriptionsProcessing :: Config App -> TestTree testSubscriptionsProcessing confTemplate =- let mkConf commitStrategy = confTemplate- & setCommitStrategy commitStrategy- in testGroup "Processing"- [ testCase "SubscriptionProcessing/async/TimeBuffer" $- testSubscriptionHighLevelProcessing (mkConf (CommitAsync (CommitTimeBuffer 200)))- , testCase "SubscriptionProcessing/sync" $- testSubscriptionHighLevelProcessing (mkConf CommitSync)- , testCase "SubscriptionProcessing/async/NoBuffer" $- testSubscriptionHighLevelProcessing (mkConf (CommitAsync CommitNoBuffer))- , testCase "SubscriptionProcessing/async/SmartBuffer" $- testSubscriptionHighLevelProcessing (mkConf (CommitAsync CommitSmartBuffer))- ]+ let mkConf commitStrategy nWorkers =+ confTemplate & setCommitStrategy commitStrategy & setWorkerThreads nWorkers+ in testGroup+ "Processing"+ [ testCase "SubscriptionProcessing/async/TimeBuffer/singleWorker"+ $ testSubscriptionHighLevelProcessing (mkConf (CommitAsync (CommitTimeBuffer 200)) 1)+ , testCase "SubscriptionProcessing/sync/singleWorker"+ $ testSubscriptionHighLevelProcessing (mkConf CommitSync 1)+ , testCase "SubscriptionProcessing/async/NoBuffer/singleWorker"+ $ testSubscriptionHighLevelProcessing (mkConf (CommitAsync CommitNoBuffer) 1)+ , testCase "SubscriptionProcessing/async/SmartBuffer/singleWorker"+ $ testSubscriptionHighLevelProcessing (mkConf (CommitAsync CommitSmartBuffer) 1)+ , testCase "SubscriptionProcessing/async/TimeBuffer/concurrentWorkers"+ $ testSubscriptionHighLevelProcessing (mkConf (CommitAsync (CommitTimeBuffer 200)) 8)+ , testCase "SubscriptionProcessing/sync/concurrentWorkers"+ $ testSubscriptionHighLevelProcessing (mkConf CommitSync 8)+ , testCase "SubscriptionProcessing/async/NoBuffer/concurrentWorkers"+ $ testSubscriptionHighLevelProcessing (mkConf (CommitAsync CommitNoBuffer) 8)+ , testCase "SubscriptionProcessing/async/SmartBuffer/concurrentWorkers"+ $ testSubscriptionHighLevelProcessing (mkConf (CommitAsync CommitSmartBuffer) 8)+ ] data ConsumptionDone = ConsumptionDone deriving (Show, Typeable) instance Exception ConsumptionDone testSubscriptionHighLevelProcessing :: Config App -> Assertion-testSubscriptionHighLevelProcessing conf = runApp $ do+testSubscriptionHighLevelProcessing conf' = runApp $ do logger <- askLoggerIO let logFunc src lev str = liftIO $ logger defaultLoc src lev str+ conf = conf' & setBatchFlushTimeout 1 & setMaxUncommittedEvents 5000 & setBatchLimit 10 runNakadiT (conf & setLogFunc logFunc) $ do counter <- newIORef 0- events <- sequence $- map genMyDataChangeEventIdx [1..nEvents] :: NakadiT App App [DataChangeEvent Foo]- publishAndConsume events counter `catch` \ (_exn :: ConsumptionDone) -> pure ()+ events <- mapM genMyDataChangeEventIdx [1 .. nEvents] :: NakadiT App App [DataChangeEvent Foo]+ publishAndConsume events counter `catch` \(_exn :: ConsumptionDone) -> pure () eventsRead <- readIORef counter liftIO $ nEvents @=? eventsRead-- where before :: (MonadUnliftIO m, MonadNakadi App m) => m SubscriptionId- before = do- recreateEvent 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 :: (MonadUnliftIO m, MonadNakadi App m) => SubscriptionId -> m ()- after subscriptionId = do- subscriptionDelete subscriptionId- eventTypeDelete myEventTypeName `catch` (ignoreExnNotFound ())+ where+ before :: (MonadUnliftIO m, MonadNakadi App m) => m SubscriptionId+ before = do+ recreateEvent myEventType+ subscription <- subscriptionCreate SubscriptionRequest+ { _owningApplication = "test-suite"+ , _eventTypes = [myEventTypeName]+ , _consumerGroup = Nothing -- ??+ , _subscriptionPosition = Just SubscriptionPositionEnd+ }+ pure $ subscription ^. L.id - nEvents :: Int- nEvents = 10000+ after :: (MonadUnliftIO m, MonadNakadi App m) => SubscriptionId -> m ()+ after subscriptionId = do+ subscriptionDelete subscriptionId+ eventTypeDelete myEventTypeName `catch` ignoreExnNotFound () - publishAndConsume :: (ToJSON a, FromJSON a, Show a)- => [DataChangeEvent a]- -> IORef Int- -> NakadiT App App ()- publishAndConsume events counter =- bracket before after $ \ subscriptionId -> do- let n = length events- publisherHandle <- async $ do- delayedPublish Nothing events- liftIO $ linkAsync publisherHandle- forever $ do- subscriptionProcess (Just consumeParameters) subscriptionId $- \ (batch :: SubscriptionEventStreamBatch (DataChangeEvent Foo)) -> do- let eventsReceived = fromMaybe mempty (batch^.L.events)- modifyIORef counter (+ (length eventsReceived))- eventsRead <- readIORef counter- when (n <= eventsRead) $ throwIO ConsumptionDone- putStrLn $ "Subscription Processing terminated, will restart."+ nEvents :: Int+ nEvents = 10000 - consumeParameters = defaultConsumeParameters- & setBatchFlushTimeout 1- & setMaxUncommittedEvents 5000- & setBatchLimit 10+ publishAndConsume+ :: (ToJSON a, FromJSON a, Show a) => [DataChangeEvent a] -> IORef Int -> NakadiT App App ()+ publishAndConsume events counter = bracket before after $ \subscriptionId -> do+ let n = length events+ publisherHandle <- async $ delayedPublish Nothing events+ liftIO $ linkAsync publisherHandle+ forever . runResourceT $ do+ subscriptionProcess subscriptionId+ $ \(batch :: SubscriptionEventStreamBatch (DataChangeEvent Foo)) -> do+ let eventsReceived = fromMaybe mempty (batch ^. L.events)+ modifyIORef counter (+ length eventsReceived)+ eventsRead <- readIORef counter+ when (n <= eventsRead) $ throwIO ConsumptionDone+ putStrLn "Subscription Processing terminated, will restart."
+ tests/Network/Nakadi/Subscriptions/Stats/Test.hs view
@@ -0,0 +1,72 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- This test tests the subscription statistics API.++module Network.Nakadi.Subscriptions.Stats.Test where++import ClassyPrelude++import Control.Lens+import Data.Aeson+import Network.Nakadi+import qualified Network.Nakadi.Lenses as L+import Network.Nakadi.Tests.Common+import Test.Tasty+import Test.Tasty.HUnit+import Control.Concurrent+import Control.Monad.Trans.Resource++testSubscriptionsStats :: Config App -> TestTree+testSubscriptionsStats conf = testGroup+ "Stats"+ [ testCase "SubscriptionStats/WithTimeLag" $ testSubscriptionStatsWithTimeLag conf+ , testCase "SubscriptionStats/WithoutTimeLag" $ testSubscriptionStatsWithoutTimeLag conf+ ]++produceSubscriptionStats :: Config App -> IO (Map EventTypeName [PartitionStat])+produceSubscriptionStats conf =+ runApp+ $ runNakadiT conf+ $ bracket before after+ $ \subscriptionId -> runResourceT $ do+ -- Note: Apparently we have to consume the subscription first in order to enable+ -- tracking of unconsumed events and time lag.+ void $ timeout (2 * 10 ^ (6 :: Int)) $ subscriptionProcess+ subscriptionId+ (\(_batch :: SubscriptionEventStreamBatch (DataChangeEvent Value)) -> pure ())+ events <- replicateM 10 (genMyDataChangeEventIdx 1)+ eventsPublish myEventTypeName events+ liftIO $ threadDelay (1 * 10 ^ (6 :: Int))+ subscriptionStats subscriptionId++testSubscriptionStatsWithoutTimeLag :: Config App -> Assertion+testSubscriptionStatsWithoutTimeLag conf = do+ stats <- produceSubscriptionStats (setShowTimeLag False conf)+ let (Just partitionStats) = lookup myEventTypeName stats+ noConsumerLagSeconds = partitionStats & map (^. L.consumerLagSeconds) & filter isNothing+ liftIO $ length partitionStats @=? length noConsumerLagSeconds++testSubscriptionStatsWithTimeLag :: Config App -> Assertion+testSubscriptionStatsWithTimeLag conf = do+ stats <- produceSubscriptionStats (setShowTimeLag True conf)+ let (Just partitionStats) = lookup myEventTypeName stats+ consumerLagSeconds = partitionStats & map (^. L.consumerLagSeconds) & filter isJust+ liftIO $ length partitionStats @=? length consumerLagSeconds++before :: (MonadUnliftIO m, MonadNakadi App m) => m SubscriptionId+before = do+ recreateEvent myEventType+ subscription <- subscriptionCreate SubscriptionRequest+ { _owningApplication = "test-suite"+ , _eventTypes = [myEventTypeName]+ , _consumerGroup = Nothing+ , _subscriptionPosition = Just SubscriptionPositionBegin+ }+ pure $ subscription ^. L.id++after :: (MonadUnliftIO m, MonadNakadi App m) => SubscriptionId -> m ()+after subscriptionId = do+ subscriptionDelete subscriptionId+ eventTypeDelete myEventTypeName `catch` ignoreExnNotFound ()
tests/Network/Nakadi/Subscriptions/Test.hs view
@@ -1,10 +1,6 @@-{-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE StandaloneDeriving #-} module Network.Nakadi.Subscriptions.Test where @@ -12,83 +8,79 @@ import Control.Lens import Network.Nakadi-import qualified Network.Nakadi.Lenses as L+import qualified Network.Nakadi.Lenses as L import Network.Nakadi.Tests.Common import Test.Tasty import Test.Tasty.HUnit-import Network.Nakadi.Subscriptions.Processing.Test+import Network.Nakadi.Subscriptions.Processing.Test+import Network.Nakadi.Subscriptions.Stats.Test testSubscriptions :: Config App -> TestTree-testSubscriptions conf = testGroup "Subscriptions"- [ testSubscriptionsProcessing conf- , testCase "SubscriptionsList" (testSubscriptionsList conf)- , testCase "SubscriptionsCreateDelete" (testSubscriptionsCreateDelete conf)+testSubscriptions conf = testGroup+ "Subscriptions"+ [ testSubscriptionsStats conf+ , testSubscriptionsProcessing conf+ , testCase "SubscriptionsList" (testSubscriptionsList conf)+ , testCase "SubscriptionsCreateDelete" (testSubscriptionsCreateDelete conf) , testCase "SubscriptionDoubleDeleteFailure" (testSubscriptionsDoubleDeleteFailure conf) ] testSubscriptionsList :: Config App -> Assertion-testSubscriptionsList conf = runApp . runNakadiT conf $ do+testSubscriptionsList conf = run $ do -- Cleanup deleteSubscriptionsByAppPrefix prefix recreateEvent myEventType -- Create new Subscriptions- maybeSubscriptionIds <- forM [1..n] $ \i -> do+ subscriptionIds <- forM [1 .. n] $ \i -> do let owningApp = ApplicationName (prefix <> tshow i) subscription <- subscriptionCreate (mySubscription (Just owningApp))- return (subscription^.L.id)- let subscriptionIds = catMaybes maybeSubscriptionIds+ return (subscription ^. L.id) liftIO $ n @=? length subscriptionIds -- Retrieve list of all Subscriptions 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+ let subscriptionsFiltered = filter (subscriptionAppHasPrefix prefix) subscriptions'+ subscriptionIdsFiltered = map (view L.id) $ subscriptionsFiltered liftIO $ n @=? length subscriptionIdsFiltered liftIO $ sort subscriptionIds @=? sort subscriptionIdsFiltered- - where n = 100- prefix = "test-suite-list-"+ where+ n = 100+ prefix = "test-suite-list-"+ run = runApp . runNakadiT conf 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+ let subscriptionsFiltered = filter (subscriptionAppHasPrefix prefix) subscriptions+ subscriptionIdsFiltered = map (view L.id) subscriptionsFiltered forM_ subscriptionIdsFiltered subscriptionDelete subscriptionAppHasPrefix :: Text -> Subscription -> Bool subscriptionAppHasPrefix prefix subscription =- let ApplicationName owningApp = subscription^.L.owningApplication- in take (length prefix) owningApp == prefix+ let ApplicationName owningApp = subscription ^. L.owningApplication+ in take (length prefix) owningApp == prefix -mySubscription :: Maybe ApplicationName -> Subscription-mySubscription maybeOwningApp = Subscription- { _id = Nothing- , _owningApplication = fromMaybe "test-suite" maybeOwningApp- , _eventTypes = [myEventTypeName]- , _consumerGroup = Nothing -- ??- , _createdAt = Nothing- , _readFrom = Just SubscriptionPositionEnd- , _initialCursors = Nothing+mySubscription :: Maybe ApplicationName -> SubscriptionRequest+mySubscription maybeOwningApp = SubscriptionRequest+ { _owningApplication = fromMaybe "test-suite" maybeOwningApp+ , _eventTypes = [myEventTypeName]+ , _consumerGroup = Nothing+ , _subscriptionPosition = Nothing } testSubscriptionsCreateDelete :: Config App -> Assertion testSubscriptionsCreateDelete conf = runApp . runNakadiT conf $ do recreateEvent myEventType subscription <- subscriptionCreate (mySubscription Nothing)- liftIO $ True @=? isJust (subscription^.L.id)- let (Just subscriptionId) = subscription^.L.id- subscriptionDelete subscriptionId+ subscriptionDelete (subscription ^. L.id) return () testSubscriptionsDoubleDeleteFailure :: Config App -> Assertion testSubscriptionsDoubleDeleteFailure conf = runApp . runNakadiT conf $ do recreateEvent myEventType subscription <- subscriptionCreate (mySubscription Nothing)- liftIO $ True @=? isJust (subscription^.L.id)- let (Just subscriptionId) = subscription^.L.id- subscriptionDelete subscriptionId- res <- try (subscriptionDelete subscriptionId)+ subscriptionDelete (subscription ^. L.id)+ res <- try (subscriptionDelete (subscription ^. L.id)) case res of Left (SubscriptionNotFound _) -> return () _ -> liftIO $ assertFailure "Expected SubscriptionNotFound exception"
tests/Network/Nakadi/Tests/Common.hs view
@@ -8,15 +8,17 @@ import ClassyPrelude -import Control.Exception.Safe (MonadThrow, throwM)+import Control.Exception.Safe ( MonadThrow+ , throwM+ ) import Control.Lens import Control.Monad.Logger import Data.Aeson-import Data.List.Split (chunksOf)-import qualified Data.Text as Text-import Data.UUID (UUID)+import Data.List.Split ( chunksOf )+import qualified Data.Text as Text+import Data.UUID ( UUID ) import Network.Nakadi-import qualified Network.Nakadi.Lenses as L+import qualified Network.Nakadi.Lenses as L import System.Random import UnliftIO.Concurrent @@ -44,24 +46,29 @@ myEventTypeSchema :: EventTypeSchema myEventTypeSchema = EventTypeSchema- { _version = Just "0.1"- , _createdAt = Nothing+ { _version = Just "0.1"+ , _createdAt = Nothing , _schemaType = SchemaTypeJson , _schema = "{ \"properties\": {\"fortune\": {\"type\": \"string\"} }, \"required\": [\"fortune\"] }" } myEventType :: EventType myEventType = EventType- { _name = myEventTypeName- , _owningApplication = Just "test-suite"- , _category = Just EventTypeCategoryData+ { _name = myEventTypeName+ , _owningApplication = Just "test-suite"+ , _category = Just EventTypeCategoryData , _enrichmentStrategies = Just [EnrichmentStrategyMetadata]- , _partitionStrategy = Just "hash"- , _compatibilityMode = Just CompatibilityModeForward- , _partitionKeyFields = Just ["fortune"]- , _schema = myEventTypeSchema- , _defaultStatistic = Nothing- , _options = Nothing+ , _partitionStrategy = Just "hash"+ , _compatibilityMode = Just CompatibilityModeForward+ , _partitionKeyFields = Just ["fortune"]+ , _schema = myEventTypeSchema+ , _defaultStatistic = Just EventTypeStatistics+ { _messagesPerMinute = 1000+ , _messageSize = 200+ , _readParallelism = 8+ , _writeParallelism = 8+ }+ , _options = Nothing } ignoreExnNotFound :: MonadThrow m => a -> NakadiException -> m a@@ -69,20 +76,19 @@ ignoreExnNotFound _ exn = throwM exn extractCursor :: Partition -> Cursor-extractCursor Partition { ..} =- Cursor { _partition = _partition- , _offset = _newestAvailableOffset }+extractCursor Partition {..} = Cursor {_partition = _partition, _offset = _newestAvailableOffset} myDataChangeEvent :: EventId -> UTCTime -> DataChangeEvent Foo-myDataChangeEvent eid now = DataChangeEvent- { _payload = Foo "Hello!"- , _metadata = EventMetadata { _eid = eid- , _occurredAt = Timestamp now- , _parentEids = Nothing- , _partition = Nothing- }+myDataChangeEvent eid now = DataChangeEvent+ { _payload = Foo "Hello!"+ , _metadata = EventMetadata+ { _eid = eid+ , _occurredAt = Timestamp now+ , _parentEids = Nothing+ , _partition = Nothing+ } , _dataType = "test.FOO"- , _dataOp = DataOpUpdate+ , _dataOp = DataOpUpdate } @@ -91,14 +97,15 @@ eid <- genRandomUUID now <- liftIO getCurrentTime pure DataChangeEvent- { _payload = Foo "Hello!"- , _metadata = EventMetadata { _eid = EventId eid- , _occurredAt = Timestamp now- , _parentEids = Nothing- , _partition = Nothing- }+ { _payload = Foo "Hello!"+ , _metadata = EventMetadata+ { _eid = EventId eid+ , _occurredAt = Timestamp now+ , _parentEids = Nothing+ , _partition = Nothing+ } , _dataType = "test.FOO"- , _dataOp = DataOpUpdate+ , _dataOp = DataOpUpdate } genMyDataChangeEventIdx :: MonadIO m => Int -> m (DataChangeEvent Foo)@@ -106,14 +113,15 @@ eid <- genRandomUUID now <- liftIO getCurrentTime pure DataChangeEvent- { _payload = Foo ("Hello " ++ Text.pack (show idx))- , _metadata = EventMetadata { _eid = EventId eid- , _occurredAt = Timestamp now- , _parentEids = Nothing- , _partition = Nothing- }+ { _payload = Foo ("Hello " ++ Text.pack (show idx))+ , _metadata = EventMetadata+ { _eid = EventId eid+ , _occurredAt = Timestamp now+ , _parentEids = Nothing+ , _partition = Nothing+ } , _dataType = "test.FOO"- , _dataOp = DataOpUpdate+ , _dataOp = DataOpUpdate } genRandomUUID :: MonadIO m => m UUID@@ -121,22 +129,16 @@ recreateEvent :: (MonadUnliftIO m, MonadNakadi b m) => EventType -> m () recreateEvent eventType = do- let eventTypeName = eventType^.L.name- subscriptionIds <- subscriptionsList Nothing (Just [eventTypeName])- <&> catMaybes . map (view L.id)+ let eventTypeName = eventType ^. L.name+ subscriptionIds <- subscriptionsList Nothing (Just [eventTypeName]) <&> map (view L.id) mapM_ subscriptionDelete subscriptionIds- eventTypeDelete eventTypeName `catch` (ignoreExnNotFound ())+ 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)+delayedPublish :: (MonadNakadi b m, MonadIO m, ToJSON a) => Maybe FlowId -> [a] -> m ()+delayedPublish maybeFlowId events = do+ liftIO $ threadDelay (10 ^ 6) let flowId = fromMaybe (FlowId "shalom") maybeFlowId config <- nakadiAsk <&> setFlowId flowId -- Publish events in batches.- runNakadiT config $- forM_ (chunksOf 100 events) (eventsPublish myEventTypeName)+ runNakadiT config $ forM_ (chunksOf 100 events) (eventsPublish myEventTypeName)
tests/Tests.hs view
@@ -71,9 +71,9 @@ integrationTests :: Config App -> ConfigIO -> [TestTree] integrationTests conf confIO =- [ testSubscriptions conf+ [ testEventTypes conf+ , testSubscriptions conf , testExamples confIO- , testEventTypes conf , testRegistry conf , testMonadicAPI conf ]