amqp-worker 0.4.0 → 1.0.0

raw patch · 9 files changed

+359/−317 lines, 9 filesdep −transformers-basePVP ok

version bump matches the API change (PVP)

Dependencies removed: transformers-base

API changes (from Hackage documentation)

Files

• CHANGELOG.md +7/−0
• README.md +90/−62
• amqp-worker.cabal +1/−4
• example/Example.hs +74/−49
• src/Network/AMQP/Worker.hs +59/−105
• src/Network/AMQP/Worker/Connection.hs +15/−14
• src/Network/AMQP/Worker/Key.hs +59/−53
• src/Network/AMQP/Worker/Message.hs +15/−9
• src/Network/AMQP/Worker/Queue.hs +39/−21

@@ -1,3 +1,10 @@+### Version 1.0.0++Major simplification of queues and keys. Users no longer have to think about topic vs direct queues.++* Creating queues automatically binds them+* Keys can be used intuitively to control direct receipt, fanout, and load balancing+ ### Version 0.4.0  * removed MonadBaseControl following Data.Pool, other internal Data.Pool updates

@@ -3,92 +3,120 @@  Type-safe AMQP workers. Compatible with RabbitMQ -    {-# LANGUAGE DeriveGeneric     #-}-    {-# LANGUAGE OverloadedStrings #-}-    module Main where+```haskell+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-} -    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)+module Main where -    data TestMessage = TestMessage-      { greeting :: Text }-      deriving (Generic, Show, Eq)+import Control.Concurrent (forkIO)+import Control.Monad.Catch (SomeException)+import Data.Aeson (FromJSON, ToJSON)+import Data.Function ((&))+import Data.Text (Text)+import GHC.Generics (Generic)+import Network.AMQP.Worker+import qualified Network.AMQP.Worker as Worker+import System.IO (BufferMode (..), hSetBuffering, stderr, stdout) -    instance FromJSON TestMessage-    instance ToJSON TestMessage+newtype Greeting = Greeting+    {message :: Text}+    deriving (Generic, Show, Eq) +instance FromJSON Greeting+instance ToJSON Greeting -    newMessages :: Key Routing TestMessage-    newMessages = key "messages" & word "new"+newGreetings :: Key Routing Greeting+newGreetings = key "greetings" & word "new" -    results :: Key Routing Text-    results = key "results"+anyGreetings :: Key Binding Greeting+anyGreetings = key "greetings" & any1 -    anyMessages :: Key Binding TestMessage-    anyMessages = key "messages" & star+example :: IO ()+example = do+    conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")+    simple conn +publishing :: Connection -> IO ()+publishing conn = do+    Worker.publish conn newGreetings $ Greeting "Hello" +-- | Create a queue to process messages+simple :: Connection -> IO ()+simple conn = do+    -- create a queue to receive them+    q <- Worker.queue conn def newGreetings -    example :: IO ()-    example = do+    -- publish a message (delivered to queue)+    Worker.publish conn newGreetings $ Greeting "Hello" -      -- connect-      conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")+    -- We cannot publish to anyGreetings because it is a binding key (with wildcards in it)+    -- Worker.publish conn anyGreetings $ TestMessage "Compiler Error" -      let handleAnyMessages = Worker.topic anyMessages "handleAnyMessage"+    -- Loop and print any values received+    Worker.worker conn def q onError (print . value) -      -- initialize the queues-      Worker.bindQueue conn (Worker.direct newMessages)-      Worker.bindQueue conn (Worker.direct results)+-- | Multiple queues with distinct names will each get copies of published messages+multiple :: Connection -> IO ()+multiple conn = do+    -- create two separate queues+    one <- Worker.queue conn "one" newGreetings+    two <- Worker.queue conn "two" newGreetings -      -- topic queue!-      Worker.bindQueue conn handleAnyMessages+    -- publish a message (delivered to both)+    Worker.publish conn newGreetings $ Greeting "Hello" -      putStrLn "Enter a message"-      msg <- getLine+    -- Each of these workers will receive the same message+    _ <- forkIO $ Worker.worker conn def one onError $ \m -> putStrLn "one" >> print (value m)+    _ <- forkIO $ Worker.worker conn def two onError $ \m -> putStrLn "two" >> print (value m) -      -- publish a message-      putStrLn "Publishing a message"-      Worker.publish conn newMessages (TestMessage $ pack msg)+    putStrLn "Press any key to exit"+    _ <- getLine+    return () -      -- 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)+-- | Create multiple workers on the same queue to load balance between them+balance :: Connection -> IO ()+balance conn = do+    -- create a single queue+    q <- Worker.queue conn def newGreetings -      putStrLn "Press any key to exit"-      _ <- getLine-      return ()+    -- publish two messages+    Worker.publish conn newGreetings $ Greeting "Hello1"+    Worker.publish conn newGreetings $ Greeting "Hello2" +    -- Each worker will receive one of the messages+    _ <- forkIO $ Worker.worker conn def q onError $ \m -> putStrLn "one" >> print (value m)+    _ <- forkIO $ Worker.worker conn def q onError $ \m -> putStrLn "two" >> print (value m) +    putStrLn "Press any key to exit"+    _ <- getLine+    return () +-- | You can bind to messages dynamically with wildcards in Binding Keys+dynamic :: Connection -> IO ()+dynamic conn = do+    -- \| anyGreetings matches `greetings.*`+    q <- Worker.queue conn def anyGreetings -    onMessage :: Connection -> Message TestMessage -> IO ()-    onMessage conn m = do-      let testMessage = value m-      putStrLn "Got Message"-      print testMessage-      Worker.publish conn results (greeting testMessage)+    -- You can only publish to a Routing Key. Publishing to anyGreetings will give a compile error+    Worker.publish conn newGreetings $ Greeting "Hello" +    -- This queue listens for anything under `greetings.`+    Worker.worker conn def q onError $ \m -> putStrLn "Got: " >> print (value m) -    onError :: WorkerException SomeException -> IO ()-    onError e = do-      putStrLn "Do something with errors"-      print e+onError :: WorkerException SomeException -> IO ()+onError e = do+    putStrLn "Do something with errors"+    print e +test :: (Connection -> IO ()) -> IO ()+test action = do+    hSetBuffering stdout LineBuffering+    hSetBuffering stderr LineBuffering+    conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")+    action conn +main :: IO ()+main = example -    main :: IO ()-    main = do-      hSetBuffering stdout LineBuffering-      hSetBuffering stderr LineBuffering-      example+```

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           amqp-worker-version:        0.4.0+version:        1.0.0 synopsis:       Type-safe AMQP workers description:    Please see the README on GitHub at <https://github.com/seanhess/amqp-worker#readme> category:       Network@@ -49,7 +49,6 @@     , mtl ==2.2.*     , resource-pool >=0.3 && <0.5     , text ==1.2.*-    , transformers-base ==0.4.*   default-language: Haskell2010  executable example@@ -71,7 +70,6 @@     , mtl ==2.2.*     , resource-pool >=0.3 && <0.5     , text ==1.2.*-    , transformers-base ==0.4.*   default-language: Haskell2010  test-suite test@@ -94,5 +92,4 @@     , mtl ==2.2.*     , resource-pool >=0.3 && <0.5     , text ==1.2.*-    , transformers-base ==0.4.*   default-language: Haskell2010

@@ -7,83 +7,108 @@ import Control.Monad.Catch (SomeException) import Data.Aeson (FromJSON, ToJSON) import Data.Function ((&))-import Data.Text (Text, pack)+import Data.Text (Text) 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-    )+import System.IO (BufferMode (..), hSetBuffering, stderr, stdout) -data TestMessage = TestMessage-    {greeting :: Text}+newtype Greeting = Greeting+    {message :: Text}     deriving (Generic, Show, Eq) -instance FromJSON TestMessage-instance ToJSON TestMessage--newMessages :: Key Routing TestMessage-newMessages = key "messages" & word "new"+instance FromJSON Greeting+instance ToJSON Greeting -results :: Key Routing Text-results = key "results"+newGreetings :: Key Routing Greeting+newGreetings = key "greetings" & word "new" -anyMessages :: Key Binding TestMessage-anyMessages = key "messages" & star+anyGreetings :: Key Binding Greeting+anyGreetings = key "greetings" & any1  example :: IO () example = do-    -- connect     conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")+    simple conn -    let handleAnyMessages = Worker.topic anyMessages "handleAnyMessage"+publishing :: Connection -> IO ()+publishing conn = do+    Worker.publish conn newGreetings $ Greeting "Hello" -    -- initialize the queues-    Worker.bindQueue conn (Worker.direct newMessages)-    Worker.bindQueue conn (Worker.direct results)+-- | Create a queue to process messages+simple :: Connection -> IO ()+simple conn = do+    -- create a queue to receive them+    q <- Worker.queue conn def newGreetings -    -- topic queue!-    Worker.bindQueue conn handleAnyMessages+    -- publish a message (delivered to queue)+    Worker.publish conn newGreetings $ Greeting "Hello" -    putStrLn "Enter a message"-    msg <- getLine+    -- We cannot publish to anyGreetings because it is a binding key (with wildcards in it)+    -- Worker.publish conn anyGreetings $ TestMessage "Compiler Error" -    -- publish a message-    putStrLn "Publishing a message"-    Worker.publish conn newMessages (TestMessage $ pack msg)+    -- Loop and print any values received+    Worker.worker conn def q onError (print . value) -    -- 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)+-- | Multiple queues with distinct names will each get copies of published messages+multiple :: Connection -> IO ()+multiple conn = do+    -- create two separate queues+    one <- Worker.queue conn "one" newGreetings+    two <- Worker.queue conn "two" newGreetings +    -- publish a message (delivered to both)+    Worker.publish conn newGreetings $ Greeting "Hello"++    -- Each of these workers will receive the same message+    _ <- forkIO $ Worker.worker conn def one onError $ \m -> putStrLn "one" >> print (value m)+    _ <- forkIO $ Worker.worker conn def two onError $ \m -> putStrLn "two" >> print (value m)+     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)+-- | Create multiple workers on the same queue to load balance between them+balance :: Connection -> IO ()+balance conn = do+    -- create a single queue+    q <- Worker.queue conn def newGreetings +    -- publish two messages+    Worker.publish conn newGreetings $ Greeting "Hello1"+    Worker.publish conn newGreetings $ Greeting "Hello2"++    -- Each worker will receive one of the messages+    _ <- forkIO $ Worker.worker conn def q onError $ \m -> putStrLn "one" >> print (value m)+    _ <- forkIO $ Worker.worker conn def q onError $ \m -> putStrLn "two" >> print (value m)++    putStrLn "Press any key to exit"+    _ <- getLine+    return ()++-- | You can bind to messages dynamically with wildcards in Binding Keys+dynamic :: Connection -> IO ()+dynamic conn = do+    -- anyGreetings matches `greetings.*`+    q <- Worker.queue conn def anyGreetings++    -- You can only publish to a Routing Key. Publishing to anyGreetings will give a compile error+    Worker.publish conn newGreetings $ Greeting "Hello"++    -- This queue listens for anything under `greetings.`+    Worker.worker conn def q onError $ \m -> putStrLn "Got: " >> print (value m)+ onError :: WorkerException SomeException -> IO () onError e = do     putStrLn "Do something with errors"     print e -main :: IO ()-main = do+test :: (Connection -> IO ()) -> IO ()+test action = do     hSetBuffering stdout LineBuffering     hSetBuffering stderr LineBuffering-    example+    conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")+    action conn++main :: IO ()+main = example

