amqp-worker 1.0.0 → 2.0.0

raw patch · 9 files changed

+252/−277 lines, 9 filesdep −data-defaultPVP ok

version bump matches the API change (PVP)

Dependencies removed: data-default

API changes (from Hackage documentation)

Files

• README.md +36/−45
• amqp-worker.cabal +1/−5
• example/Example.hs +32/−31
• src/Network/AMQP/Worker.hs +28/−29
• src/Network/AMQP/Worker/Connection.hs +38/−20
• src/Network/AMQP/Worker/Key.hs +42/−27
• src/Network/AMQP/Worker/Message.hs +65/−39
• src/Network/AMQP/Worker/Queue.hs +10/−14
• src/Network/AMQP/Worker/Worker.hs +0/−67

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

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           amqp-worker-version:        1.0.0+version:        2.0.0 synopsis:       Type-safe AMQP workers description:    Please see the README on GitHub at <https://github.com/seanhess/amqp-worker#readme> category:       Network@@ -33,7 +33,6 @@       Network.AMQP.Worker.Message       Network.AMQP.Worker.Poll       Network.AMQP.Worker.Queue-      Network.AMQP.Worker.Worker   other-modules:       Paths_amqp_worker   hs-source-dirs:@@ -43,7 +42,6 @@     , amqp >=0.20 && <1     , base >=4.9 && <5     , bytestring ==0.11.*-    , data-default ==0.7.*     , exceptions ==0.10.*     , monad-loops ==0.4.*     , mtl ==2.2.*@@ -64,7 +62,6 @@     , amqp-worker     , base >=4.9 && <5     , bytestring ==0.11.*-    , data-default ==0.7.*     , exceptions ==0.10.*     , monad-loops ==0.4.*     , mtl ==2.2.*@@ -86,7 +83,6 @@     , amqp-worker     , base >=4.9 && <5     , bytestring ==0.11.*-    , data-default ==0.7.*     , exceptions ==0.10.*     , monad-loops ==0.4.*     , mtl ==2.2.*

@@ -4,7 +4,6 @@ module Main where  import Control.Concurrent (forkIO)-import Control.Monad.Catch (SomeException) import Data.Aeson (FromJSON, ToJSON) import Data.Function ((&)) import Data.Text (Text)@@ -20,10 +19,10 @@ instance FromJSON Greeting instance ToJSON Greeting -newGreetings :: Key Routing Greeting+newGreetings :: Key Route Greeting newGreetings = key "greetings" & word "new" -anyGreetings :: Key Binding Greeting+anyGreetings :: Key Bind Greeting anyGreetings = key "greetings" & any1  example :: IO ()@@ -39,16 +38,14 @@ simple :: Connection -> IO () simple conn = do     -- create a queue to receive them-    q <- Worker.queue conn def newGreetings+    q <- Worker.queue conn "main" newGreetings      -- publish a message (delivered to queue)     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 $ TestMessage "Compiler Error"--    -- Loop and print any values received-    Worker.worker conn def q onError (print . value)+    -- wait until we receive the message+    m <- Worker.takeMessage conn q+    print (value m)  -- | Multiple queues with distinct names will each get copies of published messages multiple :: Connection -> IO ()@@ -60,28 +57,34 @@     -- 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)+    -- Each of these queues will receive the same message+    m1 <- Worker.takeMessage conn one+    m2 <- Worker.takeMessage conn two -    putStrLn "Press any key to exit"-    _ <- getLine-    return ()+    print $ value m1+    print $ value m2 --- | Create multiple workers on the same queue to load balance between them-balance :: Connection -> IO ()-balance conn = do+-- | Create workers to continually process messages+workers :: Connection -> IO ()+workers conn = do     -- create a single queue-    q <- Worker.queue conn def newGreetings+    q <- Worker.queue conn "main" newGreetings -    -- publish two messages+    -- publish some messages     Worker.publish conn newGreetings $ Greeting "Hello1"     Worker.publish conn newGreetings $ Greeting "Hello2"+    Worker.publish conn newGreetings $ Greeting "Hello3" -    -- 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)+    -- 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)+     putStrLn "Press any key to exit"     _ <- getLine     return ()@@ -90,18 +93,16 @@ dynamic :: Connection -> IO () dynamic conn = do     -- anyGreetings matches `greetings.*`-    q <- Worker.queue conn def anyGreetings+    q <- Worker.queue conn "main" anyGreetings -    -- You can only publish to a Routing Key. Publishing to anyGreetings will give a compile error+    -- You can only publish to a specific Routing Key, like `greetings.new`     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)+    -- We cannot publish to anyGreetings because it is a Binding Key (with wildcards in it)+    -- Worker.publish conn anyGreetings $ Greeting "Compiler Error" -onError :: WorkerException SomeException -> IO ()-onError e = do-    putStrLn "Do something with errors"-    print e+    m <- Worker.takeMessage conn q+    print $ value m  test :: (Connection -> IO ()) -> IO () test action = do

