nakadi-client 0.3.0.0 → 0.4.0.0

raw patch · 26 files changed

+698/−165 lines, 26 filesdep +stmdep +waidep +warpPVP ok

version bump matches the API change (PVP)

Dependencies added: stm, wai, warp

API changes (from Hackage documentation)

Files

• nakadi-client.cabal +71/−56
• src/Network/Nakadi/Config.hs +29/−3
• src/Network/Nakadi/Internal/Http.hs +31/−12
• src/Network/Nakadi/Internal/Lenses.hs +2/−0
• src/Network/Nakadi/Internal/Prelude.hs +12/−1
• src/Network/Nakadi/Internal/Retry.hs +39/−21
• src/Network/Nakadi/Internal/Types/Config.hs +18/−0
• src/Network/Nakadi/Internal/Types/Exceptions.hs +3/−1
• src/Network/Nakadi/Internal/Types/Problem.hs +1/−0
• src/Network/Nakadi/Internal/Types/Service.hs +43/−35
• src/Network/Nakadi/Internal/Types/Subscription.hs +2/−0
• src/Network/Nakadi/Subscriptions/Events.hs +3/−3
• src/Network/Nakadi/Types/Config.hs +1/−0
• src/Network/Nakadi/Types/Service.hs +1/−0
• src/Network/Nakadi/Types/Subscription.hs +1/−1
• tests/Network/Nakadi/Config/Test.hs +48/−0
• tests/Network/Nakadi/Connection/Test.hs +93/−0
• tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs +1/−1
• tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs +1/−4
• tests/Network/Nakadi/EventTypes/Test.hs +39/−14
• tests/Network/Nakadi/Internal/Http/Test.hs +58/−0
• tests/Network/Nakadi/Internal/Retry/Test.hs +104/−0
• tests/Network/Nakadi/Internal/Test.hs +13/−0
• tests/Network/Nakadi/Internal/Types/Test.hs +23/−0
• tests/Network/Nakadi/Tests/Common.hs +6/−1
• tests/Tests.hs +55/−12

@@ -1,9 +1,11 @@--- This file has been generated from package.yaml by hpack version 0.17.0.+-- This file has been generated from package.yaml by hpack version 0.20.0. -- -- see: https://github.com/sol/hpack+--+-- hash: 8c95e6f774d7ddcacd71c47566a57d2e35c231b13bd9c2c23808078cfad08542  name:           nakadi-client-version:        0.3.0.0+version:        0.4.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@@ -24,42 +26,48 @@   type: git   location: https://github.com/mtesseract/nakadi-haskell +flag devel+  manual: True+  default: False+ library   hs-source-dirs:       src   default-extensions: NoImplicitPrelude OverloadedStrings DuplicateRecordFields-  ghc-options: -Wall -fno-warn-type-defaults   build-depends:-      base >=4.7 && <5+      aeson+    , aeson-casing+    , base >=4.7 && <5+    , bytestring     , conduit-    , conduit-extra     , conduit-combinators-    , iso8601-time-    , bytestring+    , conduit-extra     , containers-    , vector     , hashable-    , aeson-    , monad-logger-    , text-    , lens-    , http-types     , http-client     , http-client-tls     , http-conduit-    , resourcet-    , uuid+    , http-types+    , iso8601-time+    , lens+    , monad-logger     , mtl-    , transformers-    , scientific+    , resourcet+    , retry     , safe-exceptions-    , unordered-containers-    , time+    , scientific     , split-    , aeson-casing-    , tasty     , template-haskell-    , retry+    , text+    , time+    , transformers+    , unordered-containers+    , uuid+    , vector+  if flag(devel)+    ghc-options: -Wall -fno-warn-type-defaults -Werror+  else+    ghc-options: -Wall -fno-warn-type-defaults   exposed-modules:       Network.Nakadi       Network.Nakadi.Config@@ -85,14 +93,14 @@       Network.Nakadi.Types.Subscription       Network.Nakadi.Registry       Network.Nakadi.Lenses+      Network.Nakadi.Internal.Http+      Network.Nakadi.Internal.Retry   other-modules:       Network.Nakadi.Internal.Config       Network.Nakadi.Internal.Conversions-      Network.Nakadi.Internal.Http       Network.Nakadi.Internal.Json       Network.Nakadi.Internal.Lenses       Network.Nakadi.Internal.Prelude-      Network.Nakadi.Internal.Retry       Network.Nakadi.Internal.TH       Network.Nakadi.Internal.Types       Network.Nakadi.Internal.Types.Config@@ -112,57 +120,64 @@   hs-source-dirs:       tests   default-extensions: NoImplicitPrelude OverloadedStrings DuplicateRecordFields-  ghc-options: -Wall -fno-warn-type-defaults -Wall -fno-warn-type-defaults+  ghc-options: -Wall -fno-warn-type-defaults   build-depends:-      base >=4.7 && <5+      aeson+    , aeson-casing+    , async+    , base+    , bytestring+    , classy-prelude     , conduit-    , conduit-extra     , conduit-combinators-    , iso8601-time-    , bytestring+    , conduit-extra     , containers-    , vector     , hashable-    , aeson-    , monad-logger-    , text-    , lens-    , http-types     , http-client     , http-client-tls     , http-conduit-    , resourcet-    , uuid+    , http-types+    , iso8601-time+    , lens+    , lens-aeson+    , monad-logger     , mtl-    , transformers-    , scientific+    , nakadi-client+    , random+    , resourcet+    , retry     , safe-exceptions-    , unordered-containers-    , time+    , say+    , scientific     , split-    , aeson-casing-    , tasty-    , template-haskell-    , retry-    , base-    , classy-prelude-    , nakadi-client+    , stm     , tasty     , tasty-hunit-    , http-client-    , http-conduit-    , aeson-    , lens-aeson+    , template-haskell     , text-    , say-    , random-    , async-    , retry+    , time+    , transformers+    , unordered-containers+    , uuid+    , vector+    , wai+    , warp+  if flag(devel)+    ghc-options: -Wall -fno-warn-type-defaults -Werror+  else+    ghc-options: -Wall -fno-warn-type-defaults   other-modules:+      Network.Nakadi.Config.Test+      Network.Nakadi.Connection.Test       Network.Nakadi.EventTypes.CursorsLag.Test       Network.Nakadi.EventTypes.ShiftedCursors.Test       Network.Nakadi.EventTypes.Test+      Network.Nakadi.Internal.Http.Test+      Network.Nakadi.Internal.Retry.Test+      Network.Nakadi.Internal.Test+      Network.Nakadi.Internal.Types.Test       Network.Nakadi.Registry.Test       Network.Nakadi.Subscriptions.Test       Network.Nakadi.Tests.Common+      Paths_nakadi_client   default-language: Haskell2010