@@ -2,125 +2,44 @@ {-# LANGUAGE OverloadedStrings #-}  -- |------ High level functions for working with message queues. Built on top of Network.AMQP. See https://hackage.haskell.org/package/amqp, which only works with RabbitMQ: https://www.rabbitmq.com/------ /Example/:------ Connect to a server, initialize a queue, publish a message, and create a worker to process them.+-- Module:      Network.AMQP.Worker+-- Copyright:   (c) 2023 Sean Hess+-- License:     BSD3+-- Maintainer:  Sean Hess <seanhess@gmail.com>+-- Stability:   experimental+-- Portability: portable ----- > {-# 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--- >--- >+-- Type safe and simplified message queues with AMQP module Network.AMQP.Worker-    ( -- * Routing Keys+    ( -- * How to use this library+      -- $use++      -- * Binding and Routing Keys       Key (..)-    , Routing     , Binding+    , Routing     , word     , key-    , star-    , hash+    , any1+    , many        -- * Connecting-    , Connection     , connect-    , disconnect-    , exchange     , AMQP.fromURI--      -- * Initializing queues-    , Queue (..)-    , direct-    , topic-    , bindQueue+    , Connection        -- * Sending Messages     , publish-    , publishToExchange -      -- * Reading Messages-    , consume-    , consumeNext-    , ConsumeResult (..)+      -- * Initializing queues+    , queue+    , queueNamed+    , Queue (..)+    , queueName+    , QueueName+    , QueuePrefix (..)++      -- * Messages     , ParseError (..)     , Message (..) @@ -140,3 +59,38 @@ import Network.AMQP.Worker.Message import Network.AMQP.Worker.Queue import Network.AMQP.Worker.Worker++-- $use+--+-- Define keys to identify how messages will be published and what the message type is+--+-- > import Network.AMQP.Worker as Worker+-- >+-- > data Greeting = Greeting+-- >   { message :: Text }+-- >   deriving (Generic, Show, Eq)+-- >+-- > instance FromJSON Greeting+-- > instance ToJSON Greeting+-- >+-- > newGreetings :: Key Routing Greeting+-- > newGreetings = key "messages" & word "greetings" & word "new"+--+-- Connect to AMQP and publish a message+--+-- >   conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")+-- >+-- >   Worker.publish conn newMessages $ TestMessage "hello"+--+-- To receive messages, first define a queue. You can bind direclty to the Routing Key to ensure it is delivered once+--+-- >   q <- Worker.queue conn def newMessages :: IO (Queue Greeting)+-- >+-- >   -- Loop and print any values received+-- >   Worker.worker conn def q onError (print . value)+--+-- You can also define dynamic Routing Keys to receive many kinds of messages+--+-- >   let newMessages = key "messages" & any1 & word "new"+-- >   q <- Worker.queue conn def newMessages :: IO (Queue Greeting)+-- >

@@ -3,20 +3,17 @@  module Network.AMQP.Worker.Connection     ( Connection (..)+    , AMQP.ConnectionOpts (..)+    , AMQP.defaultConnectionOpts     , connect     , disconnect     , withChannel     ) where -import Control.Concurrent.MVar-    ( MVar-    , newEmptyMVar-    , putMVar-    , readMVar-    , takeMVar-    )+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, readMVar, takeMVar) import Control.Monad.Catch (catch, throwM) import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.Function ((&)) import Data.Pool (Pool) import qualified Data.Pool as Pool import Data.Text (Text)@@ -25,7 +22,6 @@  type ExchangeName = Text --- | Internal connection details data Connection = Connection     { amqpConn :: MVar AMQP.Connection     , pool :: Pool Channel@@ -34,7 +30,7 @@  -- | Connect to the AMQP server. ----- >>> conn <- connect (fromURI "amqp://guest:guest@localhost:5672")+-- > conn <- connect (fromURI "amqp://guest:guest@localhost:5672") connect :: MonadIO m => AMQP.ConnectionOpts -> m Connection connect opts = liftIO $ do     -- use a default exchange name@@ -44,16 +40,21 @@     cvar <- newEmptyMVar     openConnection cvar -    let config = Pool.defaultPoolConfig (create cvar) destroy openTime numChans-     -- open a shared pool for channels-    chans <- Pool.newPool config+    chans <- Pool.newPool (config cvar)      pure $ Connection cvar chans exchangeName   where+    config cvar =+        Pool.defaultPoolConfig (create cvar) destroy openTime maxChans+            & Pool.setNumStripes (Just 1)++    openTime :: Double     openTime = 10-    numChans = 4 +    maxChans :: Int+    maxChans = 8+     openConnection cvar = do         -- open a connection and store in the mvar         conn <- AMQP.openConnection'' opts@@ -83,7 +84,7 @@     Pool.destroyAllResources $ pool c     AMQP.closeConnection conn --- | Perform an action with a channel resource+-- | Perform an action with a channel resource, and give it back at the end withChannel :: Connection -> (Channel -> IO b) -> IO b withChannel (Connection _ p _) action = do     Pool.withResource p action

@@ -1,85 +1,91 @@+{-# LANGUAGE DataKinds #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE UndecidableInstances #-}  module Network.AMQP.Worker.Key     ( Key (..)     , Binding (..)-    , BindingWord     , Routing     , key     , word-    , star-    , hash+    , any1+    , many     , keyText-    , KeySegment (..)-    , bindingKey+    , fromBind+    , toBind+    , toBindingKey+    , RequireRouting     ) where +import Data.Kind (Constraint, Type) import qualified Data.List as List import Data.Text (Text) import qualified Data.Text as Text---- | Keys describe routing and binding info for a message-newtype Key a msg = Key [a]-    deriving (Eq, Show, Semigroup, Monoid)+import GHC.TypeLits (ErrorMessage (..), TypeError) --- | Every message is sent with a specific routing key+-- | Messages are published with a specific identifier called a Routing key. Queues can use Binding Keys to control which messages are delivered to them. ----- > 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+-- Routing keys have no dynamic component and can be used to publish messages --+-- > commentsKey :: Key Routing Comment+-- > commentsKey = key "posts" & word "new"+--+-- Binding keys can contain wildcards, only used for matching messages+-- -- > commentsKey :: Key Binding Comment--- > commentsKey = key "posts" & star & word "comments" & hash+-- > commentsKey = key "posts" & any1 & word "comments" & many+newtype Key a msg = Key [Binding]+    deriving (Eq, Show, Semigroup, Monoid)++data Routing+ data Binding-    = Word BindingWord-    | Star-    | Hash+    = Word Text+    | Any+    | Many     deriving (Eq, Show) -instance KeySegment Binding where-    toText (Word t) = t-    toText Star = "*"-    toText Hash = "#"-    fromText = Word-    toBind = id+fromBind :: Binding -> Text+fromBind (Word t) = t+fromBind Any = "*"+fromBind Many = "#" -class KeySegment a where-    toText :: a -> Text-    fromText :: Text -> a-    toBind :: a -> Binding+toBind :: Text -> Binding+toBind = Word -keyText :: KeySegment a => Key a msg -> Text+keyText :: 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)+    Text.intercalate "." . List.map fromBind $ ns --- | Match any one word-star :: KeySegment a => Key a msg -> Key Binding msg-star (Key ws) = Key (map toBind ws ++ [Star])+-- | Match any one word. Equivalent to `*`. Converts to a Binding key and can no longer be used to publish messaages+any1 :: Key a msg -> Key Binding msg+any1 (Key ws) = Key (ws ++ [Any]) --- | Match any words-hash :: KeySegment a => Key a msg -> Key Binding msg-hash (Key ws) = Key (map toBind ws ++ [Hash])+-- | Match zero or more words. Equivalient to `#`. Converts to a Binding key and can no longer be used to publish messages+many :: Key a msg -> Key Binding msg+many (Key ws) = Key (ws ++ [Many]) --- | Match a specific word-word :: KeySegment a => Text -> Key a msg -> Key a msg-word w (Key ws) = Key $ ws ++ [fromText w]+-- | A specific word. Can be used to chain Routing keys or Binding keys+word :: Text -> Key a msg -> Key a msg+word w (Key ws) = Key $ ws ++ [toBind w] --- | Create a new key+-- | Start a new routing key (can also be used for bindings) key :: Text -> Key Routing msg-key t = Key [Routing t]+key t = Key [Word t] -type BindingWord = Text+-- | We can convert Routing Keys to Binding Keys safely, as they are usable for both publishing and binding+toBindingKey :: Key a msg -> Key Binding msg+toBindingKey (Key ws) = Key ws++-- | Custom error message when trying to publish to Binding keys+type family RequireRouting (a :: Type) :: Constraint where+    RequireRouting Binding =+        TypeError+            ( 'Text "Expected Routing Key but got Binding Key instead. Messages can be published only with keys that exclusivlely use `key` and `word`"+                :$$: 'Text "\n          key \"message\" & word \"new\" \n"+            )+    RequireRouting a = ()

@@ -1,18 +1,17 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TypeFamilies #-}  module Network.AMQP.Worker.Message where -import Control.Monad.Base (liftBase) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Aeson (FromJSON, ToJSON) import qualified Data.Aeson as Aeson import Data.ByteString.Lazy (ByteString) import Network.AMQP (Ack (..), DeliveryMode (..), newMsg) import qualified Network.AMQP as AMQP- import Network.AMQP.Worker.Connection (Connection, exchange, withChannel)-import Network.AMQP.Worker.Key (Key, Routing, keyText)+import Network.AMQP.Worker.Key (Key, RequireRouting, Routing, keyText) import Network.AMQP.Worker.Poll (poll) import Network.AMQP.Worker.Queue (Queue (..)) @@ -31,8 +30,6 @@  data ParseError = ParseError String ByteString -type Microseconds = Int- jsonMessage :: ToJSON a => a -> AMQP.Message jsonMessage a =     newMsg@@ -45,7 +42,7 @@ -- | 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 key (User "username")-publishToExchange :: (ToJSON a, MonadIO m) => Connection -> Key Routing a -> a -> m ()+publishToExchange :: (RequireRouting a, ToJSON msg, MonadIO m) => Connection -> Key a msg -> msg -> m () publishToExchange conn rk msg =     liftIO $ withChannel conn $ \chan -> do         _ <- AMQP.publishMsg chan (exchange conn) (keyText rk) (jsonMessage msg)@@ -53,8 +50,15 @@  -- | send a message to a queue. Enforces that the message type and queue name are correct at the type level ----- > publish conn (key "users" :: Key Routing User) (User "username")-publish :: (ToJSON a, MonadIO m) => Connection -> Key Routing a -> a -> m ()+-- > let key = key "users" :: Key Routing User+-- > publish conn key (User "username")+--+-- Publishing to a Binding Key results in an error+--+-- > -- Compiler error! This doesn't make sense+-- > let key = key "users" & many :: Key Binding User+-- > publish conn key (User "username")+publish :: (RequireRouting a, ToJSON msg, MonadIO m) => Connection -> Key a msg -> msg -> m () publish = publishToExchange  -- | Check for a message once and attempt to parse it@@ -67,7 +71,7 @@ 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+        m <- AMQP.getMsg chan Ack name         pure m      case mme of@@ -91,3 +95,5 @@ consumeNext :: (FromJSON msg, MonadIO m) => Microseconds -> Connection -> Queue msg -> m (ConsumeResult msg) consumeNext pd conn key =     poll pd $ consume conn key++type Microseconds = Int

@@ -1,45 +1,63 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE OverloadedStrings #-}  module Network.AMQP.Worker.Queue where  import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.Default (Default (..))+import Data.String (IsString) import Data.Text (Text)-import Network.AMQP (ExchangeOpts (..), QueueOpts (..))+import Network.AMQP (ExchangeOpts (..), QueueOpts) import qualified Network.AMQP as AMQP  import Network.AMQP.Worker.Connection (Connection, exchange, withChannel)-import Network.AMQP.Worker.Key (Binding, Key (..), KeySegment, Routing, bindingKey, keyText)+import Network.AMQP.Worker.Key (Binding, Key (..), keyText, toBindingKey) --- | Declare a direct queue, which will receive messages published with the exact same routing key------ > newUsers :: Queue User--- > newUsers = Worker.direct (key "users" & word "new")-direct :: Key Routing msg -> Queue msg-direct key = Queue (bindingKey key) (keyText key)+type QueueName = Text --- | Declare a topic queue, which will receive messages that match using wildcards------ > 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+newtype QueuePrefix = QueuePrefix Text+    deriving (Show, Eq, IsString) -type QueueName = Text+instance Default QueuePrefix where+    def = QueuePrefix "main" +-- | A queue is an inbox for messages to be delivered data Queue msg     = Queue (Key Binding msg) QueueName     deriving (Show, Eq) --- | Queues must be bound before you publish messages to them, or the messages will not be saved.+-- | Create a queue to receive messages matching the 'Key' with a name prefixed via `queueName`. ----- > let queue = Worker.direct (key "users" & word "new") :: Queue User--- > conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")--- > Worker.bindQueue conn queue+-- > q <- Worker.queue conn "main" $ key "messages" & any1+-- > Worker.worker conn def q onError onMessage+queue :: (MonadIO m) => Connection -> QueuePrefix -> Key a msg -> m (Queue msg)+queue conn pre key = do+    queueNamed conn (queueName pre key) key++-- | Create a queue to receive messages matching the binding key. Each queue with a unique name+-- will be delivered a separate copy of the messsage. Workers operating on the same queue,+-- or on queues with the same name will load balance+queueNamed :: (MonadIO m) => Connection -> QueueName -> Key a msg -> m (Queue msg)+queueNamed conn name key = do+    let q = Queue (toBindingKey key) name+    bindQueue conn q+    return q++-- | Name a queue with a prefix and the binding key name. Useful for seeing at+-- a glance which queues are receiving which messages+--+-- > -- "main messages.new"+-- > queueName "main" (key "messages" & word "new")+queueName :: QueuePrefix -> Key a msg -> QueueName+queueName (QueuePrefix pre) key = pre <> " " <> keyText key++-- | Queues must be bound before you publish messages to them, or the messages will not be saved.+-- Use `queue` or `queueNamed` instead 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"}+        let options = AMQP.newQueue{AMQP.queueName = name}+        let exg = AMQP.newExchange{exchangeName = exchange conn, exchangeType = "topic"}         _ <- AMQP.declareExchange chan exg         _ <- AMQP.declareQueue chan options         _ <- AMQP.bindQueue chan name (exchange conn) (keyText key)