@@ -9,25 +9,25 @@ -- Stability:   experimental -- Portability: portable ----- Type safe and simplified message queues with AMQP+-- Type safe and simplified message queues with AMQP. Compatible with RabbitMQ module Network.AMQP.Worker     ( -- * How to use this library       -- $use +      -- * Connecting+      connect+    , AMQP.fromURI+    , Connection+       -- * Binding and Routing Keys-      Key (..)-    , Binding-    , Routing-    , word+    , Key (..)+    , Bind+    , Route     , key+    , word     , any1     , many -      -- * Connecting-    , connect-    , AMQP.fromURI-    , Connection-       -- * Sending Messages     , publish @@ -40,25 +40,20 @@     , QueuePrefix (..)        -- * Messages+    , takeMessage     , ParseError (..)     , Message (..)        -- * Worker     , worker-    , WorkerException (..)-    , WorkerOptions (..)-    , Microseconds-    , Default.def     ) where -import qualified Data.Default as Default import qualified Network.AMQP as AMQP  import Network.AMQP.Worker.Connection import Network.AMQP.Worker.Key import Network.AMQP.Worker.Message import Network.AMQP.Worker.Queue-import Network.AMQP.Worker.Worker  -- $use --@@ -74,23 +69,27 @@ -- > instance ToJSON Greeting -- > -- > newGreetings :: Key Routing Greeting--- > newGreetings = key "messages" & word "greetings" & word "new"+-- > newGreetings = key "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"+-- > conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")+-- > Worker.publish conn newGreetings $ Greeting "hello" ----- To receive messages, first define a queue. You can bind direclty to the Routing Key to ensure it is delivered once+-- Create a queue to receive messages. 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)+-- > q <- Worker.queue conn "new" newMessages :: IO (Queue Greeting)+-- > m <- Worker.takeMessage conn q+-- > print (value m) ----- You can also define dynamic Routing Keys to receive many kinds of messages+-- 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)--- >+-- > let anyMessages = key "messages" & any1+-- > q <- Worker.queue conn "main" anyMessages+-- > m <- Worker.takeMessage conn q+-- > print (value m)+--+-- Create a worker to conintually process messages+--+-- > forkIO $ Worker.worker conn q $ \m -> do+-- >     print (value m)

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

@@ -8,8 +8,8 @@  module Network.AMQP.Worker.Key     ( Key (..)-    , Binding (..)-    , Routing+    , Bind (..)+    , Route     , key     , word     , any1@@ -17,8 +17,8 @@     , keyText     , fromBind     , toBind-    , toBindingKey-    , RequireRouting+    , toBindKey+    , RequireRoute     ) where  import Data.Kind (Constraint, Type)@@ -31,61 +31,76 @@ -- -- Routing keys have no dynamic component and can be used to publish messages ----- > commentsKey :: Key Routing Comment+-- > commentsKey :: Key Route Comment -- > commentsKey = key "posts" & word "new" -- -- Binding keys can contain wildcards, only used for matching messages ----- > commentsKey :: Key Binding Comment+-- > commentsKey :: Key Bind Comment -- > commentsKey = key "posts" & any1 & word "comments" & many-newtype Key a msg = Key [Binding]+newtype Key a msg = Key [Bind]     deriving (Eq, Show, Semigroup, Monoid) -data Routing+data Route -data Binding+data Bind     = Word Text     | Any     | Many     deriving (Eq, Show) -fromBind :: Binding -> Text+fromBind :: Bind -> Text fromBind (Word t) = t fromBind Any = "*" fromBind Many = "#" -toBind :: Text -> Binding+toBind :: Text -> Bind toBind = Word  keyText :: Key a msg -> Text keyText (Key ns) =     Text.intercalate "." . List.map fromBind $ ns --- | 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 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])+-- | 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] --- | Start a new routing key (can also be used for bindings)-key :: Text -> Key Routing msg-key t = Key [Word t]+-- | 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]) --- | 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+-- | 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]) +-- | 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+toBindKey (Key ws) = Key ws+ -- | Custom error message when trying to publish to Binding keys-type family RequireRouting (a :: Type) :: Constraint where-    RequireRouting Binding =+type family RequireRoute (a :: Type) :: Constraint where+    RequireRoute Bind =         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 = ()+    RequireRoute a = ()