@@ -17,8 +17,11 @@  import           Control.Lens import           Control.Retry-import           Network.HTTP.Client-import           Network.HTTP.Client.TLS+import           Network.HTTP.Client             (Manager, ManagerSettings,+                                                  responseClose, responseOpen)+import           Network.HTTP.Client.TLS         (newTlsManagerWith,+                                                  tlsManagerSettings)+import           Network.HTTP.Simple             (httpLbs) import qualified Network.Nakadi.Internal.Lenses  as L import           Network.Nakadi.Internal.Types @@ -42,8 +45,18 @@                 , _deserializationFailureCallback = Nothing                 , _streamConnectCallback          = Nothing                 , _logFunc                        = Nothing-                , _retryPolicy                    = defaultRetryPolicy }+                , _retryPolicy                    = defaultRetryPolicy+                , _http                           = defaultHttpBackend+                , _httpErrorCallback              = Nothing+                } +-- | Default 'HttpBackend' doing IO using http-client.+defaultHttpBackend :: HttpBackend+defaultHttpBackend =+  HttpBackend { _httpLbs                        = httpLbs+              , _responseOpen                   = responseOpen+              , _responseClose                  = responseClose }+ -- | Produce a new configuration, with optional HTTP manager settings -- and mandatory HTTP request template. newConfig ::@@ -75,6 +88,14 @@ setStreamConnectCallback :: StreamConnectCallback  -> Config -> Config setStreamConnectCallback cb = L.streamConnectCallback .~ Just cb +-- | 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 -> Config -> Config+setHttpErrorCallback cb = L.httpErrorCallback .~ Just cb+ -- | Install a logger callback in the provided configuration. setLogFunc :: LogFunc -> Config -> Config setLogFunc logFunc = L.logFunc .~ Just logFunc@@ -82,6 +103,11 @@ -- | Set a custom retry policy in the provided configuration. setRetryPolicy :: RetryPolicyM IO -> Config -> Config setRetryPolicy = (L.retryPolicy .~)++-- | Set a custom HTTP Backend in the provided configuration. Can be+-- used for testing.+setHttpBackend :: HttpBackend -> Config -> Config+setHttpBackend = (L.http .~)  -- | Default parameters for event consumption. defaultConsumeParameters :: ConsumeParameters

@@ -50,7 +50,10 @@ import qualified Data.ByteString.Lazy            as ByteString.Lazy import qualified Data.Conduit.Binary             as Conduit import qualified Data.Text                       as Text-import           Network.HTTP.Client             (responseClose, responseOpen)+import           Network.HTTP.Client             (BodyReader,+                                                  HttpException (..),+                                                  HttpExceptionContent (..),+                                                  checkResponse, responseStatus) import           Network.HTTP.Client.Conduit     (bodyReaderSource) import           Network.HTTP.Simple import           Network.HTTP.Types@@ -74,15 +77,29 @@     where callback = fromMaybe dummyCallback _deserializationFailureCallback           dummyCallback _ _ = return () +-- | 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)+ httpBuildRequest ::   (MonadIO m, MonadCatch m)   => Config -- ^ Configuration, contains the impure request modifier   -> (Request -> Request) -- ^ Pure request modifier-  -> m Request -- ^ Resulting request to excecute-httpBuildRequest Config { .. } requestDef = do+  -> m Request -- ^ Resulting request to execute+httpBuildRequest Config { .. } requestDef =   let manager = _manager-      request = requestDef _requestTemplate & setRequestManager manager-  tryAny (liftIO (_requestModifier request)) >>= \case+      request = requestDef _requestTemplate+                   & setRequestManager manager+                   & (\req -> req { checkResponse = checkNakadiResponse })+  in modifyRequest _requestModifier request+++-- | Modify the Request based on a user function in the configuration.+modifyRequest :: (MonadIO m, MonadCatch m) => (Request -> IO Request) -> Request -> m Request+modifyRequest rm request =+  tryAny (liftIO (rm request)) >>= \case     Right modifiedRequest -> return modifiedRequest     Left  exn             -> throwIO $ RequestModificationException exn @@ -93,9 +110,9 @@   => Config   -> (Request -> Request)   -> m (Response ByteString.Lazy.ByteString)-httpExecRequest config requestDef =-  httpBuildRequest config requestDef-  >>= (retryAction undefined . (liftIO . httpLbs))+httpExecRequest config requestDef = do+  req <- httpBuildRequest config requestDef+  retryAction config req (liftIO . (config^.L.http.L.httpLbs))  -- | Executes an HTTP request using the provided configuration and a -- pure request modifier. Returns the HTTP response and separately the@@ -150,10 +167,12 @@   -> (Request -> Request)   -> m (b, ConduitM () a (ReaderT r m) ()) httpJsonBodyStream config successStatus f exceptionMap requestDef = do-  let manager        = config^.L.manager-      request        = requestDef (config^.L.requestTemplate)-                       & setRequestManager manager-  (_, response) <- allocate (retryAction config (responseOpen request manager)) responseClose+  request <- httpBuildRequest config requestDef+  let manager           = config^.L.manager+      responseOpen      = config^.L.http.L.responseOpen+      responseClose     = config^.L.http.L.responseClose+      responseOpenRetry = retryAction config request (flip responseOpen manager)+  (_, response) <- allocate responseOpenRetry responseClose   let response_  = void response       bodySource = bodyReaderSource (getResponseBody response)       status     = getResponseStatus response

