diff --git a/nakadi-client.cabal b/nakadi-client.cabal --- a/nakadi-client.cabal +++ b/nakadi-client.cabal @@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: 74ff60305e7890a09afb58a9d855f782f960dab69c2ca9901570da82116a91f2 +-- hash: 15f2229a93e024c41a3489a776b98ade41ee2e1b9f5302df4afd82ac5844392e name: nakadi-client -version: 0.6.1.0 +version: 0.7.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 @@ -13,7 +13,7 @@ bug-reports: https://github.com/mtesseract/nakadi-haskell/issues author: Moritz Clasmeier maintainer: mtesseract@silverratio.net -copyright: (c) 2017, 2018 Moritz Clasmeier +copyright: (c) 2017, 2018, 2019 Moritz Clasmeier license: BSD3 license-file: LICENSE build-type: Simple @@ -115,6 +115,7 @@ , http-types , iso8601-time , lens + , modern-uri >=0.2.1.0 && <0.4 , monad-control , monad-logger , mtl @@ -158,6 +159,7 @@ Network.Nakadi.Internal.Http.Test Network.Nakadi.Internal.Retry.Test Network.Nakadi.Internal.Test + Network.Nakadi.Internal.Types.Problem.Test Network.Nakadi.Internal.Types.Test Network.Nakadi.MonadicAPI.Test Network.Nakadi.Registry.Test @@ -192,6 +194,7 @@ , iso8601-time , lens , lens-aeson + , modern-uri >=0.2.1.0 && <0.4 , monad-control , monad-logger , mtl 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 @@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Config Description : Nakadi Client Configuration -Copyright : (c) Moritz Clasmeier 2017, 2018 +Copyright : (c) Moritz Clasmeier 2017, 2018, 2019 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental @@ -32,6 +32,7 @@ , setStreamKeepAliveLimit , setFlowId , setCommitStrategy + , setCommitTimeout , setShowTimeLag ) where @@ -114,6 +115,9 @@ setCommitStrategy :: CommitStrategy -> Config m -> Config m setCommitStrategy = (L.commitStrategy .~) +setCommitTimeout :: CommitTimeout -> Config m -> Config m +setCommitTimeout = (L.commitTimeout ?~) + -- | 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 @@ -146,4 +150,4 @@ setStreamKeepAliveLimit = (L.streamKeepAliveLimit ?~) setShowTimeLag :: Bool -> Config m -> Config m -setShowTimeLag flag = L.subscriptionStats ?~ SubscriptionStatsConf {_showTimeLag = flag} +setShowTimeLag flag = L.subscriptionStats ?~ SubscriptionStatsConf { _showTimeLag = flag } diff --git a/src/Network/Nakadi/Internal/Config.hs b/src/Network/Nakadi/Internal/Config.hs --- a/src/Network/Nakadi/Internal/Config.hs +++ b/src/Network/Nakadi/Internal/Config.hs @@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Internal.Config Description : Nakadi Client Configuration (Internal) -Copyright : (c) Moritz Clasmeier 2017, 2018 +Copyright : (c) Moritz Clasmeier 2017, 2018, 2019 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental @@ -33,6 +33,7 @@ , ("stream_timeout", ) . encodeUtf8 . tshow <$> _streamTimeout , ("max_uncommitted_events", ) . encodeUtf8 . tshow <$> _maxUncommittedEvents , ("stream_keep_alive_limit", ) . encodeUtf8 . tshow <$> _streamKeepAliveLimit + , ("commit_timeout", ) . encodeUtf8 . tshow . unCommitTimeout <$> _commitTimeout ] newConfigFromEnv :: (MonadIO m, MonadThrow m, MonadMask b, MonadIO b) => m (Config b) @@ -50,7 +51,7 @@ -- | Default worker configuration. This specifies single-threaded consumption of subscriptions. defaultWorkerConfig :: WorkerConfig -defaultWorkerConfig = WorkerConfig {_nThreads = 1} +defaultWorkerConfig = WorkerConfig { _nThreads = 1 } -- | Producs a new configuration, with mandatory HTTP manager, default -- consumption parameters and HTTP request template. @@ -59,24 +60,24 @@ => 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 - } +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 + , _commitTimeout = Nothing + , _subscriptionStats = Nothing + , _maxUncommittedEvents = Nothing + , _batchLimit = Nothing + , _streamLimit = Nothing + , _batchFlushTimeout = Nothing + , _streamTimeout = Nothing + , _streamKeepAliveLimit = Nothing + , _worker = defaultWorkerConfig + } 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 @@ -47,6 +47,7 @@ , _httpErrorCallback :: Maybe (HttpErrorCallback m) , _flowId :: Maybe FlowId , _commitStrategy :: CommitStrategy + , _commitTimeout :: Maybe CommitTimeout , _subscriptionStats :: Maybe SubscriptionStatsConf , _maxUncommittedEvents :: Maybe Int32 , _batchLimit :: Maybe Int32 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 @@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Internal.Types.Problem Description : Nakadi Client Problem Type (Internal) -Copyright : (c) Moritz Clasmeier 2017 +Copyright : (c) Moritz Clasmeier 2017, 2019 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental @@ -15,50 +15,83 @@ {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE StrictData #-} {-# LANGUAGE TupleSections #-} +{-# LANGUAGE LambdaCase #-} module Network.Nakadi.Internal.Types.Problem where import Data.Aeson -import Data.HashMap.Lazy (HashMap) -import Data.Int -import Data.Text (Text) +import Data.Aeson.Types +import Data.HashMap.Lazy ( HashMap ) +import Data.Text ( Text ) import Prelude +import Network.HTTP.Types.Status as HTTP + ( Status ) -import qualified Data.HashMap.Lazy as HashMap +import qualified Data.HashMap.Lazy as HashMap import Data.Maybe +import qualified Text.URI as URI +import Text.URI ( URI ) import GHC.Generics -- | Type for RFC7807 @Problem@ objects. data Problem = Problem - { problemType :: Text - , problemTitle :: Text - , problemStatus :: Int32 - , problemDetail :: Maybe Text - , problemInstance :: Maybe Text - , problemCustom :: Maybe (HashMap Text Value) + { problemType :: Maybe URI -- ^ (string) - A URI reference [RFC3986] that identifies the + -- problem type. This specification encourages that, when + -- dereferenced, it provide human-readable documentation for the + -- problem type (e.g., using HTML [W3C.REC-html5-20141028]). When + -- this member is not present, its value is assumed to be + -- "about:blank". + , problemTitle :: Text -- ^ (string) - A short, human-readable summary of the problem + -- type. It SHOULD NOT change from occurrence to occurrence of the + -- problem, except for purposes of localization (e.g., using + -- proactive content negotiation; see [RFC7231], Section 3.4). + , problemStatus :: Maybe HTTP.Status -- ^ "status" (number) - The HTTP status code ([RFC7231], Section 6) + -- generated by the origin server for this occurrence of the problem. + , problemDetail :: Maybe Text -- ^ (string) - A human-readable explanation specific to this + -- occurrence of the problem. + , problemInstance :: Maybe URI -- ^ (string) - A URI reference that identifies the specific + -- occurrence of the problem. It may or may not yield further + -- information if dereferenced. + , problemCustom :: HashMap Text Value } deriving (Show, Eq, Generic) instance ToJSON Problem where - toJSON Problem { .. } = - let hm = HashMap.fromList ([ ("type", String problemType) - , ("title", String problemTitle) - , ("status", Number (fromIntegral problemStatus)) ] - ++ catMaybes [ ("detail",) . String <$> problemDetail - , ("instance",) . String <$> problemInstance ]) - in Object (maybe hm (HashMap.union hm) problemCustom) + toJSON Problem {..} = + let hm = HashMap.fromList + (("title", String problemTitle) : catMaybes + [ ("type", ) . String . URI.render <$> problemType + , ("status", ) . Number . fromIntegral . fromEnum <$> problemStatus + , ("detail", ) . String <$> problemDetail + , ("instance", ) . String . URI.render <$> problemInstance + ] + ) + in Object (HashMap.union hm problemCustom) instance FromJSON Problem where - parseJSON = withObject "Problem" $ \obj -> do - let custom = HashMap.filterWithKey (\ k _ -> k `notElem` ["type", "title", "status"]) obj - type' <- obj .: "type" - title <- obj .: "title" - status <- obj .: "status" - detail <- obj .:? "detail" - instance' <- obj .:? "instance" - pure Problem { problemType = type' - , problemTitle = title - , problemStatus = status - , problemDetail = detail - , problemInstance = instance' - , problemCustom = if HashMap.null custom then Nothing else Just custom - } + parseJSON val = withObject "Problem" parser val + + where + parser obj = do + let custom = HashMap.filterWithKey + (\k _ -> k `notElem` ["type", "title", "status", "detail", "instance"]) + obj + typeURI <- obj .:? "type" >>= \case + Nothing -> pure Nothing + Just uriText -> Just <$> parseURI uriText + title <- obj .: "title" + status <- obj .:? "status" + detail <- obj .:? "detail" + instanceURI <- obj .:? "instance" >>= \case + Nothing -> pure Nothing + Just uriText -> Just <$> parseURI uriText + pure Problem { problemType = typeURI + , problemTitle = title + , problemStatus = toEnum <$> status + , problemDetail = detail + , problemInstance = instanceURI + , problemCustom = custom + } + + parseURI uriText = case URI.mkURI uriText of + Right uri -> pure uri + Left _exn -> typeMismatch "Failed to parse type URI" val diff --git a/src/Network/Nakadi/Internal/Types/Subscriptions.hs b/src/Network/Nakadi/Internal/Types/Subscriptions.hs --- a/src/Network/Nakadi/Internal/Types/Subscriptions.hs +++ b/src/Network/Nakadi/Internal/Types/Subscriptions.hs @@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Types.Subscriptions Description : Nakadi Service Types -Copyright : (c) Moritz Clasmeier 2018 +Copyright : (c) Moritz Clasmeier 2018, 2019 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental @@ -14,7 +14,9 @@ module Network.Nakadi.Internal.Types.Subscriptions where import Data.Int - +import Data.Eq ( Eq ) +import Data.Ord ( Ord ) +import Prelude ( Show ) -- | This type encodes the supported strategies for subscription -- cursor committing. @@ -38,3 +40,7 @@ -- number of events processed since the -- last commit crosses a threshold derived -- from @maxUncommittedEvents@. + +-- | This type is used for specifying subscription commit timeouts in seconds. +newtype CommitTimeout = CommitTimeout { unCommitTimeout :: Int32 } + deriving (Eq, Ord, Show) diff --git a/src/Network/Nakadi/Types/Subscriptions.hs b/src/Network/Nakadi/Types/Subscriptions.hs --- a/src/Network/Nakadi/Types/Subscriptions.hs +++ b/src/Network/Nakadi/Types/Subscriptions.hs @@ -1,7 +1,7 @@ {-| Module : Network.Nakadi.Types.Subscriptions Description : Nakadi Service Subscription Types -Copyright : (c) Moritz Clasmeier 2018 +Copyright : (c) Moritz Clasmeier 2018, 2019 License : BSD3 Maintainer : mtesseract@silverratio.net Stability : experimental @@ -14,6 +14,8 @@ module Network.Nakadi.Types.Subscriptions ( CommitStrategy(..) , CommitBufferingStrategy(..) - ) where + , CommitTimeout(..) + ) +where import Network.Nakadi.Internal.Types.Subscriptions diff --git a/tests/Network/Nakadi/Internal/Http/Test.hs b/tests/Network/Nakadi/Internal/Http/Test.hs --- a/tests/Network/Nakadi/Internal/Http/Test.hs +++ b/tests/Network/Nakadi/Internal/Http/Test.hs @@ -5,36 +5,50 @@ module Network.Nakadi.Internal.Http.Test ( testHttp - ) where + ) +where import ClassyPrelude import Control.Arrow import Control.Lens -import qualified Data.ByteString.Lazy as LB +import qualified Data.ByteString.Lazy as LB import Network.HTTP.Client import Network.HTTP.Types import Network.Nakadi import Network.Nakadi.Internal.Http -import qualified Network.Nakadi.Lenses as L +import qualified Network.Nakadi.Lenses as L import Test.Tasty import Test.Tasty.HUnit +import qualified Data.Map as Map +import Control.Monad.Trans.Resource ( runResourceT ) +import Prelude ( read ) +import qualified Data.Text as Text +import qualified Data.UUID as UUID +import Data.Maybe (fromJust) +import Conduit testHttp :: TestTree -testHttp = testGroup "Http" - [ testCase "HttpRequestModifier" testHttpRequestModifier - , testCase "FlowIdInclusion" testFlowIdInclusion - , testCase "FlowIdMissing" testFlowIdMissing - , testCase "FlowIdInclusionHttp" testFlowIdInclusionHttp - , testCase "FlowIdMissingHttp" testFlowIdMissingHttp +testHttp = testGroup + "Http" + [ testCase "HttpRequestModifier" testHttpRequestModifier + , testCase "FlowIdInclusion" testFlowIdInclusion + , testCase "FlowIdMissing" testFlowIdMissing + , testCase "FlowIdInclusionHttp" testFlowIdInclusionHttp + , testCase "FlowIdMissingHttp" testFlowIdMissingHttp + , testCase "CommitTimeoutInclusionHttp" testCommitTimeoutInclusionHttp + , testCase "CommitTimeoutMissingHttp" testCommitTimeoutMissingHttp ] extractFlowId :: Request -> Maybe FlowId extractFlowId = - requestHeaders - >>> filter (\ (key, _) -> key == "X-Flow-Id") - >>> listToMaybe - >>> fmap (FlowId . decodeUtf8 . snd) + requestHeaders >>> filter (\(key, _) -> key == "X-Flow-Id") >>> listToMaybe >>> fmap + (FlowId . decodeUtf8 . snd) +extractCommitTimeout :: Request -> Maybe CommitTimeout +extractCommitTimeout = + queryString >>> parseSimpleQuery >>> Map.fromList >>> Map.lookup "commit_timeout" >>> fmap + (CommitTimeout . read . Text.unpack . decodeUtf8) + headers :: RequestHeaders headers = [("test-header", "header-value")] @@ -43,8 +57,7 @@ testHttpRequestModifier :: Assertion testHttpRequestModifier = do - let conf = newConfigIO defaultRequest - & setRequestModifier dummyRequestModifier + let conf = newConfigIO defaultRequest & setRequestModifier dummyRequestModifier request <- runNakadiT conf $ httpBuildRequest id requestHeaders request @=? headers @@ -66,10 +79,27 @@ -> Request -> Maybe Manager -> b (Response LB.ByteString) -mockHttpLbs tv _config request _manager = do +mockHttpLbs tv _config request _manager = do liftIO . atomically $ writeTVar tv (Just request) throwString "Mock" +mockHttpResponseOpen + :: b ~ IO + => TVar (Maybe Request) + -> Config b + -> Request + -> Maybe Manager + -> b (Response (ConduitM () ByteString b ())) +mockHttpResponseOpen tv _config request _manager = do + liftIO . atomically $ writeTVar tv (Just request) + throwString "Mock" + +mockHttpResponseClose + :: b ~ IO + => Response () + -> b () +mockHttpResponseClose _ = pure () + testFlowIdInclusionHttp :: Assertion testFlowIdInclusionHttp = do tv <- atomically $ newTVar Nothing @@ -77,7 +107,7 @@ httpBackend = httpBackendIO & L.httpLbs .~ mockHttpLbs tv config = newConfig httpBackend defaultRequest & setFlowId flowId Left (StringException _ _) <- try $ runNakadiT config eventTypesList - Just requestExecuted <- liftIO . atomically $ readTVar tv + Just requestExecuted <- liftIO . atomically $ readTVar tv Just flowId @=? extractFlowId requestExecuted testFlowIdMissingHttp :: Assertion @@ -86,5 +116,33 @@ let httpBackend = httpBackendIO & L.httpLbs .~ mockHttpLbs tv config = newConfig httpBackend defaultRequest Left (StringException _ _) <- try $ runNakadiT config eventTypesList - Just requestExecuted <- liftIO . atomically $ readTVar tv + Just requestExecuted <- liftIO . atomically $ readTVar tv Nothing @=? extractFlowId requestExecuted + +dummySubscriptionId :: SubscriptionId +dummySubscriptionId = SubscriptionId (fromJust (UUID.fromString "975F8AEE-1F22-4798-8864-CC418CDF66EB")) + +testCommitTimeoutInclusionHttp :: Assertion +testCommitTimeoutInclusionHttp = do + tv <- atomically $ newTVar Nothing + let commitTimeout = CommitTimeout 42 + httpBackend = httpBackendIO + & L.httpResponseOpen .~ mockHttpResponseOpen tv + & L.httpResponseClose .~ mockHttpResponseClose + config = newConfig httpBackend defaultRequest & setCommitTimeout commitTimeout + Left (StringException _ _) <- try $ runResourceT . runNakadiT config $ subscriptionProcess + dummySubscriptionId (\(_ :: SubscriptionEventStreamBatch Int) -> pure ()) + Just requestExecuted <- liftIO . atomically $ readTVar tv + Just commitTimeout @=? extractCommitTimeout requestExecuted + +testCommitTimeoutMissingHttp :: Assertion +testCommitTimeoutMissingHttp = do + tv <- atomically $ newTVar Nothing + let httpBackend = httpBackendIO + & L.httpResponseOpen .~ mockHttpResponseOpen tv + & L.httpResponseClose .~ mockHttpResponseClose + config = newConfig httpBackend defaultRequest + Left (StringException _ _) <- try $ runResourceT . runNakadiT config $ subscriptionProcess + dummySubscriptionId (\(_ :: SubscriptionEventStreamBatch Int) -> pure ()) + Just requestExecuted <- liftIO . atomically $ readTVar tv + Nothing @=? extractCommitTimeout requestExecuted diff --git a/tests/Network/Nakadi/Internal/Types/Problem/Test.hs b/tests/Network/Nakadi/Internal/Types/Problem/Test.hs new file mode 100644 --- /dev/null +++ b/tests/Network/Nakadi/Internal/Types/Problem/Test.hs @@ -0,0 +1,50 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE QuasiQuotes #-} + +module Network.Nakadi.Internal.Types.Problem.Test where + +import ClassyPrelude +import Test.Tasty +import Test.Tasty.HUnit +import Data.Aeson +import Network.Nakadi +import Data.Aeson.QQ +import Data.Maybe ( fromJust ) + +testProblem :: TestTree +testProblem = testGroup + "Types/Problem" + [ testCase "Minimal Deserialization" testMinimalProblemDeserialization + , testCase "Complete Deserialization" testCompleteProblemDeserialization + , testCase "Custom Deserialization" testCustomProblemDeserialization + ] + +testMinimalProblemDeserialization :: Assertion +testMinimalProblemDeserialization = assertBool "Minimal Problem can be deserialized" + (problemJSON == toJSON problem) + where + problemJSON = [aesonQQ|{"title": "Something went wrong"}|] :: Value + problemBS = encode problemJSON + problem = fromJust (decode problemBS) :: Problem + +testCompleteProblemDeserialization :: Assertion +testCompleteProblemDeserialization = assertBool "Complete Problem can be deserialized" + (problemJSON == toJSON problem) + where + problemJSON = + [aesonQQ|{"title": "Something went wrong", + "type": "/some/type/uri", + "status": 500, + "detail": "Server is dusty", + "instance": "https://another/uri/"}|] :: Value + problemBS = encode problemJSON + problem = fromJust (decode problemBS) :: Problem + +testCustomProblemDeserialization :: Assertion +testCustomProblemDeserialization = assertBool "Custom Problem can be deserialized" + (problemJSON == toJSON problem) + where + problemJSON + = [aesonQQ|{"title": "Something went wrong", "type": "/some/type/uri", "answer": 42}|] :: Value + problemBS = encode problemJSON + problem = fromJust (decode problemBS) :: Problem diff --git a/tests/Network/Nakadi/Internal/Types/Test.hs b/tests/Network/Nakadi/Internal/Types/Test.hs --- a/tests/Network/Nakadi/Internal/Types/Test.hs +++ b/tests/Network/Nakadi/Internal/Types/Test.hs @@ -11,9 +11,10 @@ import Network.Nakadi.Types.Service import Data.Aeson.QQ import Data.Maybe (fromJust) +import Network.Nakadi.Internal.Types.Problem.Test testTypes :: TestTree -testTypes = testGroup "Types" [testService] +testTypes = testGroup "Types" [testService, testProblem] testService :: TestTree testService = testGroup diff --git a/tests/Network/Nakadi/Subscriptions/Processing/Test.hs b/tests/Network/Nakadi/Subscriptions/Processing/Test.hs --- a/tests/Network/Nakadi/Subscriptions/Processing/Test.hs +++ b/tests/Network/Nakadi/Subscriptions/Processing/Test.hs @@ -21,7 +21,9 @@ testSubscriptionsProcessing :: Config App -> TestTree testSubscriptionsProcessing confTemplate = let mkConf commitStrategy nWorkers = - confTemplate & setCommitStrategy commitStrategy & setWorkerThreads nWorkers + confTemplate + & setCommitStrategy commitStrategy & setWorkerThreads nWorkers + & setCommitTimeout (CommitTimeout 10) in testGroup "Processing" [ testCase "SubscriptionProcessing/async/TimeBuffer/singleWorker"