diff --git a/nakadi-client.cabal b/nakadi-client.cabal --- a/nakadi-client.cabal +++ b/nakadi-client.cabal @@ -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 diff --git a/src/Network/Nakadi/Config.hs b/src/Network/Nakadi/Config.hs --- a/src/Network/Nakadi/Config.hs +++ b/src/Network/Nakadi/Config.hs @@ -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 diff --git a/src/Network/Nakadi/Internal/Http.hs b/src/Network/Nakadi/Internal/Http.hs --- a/src/Network/Nakadi/Internal/Http.hs +++ b/src/Network/Nakadi/Internal/Http.hs @@ -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 diff --git a/src/Network/Nakadi/Internal/Lenses.hs b/src/Network/Nakadi/Internal/Lenses.hs --- a/src/Network/Nakadi/Internal/Lenses.hs +++ b/src/Network/Nakadi/Internal/Lenses.hs @@ -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 diff --git a/src/Network/Nakadi/Internal/Prelude.hs b/src/Network/Nakadi/Internal/Prelude.hs --- a/src/Network/Nakadi/Internal/Prelude.hs +++ b/src/Network/Nakadi/Internal/Prelude.hs @@ -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 diff --git a/src/Network/Nakadi/Internal/Retry.hs b/src/Network/Nakadi/Internal/Retry.hs --- a/src/Network/Nakadi/Internal/Retry.hs +++ b/src/Network/Nakadi/Internal/Retry.hs @@ -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 diff --git a/src/Network/Nakadi/Internal/Types/Config.hs b/src/Network/Nakadi/Internal/Types/Config.hs --- a/src/Network/Nakadi/Internal/Types/Config.hs +++ b/src/Network/Nakadi/Internal/Types/Config.hs @@ -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 diff --git a/src/Network/Nakadi/Internal/Types/Exceptions.hs b/src/Network/Nakadi/Internal/Types/Exceptions.hs --- a/src/Network/Nakadi/Internal/Types/Exceptions.hs +++ b/src/Network/Nakadi/Internal/Types/Exceptions.hs @@ -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 diff --git a/src/Network/Nakadi/Internal/Types/Problem.hs b/src/Network/Nakadi/Internal/Types/Problem.hs --- a/src/Network/Nakadi/Internal/Types/Problem.hs +++ b/src/Network/Nakadi/Internal/Types/Problem.hs @@ -13,6 +13,7 @@ {-# LANGUAGE ApplicativeDo #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE StrictData #-} {-# LANGUAGE TupleSections #-} module Network.Nakadi.Internal.Types.Problem where diff --git a/src/Network/Nakadi/Internal/Types/Service.hs b/src/Network/Nakadi/Internal/Types/Service.hs --- a/src/Network/Nakadi/Internal/Types/Service.hs +++ b/src/Network/Nakadi/Internal/Types/Service.hs @@ -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. diff --git a/src/Network/Nakadi/Internal/Types/Subscription.hs b/src/Network/Nakadi/Internal/Types/Subscription.hs --- a/src/Network/Nakadi/Internal/Types/Subscription.hs +++ b/src/Network/Nakadi/Internal/Types/Subscription.hs @@ -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 diff --git a/src/Network/Nakadi/Subscriptions/Events.hs b/src/Network/Nakadi/Subscriptions/Events.hs --- a/src/Network/Nakadi/Subscriptions/Events.hs +++ b/src/Network/Nakadi/Subscriptions/Events.hs @@ -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 diff --git a/src/Network/Nakadi/Types/Config.hs b/src/Network/Nakadi/Types/Config.hs --- a/src/Network/Nakadi/Types/Config.hs +++ b/src/Network/Nakadi/Types/Config.hs @@ -12,6 +12,7 @@ module Network.Nakadi.Types.Config ( Config + , HttpBackend(..) , ConsumeParameters , StreamConnectCallback ) where diff --git a/src/Network/Nakadi/Types/Service.hs b/src/Network/Nakadi/Types/Service.hs --- a/src/Network/Nakadi/Types/Service.hs +++ b/src/Network/Nakadi/Types/Service.hs @@ -45,6 +45,7 @@ , BatchFlushTimeout(..) , CursorCommitResultType(..) , CursorCommitResult(..) + , CursorCommitResults(..) , SchemaType(..) , EventTypeSchema(..) , PaginationLink(..) diff --git a/src/Network/Nakadi/Types/Subscription.hs b/src/Network/Nakadi/Types/Subscription.hs --- a/src/Network/Nakadi/Types/Subscription.hs +++ b/src/Network/Nakadi/Types/Subscription.hs @@ -11,7 +11,7 @@ -} module Network.Nakadi.Types.Subscription - ( SubscriptionEventStreamContext + ( SubscriptionEventStreamContext(..) ) where import Network.Nakadi.Internal.Types.Subscription diff --git a/tests/Network/Nakadi/Config/Test.hs b/tests/Network/Nakadi/Config/Test.hs new file mode 100644 --- /dev/null +++ b/tests/Network/Nakadi/Config/Test.hs @@ -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 diff --git a/tests/Network/Nakadi/Connection/Test.hs b/tests/Network/Nakadi/Connection/Test.hs new file mode 100644 --- /dev/null +++ b/tests/Network/Nakadi/Connection/Test.hs @@ -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" diff --git a/tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs b/tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs --- a/tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs +++ b/tests/Network/Nakadi/EventTypes/CursorsLag/Test.hs @@ -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 $ diff --git a/tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs b/tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs --- a/tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs +++ b/tests/Network/Nakadi/EventTypes/ShiftedCursors/Test.hs @@ -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 diff --git a/tests/Network/Nakadi/EventTypes/Test.hs b/tests/Network/Nakadi/EventTypes/Test.hs --- a/tests/Network/Nakadi/EventTypes/Test.hs +++ b/tests/Network/Nakadi/EventTypes/Test.hs @@ -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 diff --git a/tests/Network/Nakadi/Internal/Http/Test.hs b/tests/Network/Nakadi/Internal/Http/Test.hs new file mode 100644 --- /dev/null +++ b/tests/Network/Nakadi/Internal/Http/Test.hs @@ -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 () diff --git a/tests/Network/Nakadi/Internal/Retry/Test.hs b/tests/Network/Nakadi/Internal/Retry/Test.hs new file mode 100644 --- /dev/null +++ b/tests/Network/Nakadi/Internal/Retry/Test.hs @@ -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) diff --git a/tests/Network/Nakadi/Internal/Test.hs b/tests/Network/Nakadi/Internal/Test.hs new file mode 100644 --- /dev/null +++ b/tests/Network/Nakadi/Internal/Test.hs @@ -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 + ] diff --git a/tests/Network/Nakadi/Internal/Types/Test.hs b/tests/Network/Nakadi/Internal/Types/Test.hs new file mode 100644 --- /dev/null +++ b/tests/Network/Nakadi/Internal/Types/Test.hs @@ -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\"}]}" diff --git a/tests/Network/Nakadi/Tests/Common.hs b/tests/Network/Nakadi/Tests/Common.hs --- a/tests/Network/Nakadi/Tests/Common.hs +++ b/tests/Network/Nakadi/Tests/Common.hs @@ -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 diff --git a/tests/Tests.hs b/tests/Tests.hs --- a/tests/Tests.hs +++ b/tests/Tests.hs @@ -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