amqp-worker 0.2.5 → 0.3.2

raw patch · 11 files changed

+474/−282 lines, 11 filesdep +serverdep −tastydep −tasty-hunitdep ~basePVP ok

version bump matches the API change (PVP)

Dependencies added: server

Dependencies removed: tasty, tasty-hunit

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• README.md +94/−0
• amqp-worker.cabal +82/−55
• src/Network/AMQP/Worker.hs +72/−54
• src/Network/AMQP/Worker/Connection.hs +34/−21
• src/Network/AMQP/Worker/Key.hs +82/−41
• src/Network/AMQP/Worker/Message.hs +24/−26
• src/Network/AMQP/Worker/Poll.hs +4/−4
• src/Network/AMQP/Worker/Queue.hs +27/−47
• src/Network/AMQP/Worker/Worker.hs +5/−6
• test/Example.hs +40/−19
• test/Spec.hs +10/−9

@@ -0,0 +1,94 @@+AMQP Worker+-----------++Type-safe AMQP workers. Compatible with RabbitMQ++    {-# LANGUAGE DeriveGeneric     #-}+    {-# LANGUAGE OverloadedStrings #-}+    module Main where++    import           Control.Concurrent      (forkIO)+    import           Control.Monad.Catch     (SomeException)+    import           Data.Aeson              (FromJSON, ToJSON)+    import           Data.Function           ((&))+    import           Data.Text               (Text, pack)+    import           GHC.Generics            (Generic)+    import           Network.AMQP.Worker     (Connection, Message (..),+                                              WorkerException, def, fromURI)+    import qualified Network.AMQP.Worker     as Worker+    import           Network.AMQP.Worker.Key+    import           System.IO               (BufferMode (..), hSetBuffering,+                                              stderr, stdout)++    data TestMessage = TestMessage+      { greeting :: Text }+      deriving (Generic, Show, Eq)++    instance FromJSON TestMessage+    instance ToJSON TestMessage+++    newMessages :: Key Routing TestMessage+    newMessages = key "messages" & word "new"++    results :: Key Routing Text+    results = key "results"++    anyMessages :: Key Binding TestMessage+    anyMessages = key "messages" & star++++    example :: IO ()+    example = do++      -- connect+      conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")++      let handleAnyMessages = Worker.topic anyMessages "handleAnyMessage"++      -- initialize the queues+      Worker.bindQueue conn (Worker.direct newMessages)+      Worker.bindQueue conn (Worker.direct results)++      -- topic queue!+      Worker.bindQueue conn handleAnyMessages++      putStrLn "Enter a message"+      msg <- getLine++      -- publish a message+      putStrLn "Publishing a message"+      Worker.publish conn newMessages (TestMessage $ pack msg)++      -- create a worker, the program loops here+      _ <- forkIO $ Worker.worker conn def (Worker.direct newMessages) onError (onMessage conn)+      _ <- forkIO $ Worker.worker conn def (handleAnyMessages) onError (onMessage conn)++      putStrLn "Press any key to exit"+      _ <- getLine+      return ()+++++    onMessage :: Connection -> Message TestMessage -> IO ()+    onMessage conn m = do+      let testMessage = value m+      putStrLn "Got Message"+      print testMessage+      Worker.publish conn results (greeting testMessage)+++    onError :: WorkerException SomeException -> IO ()+    onError e = do+      putStrLn "Do something with errors"+      print e++++    main :: IO ()+    main = do+      hSetBuffering stdout LineBuffering+      hSetBuffering stderr LineBuffering+      example

@@ -1,40 +1,49 @@--- Initial amqp-worker.cabal generated by cabal init.  For further--- documentation, see http://haskell.org/cabal/users-guide/+cabal-version: 1.12 -name:                amqp-worker-version:             0.2.5-synopsis:            High level functions for working with message queues-description:         High level functions for working with message queues-license:             BSD3-license-file:        LICENSE-author:              Sean Hess-maintainer:          seanhess@gmail.com--- copyright:-category:            Web-build-type:          Simple--- extra-source-files:-cabal-version:       >=1.10+-- This file has been generated from package.yaml by hpack version 0.31.1.+--+-- see: https://github.com/sol/hpack+--+-- hash: d15089fce32e028d3ee60745261d658588b69431f1163607971dcd4756ce40b2 +name:           amqp-worker+version:        0.3.2+description:    Please see the README on GitHub at <https://github.com/seanhess/server#readme>+homepage:       https://github.com/seanhess/amqp-worker#readme+bug-reports:    https://github.com/seanhess/amqp-worker/issues+author:         Sean Hess+maintainer:     seanhess@gmail.com+copyright:      Orbital Labs+license:        BSD3+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    README.md++source-repository head+  type: git+  location: https://github.com/seanhess/amqp-worker+ library-  -- other-extensions:   exposed-modules:-    Network.AMQP.Worker-  other-modules:+      Network.AMQP.Worker       Network.AMQP.Worker.Connection-    , Network.AMQP.Worker.Key-    , Network.AMQP.Worker.Message-    , Network.AMQP.Worker.Poll-    , Network.AMQP.Worker.Queue-    , Network.AMQP.Worker.Worker-  hs-source-dirs:      src-  default-language:    Haskell2010+      Network.AMQP.Worker.Key+      Network.AMQP.Worker.Message+      Network.AMQP.Worker.Poll+      Network.AMQP.Worker.Queue+      Network.AMQP.Worker.Worker+  other-modules:+      Paths_amqp_worker+  hs-source-dirs:+      src   build-depends:-    base >=4.9 && <5-    , aeson+      aeson     , amqp+    , base >=4.9 && <5     , bytestring-    , exceptions     , data-default+    , exceptions     , monad-control     , monad-loops     , mtl@@ -42,37 +51,55 @@     , split     , text     , transformers-base-+  default-language: Haskell2010  executable example-  hs-source-dirs:      test-  main-is:             Example.hs-  ghc-options:         -threaded -rtsopts -with-rtsopts=-N -Wall-  build-depends:       base-                     , aeson-                     , amqp-worker-                     , exceptions-                     , text-  default-language:    Haskell2010---test-suite test-  type:                exitcode-stdio-1.0-  hs-source-dirs:      test-  main-is:             Spec.hs+  main-is: Example.hs+  other-modules:+      Spec+      Paths_amqp_worker+  hs-source-dirs:+      test+  ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall   build-depends:-      base-    , bytestring-    , aeson+      aeson     , amqp     , amqp-worker-    , tasty-    , tasty-hunit+    , base >=4.9 && <5+    , bytestring+    , data-default+    , exceptions+    , monad-control+    , monad-loops+    , mtl+    , resource-pool+    , split     , text-  ghc-options:         -threaded -rtsopts -with-rtsopts=-N-  default-language:    Haskell2010-+    , transformers-base+  default-language: Haskell2010 -source-repository head-  type:     git-  location: https://github.com/seanhess/amqp-worker+test-suite test+  type: exitcode-stdio-1.0+  main-is: Spec.hs+  other-modules:+      Example+      Paths_amqp_worker+  hs-source-dirs:+      test+  ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall+  build-depends:+      aeson+    , amqp+    , base >=4.9 && <5+    , bytestring+    , data-default+    , exceptions+    , monad-control+    , monad-loops+    , mtl+    , resource-pool+    , server+    , split+    , text+    , transformers-base+  default-language: Haskell2010

@@ -9,88 +9,111 @@ -- -- Connect to a server, initialize a queue, publish a message, and create a worker to process them. ----- >+-- > {-# LANGUAGE DeriveGeneric     #-} -- > {-# LANGUAGE OverloadedStrings #-}--- > {-# LANGUAGE DeriveGeneric #-} -- > module Main where--- >--- > import Control.Monad.Catch (SomeException)--- > import Data.Aeson (FromJSON, ToJSON)--- > import Data.Text (Text)--- > import GHC.Generics (Generic)--- > import qualified Network.AMQP.Worker as Worker--- > import Network.AMQP.Worker (fromURI, Exchange, Queue, Direct, def, WorkerException, Message(..), Connection)--- >+-- > +-- > import           Control.Concurrent      (forkIO)+-- > import           Control.Monad.Catch     (SomeException)+-- > import           Data.Aeson              (FromJSON, ToJSON)+-- > import           Data.Function           ((&))+-- > import           Data.Text               (Text, pack)+-- > import           GHC.Generics            (Generic)+-- > import           Network.AMQP.Worker     (Connection, Message (..),+-- >                                           WorkerException, def, fromURI)+-- > import qualified Network.AMQP.Worker     as Worker+-- > import           Network.AMQP.Worker.Key+-- > import           System.IO               (BufferMode (..), hSetBuffering,+-- >                                           stderr, stdout)+-- >  -- > data TestMessage = TestMessage--- >  { greeting :: Text }--- >  deriving (Generic, Show, Eq)--- >+-- >   { greeting :: Text }+-- >   deriving (Generic, Show, Eq)+-- >  -- > instance FromJSON TestMessage -- > instance ToJSON TestMessage--- >--- >--- > exchange :: Exchange--- > exchange = Worker.exchange "testExchange"--- >--- >--- > queue :: Queue Direct TestMessage--- > queue = Worker.queue exchange "testQueue"--- >--- >--- > results :: Queue Direct Text--- > results = Worker.queue exchange "resultQueue"--- >--- >+-- > +-- > +-- > newMessages :: Key Routing TestMessage+-- > newMessages = key "messages" & word "new"+-- > +-- > results :: Key Routing Text+-- > results = key "results"+-- > +-- > anyMessages :: Key Binding TestMessage+-- > anyMessages = key "messages" & star+-- > +-- > +-- >  -- > example :: IO () -- > example = do+-- >  -- >   -- connect -- >   conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")--- >+-- > +-- >   let handleAnyMessages = Worker.topic anyMessages "handleAnyMessage"+-- >  -- >   -- initialize the queues--- >   Worker.initQueue conn queue--- >   Worker.initQueue conn results--- >+-- >   Worker.bindQueue conn (Worker.direct newMessages)+-- >   Worker.bindQueue conn (Worker.direct results)+-- > +-- >   -- topic queue!+-- >   Worker.bindQueue conn handleAnyMessages+-- > +-- >   putStrLn "Enter a message"+-- >   msg <- getLine+-- >  -- >   -- publish a message--- >   Worker.publish conn queue (TestMessage "hello world")--- >+-- >   putStrLn "Publishing a message"+-- >   Worker.publish conn newMessages (TestMessage $ pack msg)+-- >  -- >   -- create a worker, the program loops here--- >   Worker.worker def conn queue onError (onMessage conn)--- >--- >+-- >   _ <- forkIO $ Worker.worker conn def (Worker.direct newMessages) onError (onMessage conn)+-- >   _ <- forkIO $ Worker.worker conn def (handleAnyMessages) onError (onMessage conn)+-- > +-- >   putStrLn "Press any key to exit"+-- >   _ <- getLine+-- >   return ()+-- > +-- >  -- > onMessage :: Connection -> Message TestMessage -> IO () -- > onMessage conn m = do -- >   let testMessage = value m -- >   putStrLn "Got Message" -- >   print testMessage -- >   Worker.publish conn results (greeting testMessage)--- >--- >+-- > +-- >  -- > onError :: WorkerException SomeException -> IO () -- > onError e = do -- >   putStrLn "Do something with errors" -- >   print e+-- > +-- >   module Network.AMQP.Worker   (-  -- * Declaring Queues and Exchanges-    exchange-  , ExchangeName-  , queue-  , topicQueue-  , Exchange(..)-  , Queue(..)-  , Direct, Topic +  -- * Routing Keys+    Key(..)+  , Routing+  , Binding+  , word, key, star, hash+   -- * Connecting   , Connection   , connect   , disconnect+  , exchange   , AMQP.fromURI -  -- * Initializing exchanges and queues-  , initQueue+  -- * Initializing queues+  , Queue(..)+  , direct+  , topic+  , bindQueue -  -- * Publishing Messages+  -- * Sending Messages   , publish   , publishToExchange @@ -108,11 +131,6 @@   , Microseconds   , Default.def -  -- * Routing Keys-  , RoutingKey(..)-  , BindingKey(..)-  , BindingName(..)-  , QueueKey(..)    ) where 

@@ -1,29 +1,42 @@-{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleContexts  #-}+{-# LANGUAGE OverloadedStrings #-} module Network.AMQP.Worker.Connection-  ( Connection+  ( Connection(..)   , connect   , disconnect   , withChannel   ) where -import Control.Concurrent.MVar (MVar, putMVar, newEmptyMVar, readMVar, takeMVar)-import Control.Monad.Catch (throwM, catch)-import Data.Pool (Pool)-import qualified Data.Pool as Pool-import qualified Network.AMQP as AMQP-import Network.AMQP (Channel, AMQPException(..))-import Control.Monad.Trans.Control (MonadBaseControl)+import           Control.Concurrent.MVar     (MVar, newEmptyMVar, putMVar,+                                              readMVar, takeMVar)+import           Control.Monad.Catch         (catch, throwM)+import           Control.Monad.IO.Class      (MonadIO, liftIO)+import           Control.Monad.Trans.Control (MonadBaseControl)+import           Data.Pool                   (Pool)+import qualified Data.Pool                   as Pool+import           Data.Text                   (Text)+import           Network.AMQP                (AMQPException (..), Channel)+import qualified Network.AMQP                as AMQP -data Connection =-  Connection (MVar AMQP.Connection) (Pool Channel)+type ExchangeName = Text ++-- | Internal connection details+data Connection = Connection+  { amqpConn :: MVar AMQP.Connection+  , pool     :: Pool Channel+  , exchange :: ExchangeName+  }+ -- | Connect to the AMQP server. -- -- > conn <- connect (fromURI "amqp://guest:guest@localhost:5672")----connect :: AMQP.ConnectionOpts -> IO Connection-connect opts = do+connect :: MonadIO m => AMQP.ConnectionOpts -> m Connection+connect opts = liftIO $ do +    -- use a default exchange name+    let exchangeName = "amq.topic"+     -- create a single connection in an mvar     cvar <- newEmptyMVar     openConnection cvar@@ -32,7 +45,7 @@     chans <- Pool.createPool (create cvar) destroy numStripes openTime numChans  -    pure $ Connection cvar chans+    pure $ Connection cvar chans exchangeName   where     numStripes = 1     openTime = 10@@ -64,15 +77,15 @@       AMQP.closeChannel chan  -disconnect :: Connection -> IO ()-disconnect (Connection c p) = do-    conn <- readMVar c-    Pool.destroyAllResources p+disconnect :: MonadIO m => Connection -> m ()+disconnect c = liftIO $ do+    conn <- readMVar $ amqpConn c+    Pool.destroyAllResources $ pool c     AMQP.closeConnection conn   +-- | Perform an action with a channel resource withChannel :: MonadBaseControl IO m => Connection -> (Channel -> m b) -> m b-withChannel (Connection _ p) action = do+withChannel (Connection _ p _) action = do     Pool.withResource p action-

@@ -1,62 +1,103 @@ {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} module Network.AMQP.Worker.Key-  ( QueueKey(..)-  , RoutingKey(..)-  , BindingKey(..)-  , BindingName(..)+  ( Key(..)+  , Binding(..)+  , BindingWord+  , Routing+  , key+  , word, star, hash+  , keyText+  , KeySegment(..)+  , bindingKey   ) where -import qualified Data.List as List-import qualified Data.List.Split as List-import Data.String (IsString(..)) import Data.Text (Text) import qualified Data.Text as Text+import qualified Data.List as List --- | A name used to address queues-newtype RoutingKey = RoutingKey Text-  deriving (Show, Eq) -instance IsString RoutingKey where-  fromString = RoutingKey . Text.pack -instance QueueKey RoutingKey where-  showKey (RoutingKey t) = t +-- | Keys describe routing and binding info for a message+newtype Key a msg = Key [a]+  deriving (Eq, Show, Semigroup, Monoid) ++-- | Every message is sent with a specific routing key+--+-- > newCommentKey :: Key Routing Comment+-- > newCommentKey = key "posts" & word "1" & word "comments" & word "new"+newtype Routing = Routing Text+    deriving (Eq, Show)++instance KeySegment Routing where+  toText (Routing s) = s+  fromText = Routing+  toBind (Routing s) = Word s+++ -- | A dynamic binding address for topic queues ----- > commentsKey :: BindingKey--- > commentsKey = "posts.*.comments"-newtype BindingKey = BindingKey [BindingName]-  deriving (Eq)+-- > commentsKey :: Key Binding Comment+-- > commentsKey = key "posts" & star & word "comments" & hash+data Binding+    = Word BindingWord+    | Star+    | Hash+    deriving (Eq, Show) -instance QueueKey BindingKey where-  showKey (BindingKey ns) =-    Text.intercalate "." . List.map bindingNameText $ ns -instance IsString BindingKey where-  fromString s =-    let segments = List.splitOn "." s-        names = List.map fromString segments-    in BindingKey names+instance KeySegment Binding where+    toText (Word t) = t+    toText Star = "*"+    toText Hash = "#"+    fromText = Word+    toBind = id  -data BindingName-  = Name Text-  | Star-  | Hash-  deriving (Eq) -instance IsString BindingName where-  fromString "*" = Star-  fromString "#" = Hash-  fromString n = Name (Text.pack n)+class KeySegment a where+    toText   :: a -> Text+    fromText :: Text -> a+    toBind :: a -> Binding -bindingNameText :: BindingName -> Text-bindingNameText (Name t) = t-bindingNameText Star = "*"-bindingNameText Hash = "#"  -class QueueKey key where-  showKey :: key -> Text+++++keyText :: KeySegment a => Key a msg -> Text+keyText (Key ns) =+    Text.intercalate "." . List.map toText $ ns+++-- | Convert any key to a binding key+bindingKey :: KeySegment a => Key a msg -> Key Binding msg+bindingKey (Key rs) = Key (map toBind rs)++++-- | Match any one word+star :: KeySegment a => Key a msg -> Key Binding msg+star (Key ws) = Key (map toBind ws ++ [Star])++-- | Match any words+hash :: KeySegment a => Key a msg -> Key Binding msg+hash (Key ws) = Key (map toBind ws ++ [Hash])++-- | Match a specific word+word :: KeySegment a => Text -> Key a msg -> Key a msg+word w (Key ws) = Key $ ws ++ [fromText w]++-- | Create a new key+key :: Text -> Key Routing msg+key t = Key [Routing t]+++++type BindingWord = Text

@@ -2,18 +2,18 @@ {-# LANGUAGE FlexibleContexts #-} module Network.AMQP.Worker.Message where -import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.Base (liftBase) import Data.Aeson (ToJSON, FromJSON) import qualified Data.Aeson as Aeson import Data.ByteString.Lazy (ByteString)-import Network.AMQP (newMsg, DeliveryMode(..), Ack(..), QueueOpts(..))+import Network.AMQP (newMsg, DeliveryMode(..), Ack(..)) import qualified Network.AMQP as AMQP -import Network.AMQP.Worker.Key (RoutingKey(..))+import Network.AMQP.Worker.Key (Key, Routing, keyText) import Network.AMQP.Worker.Poll (poll)-import Network.AMQP.Worker.Connection (Connection, withChannel)-import Network.AMQP.Worker.Queue (Queue(..), Exchange(..), ExchangeName, Direct)+import Network.AMQP.Worker.Connection (Connection, withChannel, exchange)+import Network.AMQP.Worker.Queue (Queue(..))  -- types -------------------------- @@ -45,21 +45,19 @@  -- | publish a message to a routing key, without making sure a queue exists to handle it or if it is the right type of message body ----- > publishToExchange conn "users.admin.created" (User "username")-publishToExchange :: (ToJSON a, MonadBaseControl IO m) => Connection -> ExchangeName -> RoutingKey -> a -> m ()-publishToExchange conn exg (RoutingKey rk) msg =-  withChannel conn $ \chan -> do-    _ <- liftBase $ AMQP.publishMsg chan exg rk (jsonMessage msg)+-- > publishToExchange conn key (User "username")+publishToExchange :: (ToJSON a, MonadIO m) => Connection -> Key Routing a -> a -> m ()+publishToExchange conn rk msg =+  liftIO $ withChannel conn $ \chan -> do+    _ <- AMQP.publishMsg chan (exchange conn) (keyText rk) (jsonMessage msg)     return ()  --- | publish a message to a queue. Enforces that the message type and queue name are correct at the type level+-- | send a message to a queue. Enforces that the message type and queue name are correct at the type level ----- > let queue = Worker.queue exchange "users" :: Queue Direct User--- > publish conn queue (User "username")-publish :: (ToJSON msg, MonadBaseControl IO m) => Connection -> Queue Direct msg -> msg -> m ()-publish conn (Queue (Exchange exg) key _) =-  publishToExchange conn (AMQP.exchangeName exg) key+-- > publish conn (key "users" :: Key Routing User) (User "username")+publish :: (ToJSON a, MonadIO m) => Connection -> Key Routing a -> a -> m ()+publish = publishToExchange   -- | Check for a message once and attempt to parse it@@ -68,19 +66,19 @@ -- > case res of -- >   Just (Parsed m) -> print m -- >   Just (Error e) -> putStrLn "could not parse message"--- >   Notihng -> putStrLn "No messages on the queue"-consume :: (FromJSON msg, MonadBaseControl IO m) => Connection -> Queue key msg -> m (Maybe (ConsumeResult msg))-consume conn (Queue _ _ options) = do-  mme <- withChannel conn $ \chan -> do-    m <- liftBase $ AMQP.getMsg chan Ack (queueName options)-    pure m+-- >   Nothing -> putStrLn "No messages on the queue"+consume :: (FromJSON msg, MonadIO m) => Connection -> Queue msg -> m (Maybe (ConsumeResult msg))+consume conn (Queue _ name) = do+  mme <- liftIO $ withChannel conn $ \chan -> do+            m <- liftBase $ AMQP.getMsg chan Ack name+            pure m    case mme of     Nothing ->       return Nothing      Just (msg, env) -> do-      liftBase $ AMQP.ackEnv env+      liftIO $ AMQP.ackEnv env       let bd = AMQP.msgBody msg       case Aeson.eitherDecode bd of         Left err ->@@ -97,6 +95,6 @@ -- > case res of -- >   (Parsed m) -> print m -- >   (Error e) -> putStrLn "could not parse message"-consumeNext :: (FromJSON msg, MonadBaseControl IO m) => Microseconds -> Connection -> Queue key msg -> m (ConsumeResult msg)-consumeNext pd conn queue =-    poll pd $ consume conn queue+consumeNext :: (FromJSON msg, MonadIO m) => Microseconds -> Connection -> Queue msg -> m (ConsumeResult msg)+consumeNext pd conn key =+    poll pd $ consume conn key

@@ -3,15 +3,15 @@ module Network.AMQP.Worker.Poll where  import Control.Concurrent (threadDelay)-import Control.Monad.Trans.Control (MonadBaseControl)-import Control.Monad.Base (liftBase)+import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.Loops (untilJust) -poll :: (MonadBaseControl IO m) => Int -> m (Maybe a) -> m a+-- | Keep trying action every N microseconds until it returns Just a+poll :: (MonadIO m) => Int -> m (Maybe a) -> m a poll us action = untilJust $ do     ma <- action     case ma of       Just a -> return $ Just a       Nothing -> do-        liftBase $ threadDelay us+        liftIO $ threadDelay us         return Nothing

@@ -1,74 +1,54 @@ {-# LANGUAGE OverloadedStrings #-} module Network.AMQP.Worker.Queue where +import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Text (Text) import qualified Network.AMQP as AMQP-import Network.AMQP (ExchangeOpts(..), QueueOpts(..))+import Network.AMQP (QueueOpts(..), ExchangeOpts(..)) -import Network.AMQP.Worker.Key (RoutingKey(..), BindingKey(..), QueueKey(..))-import Network.AMQP.Worker.Connection (Connection, withChannel)+import Network.AMQP.Worker.Key (Key(..), Routing, Binding, keyText, bindingKey, KeySegment)+import Network.AMQP.Worker.Connection (Connection, withChannel, exchange)  -type ExchangeName = Text  --- | Declare an exchange------ In AMQP, exchanges can be fanout, direct, or topic. This library attempts to simplify this choice by making all exchanges be topic exchanges, and allowing the user to specify topic or direct behavior on the queue itself. See @queue@------ > exchange :: Exchange--- > exchange = Worker.exchange "testExchange" -exchange :: ExchangeName -> Exchange-exchange nm =-  Exchange $ AMQP.newExchange { exchangeName = nm, exchangeType = "topic" }---- | Declare a direct queue with the name @RoutingKey@. Direct queues work as you expect: you can publish messages to them and read from them.+-- | Declare a direct queue, which will receive messages published with the exact same routing key ----- > queue :: Queue Direct MyMessageType--- > queue = Worker.queue exchange "testQueue"+-- > newUsers :: Queue User+-- > newUsers = Worker.direct (key "users" & word "new") -queue :: Exchange -> RoutingKey -> Queue Direct msg-queue exg (RoutingKey key) =-  Queue exg (RoutingKey key) $ AMQP.newQueue { queueName = key }+direct :: Key Routing msg -> Queue msg+direct key = Queue (bindingKey key) (keyText key) --- | Declare a topic queue. Topic queues allow you to bind a queue to a dynamic address with wildcards++-- | Declare a topic queue, which will receive messages that match using wildcards ----- > queue :: Queue Topic MyMessageType--- > queue = Worker.topicQueue exchange "new-users.*"+-- > anyUsers :: Queue User+-- > anyUsers = Worker.topic "anyUsers" (key "users" & star)+topic :: KeySegment a => Key a msg -> QueueName -> Queue msg+topic key name = Queue (bindingKey key) name -topicQueue :: Exchange -> BindingKey -> Queue Topic msg-topicQueue exg key =-  Queue exg key $ AMQP.newQueue { queueName = showKey key } +type QueueName = Text -data Exchange =-  Exchange AMQP.ExchangeOpts+data Queue msg =+  Queue (Key Binding msg) QueueName   deriving (Show, Eq) --- | Queues consist of an exchange, a type (Direct or Topic), and a message type------ > queue :: Queue Direct MyMessageType-data Queue queueType msg =-  Queue Exchange queueType AMQP.QueueOpts-  deriving (Show, Eq) -type Direct = RoutingKey -type Topic = BindingKey------ | Register a queue and its exchange with the AMQP server. Call this before publishing or reading+-- | Queues must be bound before you publish messages to them, or the messages will not be saved. ----- > let queue = Worker.queue exchange "my-queue" :: Queue Direct Text+-- > let queue = Worker.direct (key "users" & word "new") :: Queue User -- > conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")--- > Worker.initQueue conn queue--initQueue :: (QueueKey key) => Connection -> Queue key msg -> IO ()-initQueue conn (Queue (Exchange exg) key options) =-  withChannel conn $ \chan -> do+-- > Worker.bindQueue conn queue+bindQueue :: (MonadIO m) => Connection -> Queue msg -> m ()+bindQueue conn (Queue key name) =+  liftIO $ withChannel conn $ \chan -> do+    let options = AMQP.newQueue { queueName = name }+    let exg = AMQP.newExchange { exchangeName = (exchange conn), exchangeType = "topic" }     _ <- AMQP.declareExchange chan exg     _ <- AMQP.declareQueue chan options-    _ <- AMQP.bindQueue chan (AMQP.queueName options) (AMQP.exchangeName exg) (showKey key)+    _ <- AMQP.bindQueue chan name (exchange conn) (keyText key)     return ()

@@ -3,9 +3,8 @@  import Control.Concurrent (threadDelay) import Control.Exception (SomeException(..))-import Control.Monad.Base (liftBase)+import Control.Monad.IO.Class (liftIO, MonadIO) import Control.Monad.Catch (Exception(..), catch, MonadCatch)-import Control.Monad.Trans.Control (MonadBaseControl) import Control.Monad (forever) import Data.Aeson (FromJSON) import Data.ByteString.Lazy (ByteString)@@ -20,7 +19,7 @@ -- | Create a worker which loops, checks for messages, and handles errors -- -- > startWorker conn queue = do--- >   Worker.worker def conn queue onError onMessage+-- >   Worker.worker conn def queue onError onMessage -- > -- >   where -- >     onMessage :: Message User@@ -32,8 +31,8 @@ -- >     onError e = do -- >       putStrLn "Do something with errors" -worker :: (FromJSON a, MonadBaseControl IO m, MonadCatch m) => WorkerOptions -> Connection -> Queue key a -> (WorkerException SomeException -> m ()) -> (Message a -> m ()) -> m ()-worker opts conn queue onError action =+worker :: (FromJSON a, MonadIO m, MonadCatch m) => Connection -> WorkerOptions -> Queue a -> (WorkerException SomeException -> m ()) -> (Message a -> m ()) -> m ()+worker conn opts queue onError action =   forever $ do     eres <- consumeNext (pollDelay opts) conn queue     case eres of@@ -44,7 +43,7 @@         catch           (action msg)           (onError . OtherException (body msg))-    liftBase $ threadDelay (loopDelay opts)+    liftIO $ threadDelay (loopDelay opts)   

@@ -1,14 +1,19 @@+{-# LANGUAGE DeriveGeneric     #-} {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE DeriveGeneric #-} module Main where -import Control.Monad.Catch (SomeException)-import Data.Aeson (FromJSON, ToJSON)-import Data.Text (Text)-import GHC.Generics (Generic)-import qualified Network.AMQP.Worker as Worker-import Network.AMQP.Worker (fromURI, Exchange, Queue, Direct, def, WorkerException, Message(..), Connection)-import System.IO (hSetBuffering, stdout, stderr, BufferMode(..))+import           Control.Concurrent      (forkIO)+import           Control.Monad.Catch     (SomeException)+import           Data.Aeson              (FromJSON, ToJSON)+import           Data.Function           ((&))+import           Data.Text               (Text, pack)+import           GHC.Generics            (Generic)+import           Network.AMQP.Worker     (Connection, Message (..),+                                          WorkerException, def, fromURI)+import qualified Network.AMQP.Worker     as Worker+import           Network.AMQP.Worker.Key+import           System.IO               (BufferMode (..), hSetBuffering,+                                          stderr, stdout)  data TestMessage = TestMessage   { greeting :: Text }@@ -18,32 +23,48 @@ instance ToJSON TestMessage  -exchange :: Exchange-exchange = Worker.exchange "testExchange"+newMessages :: Key Routing TestMessage+newMessages = key "messages" & word "new" +results :: Key Routing Text+results = key "results" -queue :: Queue Direct TestMessage-queue = Worker.queue exchange "testQueue"+anyMessages :: Key Binding TestMessage+anyMessages = key "messages" & star  -results :: Queue Direct Text-results = Worker.queue exchange "resultQueue" - example :: IO () example = do+   -- connect   conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672") +  let handleAnyMessages = Worker.topic anyMessages "handleAnyMessage"+   -- initialize the queues-  Worker.initQueue conn queue-  Worker.initQueue conn results+  Worker.bindQueue conn (Worker.direct newMessages)+  Worker.bindQueue conn (Worker.direct results) +  -- topic queue!+  Worker.bindQueue conn handleAnyMessages++  putStrLn "Enter a message"+  msg <- getLine+   -- publish a message-  Worker.publish conn queue (TestMessage "hello world")+  putStrLn "Publishing a message"+  Worker.publish conn newMessages (TestMessage $ pack msg)    -- create a worker, the program loops here-  Worker.worker def conn queue onError (onMessage conn)+  _ <- forkIO $ Worker.worker conn def (Worker.direct newMessages) onError (onMessage conn)+  _ <- forkIO $ Worker.worker conn def (handleAnyMessages) onError (onMessage conn)++  putStrLn "Press any key to exit"+  _ <- getLine+  return ()++   onMessage :: Connection -> Message TestMessage -> IO ()

@@ -1,17 +1,18 @@-module Main where+module Spec where -import Test.Tasty-import Test.Tasty.HUnit+-- import Test.Tasty+-- import Test.Tasty.HUnit  main :: IO () main = do-  defaultMain tests+  putStrLn "Not implemented"+  -- defaultMain tests  -tests :: TestTree-tests = testGroup "Tests"-  [ testCase "TODO" testExample ]+-- tests :: TestTree+-- tests = testGroup "Tests"+--   [ testCase "TODO" testExample ]  -testExample :: Assertion-testExample = pure ()+-- testExample :: Assertion+-- testExample = pure ()