amqp-streamly 0.2.1 → 0.3.0

raw patch · 5 files changed

+123/−107 lines, 5 filesdep +streamly-coredep ~amqpdep ~streamlysetup-changedPVP ok

version bump matches the API change (PVP)

Dependencies added: streamly-core

Dependency ranges changed: amqp, streamly

API changes (from Hackage documentation)

Files

• ChangeLog.md +4/−0
• Setup.hs +0/−2
• amqp-streamly.cabal +8/−6
• src/Network/AMQP/Streamly.hs +42/−38
• test/Network/AMQP/StreamlySpec.hs +69/−61

@@ -1,5 +1,9 @@ # Changelog for amqp-streamly +## 0.3.0+ * @hughjfchen add support form amqp 0.22.2 and streamly 0.9 - 0.10+ * Move to nix+ ## 0.2.0  * Upgrade to amqp 0.20 and stackage 16.4 

@@ -1,2 +0,0 @@-import           Distribution.Simple-main = defaultMain

@@ -7,7 +7,7 @@ -- hash: 0e93b864e2ba6a1bd099a1c10b3f25dea8b453a814deb8e26158692a76eca769  name:           amqp-streamly-version:        0.2.1+version:        0.3.0 synopsis:       A simple streamly wrapper for amqp description:    A simple streamly wrapper for amqp. Provides two functions `produce` and `consume`. category:       streamly, conduit, amqp, rabbitmq@@ -35,13 +35,14 @@   hs-source-dirs:       src   build-depends:-      amqp >=0.19.1 && <0.23+      amqp >=0.22.2 && <0.23     , base >=4.7 && <5-    , streamly >=0.7.2 && <=0.9+    , streamly-core >=0.2.0 && <0.3+    , streamly >=0.9 && <0.11     , text >=1.2.4.0 && <3   default-language: Haskell2010 -test-suite amqp-streamly-test+test-suite spec   type: exitcode-stdio-1.0   main-is: Spec.hs   other-modules:@@ -51,13 +52,14 @@       test   ghc-options: -threaded -rtsopts -with-rtsopts=-N   build-depends:-      amqp >=0.19.1 && <0.23+      amqp >=0.22.2 && <0.23     , amqp-streamly     , base >=4.7 && <5     , bytestring     , hspec     , process-    , streamly >=0.7.2 && <=0.9+    , streamly-core >=0.2.0 && <0.3+    , streamly >=0.9 && <0.11     , testcontainers     , text >=1.2.4.0 && <3   default-language: Haskell2010