@@ -37,11 +37,13 @@   nakadiConfig :: Lens' s a  makeNakadiLenses ''Config+makeNakadiLenses ''HttpBackend makeNakadiLenses ''Cursor makeNakadiLenses ''EventStreamBatch makeNakadiLenses ''SubscriptionEventStreamBatch makeNakadiLenses ''Event makeNakadiLenses ''Metadata+makeNakadiLenses ''MetadataEnriched makeNakadiLenses ''Partition makeNakadiLenses ''CursorDistanceQuery makeNakadiLenses ''CursorDistanceResult

@@ -25,6 +25,8 @@   , encodeUtf8   , decodeUtf8   , identity+  , undefined+  , error   , MonadIO   , Request   , Response@@ -43,10 +45,19 @@ import qualified Data.Text                  as Text import           Data.Text.Encoding import           Network.HTTP.Client        (Request, Response)-import           Prelude                    hiding (id)+import           Prelude                    hiding (id, undefined, error)+import qualified Prelude  tshow :: Show a => a -> Text tshow = Text.pack . show  identity :: a -> a identity x = x++{-# WARNING undefined "usage of 'undefined' forbidden" #-}+undefined :: a+undefined = Prelude.undefined++{-# WARNING error "usage of 'error' forbidden" #-}+error :: String -> a+error = Prelude.error

@@ -11,10 +11,11 @@ -}  {-# LANGUAGE LambdaCase          #-}-{-# LANGUAGE RecordWildCards     #-} {-# LANGUAGE ScopedTypeVariables #-} -module Network.Nakadi.Internal.Retry where+module Network.Nakadi.Internal.Retry+  ( retryAction+  ) where  import           Network.Nakadi.Internal.Prelude @@ -26,37 +27,54 @@ import qualified Network.Nakadi.Internal.Lenses  as L import           Network.Nakadi.Internal.Types +-- | Invokes the HTTP Error Callback set in the configuration for the+-- provided 'Request', 'HttpException' and 'RetryStatus'. If no+-- callback is set, this is no-op.+invokeHttpErrorCallback :: MonadIO m => Config -> Request -> HttpException -> RetryStatus -> m ()+invokeHttpErrorCallback config req exn retryStatus = liftIO $+  case config^.L.httpErrorCallback of+    Just cb -> do+      finalFailure <- applyPolicy (config^.L.retryPolicy) retryStatus >>= \case+        Just _  -> pure False+        Nothing -> pure True+      cb req exn retryStatus finalFailure+    Nothing -> pure ()+ -- | Try to execute the provided IO action using the provided retry -- policy. If executing the IO action raises specific exceptions of -- type 'HttpException', the action will be potentially retried -- (depending on the retry policy).-retryAction' :: (MonadIO m, MonadMask m) => RetryPolicyM IO -> m a -> m a-retryAction' policy ma =-  let nakadiRetryPolicy = RetryPolicyM $ \retryStatus ->+retryAction ::+  (MonadIO m, MonadMask m)+  => Config+  -> Request+  -> (Request -> m a)+  -> m a+retryAction config req ma =+  let policy = config^.L.retryPolicy+      nakadiRetryPolicy = RetryPolicyM $ \retryStatus ->         liftIO (getRetryPolicyM policy retryStatus)-      handlerHttpWrapped = const $ Handler handlerHttp-  in recovering nakadiRetryPolicy [handlerHttpWrapped] (const ma)+  in recovering nakadiRetryPolicy [handlerHttp] (const (ma req)) -  where handlerHttp (HttpExceptionRequest _ exceptionContent) =+  where handlerHttp retryStatus = Handler $ \exn -> do+          invokeHttpErrorCallback config req exn retryStatus+          pure $ shouldRetry exn++        shouldRetry (HttpExceptionRequest _ exceptionContent) =           case exceptionContent of             StatusCodeException response _ ->-              return $ responseStatus response `elem` [status500, status503]+              responseStatus response `elem` [status500, status503]             ResponseTimeout ->-              return True+              True             ConnectionTimeout ->-              return True+              True             ConnectionFailure _ ->-              return True+              True             InternalException _ ->-              return True+              True             ConnectionClosed ->-              return True+              True             _ ->-              return False--        handlerHttp _ = return False+              False --- | Try to execute the provided IO action, using the retry policy--- from the provided configuration.-retryAction :: (MonadIO m, MonadMask m) => Config -> m a -> m a-retryAction config = retryAction' (config^.L.retryPolicy)+        shouldRetry _ = False

@@ -10,6 +10,8 @@ Internal configuration specific types. -} +{-# LANGUAGE StrictData #-}+ module Network.Nakadi.Internal.Types.Config where  import           Network.Nakadi.Internal.Prelude@@ -17,12 +19,17 @@ import           Control.Retry import           Network.HTTP.Client +import qualified Data.ByteString.Lazy            as LB (ByteString) import           Network.Nakadi.Types.Logger  -- | Config  type StreamConnectCallback = Maybe LogFunc -> Response () -> IO () +-- | Type synonym for user-provided callbacks which are used for HTTP+-- Errror propagation.+type HttpErrorCallback = Request -> HttpException -> RetryStatus -> Bool -> IO ()+ data Config = Config   { _requestTemplate                :: Request   , _requestModifier                :: Request -> IO Request@@ -32,6 +39,17 @@   , _streamConnectCallback          :: Maybe StreamConnectCallback   , _logFunc                        :: Maybe LogFunc   , _retryPolicy                    :: RetryPolicyM IO+  , _http                           :: HttpBackend+  , _httpErrorCallback              :: Maybe HttpErrorCallback+  }++-- | Type encapsulating the HTTP backend functions used by this+-- package. By default the corresponding functions from the+-- http-client package are used. Useful, for e.g., testing.+data HttpBackend = HttpBackend+  { _httpLbs       :: Request -> IO (Response LB.ByteString)+  , _responseOpen  :: Request -> Manager -> IO (Response BodyReader)+  , _responseClose :: Response BodyReader -> IO ()   }  -- | ConsumeParameters

@@ -10,6 +10,8 @@ Nakadi client exceptions. -} +{-# LANGUAGE StrictData #-}+ module Network.Nakadi.Internal.Types.Exceptions where  import           Network.Nakadi.Internal.Prelude@@ -31,7 +33,7 @@                      | TooManyRequests Problem                      | BadRequest Problem                      | SubscriptionNotFound Problem-                     | CursorAlreadyCommitted [CursorCommitResult]+                     | CursorAlreadyCommitted CursorCommitResults                      | CursorResetInProgress Problem                      | EventTypeNotFound Problem                      | SubscriptionExistsAlready Subscription

@@ -13,6 +13,7 @@ {-# LANGUAGE ApplicativeDo   #-} {-# LANGUAGE DeriveGeneric   #-} {-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE StrictData      #-} {-# LANGUAGE TupleSections   #-}  module Network.Nakadi.Internal.Types.Problem where

@@ -14,6 +14,7 @@ {-# LANGUAGE DeriveGeneric         #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE LambdaCase            #-}+{-# LANGUAGE StrictData            #-} {-# LANGUAGE TemplateHaskell       #-}  module Network.Nakadi.Internal.Types.Service where@@ -219,13 +220,25 @@   parseJSON (String s) = return $ FlowId s   parseJSON invalid    = typeMismatch "FlowId" invalid +-- | ID of an Event++newtype EventId = EventId+  { unEventId :: UUID -- ^ Wrapped UUID+  } deriving (Show, Eq, Ord, Generic, Hashable)++instance ToJSON EventId where+  toJSON = String . tshow . unEventId++instance FromJSON EventId where+  parseJSON = parseUUID "EventId" EventId+ -- | Metadata  data Metadata = Metadata-  { _eid        :: Text -- ^ Event ID+  { _eid        :: EventId -- ^ Event ID   , _occurredAt :: Timestamp -- ^ Occurred-At timestamp-  , _parentEids :: Maybe [Text] -- ^ Event IDs of the Events which triggered this event-  , _partition  :: Maybe Text -- ^ Partition on which this Event is stored+  , _parentEids :: Maybe [EventId] -- ^ Event IDs of the Events which triggered this event+  , _partition  :: Maybe PartitionName -- ^ Partition on which this Event is stored   } deriving (Eq, Show, Generic)  deriveJSON nakadiJsonOptions ''Metadata@@ -242,36 +255,6 @@  deriveJSON nakadiJsonOptions ''Event --- | EventStreamBatch--data EventStreamBatch a = EventStreamBatch-  { _cursor :: Cursor -- ^ Cursor for this batch-  , _events :: Maybe (Vector (Event a)) -- ^ Events in this batch-  } deriving (Show, Generic)--deriveJSON nakadiJsonOptions ''EventStreamBatch---- | SubscriptionEventStreamBatch--data SubscriptionEventStreamBatch a = SubscriptionEventStreamBatch-  { _cursor :: SubscriptionCursor -- ^ cursor for this subscription batch-  , _events :: Maybe (Vector (Event a)) -- ^ Events for this subscription batch-  } deriving (Show, Generic)--deriveJSON nakadiJsonOptions ''SubscriptionEventStreamBatch---- | ID of an Event--newtype EventId = EventId-  { unEventId :: UUID -- ^ Wrapped UUID-  } deriving (Show, Eq, Ord, Generic, Hashable)--instance ToJSON EventId where-  toJSON = String . tshow . unEventId--instance FromJSON EventId where-  parseJSON = parseUUID "EventId" EventId- -- | Partition Data  data Partition = Partition@@ -453,6 +436,12 @@  deriveJSON nakadiJsonOptions ''CursorCommitResult +newtype CursorCommitResults = CursorCommitResults+  { _items :: [CursorCommitResult]+  } deriving (Show, Eq, Ord)++deriveJSON nakadiJsonOptions ''CursorCommitResults+ -- | SchemaType  data SchemaType = SchemaTypeJson@@ -735,12 +724,14 @@ -- | Type of enriched metadata values.  data MetadataEnriched = MetadataEnriched-  { _eid        :: Text+  { _eid        :: EventId   , _eventType  :: EventTypeName   , _occurredAt :: Timestamp   , _receivedAt :: Timestamp   , _version    :: SchemaVersion-  , _parentEids :: Maybe [Text]+  , _parentEids :: Maybe [EventId]+  , _flowId     :: Maybe FlowId+  , _partition  :: Maybe PartitionName   } deriving (Eq, Show, Generic)  deriveJSON nakadiJsonOptions ''MetadataEnriched@@ -759,6 +750,23 @@                                         , ("_metadata", "metadata") ]   }  ''EventEnriched +-- | EventStreamBatch++data EventStreamBatch a = EventStreamBatch+  { _cursor :: Cursor -- ^ Cursor for this batch+  , _events :: Maybe (Vector (EventEnriched a)) -- ^ Events in this batch+  } deriving (Show, Generic)++deriveJSON nakadiJsonOptions ''EventStreamBatch++-- | SubscriptionEventStreamBatch++data SubscriptionEventStreamBatch a = SubscriptionEventStreamBatch+  { _cursor :: SubscriptionCursor -- ^ cursor for this subscription batch+  , _events :: Maybe (Vector (EventEnriched a)) -- ^ Events for this subscription batch+  } deriving (Show, Generic)++deriveJSON nakadiJsonOptions ''SubscriptionEventStreamBatch  -- | Type for "data_op" as contained in the DataChangeEvent. 

@@ -11,6 +11,8 @@ Service API but custom to this package. -} +{-# LANGUAGE StrictData #-}+ module Network.Nakadi.Internal.Types.Subscription where  import           Network.Nakadi.Internal.Types.Config

@@ -59,7 +59,7 @@   let consumeParams = fromMaybe (config^.L.consumeParameters) maybeParams       queryParams = buildSubscriptionConsumeQueryParameters consumeParams -      addFlowId     = case _flowId consumeParams of+      addFlowId     = case consumeParams^.L.flowId of                         Just flowId -> setRequestHeader "X-Flow-Id" [encodeUtf8 flowId]                         Nothing     -> identity   httpJsonBodyStream config ok200 buildSubscriptionEventStream@@ -122,8 +122,8 @@   -> ConduitM ()               Void               (ReaderT SubscriptionEventStreamContext m)-              r              -- ^ Subscription Conduit to run-  -> m r                     -- ^ Result of the Conduit+              s              -- ^ Subscription Conduit to run+  -> m s                     -- ^ Result of the Conduit runSubscriptionR subscriptionEventStream conduit = do   config <- asks (view L.nakadiConfig)   runSubscription config subscriptionEventStream conduit

@@ -12,6 +12,7 @@  module Network.Nakadi.Types.Config   ( Config+  , HttpBackend(..)   , ConsumeParameters   , StreamConnectCallback   ) where

@@ -45,6 +45,7 @@   , BatchFlushTimeout(..)   , CursorCommitResultType(..)   , CursorCommitResult(..)+  , CursorCommitResults(..)   , SchemaType(..)   , EventTypeSchema(..)   , PaginationLink(..)

@@ -11,7 +11,7 @@ -}  module Network.Nakadi.Types.Subscription-  ( SubscriptionEventStreamContext+  ( SubscriptionEventStreamContext(..)   ) where  import           Network.Nakadi.Internal.Types.Subscription

@@ -0,0 +1,48 @@+module Network.Nakadi.Config.Test where++import           ClassyPrelude+import           Control.Lens         ((<&>))+import           Control.Retry+import qualified Data.ByteString.Lazy as LB+import           Network.HTTP.Client+import           Network.Nakadi+import           System.IO.Unsafe+import           Test.Tasty+import           Test.Tasty.HUnit++testConfig :: TestTree+testConfig = testGroup "Config"+  [ testCase "Use Custom HttpBackend" testCustomHttpBackend ]++{-# NOINLINE requestsExecuted #-}+requestsExecuted :: TVar [Request]+requestsExecuted = unsafePerformIO . newTVarIO $ []++dummyHttpLbs :: Request -> IO (Response LB.ByteString)+dummyHttpLbs req = do+  atomically $ modifyTVar requestsExecuted (req :)+  throwIO (HttpExceptionRequest req ResponseTimeout)++dummyResponseOpen :: Request -> Manager -> IO (Response (IO ByteString))+dummyResponseOpen req _manager = do+  atomically $ modifyTVar requestsExecuted (req :)+  throwIO (HttpExceptionRequest req ResponseTimeout)++dummyResponseClose :: Response BodyReader -> IO ()+dummyResponseClose _response = return ()++dummyHttpBackend :: HttpBackend+dummyHttpBackend = HttpBackend dummyHttpLbs dummyResponseOpen dummyResponseClose++testCustomHttpBackend :: Assertion+testCustomHttpBackend = do+  let trivialRetryPolicy = limitRetries 0+  conf <- newConfig Nothing defaultRequest+          <&> setHttpBackend dummyHttpBackend+          <&> setRetryPolicy trivialRetryPolicy+  res0 <- try $ registryPartitionStrategies conf -- This should use httpLbs+  case res0 of+    Left (HttpExceptionRequest _ ResponseTimeout) -> return ()+    _ -> assertFailure "Expected ResponseTimeout exception from dummy HttpBackend"+  requests <- atomically . readTVar $ requestsExecuted+  1 @=? length requests

@@ -0,0 +1,93 @@+module Network.Nakadi.Connection.Test where++import           ClassyPrelude+import           Network.HTTP.Client      (HttpException (..),+                                           HttpExceptionContent (..),+                                           Request (..), parseRequest,+                                           responseTimeoutMicro)+import           Network.HTTP.Types+import           Network.Nakadi+import qualified Network.Wai              as Wai+import           Network.Wai.Handler.Warp+import           System.IO.Unsafe+import           Test.Tasty+import           Test.Tasty.HUnit++testConnection :: TestTree+testConnection = testGroup "Connection"+  [ testCase "SimpleRetry" testSimpleRetry+  , testCase "ResponseTimeout Success" testResponseTimeoutSuccess+  , testCase "ResponseTimeout Fail" testResponseTimeoutFail+  ]++{-# NOINLINE requestCounter #-}+requestCounter :: TVar Int+requestCounter = unsafePerformIO $ newTVarIO 0++testServerRetryPort :: Port+testServerRetryPort = 5001++testServerResponseTimeoutPort :: Port+testServerResponseTimeoutPort = 5002++{-# NOINLINE testServerRequest #-}+testServerRequest :: Request+testServerRequest = unsafePerformIO $ parseRequest "http://localhost"++testServerRetryApp ::+  Int+  -> Wai.Request+  -> (Wai.Response -> IO Wai.ResponseReceived)+  -> IO Wai.ResponseReceived+testServerRetryApp noOfFailures req respond =+  case Wai.rawPathInfo req of+    "/event-types" -> do+      reqIdx <- atomically $ do+        modifyTVar requestCounter (+ 1)+        readTVar requestCounter+      let response = if reqIdx <= noOfFailures+                     then Wai.responseLBS status503 [] ""+                     else Wai.responseLBS status200 [] "[]"+      respond response+    _ ->+      respond $ Wai.responseLBS status404 [] ""++testServerResponseTimeoutApp ::+  Wai.Request+  -> (Wai.Response -> IO Wai.ResponseReceived)+  -> IO Wai.ResponseReceived+testServerResponseTimeoutApp req respond =+  case Wai.rawPathInfo req of+    "/event-types" -> do+      threadDelay (4 * 10^6) -- Wait for 4s+      respond $ Wai.responseLBS status200 [] "[]"+    _ ->+      respond $ Wai.responseLBS status404 [] ""++testSimpleRetry :: Assertion+testSimpleRetry = do+  conf <- newConfig Nothing testServerRequest { port = testServerRetryPort }+  withAsync (run testServerRetryPort (testServerRetryApp 2)) $ \_serverHandle -> do+    events <- eventTypesList conf+    [] @=? events++testResponseTimeoutSuccess :: Assertion+testResponseTimeoutSuccess = do+  let timeout = responseTimeoutMicro (5 * 10^6) -- Accept delay of 5s+  conf <- newConfig Nothing testServerRequest { port = testServerResponseTimeoutPort+                                              , responseTimeout = timeout }+  withAsync (run testServerResponseTimeoutPort testServerResponseTimeoutApp) $ \_serverHandle -> do+    events <- eventTypesList conf+    [] @=? events++testResponseTimeoutFail :: Assertion+testResponseTimeoutFail = do+  res <- try $ do+    let timeout = responseTimeoutMicro (3 * 10^6) -- Accept delay of 3s+    conf <- newConfig Nothing testServerRequest { port = testServerResponseTimeoutPort+                                                , responseTimeout = timeout }+    withAsync (run testServerResponseTimeoutPort testServerResponseTimeoutApp) $ \_serverHandle -> do+      eventTypesList conf+  case res of+    Left (HttpExceptionRequest _request ResponseTimeout) -> return ()+    _ -> assertFailure "Expected HttpExceptionRequest with content ResponseTimeout"

@@ -36,7 +36,7 @@ testCursorsLagN :: Config -> Int64 -> Assertion testCursorsLagN conf n = do   now <- getCurrentTime-  eid <- tshow <$> genRandomUUID+  eid <- EventId <$> genRandomUUID   recreateEvent conf myEventTypeName myEventType   partitions <- eventTypePartitions conf myEventTypeName   let cursorsMap = Map.fromList $

@@ -10,9 +10,6 @@  import           ClassyPrelude -import           Conduit-import           Control.Concurrent.Async    (link)-import           Data.Function               ((&)) import           Network.Nakadi import           Network.Nakadi.Tests.Common import           Test.Tasty@@ -35,7 +32,7 @@ testShiftedCursorsN :: Config -> Int64 -> Assertion testShiftedCursorsN conf n = do   now <- getCurrentTime-  eid <- tshow <$> genRandomUUID+  eid <- EventId <$> genRandomUUID   recreateEvent conf myEventTypeName myEventType   partitions <- eventTypePartitions conf myEventTypeName   let cursors = map extractCursor partitions

@@ -32,6 +32,7 @@   , testCase "EventTypeCursorDistances0" (testEventTypeCursorDistances0 conf)   , testCase "EventTypeCursorDistances10" (testEventTypeCursorDistances10 conf)   , testCase "EventTypePublishData" (testEventTypePublishData conf)+  , testCase "EventTypeParseFlowId" (testEventTypeParseFlowId conf)   , testCase "EventTypeDeserializationFailure" (testEventTypeDeserializationFailure conf)   , testEventTypesShiftedCursors conf   , testEventTypesCursorsLag conf@@ -84,7 +85,7 @@    forM_ [1..10] $ \_ -> do     now <- getCurrentTime-    eid <- tshow <$> genRandomUUID+    eid <- EventId <$> genRandomUUID     eventPublish conf myEventTypeName Nothing [myDataChangeEvent eid now]    cursorPairs <- forM cursors $ \cursor@Cursor { .. } -> do@@ -106,7 +107,7 @@ testEventTypePublishData :: Config -> Assertion testEventTypePublishData conf = do   now <- getCurrentTime-  eid <- tshow <$> genRandomUUID+  eid <- EventId <$> genRandomUUID   eventTypeDelete conf myEventTypeName `catch` (ignoreExnNotFound ())   eventTypeCreate conf myEventType   let event = DataChangeEvent { _payload = Foo "Hello!"@@ -118,21 +119,49 @@                               , _dataType = "test.FOO"                               , _dataOp = DataOpUpdate                               }-  withAsync (delayedPublish [event]) $ \asyncHandle -> do+  withAsync (delayedPublish conf Nothing [event]) $ \asyncHandle -> do     link asyncHandle     eventConsumed :: Maybe (EventStreamBatch Foo) <- runResourceT $ do       source <- eventSource conf (Just consumeParametersSingle) myEventTypeName Nothing       runConduit $ source .| headC     isJust eventConsumed @=? True -  where delayedPublish events = do-          threadDelay (10^6)-          eventPublish conf myEventTypeName Nothing events+testEventTypeParseFlowId :: Config -> Assertion+testEventTypeParseFlowId conf = do+  now <- getCurrentTime+  eid <- EventId <$> genRandomUUID+  eventTypeDelete conf myEventTypeName `catch` (ignoreExnNotFound ())+  eventTypeCreate conf myEventType+  let event = DataChangeEvent { _payload = Foo "Hello!"+                              , _metadata = Metadata { _eid = eid+                                                     , _occurredAt = Timestamp now+                                                     , _parentEids = Nothing+                                                     , _partition = Nothing+                                                     }+                              , _dataType = "test.FOO"+                              , _dataOp = DataOpUpdate+                              }+      expectedFlowId = Just $ FlowId "12345"+  withAsync (delayedPublish conf expectedFlowId [event]) $ \asyncHandle -> do+    link asyncHandle+    eventConsumed :: Maybe (EventStreamBatch Foo) <- runResourceT $ do+      source <- eventSource conf (Just consumeParametersSingle) myEventTypeName Nothing+      runConduit $ source .| headC+    isJust eventConsumed @=? True+    let events = eventConsumed >>= (\batch -> batch^.L.events)+    isJust events @=? True +    case events of+      Nothing -> assertFailure "Received no events"+      Just v -> case toList v of+        [EventEnriched _ x] ->+          x^.L.flowId @=? expectedFlowId+        _ -> assertFailure "Received not a singleton event list"+ testEventTypeDeserializationFailure :: Config -> Assertion testEventTypeDeserializationFailure conf' = do   now <- getCurrentTime-  eid <- tshow <$> genRandomUUID+  eid <- EventId <$> genRandomUUID   eventTypeDelete conf myEventTypeName `catch` (ignoreExnNotFound ())   eventTypeCreate conf myEventType   let event = DataChangeEvent { _payload = Foo "Hello!"@@ -144,7 +173,7 @@                               , _dataType = "test.FOO"                               , _dataOp = DataOpUpdate                               }-  withAsync (delayedPublish [event]) $ \asyncHandle -> do+  withAsync (delayedPublish conf Nothing [event]) $ \asyncHandle -> do     link asyncHandle     eventConsumed :: Maybe (EventStreamBatch WrongFoo) <- runResourceT $ do       source <- eventSource conf (Just consumeParametersSingle) myEventTypeName Nothing@@ -154,14 +183,10 @@   counter <- atomically $ readTVar deserializationFailureCounter   1 @=? counter -  where delayedPublish events = do-          threadDelay (10^6)-          eventPublish conf myEventTypeName Nothing events--        conf = conf'+  where conf = conf'                & setDeserializationFailureCallback (deserializationFailureCb deserializationFailureCounter) -        deserializationFailureCb counter _ errMsg = do+        deserializationFailureCb counter _ _ =           atomically $ modifyTVar counter (+ 1)          deserializationFailureCounter = unsafePerformIO $ newTVarIO 0

@@ -0,0 +1,58 @@+{-# LANGUAGE OverloadedStrings     #-}+{-# LANGUAGE ScopedTypeVariables   #-}+{-# LANGUAGE RecordWildCards       #-}++module Network.Nakadi.Internal.Http.Test+  ( testHttp+  ) where++import           Network.HTTP.Client+import           Network.HTTP.Types+import           Test.Tasty+import           Test.Tasty.HUnit+import           Control.Monad+import           Control.Monad.Reader+import           ClassyPrelude+import           Network.Nakadi.Internal.Http+import           Network.HTTP.Client.Internal (CookieJar (..), Request(..), Response (..), ResponseClose (..))+import           Network.Nakadi+import           Conduit++testHttp :: TestTree+testHttp = testGroup "Http"+  [ testCase "HttpRequestModifier" testHttpRequestModifier+  ]++resp :: Response (IO ByteString)+resp = Response+    { responseStatus = status200+    , responseVersion = http11+    , responseHeaders = []+    , responseBody = pure ""+    , responseCookieJar = CJ []+    , responseClose' = ResponseClose (pure ())+    }++headers :: RequestHeaders+headers = [("test-header", "header-value")]++dummyResponseOpen :: Request -> Manager -> IO (Response (IO ByteString))+dummyResponseOpen Request { .. } _ = do+  requestHeaders @=? headers+  pure resp++dummyHttpBackend :: HttpBackend+dummyHttpBackend = defaultHttpBackend {+    _responseOpen = dummyResponseOpen+}++dummyRequestModifier :: Request -> IO Request+dummyRequestModifier request = pure (request { requestHeaders = headers })++testHttpRequestModifier :: Assertion+testHttpRequestModifier = do+    conf <- newConfig Nothing defaultRequest+    let config = conf {_http = dummyHttpBackend, _requestModifier = dummyRequestModifier }+    (_, source) <- runResourceT $ httpJsonBodyStream config ok200 (const (Right ())) [] id+    (_ :: Maybe Text) <- runResourceT $ runReaderT (runConduit $ source .| headC) ()+    return ()

@@ -0,0 +1,104 @@+{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Network.Nakadi.Internal.Retry.Test+  ( testRetry+  ) where++import           ClassyPrelude+import           Control.Lens+import           Control.Retry+import qualified Data.ByteString.Lazy          as LB+import           Network.HTTP.Client+import           Network.HTTP.Client.Internal  (CookieJar (..), Request (..),+                                                Response (..),+                                                ResponseClose (..))+import           Network.HTTP.Types+import           Network.Nakadi+import           Network.Nakadi.Internal.Retry+import           Test.Tasty+import           Test.Tasty.HUnit++testRetry :: TestTree+testRetry = testGroup "Retry"+  [ testCase "HttpErrorCallback not called on success" testHttpErrorCallback0+  , testCase "HttpErrorCallback called on failure (n = 1)" testHttpErrorCallback1+  , testCase "HttpErrorCallback called on failure (n = maxRetries)" testHttpErrorCallbackMax+  , testCase "HttpErrorCallback called on failure (n = maxRetries + 1)" testHttpErrorCallbackMaxPlusOne+  ]++prepareMockResponse ::+  Monoid a+  => Int -- ^ Fail this many times prior to success+  -> IO (IO (Response a))+prepareMockResponse successNum = do+  counter <- newTVarIO 0+  return $ do+    current <- atomically $ do+      n <- readTVar counter+      modifyTVar counter (+ 1)+      return n+    if current >= successNum+    then return responseSuccess+    else throwM (HttpExceptionRequest defaultRequest (StatusCodeException responseFailure mempty))++  where responseTemplate = Response+          { responseStatus    = status200+          , responseVersion   = http11+          , responseHeaders   = []+          , responseBody      = mempty+          , responseCookieJar = CJ []+          , responseClose'    = ResponseClose (pure ())+          }+        responseSuccess = responseTemplate+        responseFailure = void $ responseTemplate { responseStatus = status503 }++maxRetries :: Int+maxRetries = 7++httpErrorCallback :: TVar Int -> Request -> HttpException -> RetryStatus -> Bool -> IO ()+httpErrorCallback tvCounter _request _exn _retryStatus finalFailure = do+  current <- atomically $ do+    n <- readTVar tvCounter+    modifyTVar tvCounter (+ 1)+    return n+  if current == maxRetries+     then True @=? finalFailure+     else False @=? finalFailure++mockHttpBackend :: Int -> IO HttpBackend+mockHttpBackend numFailures = do+  responder <- prepareMockResponse numFailures+  return HttpBackend+    { _httpLbs       = mockHttpLbs responder+    , _responseOpen  = responseOpen+    , _responseClose = responseClose+    }++    where mockHttpLbs responder _request = responder++-- | Tests that the callback is called exactly numFailures times+-- before the request succeeds — depending on the retry policy.+testHttpErrorCallbackN :: Int -> Assertion+testHttpErrorCallbackN numFailures = do+  httpBackend <- mockHttpBackend numFailures+  counter <- newTVarIO 0+  conf <- newConfig Nothing defaultRequest+          <&> setHttpErrorCallback (httpErrorCallback counter)+          <&> setRetryPolicy (fullJitterBackoff 2 ++ limitRetries maxRetries)+  _response :: Either HttpException (Response LB.ByteString) <- try $+    retryAction conf defaultRequest (_httpLbs httpBackend)+  current <- readTVarIO counter+  numFailures @=? current++testHttpErrorCallback0 :: Assertion+testHttpErrorCallback0 = testHttpErrorCallbackN 0++testHttpErrorCallback1 :: Assertion+testHttpErrorCallback1 = testHttpErrorCallbackN 1++testHttpErrorCallbackMax :: Assertion+testHttpErrorCallbackMax = testHttpErrorCallbackN maxRetries++testHttpErrorCallbackMaxPlusOne :: Assertion+testHttpErrorCallbackMaxPlusOne = testHttpErrorCallbackN (maxRetries + 1)

@@ -0,0 +1,13 @@+module Network.Nakadi.Internal.Test where++import           Network.Nakadi.Internal.Http.Test+import           Network.Nakadi.Internal.Retry.Test+import           Network.Nakadi.Internal.Types.Test+import           Test.Tasty++testInternal :: TestTree+testInternal = testGroup "Internal"+  [ testTypes+  , testHttp+  , testRetry+  ]

@@ -0,0 +1,23 @@+{-# LANGUAGE OverloadedStrings     #-}++module Network.Nakadi.Internal.Types.Test where++import ClassyPrelude+import Test.Tasty+import Test.Tasty.HUnit+import Data.Aeson+import Network.Nakadi.Types.Service++testTypes :: TestTree+testTypes = testGroup "Types"+  [ testService+  ]++testService :: TestTree+testService = testGroup "Service"+  [ testCase "JSON-decoding CursorCommitResults" testDecodeCursorCommitResults+  ]++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\"}]}"

@@ -56,7 +56,7 @@   Cursor { _partition = _partition          , _offset    = _newestAvailableOffset } -myDataChangeEvent :: Text -> UTCTime -> DataChangeEvent Foo+myDataChangeEvent :: EventId -> UTCTime -> DataChangeEvent Foo myDataChangeEvent eid now =  DataChangeEvent   { _payload = Foo "Hello!"   , _metadata = Metadata { _eid        = eid@@ -75,3 +75,8 @@ recreateEvent conf eventTypeName eventType = do   eventTypeDelete conf eventTypeName `catch` (ignoreExnNotFound ())   eventTypeCreate conf eventType++delayedPublish :: (ToJSON a) => Config -> Maybe FlowId -> [a] -> IO ()+delayedPublish conf flowId events  = do+  threadDelay (10^6)+  eventPublish conf myEventTypeName flowId events

@@ -1,28 +1,71 @@ import           ClassyPrelude -import           Control.Lens import           Network.HTTP.Client import           Network.Nakadi+import           Network.Nakadi.Config.Test+import           Network.Nakadi.Connection.Test import           Network.Nakadi.EventTypes.Test+import           Network.Nakadi.Internal.Test import           Network.Nakadi.Registry.Test import           Network.Nakadi.Subscriptions.Test import           System.Environment+import           System.Exit+import           System.IO                         (hFlush) import           Test.Tasty -createConfig :: IO Config-createConfig = do-  nakadiEndpoint <- fromMaybe (error "TEST_NAKADI_ENDPOINT not set") <$>-                    lookupEnv "TEST_NAKADI_ENDPOINT"-  request <- parseRequest nakadiEndpoint-  newConfig Nothing request- main :: IO () main = do-  conf <- createConfig-  defaultMain (tests conf)+  putStrLn ""+  maybeNakadiEndpoint <- lookupEnv "TEST_NAKADI_ENDPOINT"+  (label, runTests) <- case maybeNakadiEndpoint of+    Nothing -> do+      hPut stderr . encodeUtf8 $+        "***************************************************************\n\+        \** The environment variable TEST_NAKADI_ENDPOINT is not set. **\n\+        \**                                                           **\n\+        \**              SKIPPING INTEGRATION TESTS                   **\n\+        \**                                                           **\n\+        \** If you want to run the integration tests contained in     **\n\+        \** this test suite, you need to run a Nakadi server which    **\n\+        \** allows access without authentication. Then, set the       **\n\+        \** environment variable TEST_NAKADI_ENDPOINT to the base     **\n\+        \** endpoint of this Nakadi server.                           **\n\+        \**                                                           **\n\+        \** To run a local Nakadi server clone the repository         **\n\+        \**                                                           **\n\+        \**     https://github.com/zalando/nakadi                     **\n\+        \**                                                           **\n\+        \** and follow the contained instructions; the command        **\n\+        \**                                                           **\n\+        \**     ./gradlew startNakadi                                 **\n\+        \**                                                           **\n\+        \** starts a local Nakadi server at http://localhost:8080.    **\n\+        \**                                                           **\n\+        \** Thus, set TEST_NAKADI_ENDPOINT to http://localhost:8080.  **\n\+        \***************************************************************\n"+      return ("nakadi-client Test Suite (w/o integration tests)", unitTests)+    Just nakadiEndpoint -> do+      let nakadiEndpointT = pack nakadiEndpoint+      case parseRequest nakadiEndpoint of+        Just request -> do+          conf <- newConfig Nothing request+          return ("nakadi-client Test Suite", unitTests ++ integrationTests conf)+        Nothing -> do+          hPut stderr . encodeUtf8 $+            "Failed to parse Nakadi URL in TEST_NAKADI_ENDPOINT (" <> nakadiEndpointT <> ")"+          hFlush stderr+          exitFailure+  defaultMain (testGroup label runTests) -tests :: Config -> TestTree-tests conf = testGroup "Test Suite"+unitTests :: [TestTree]+unitTests =+  [ testInternal+  , testConfig+  , testConnection+  ]++integrationTests :: Config -> [TestTree]+integrationTests conf =   [ testEventTypes conf   , testRegistry conf   , testSubscriptions conf