amqp-streamly (empty) → 0.1.0

raw patch · 8 files changed

+288/−0 lines, 8 filesdep +amqpdep +amqp-streamlydep +basesetup-changed

Dependencies added: amqp, amqp-streamly, base, bytestring, hspec, process, streamly, testcontainers, text

Files

• ChangeLog.md +4/−0
• LICENSE +30/−0
• README.md +18/−0
• Setup.hs +2/−0
• amqp-streamly.cabal +63/−0
• src/Network/AMQP/Streamly.hs +76/−0
• test/Network/AMQP/StreamlySpec.hs +94/−0
• test/Spec.hs +1/−0

@@ -0,0 +1,4 @@+# Changelog for amqp-streamly++## 0.1.0+ * Creation with `produce` and `consume`

@@ -0,0 +1,30 @@+Copyright Gautier DI FOLCO (c) 2020++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Gautier DI FOLCO nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,18 @@+# amqp-streamly++A simple wrapper around [amqp](https://hackage.haskell.org/package/amqp/).++## Usage+You can either build a producer, which will publish all the messages of+a stream:++```+Streamly.drain $ produce channel sendInstructionsStream+```++Or a consumer, which will contain the `Message`s and `Envelope`s of+a queue:++```+Streamly.drain $ consume channel aQueue NoAck+```

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

@@ -0,0 +1,63 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.33.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: 12d7b6fb1d3da6a6da9f0769f11d261769c8ee0f8ec6e302b18d756e3ac46127++name:           amqp-streamly+version:        0.1.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+homepage:       https://github.com/blackheaven/amqp-streamly#readme+bug-reports:    https://github.com/blackheaven/amqp-streamly/issues+author:         Gautier DI FOLCO+maintainer:     gautier.difolco@gmail.com+copyright:      Gautier DI FOLCO+license:        BSD3+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    README.md+    ChangeLog.md++source-repository head+  type: git+  location: https://github.com/blackheaven/amqp-streamly++library+  exposed-modules:+      Network.AMQP.Streamly+  other-modules:+      Paths_amqp_streamly+  hs-source-dirs:+      src+  build-depends:+      amqp >=0.19.1 && <0.20+    , base >=4.7 && <5+    , streamly >=0.7.2 && <0.8+    , text >=1.2.4.0 && <1.3+  default-language: Haskell2010++test-suite amqp-streamly-test+  type: exitcode-stdio-1.0+  main-is: Spec.hs+  other-modules:+      Network.AMQP.StreamlySpec+      Paths_amqp_streamly+  hs-source-dirs:+      test+  ghc-options: -threaded -rtsopts -with-rtsopts=-N+  build-depends:+      amqp >=0.19.1 && <0.20+    , amqp-streamly+    , base >=4.7 && <5+    , bytestring >=0.10.10.0 && <0.11+    , hspec >=2.7.1 && <2.8+    , process >=1.6.8.0 && <1.7+    , streamly >=0.7.2 && <0.8+    , testcontainers ==0.1.*+    , text >=1.2.4.0 && <1.3+  default-language: Haskell2010

@@ -0,0 +1,76 @@+{-# LANGUAGE FlexibleContexts #-}++module Network.AMQP.Streamly+  (+    -- * How to use this library+    -- $use+    SendInstructions(..)+  , produce+  , consume+  )+where++import           Control.Concurrent.MVar+import           Control.Monad.IO.Class         ( MonadIO+                                                , liftIO+                                                )++import           Data.Text                      ( Text )+import           Network.AMQP+import           Streamly+import qualified Streamly.Internal.Prelude     as S+import qualified Streamly.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)++-- | The Queue name+type Queue = Text++-- | Publish the produced messages+produce+  :: (IsStream t, MonadAsync m) => Channel -> t m SendInstructions -> t m ()+produce channel = S.mapM send+ 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+  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+++-- $use+--+-- This section contains basic step-by-step usage of the library.+--+-- You can either build a producer, which will publish all the messages of+-- a stream:+--+-- > Streamly.drain $ produce channel sendInstructionsStream+--+-- Or a consumer, which will contain the @Message@s and @Envelope@s of+-- a queue:+--+-- > Streamly.drain $ consume channel aQueue NoAck

@@ -0,0 +1,94 @@+{-# LANGUAGE OverloadedStrings #-}++module Network.AMQP.StreamlySpec+  ( 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 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++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"+        let arbitraryRoutingKey = "arbitraryRoutingKey"+        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+        S.toList+          (   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++mkChannel :: T.Text -> T.Text -> String -> IO Channel+mkChannel login password ip = do+  connection <- openConnection ip "/" login password+  openChannel connection++getIp :: Container -> IO String+getIp (Container id _ _) = format <$> readProcess+  "docker"+  [ "inspect"+  , "-f"+  , "'{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}'"+  , T.unpack id+  ]+  ""+  where format = init . init . tail++containers :: MonadDocker m => m Channel+containers = do+  let (login, password) = ("guest", "guest")+  container <- run (fromTag "rabbitmq:3.8.4") defaultContainerRequest+  liftIO $ threadDelay $ 15 * 1000 * 1000+  ip <- liftIO $ getIp container+  liftIO $ mkChannel login password ip

@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}