@@ -1,65 +1,69 @@ {-# LANGUAGE FlexibleContexts #-}  module Network.AMQP.Streamly-  (-    -- * How to use this library+  ( -- * How to use this library     -- $use-    SendInstructions(..)-  , produce-  , consume+    SendInstructions (..),+    produce,+    consume,   ) where -import           Control.Concurrent.MVar-import           Control.Monad.IO.Class         ( MonadIO-                                                , liftIO-                                                )--import           Data.Text                      ( Text )-import           Network.AMQP-import           Streamly.Prelude-import qualified Streamly.Internal.Data.Stream.IsStream.Common     as S-import qualified Streamly.Prelude              as S+import Control.Concurrent.MVar+import Control.Monad.IO.Class+  ( MonadIO,+    liftIO,+  )+import Data.Text (Text)+import Network.AMQP+import Streamly.Data.Stream+import qualified Streamly.Data.Stream as S+import Streamly.Data.Stream.Prelude+import qualified Streamly.Data.Stream.Prelude as S  -- | Informations to be sent -- -- See @Network.AMQP.publishMsg'@ for options-data SendInstructions = SendInstructions { exchange :: Text, routingKey :: Text, mandatory :: Bool, message :: Message } deriving (Show)+data SendInstructions = SendInstructions {exchange :: Text, routingKey :: Text, mandatory :: Bool, message :: Message} deriving (Show)  -- | The Queue name type Queue = Text  -- | Publish the produced messages-produce-  :: (IsStream t, MonadAsync m) => Channel -> t m SendInstructions -> t m ()+produce ::+  (MonadIO m) =>+  Channel ->+  Stream m SendInstructions ->+  Stream m () produce channel = S.mapM send- where-  send i = liftIO $ do-    publishMsg' channel (exchange i) (routingKey i) (mandatory i) (message i)-    return ()+  where+    send i = liftIO $ do+      publishMsg' channel (exchange i) (routingKey i) (mandatory i) (message i)+      return ()  -- | Stream messages from a queue -- -- See @Network.AMQP.consumeMsgs@ for options-consume-  :: (IsStream t, MonadAsync m)-  => Channel-  -> Queue-  -> Ack-  -> t m (Message, Envelope)-consume channel queue ack = S.concatM $ liftIO $ do+consume ::+  (MonadIO m) =>+  Channel ->+  Queue ->+  Ack ->+  Stream m (Message, Envelope)+consume channel queue ack = S.concatEffect $ liftIO $ do   mvar <- newEmptyMVar   consumeMsgs channel queue Ack $ putMVar mvar   return $ S.repeatM $ taking mvar- where-  taking :: MonadIO m => MVar (Message, Envelope) -> m (Message, Envelope)-  taking mvar = liftIO $ if ack == NoAck-    then do-      retrieved <- takeMVar mvar-      ackEnv $ snd retrieved-      return retrieved-    else takeMVar mvar-+  where+    taking :: (MonadIO m) => MVar (Message, Envelope) -> m (Message, Envelope)+    taking mvar =+      liftIO $+        if ack == NoAck+          then do+            retrieved <- takeMVar mvar+            ackEnv $ snd retrieved+            return retrieved+          else takeMVar mvar  -- $use --

@@ -1,75 +1,80 @@-{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}  module Network.AMQP.StreamlySpec-  ( main-  , spec+  ( main,+    spec,   ) where -import           Network.AMQP.Streamly--import           Control.Concurrent             ( threadDelay )-import           Control.Monad.IO.Class         ( liftIO )--import           Network.AMQP-import           Test.Hspec-import           TestContainers.Hspec+import Control.Concurrent (threadDelay)+import Control.Monad.IO.Class (liftIO)+import qualified Data.ByteString.Lazy.Char8 as B+import Data.Maybe (fromJust)+import qualified Data.Text as T+import Network.AMQP+import Network.AMQP.Streamly+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream as S+import System.Process (readProcess)+import Test.Hspec import qualified TestContainers.Docker as D-import qualified Data.ByteString.Lazy.Char8    as B-import           Data.Maybe                     ( fromJust )-import qualified Data.Text                     as T-import           System.Process                 ( readProcess )-import qualified Streamly.Prelude              as S+import TestContainers.Hspec  main :: IO () main = hspec spec  spec :: Spec spec =-  around (withContainers containers)-    $ it "producing 5 messages should be consumed in the right order"-    $ \channel -> flip shouldReturn messages $ do-        let arbitraryExchange   = "arbitraryExchange"+  around (withContainers containers) $+    it "producing 5 messages should be consumed in the right order" $+      \channel -> flip shouldReturn messages $ do+        let arbitraryExchange = "arbitraryExchange"         let arbitraryRoutingKey = "arbitraryRoutingKey"-        let arbitraryQueue      = "arbitraryQueue"+        let arbitraryQueue = "arbitraryQueue"         let fixedInstructions =               SendInstructions arbitraryExchange arbitraryRoutingKey True-        liftIO $ declareExchange-          channel-          newExchange { exchangeName = arbitraryExchange-                      , exchangeType = "fanout"-                      }-        liftIO $ declareQueue channel newQueue { queueName = arbitraryQueue }-        liftIO $ bindQueue channel-                           arbitraryQueue-                           arbitraryExchange-                           arbitraryRoutingKey-        S.drain $ produce channel $ S.fromList $ map fixedInstructions messages+        liftIO $+          declareExchange+            channel+            newExchange+              { exchangeName = arbitraryExchange,+                exchangeType = "fanout"+              }+        liftIO $ declareQueue channel newQueue {queueName = arbitraryQueue}+        liftIO $+          bindQueue+            channel+            arbitraryQueue+            arbitraryExchange+            arbitraryRoutingKey+        S.fold Fold.drain $ produce channel $ S.fromList $ map fixedInstructions messages         S.toList-          (   S.take (length messages)-          $   fst-          <$> consume channel arbitraryQueue NoAck+          ( S.take (length messages) $+              fst+                <$> consume channel arbitraryQueue NoAck           )  messages :: [Message] messages = map toMessage ["Lorem", "ipsum", "dolor", "sit", "amet"]- where-  toMessage x = Message (B.pack x)-                        Nothing-                        Nothing-                        Nothing-                        Nothing-                        Nothing-                        Nothing-                        Nothing-                        Nothing-                        Nothing-                        Nothing-                        Nothing-                        Nothing-                        Nothing-                        Nothing+  where+    toMessage x =+      Message+        (B.pack x)+        Nothing+        Nothing+        Nothing+        Nothing+        Nothing+        Nothing+        Nothing+        Nothing+        Nothing+        Nothing+        Nothing+        Nothing+        Nothing+        Nothing  mkChannel :: T.Text -> T.Text -> String -> IO Channel mkChannel login password ip = do@@ -77,17 +82,20 @@   openChannel connection  getIp :: D.Container -> IO String-getIp c = format <$> readProcess-  "docker"-  [ "inspect"-  , "-f"-  , "'{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}'"-  , T.unpack $ D.containerId c-  ]-  ""-  where format = init . init . tail+getIp c =+  format+    <$> readProcess+      "docker"+      [ "inspect",+        "-f",+        "'{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}'",+        T.unpack $ D.containerId c+      ]+      ""+  where+    format = init . init . tail -containers :: MonadDocker m => m Channel+containers :: (MonadDocker m) => m Channel containers = do   let (login, password) = ("guest", "guest")   container <- run (containerRequest $ fromTag "rabbitmq:3.8.4")