@@ -1,22 +1,24 @@+{-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TypeFamilies #-}  module Network.AMQP.Worker.Message where +import Control.Exception (Exception, throwIO)+import Control.Monad (forever) 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, RequireRouting, Routing, keyText)+import Network.AMQP.Worker.Connection (Connection (..), withChannel)+import Network.AMQP.Worker.Key (Key, RequireRoute, Route, keyText) import Network.AMQP.Worker.Poll (poll) import Network.AMQP.Worker.Queue (Queue (..)) --- types --------------------------- -- | a parsed message from the queue data Message a = Message     { body :: ByteString@@ -24,42 +26,69 @@     }     deriving (Show, Eq) -data ConsumeResult a-    = Parsed (Message a)-    | Error ParseError--data ParseError = ParseError String ByteString--jsonMessage :: ToJSON a => a -> AMQP.Message-jsonMessage a =-    newMsg-        { AMQP.msgBody = Aeson.encode a-        , AMQP.msgContentType = Just "application/json"-        , AMQP.msgContentEncoding = Just "UTF-8"-        , AMQP.msgDeliveryMode = Just Persistent-        }+-- | 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+-- > publish conn newUsers (User "username")+--+-- Publishing to a Binding Key results in an error+--+-- > -- Compiler error! This doesn't make sense+-- > let users = key "users" & many :: Key Binding User+-- > publish conn users (User "username")+publish :: (RequireRoute a, ToJSON msg, MonadIO m) => Connection -> Key a msg -> msg -> m ()+publish = publishToExchange --- | 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+-- | publish a message to a routing key, without making sure a queue exists to handle it -- -- > publishToExchange conn key (User "username")-publishToExchange :: (RequireRouting a, ToJSON msg, MonadIO m) => Connection -> Key a msg -> msg -> m ()+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 (exchange conn) (keyText rk) (jsonMessage msg)+        _ <- AMQP.publishMsg chan conn.exchange (keyText rk) (jsonMessage msg)         return ()+  where+    jsonMessage :: ToJSON a => a -> AMQP.Message+    jsonMessage a =+        newMsg+            { AMQP.msgBody = Aeson.encode a+            , AMQP.msgContentType = Just "application/json"+            , AMQP.msgContentEncoding = Just "UTF-8"+            , AMQP.msgDeliveryMode = Just Persistent+            } --- | send a message to a queue. Enforces that the message type and queue name are correct at the type level+-- | Wait until a message is read from the queue. Throws an exception if the incoming message doesn't match the key type ----- > let key = key "users" :: Key Routing User--- > publish conn key (User "username")+-- > m <- Worker.takeMessage conn queue+-- > print (value m)+takeMessage :: (MonadIO m, FromJSON a) => Connection -> Queue a -> m (Message a)+takeMessage conn q = do+    let delay = 10000 :: Microseconds+    res <- consumeNext delay conn q+    case res of+        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. ----- Publishing to a Binding Key results in an error+-- > Worker.worker conn queue $ \m -> do+-- >   print (value m)+worker :: (FromJSON a, MonadIO m) => Connection -> Queue a -> (Message a -> m ()) -> m ()+worker conn queue action =+    forever $ do+        m <- takeMessage conn queue+        action m++-- | Block while checking for messages every N microseconds. Return once you find one. ----- > -- 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+-- > res <- consumeNext conn queue+-- > case res of+-- >   (Parsed m) -> print m+-- >   (Error e) -> putStrLn "could not parse message"+consumeNext :: (FromJSON msg, MonadIO m) => Microseconds -> Connection -> Queue msg -> m (ConsumeResult msg)+consumeNext pd conn key =+    poll pd $ consume conn key  -- | Check for a message once and attempt to parse it --@@ -86,14 +115,11 @@                 Right v ->                     return $ Just $ Parsed (Message bd v) --- | Block while checking for messages every N microseconds. Return once you find one.------ > res <- consumeNext conn queue--- > case res of--- >   (Parsed m) -> print m--- >   (Error e) -> putStrLn "could not parse message"-consumeNext :: (FromJSON msg, MonadIO m) => Microseconds -> Connection -> Queue msg -> m (ConsumeResult msg)-consumeNext pd conn key =-    poll pd $ consume conn key+data ConsumeResult a+    = Parsed (Message a)+    | Error ParseError++data ParseError = ParseError String ByteString+    deriving (Show, Exception)  type Microseconds = Int

@@ -1,29 +1,25 @@ {-# 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, exchange, withChannel)-import Network.AMQP.Worker.Key (Binding, Key (..), keyText, toBindingKey)+import Network.AMQP.Worker.Connection (Connection (..), withChannel)+import Network.AMQP.Worker.Key (Bind, Key (..), keyText, toBindKey)  type QueueName = Text -newtype QueuePrefix = QueuePrefix Text-    deriving (Show, Eq, IsString)--instance Default QueuePrefix where-    def = QueuePrefix "main"+type QueuePrefix = Text  -- | A queue is an inbox for messages to be delivered data Queue msg-    = Queue (Key Binding msg) QueueName+    = Queue (Key Bind msg) QueueName     deriving (Show, Eq)  -- | Create a queue to receive messages matching the 'Key' with a name prefixed via `queueName`.@@ -35,11 +31,11 @@     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+-- 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 queueNamed :: (MonadIO m) => Connection -> QueueName -> Key a msg -> m (Queue msg) queueNamed conn name key = do-    let q = Queue (toBindingKey key) name+    let q = Queue (toBindKey key) name     bindQueue conn q     return q @@ -49,7 +45,7 @@ -- > -- "main messages.new" -- > queueName "main" (key "messages" & word "new") queueName :: QueuePrefix -> Key a msg -> QueueName-queueName (QueuePrefix pre) key = pre <> " " <> keyText key+queueName 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@@ -57,7 +53,7 @@ bindQueue conn (Queue key name) =     liftIO $ withChannel conn $ \chan -> do         let options = AMQP.newQueue{AMQP.queueName = name}-        let exg = AMQP.newExchange{exchangeName = exchange conn, exchangeType = "topic"}+        let exg = AMQP.newExchange{exchangeName = conn.exchange, exchangeType = "topic"}         _ <- AMQP.declareExchange chan exg         _ <- AMQP.declareQueue chan options         _ <- AMQP.bindQueue chan name (exchange conn) (keyText key)

@@ -1,67 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}--module Network.AMQP.Worker.Worker where--import Control.Concurrent (threadDelay)-import Control.Exception (SomeException (..))-import Control.Monad (forever)-import Control.Monad.Catch (Exception (..), MonadCatch, catch)-import Control.Monad.IO.Class (MonadIO, liftIO)-import Data.Aeson (FromJSON)-import Data.ByteString.Lazy (ByteString)-import Data.Default (Default (..))--import Network.AMQP.Worker.Connection (Connection)-import Network.AMQP.Worker.Message (ConsumeResult (..), Message (..), Microseconds, ParseError (..), consumeNext)-import Network.AMQP.Worker.Queue (Queue (..))---- | Create a worker which loops, checks for messages, and handles errors------ > startWorker conn queue = do--- >   Worker.worker conn def queue onError onMessage--- >--- >   where--- >     onMessage :: Message User--- >     onMessage m = do--- >       putStrLn "handle user message"--- >       print (value m)--- >--- >     onError :: WorkerException SomeException -> IO ()--- >     onError e = do--- >       putStrLn "Do something with errors"-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-            Error (ParseError reason bd) ->-                onError (MessageParseError bd reason)-            Parsed msg ->-                catch-                    (action msg)-                    (onError . OtherException (body msg))-        liftIO $ threadDelay (loopDelay opts)---- | Options for worker-data WorkerOptions = WorkerOptions-    { pollDelay :: Microseconds-    -- ^ Delay between checks to consume. Defaults to 10ms-    , loopDelay :: Microseconds-    -- ^ Delay between calls to job. Defaults to 0-    }-    deriving (Show, Eq)--instance Default WorkerOptions where-    def =-        WorkerOptions-            { pollDelay = 10 * 1000-            , loopDelay = 0-            }---- | Exceptions created while processing-data WorkerException e-    = MessageParseError ByteString String-    | OtherException ByteString e-    deriving (Show, Eq)--instance (Exception e) => Exception (WorkerException e)