amqp-worker 2.0.0 → 2.0.1

raw patch · 8 files changed

+121/−138 lines, 8 filesdep +data-defaultdep ~bytestringdep ~mtldep ~resource-poolPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: data-default

Dependency ranges changed: bytestring, mtl, resource-pool, text

API changes (from Hackage documentation)

Files

• README.md +44/−35
• amqp-worker.cabal +17/−14
• example/Example.hs +3/−3
• src/Network/AMQP/Worker.hs +13/−11
• src/Network/AMQP/Worker/Connection.hs +20/−38
• src/Network/AMQP/Worker/Key.hs +8/−23
• src/Network/AMQP/Worker/Message.hs +5/−7
• src/Network/AMQP/Worker/Queue.hs +11/−7

@@ -1,9 +1,7 @@ AMQP Worker ----------- -Type-safe and simplified message queues with AMQP. Compatible with RabbitMQ.--[View on Hackage](https://hackage.haskell.org/package/amqp-worker)+Type-safe AMQP workers. Compatible with RabbitMQ  ```haskell {-# LANGUAGE DeriveGeneric #-}@@ -12,12 +10,14 @@ module Main where  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)  newtype Greeting = Greeting     {message :: Text}@@ -26,7 +26,7 @@ instance FromJSON Greeting instance ToJSON Greeting -newGreetings :: Key Route Greeting+newGreetings :: Key Bind Greeting newGreetings = key "greetings" & word "new"  anyGreetings :: Key Bind Greeting@@ -45,15 +45,17 @@ simple :: Connection -> IO () simple conn = do     -- create a queue to receive them-    q <- Worker.queue conn "main" newGreetings+    q <- Worker.queue conn def newGreetings      -- publish a message (delivered to queue)     Worker.publish conn newGreetings $ Greeting "Hello" -    -- wait until we receive the message-    m <- Worker.takeMessage conn q-    print (value m)+    -- We cannot publish to anyGreetings because it is a binding key (with wildcards in it)+    -- Worker.publish conn anyGreetings $ TestMessage "Compiler Error" +    -- Loop and print any values received+    Worker.worker conn def q onError (print . value)+ -- | Multiple queues with distinct names will each get copies of published messages multiple :: Connection -> IO () multiple conn = do@@ -64,33 +66,27 @@     -- publish a message (delivered to both)     Worker.publish conn newGreetings $ Greeting "Hello" -    -- Each of these queues will receive the same message-    m1 <- Worker.takeMessage conn one-    m2 <- Worker.takeMessage conn two+    -- 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) -    print $ value m1-    print $ value m2+    putStrLn "Press any key to exit"+    _ <- getLine+    return () --- | Create workers to continually process messages-workers :: Connection -> IO ()-workers conn = do+-- | 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 "main" newGreetings+    q <- Worker.queue conn def newGreetings -    -- publish some messages+    -- publish two messages     Worker.publish conn newGreetings $ Greeting "Hello1"     Worker.publish conn newGreetings $ Greeting "Hello2"-    Worker.publish conn newGreetings $ Greeting "Hello3" -    -- Create a worker to process any messages on the queue-    _ <- forkIO $ Worker.worker conn q $ \m -> do-        putStrLn "one"-        print (value m)--    -- Listening to the same queue with N workers will load balance them-    _ <- forkIO $ Worker.worker conn q $ \m -> do-        putStrLn "two"-        print (value m)+    -- 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@@ -99,15 +95,28 @@ -- | You can bind to messages dynamically with wildcards in Binding Keys dynamic :: Connection -> IO () dynamic conn = do-    -- anyGreetings matches `greetings.*`-    q <- Worker.queue conn "main" anyGreetings+    -- \| anyGreetings matches `greetings.*`+    q <- Worker.queue conn def anyGreetings -    -- You can only publish to a specific Routing Key, like `greetings.new`+    -- You can only publish to a Routing Key. Publishing to anyGreetings will give a compile error     Worker.publish conn newGreetings $ Greeting "Hello" -    -- We cannot publish to anyGreetings because it is a Binding Key (with wildcards in it)-    -- Worker.publish conn anyGreetings $ Greeting "Compiler Error"+    -- This queue listens for anything under `greetings.`+    Worker.worker conn def q onError $ \m -> putStrLn "Got: " >> print (value m) -    m <- Worker.takeMessage conn q-    print $ value m+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+ ```

@@ -1,11 +1,11 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.35.3.+-- This file has been generated from package.yaml by hpack version 0.37.0. -- -- see: https://github.com/sol/hpack  name:           amqp-worker-version:        2.0.0+version:        2.0.1 synopsis:       Type-safe AMQP workers description:    Please see the README on GitHub at <https://github.com/seanhess/amqp-worker#readme> category:       Network@@ -41,12 +41,13 @@       aeson >=2.0 && <2.3     , amqp >=0.20 && <1     , base >=4.9 && <5-    , bytestring ==0.11.*+    , bytestring >=0.11 && <0.13+    , data-default >=0.7 && <0.9     , exceptions ==0.10.*     , monad-loops ==0.4.*-    , mtl ==2.2.*-    , resource-pool >=0.3 && <0.5-    , text ==1.2.*+    , mtl >=2.2 && <2.4+    , resource-pool ==0.4.*+    , text ==2.*   default-language: Haskell2010  executable example@@ -61,12 +62,13 @@     , amqp >=0.20 && <1     , amqp-worker     , base >=4.9 && <5-    , bytestring ==0.11.*+    , bytestring >=0.11 && <0.13+    , data-default >=0.7 && <0.9     , exceptions ==0.10.*     , monad-loops ==0.4.*-    , mtl ==2.2.*-    , resource-pool >=0.3 && <0.5-    , text ==1.2.*+    , mtl >=2.2 && <2.4+    , resource-pool ==0.4.*+    , text ==2.*   default-language: Haskell2010  test-suite test@@ -82,10 +84,11 @@     , amqp >=0.20 && <1     , amqp-worker     , base >=4.9 && <5-    , bytestring ==0.11.*+    , bytestring >=0.11 && <0.13+    , data-default >=0.7 && <0.9     , exceptions ==0.10.*     , monad-loops ==0.4.*-    , mtl ==2.2.*-    , resource-pool >=0.3 && <0.5-    , text ==1.2.*+    , mtl >=2.2 && <2.4+    , resource-pool ==0.4.*+    , text ==2.*   default-language: Haskell2010

@@ -38,7 +38,7 @@ simple :: Connection -> IO () simple conn = do     -- create a queue to receive them-    q <- Worker.queue conn "main" newGreetings+    q <- Worker.queue conn def newGreetings      -- publish a message (delivered to queue)     Worker.publish conn newGreetings $ Greeting "Hello"@@ -68,7 +68,7 @@ workers :: Connection -> IO () workers conn = do     -- create a single queue-    q <- Worker.queue conn "main" newGreetings+    q <- Worker.queue conn def newGreetings      -- publish some messages     Worker.publish conn newGreetings $ Greeting "Hello1"@@ -93,7 +93,7 @@ dynamic :: Connection -> IO () dynamic conn = do     -- anyGreetings matches `greetings.*`-    q <- Worker.queue conn "main" anyGreetings+    q <- Worker.queue conn def anyGreetings      -- You can only publish to a specific Routing Key, like `greetings.new`     Worker.publish conn newGreetings $ Greeting "Hello"

@@ -9,25 +9,25 @@ -- Stability:   experimental -- Portability: portable ----- Type safe and simplified message queues with AMQP. Compatible with RabbitMQ+-- Type safe and simplified message queues with AMQP module Network.AMQP.Worker     ( -- * How to use this library       -- $use -      -- * Connecting-      connect-    , AMQP.fromURI-    , Connection-       -- * Binding and Routing Keys-    , Key (..)+      Key (..)     , Bind     , Route-    , key     , word+    , key     , any1     , many +      -- * Connecting+    , connect+    , AMQP.fromURI+    , Connection+       -- * Sending Messages     , publish @@ -40,14 +40,16 @@     , QueuePrefix (..)        -- * Messages-    , takeMessage     , ParseError (..)     , Message (..)+    , takeMessage        -- * Worker     , worker+    , Default.def     ) where +import qualified Data.Default as Default import qualified Network.AMQP as AMQP  import Network.AMQP.Worker.Connection@@ -84,8 +86,8 @@ -- -- Define dynamic Routing Keys to receive many kinds of messages ----- > let anyMessages = key "messages" & any1--- > q <- Worker.queue conn "main" anyMessages+-- > let newMessages = key "messages" & any1 & word "new"+-- > q <- Worker.queue conn def newMessages :: IO (Queue Greeting) -- > m <- Worker.takeMessage conn q -- > print (value m) --

@@ -1,19 +1,13 @@-{-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-}  module Network.AMQP.Worker.Connection-    ( connect-    , connect'-    , disconnect-    , withChannel-    , Connection (..)-    , WorkerOpts (..)-    , ExchangeName+    ( Connection (..)     , AMQP.ConnectionOpts (..)     , AMQP.defaultConnectionOpts-    , AMQP.fromURI+    , connect+    , disconnect+    , withChannel     ) where  import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, readMVar, takeMVar)@@ -34,32 +28,14 @@     , exchange :: ExchangeName     } -data WorkerOpts = WorkerOpts-    { exchange :: ExchangeName-    -- ^ Everything goes on one exchange-    , openTime :: Double-    -- ^ Number of seconds connections in the pool remain open for re-use-    , maxChannels :: Int-    -- ^ Number of concurrent connectinos available in the pool-    , numStripes :: Maybe Int-    }-    deriving (Show, Eq)---- | Connect to the AMQP server using simple defaults+-- | Connect to the AMQP server. -- -- > conn <- connect (fromURI "amqp://guest:guest@localhost:5672") connect :: MonadIO m => AMQP.ConnectionOpts -> m Connection-connect opts =-    connect' opts $-        WorkerOpts-            { exchange = "amq.topic"-            , openTime = 10-            , maxChannels = 8-            , numStripes = Just 1-            }+connect opts = liftIO $ do+    -- use a default exchange name+    let exchangeName = "amq.topic" -connect' :: MonadIO m => AMQP.ConnectionOpts -> WorkerOpts -> m Connection-connect' copt wopt = liftIO $ do     -- create a single connection in an mvar     cvar <- newEmptyMVar     openConnection cvar@@ -67,15 +43,21 @@     -- open a shared pool for channels     chans <- Pool.newPool (config cvar) -    pure $ Connection cvar chans wopt.exchange+    pure $ Connection cvar chans exchangeName   where     config cvar =-        Pool.defaultPoolConfig (create cvar) destroy wopt.openTime wopt.maxChannels-            & Pool.setNumStripes wopt.numStripes+        Pool.defaultPoolConfig (create cvar) destroy openTime maxChans+            & Pool.setNumStripes (Just 1) +    openTime :: Double+    openTime = 10++    maxChans :: Int+    maxChans = 8+     openConnection cvar = do         -- open a connection and store in the mvar-        conn <- AMQP.openConnection'' copt+        conn <- AMQP.openConnection'' opts         putMVar cvar conn      reopenConnection cvar = do@@ -85,9 +67,9 @@      create cvar = do         conn <- readMVar cvar-        catch (AMQP.openChannel conn) (createEx cvar)+        chan <- catch (AMQP.openChannel conn) (createEx cvar)+        return chan -    -- Reopen closed connections     createEx cvar (ConnectionClosedException _ _) = do         reopenConnection cvar         create cvar

@@ -61,36 +61,21 @@ keyText (Key ns) =     Text.intercalate "." . List.map fromBind $ ns --- | Start a new routing key (can also be used for bindings)------ > key "messages"--- > -- matches "messages"-key :: Text -> Key Route msg-key t = Key [Word t]---- | A specific word. Can be used to chain Routing keys or Binding keys------ > key "messages" & word "new"--- > -- matches "messages.new"-word :: Text -> Key a msg -> Key a msg-word w (Key ws) = Key $ ws ++ [toBind w]- -- | Match any one word. Equivalent to `*`. Converts to a Binding key and can no longer be used to publish messaages------ > key "messages" & any1--- > -- matches "messages.new"--- > -- matches "messages.update" any1 :: Key a msg -> Key Bind msg any1 (Key ws) = Key (ws ++ [Any])  -- | Match zero or more words. Equivalient to `#`. Converts to a Binding key and can no longer be used to publish messages------ > key "messages" & many--- > -- matches "messages"--- > -- matches "messages.new"--- > -- matches "messages.1234.update" many :: Key a msg -> Key Bind msg many (Key ws) = Key (ws ++ [Many])++-- | 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]++-- | Start a new routing key (can also be used for bindings)+key :: Text -> Key Route msg+key t = Key [Word t]  -- | We can convert Route Keys to Bind Keys safely, as they are usable for both publishing and binding toBindKey :: Key a msg -> Key Bind msg

@@ -1,6 +1,5 @@ {-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TypeFamilies #-} @@ -14,7 +13,7 @@ import Data.ByteString.Lazy (ByteString) import Network.AMQP (Ack (..), DeliveryMode (..), newMsg) import qualified Network.AMQP as AMQP-import Network.AMQP.Worker.Connection (Connection (..), withChannel)+import Network.AMQP.Worker.Connection (Connection, exchange, withChannel) import Network.AMQP.Worker.Key (Key, RequireRoute, Route, keyText) import Network.AMQP.Worker.Poll (poll) import Network.AMQP.Worker.Queue (Queue (..))@@ -28,7 +27,7 @@  -- | send a message to a queue. Enforces that the message type and queue name are correct at the type level ----- > let newUsers = key "users" & word "new" :: Key Route User+-- > let newUsers = key "users" & word "new":: Key Route User -- > publish conn newUsers (User "username") -- -- Publishing to a Binding Key results in an error@@ -45,7 +44,7 @@ publishToExchange :: (RequireRoute a, ToJSON msg, MonadIO m) => Connection -> Key a msg -> msg -> m () publishToExchange conn rk msg =     liftIO $ withChannel conn $ \chan -> do-        _ <- AMQP.publishMsg chan conn.exchange (keyText rk) (jsonMessage msg)+        _ <- AMQP.publishMsg chan (exchange conn) (keyText rk) (jsonMessage msg)         return ()   where     jsonMessage :: ToJSON a => a -> AMQP.Message@@ -57,7 +56,7 @@             , AMQP.msgDeliveryMode = Just Persistent             } --- | Wait until a message is read from the queue. Throws an exception if the incoming message doesn't match the key type+-- \| Wait until a message is read from the queue. Throws an exception if the message doesn't match the declared type -- -- > m <- Worker.takeMessage conn queue -- > print (value m)@@ -69,8 +68,7 @@         Error e -> liftIO $ throwIO e         Parsed msg -> pure msg --- | Create a worker which loops and handles messages. Throws an exception if the incoming message doesn't match the key type.--- It is recommended that you catch errors in your handler and allow message parsing errors to crash your program.+-- | Create a worker which loops and handles messages -- -- > Worker.worker conn queue $ \m -> do -- >   print (value m)

@@ -1,22 +1,26 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE OverloadedRecordDot #-} {-# 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 qualified Network.AMQP as AMQP -import Network.AMQP.Worker.Connection (Connection (..), withChannel)+import Network.AMQP.Worker.Connection (Connection, exchange, withChannel) import Network.AMQP.Worker.Key (Bind, Key (..), keyText, toBindKey)  type QueueName = Text -type QueuePrefix = Text+newtype QueuePrefix = QueuePrefix Text+    deriving (Show, Eq, IsString) +instance Default QueuePrefix where+    def = QueuePrefix "main"+ -- | A queue is an inbox for messages to be delivered data Queue msg     = Queue (Key Bind msg) QueueName@@ -31,8 +35,8 @@     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 will load balance if operating on the--- same queue, or on queues with the same 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 (toBindKey key) name@@ -45,7 +49,7 @@ -- > -- "main messages.new" -- > queueName "main" (key "messages" & word "new") queueName :: QueuePrefix -> Key a msg -> QueueName-queueName pre key = pre <> " " <> keyText key+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@@ -53,7 +57,7 @@ bindQueue conn (Queue key name) =     liftIO $ withChannel conn $ \chan -> do         let options = AMQP.newQueue{AMQP.queueName = name}-        let exg = AMQP.newExchange{exchangeName = conn.exchange, exchangeType = "topic"}+        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)