nakadi-client 0.6.1.0 → 0.7.0.0

raw patch · 11 files changed

+245/−84 lines, 11 filesdep +modern-uriPVP ok

version bump matches the API change (PVP)

Dependencies added: modern-uri

API changes (from Hackage documentation)

Files

• nakadi-client.cabal +6/−3
• src/Network/Nakadi/Config.hs +6/−2
• src/Network/Nakadi/Internal/Config.hs +24/−23
• src/Network/Nakadi/Internal/Types/Config.hs +1/−0
• src/Network/Nakadi/Internal/Types/Problem.hs +65/−32
• src/Network/Nakadi/Internal/Types/Subscriptions.hs +8/−2
• src/Network/Nakadi/Types/Subscriptions.hs +4/−2
• tests/Network/Nakadi/Internal/Http/Test.hs +76/−18
• tests/Network/Nakadi/Internal/Types/Problem/Test.hs +50/−0
• tests/Network/Nakadi/Internal/Types/Test.hs +2/−1
• tests/Network/Nakadi/Subscriptions/Processing/Test.hs +3/−1

@@ -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

@@ -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 }

@@ -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+                                       }

@@ -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

@@ -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

@@ -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)

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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"