amqp-worker 0.3.2 → 0.4.0
raw patch · 12 files changed
+342/−384 lines, 12 filesdep −monad-controldep −serverdep −splitdep ~aesondep ~amqpdep ~bytestringPVP ok
version bump matches the API change (PVP)
Dependencies removed: monad-control, server, split
Dependency ranges changed: aeson, amqp, bytestring, data-default, exceptions, monad-loops, mtl, resource-pool, text, transformers-base
API changes (from Hackage documentation)
- Network.AMQP.Worker.Connection: withChannel :: MonadBaseControl IO m => Connection -> (Channel -> m b) -> m b
+ Network.AMQP.Worker.Connection: withChannel :: Connection -> (Channel -> IO b) -> IO b
Files
- CHANGELOG.md +4/−0
- amqp-worker.cabal +38/−45
- example/Example.hs +89/−0
- src/Network/AMQP/Worker.hs +62/−64
- src/Network/AMQP/Worker/Connection.hs +42/−44
- src/Network/AMQP/Worker/Key.hs +20/−38
- src/Network/AMQP/Worker/Message.hs +37/−44
- src/Network/AMQP/Worker/Poll.hs +4/−4
- src/Network/AMQP/Worker/Queue.hs +14/−22
- src/Network/AMQP/Worker/Worker.hs +31/−33
- test/Example.hs +0/−89
- test/Spec.hs +1/−1
+ CHANGELOG.md view
@@ -0,0 +1,4 @@+### Version 0.4.0++* removed MonadBaseControl following Data.Pool, other internal Data.Pool updates+* updated to GHC 9.2 compat
amqp-worker.cabal view
@@ -1,14 +1,14 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.31.1.+-- This file has been generated from package.yaml by hpack version 0.35.3. -- -- 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>+version: 0.4.0+synopsis: Type-safe AMQP workers+description: Please see the README on GitHub at <https://github.com/seanhess/amqp-worker#readme>+category: Network homepage: https://github.com/seanhess/amqp-worker#readme bug-reports: https://github.com/seanhess/amqp-worker/issues author: Sean Hess@@ -19,6 +19,7 @@ build-type: Simple extra-source-files: README.md+ CHANGELOG.md source-repository head type: git@@ -38,68 +39,60 @@ hs-source-dirs: src build-depends:- aeson- , amqp+ aeson >=2.0 && <2.3+ , amqp >=0.20 && <1 , base >=4.9 && <5- , bytestring- , data-default- , exceptions- , monad-control- , monad-loops- , mtl- , resource-pool- , split- , text- , transformers-base+ , bytestring ==0.11.*+ , data-default ==0.7.*+ , exceptions ==0.10.*+ , monad-loops ==0.4.*+ , mtl ==2.2.*+ , resource-pool >=0.3 && <0.5+ , text ==1.2.*+ , transformers-base ==0.4.* default-language: Haskell2010 executable example main-is: Example.hs other-modules:- Spec Paths_amqp_worker hs-source-dirs:- test+ example ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall build-depends:- aeson- , amqp+ aeson >=2.0 && <2.3+ , amqp >=0.20 && <1 , amqp-worker , base >=4.9 && <5- , bytestring- , data-default- , exceptions- , monad-control- , monad-loops- , mtl- , resource-pool- , split- , text- , transformers-base+ , bytestring ==0.11.*+ , data-default ==0.7.*+ , exceptions ==0.10.*+ , monad-loops ==0.4.*+ , mtl ==2.2.*+ , resource-pool >=0.3 && <0.5+ , text ==1.2.*+ , transformers-base ==0.4.* default-language: Haskell2010 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+ aeson >=2.0 && <2.3+ , amqp >=0.20 && <1+ , amqp-worker , base >=4.9 && <5- , bytestring- , data-default- , exceptions- , monad-control- , monad-loops- , mtl- , resource-pool- , server- , split- , text- , transformers-base+ , bytestring ==0.11.*+ , data-default ==0.7.*+ , exceptions ==0.10.*+ , monad-loops ==0.4.*+ , mtl ==2.2.*+ , resource-pool >=0.3 && <0.5+ , text ==1.2.*+ , transformers-base ==0.4.* default-language: Haskell2010
+ example/Example.hs view
@@ -0,0 +1,89 @@+{-# 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
src/Network/AMQP/Worker.hs view
@@ -1,5 +1,5 @@-{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-} -- | --@@ -12,7 +12,7 @@ -- > {-# LANGUAGE DeriveGeneric #-} -- > {-# LANGUAGE OverloadedStrings #-} -- > module Main where--- > +-- > -- > import Control.Concurrent (forkIO) -- > import Control.Monad.Catch (SomeException) -- > import Data.Aeson (FromJSON, ToJSON)@@ -25,120 +25,118 @@ -- > 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--- > --- > -+-- >+-- > module Network.AMQP.Worker- (-- -- * Routing Keys- Key(..)- , Routing- , Binding- , word, key, star, hash-- -- * Connecting- , Connection- , connect- , disconnect- , exchange- , AMQP.fromURI-- -- * Initializing queues- , Queue(..)- , direct- , topic- , bindQueue+ ( -- * Routing Keys+ Key (..)+ , Routing+ , Binding+ , word+ , key+ , star+ , hash - -- * Sending Messages- , publish- , publishToExchange+ -- * Connecting+ , Connection+ , connect+ , disconnect+ , exchange+ , AMQP.fromURI - -- * Reading Messages- , consume- , consumeNext- , ConsumeResult(..)- , ParseError(..)- , Message(..)+ -- * Initializing queues+ , Queue (..)+ , direct+ , topic+ , bindQueue - -- * Worker- , worker- , WorkerException(..)- , WorkerOptions(..)- , Microseconds- , Default.def+ -- * Sending Messages+ , publish+ , publishToExchange + -- * Reading Messages+ , consume+ , consumeNext+ , ConsumeResult (..)+ , ParseError (..)+ , Message (..) - ) where+ -- * Worker+ , worker+ , WorkerException (..)+ , WorkerOptions (..)+ , Microseconds+ , Default.def+ ) where import qualified Data.Default as Default import qualified Network.AMQP as AMQP -import Network.AMQP.Worker.Key import Network.AMQP.Worker.Connection-import Network.AMQP.Worker.Queue+import Network.AMQP.Worker.Key import Network.AMQP.Worker.Message+import Network.AMQP.Worker.Queue import Network.AMQP.Worker.Worker
src/Network/AMQP/Worker/Connection.hs view
@@ -1,39 +1,42 @@-{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE OverloadedStrings #-}+ module Network.AMQP.Worker.Connection- ( Connection(..)- , connect- , disconnect- , withChannel- ) where+ ( Connection (..)+ , connect+ , disconnect+ , withChannel+ ) where -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+import Control.Concurrent.MVar+ ( MVar+ , newEmptyMVar+ , putMVar+ , readMVar+ , takeMVar+ )+import Control.Monad.Catch (catch, throwM)+import Control.Monad.IO.Class (MonadIO, liftIO)+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 type ExchangeName = Text - -- | Internal connection details data Connection = Connection- { amqpConn :: MVar AMQP.Connection- , pool :: Pool Channel- , exchange :: ExchangeName- }+ { amqpConn :: MVar AMQP.Connection+ , pool :: Pool Channel+ , exchange :: ExchangeName+ } -- | 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 let exchangeName = "amq.topic" @@ -41,41 +44,38 @@ cvar <- newEmptyMVar openConnection cvar - -- open a shared pool for channels- chans <- Pool.createPool (create cvar) destroy numStripes openTime numChans+ let config = Pool.defaultPoolConfig (create cvar) destroy openTime numChans + -- open a shared pool for channels+ chans <- Pool.newPool config pure $ Connection cvar chans exchangeName where- numStripes = 1 openTime = 10 numChans = 4 openConnection cvar = do- -- open a connection and store in the mvar- conn <- AMQP.openConnection'' opts- putMVar cvar conn+ -- open a connection and store in the mvar+ conn <- AMQP.openConnection'' opts+ putMVar cvar conn reopenConnection cvar = do- -- clear the mvar and reopen- _ <- takeMVar cvar- openConnection cvar-+ -- clear the mvar and reopen+ _ <- takeMVar cvar+ openConnection cvar create cvar = do- conn <- readMVar cvar- chan <- catch (AMQP.openChannel conn) (createEx cvar)- return chan+ conn <- readMVar cvar+ chan <- catch (AMQP.openChannel conn) (createEx cvar)+ return chan createEx cvar (ConnectionClosedException _ _) = do- reopenConnection cvar- create cvar-+ reopenConnection cvar+ create cvar createEx _ ex = throwM ex destroy chan = do- AMQP.closeChannel chan-+ AMQP.closeChannel chan disconnect :: MonadIO m => Connection -> m () disconnect c = liftIO $ do@@ -83,9 +83,7 @@ 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 -> (Channel -> IO b) -> IO b withChannel (Connection _ p _) action = do Pool.withResource p action
src/Network/AMQP/Worker/Key.hs view
@@ -1,29 +1,28 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}+ module Network.AMQP.Worker.Key- ( Key(..)- , Binding(..)- , BindingWord- , Routing- , key- , word, star, hash- , keyText- , KeySegment(..)- , bindingKey- ) where+ ( Key (..)+ , Binding (..)+ , BindingWord+ , Routing+ , key+ , word+ , star+ , hash+ , keyText+ , KeySegment (..)+ , bindingKey+ ) where +import qualified Data.List as List import Data.Text (Text) import qualified Data.Text as Text-import qualified Data.List as List --- -- | Keys describe routing and binding info for a message newtype Key a msg = Key [a]- deriving (Eq, Show, Semigroup, Monoid)-+ deriving (Eq, Show, Semigroup, Monoid) -- | Every message is sent with a specific routing key --@@ -33,11 +32,9 @@ deriving (Eq, Show) instance KeySegment Routing where- toText (Routing s) = s- fromText = Routing- toBind (Routing s) = Word s--+ toText (Routing s) = s+ fromText = Routing+ toBind (Routing s) = Word s -- | A dynamic binding address for topic queues --@@ -49,7 +46,6 @@ | Hash deriving (Eq, Show) - instance KeySegment Binding where toText (Word t) = t toText Star = "*"@@ -57,30 +53,19 @@ fromText = Word toBind = id -- class KeySegment a where- toText :: a -> Text+ toText :: a -> Text fromText :: Text -> a toBind :: a -> Binding ------ 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])@@ -96,8 +81,5 @@ -- | Create a new key key :: Text -> Key Routing msg key t = Key [Routing t]--- type BindingWord = Text
src/Network/AMQP/Worker/Message.hs view
@@ -1,57 +1,55 @@-{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+ module Network.AMQP.Worker.Message where -import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.Base (liftBase)-import Data.Aeson (ToJSON, FromJSON)+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 (newMsg, DeliveryMode(..), Ack(..))+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.Poll (poll)-import Network.AMQP.Worker.Connection (Connection, withChannel, exchange)-import Network.AMQP.Worker.Queue (Queue(..))+import Network.AMQP.Worker.Queue (Queue (..)) -- types -------------------------- -- | a parsed message from the queue data Message a = Message- { body :: ByteString- , value :: a- } deriving (Show, Eq)-+ { body :: ByteString+ , value :: a+ }+ deriving (Show, Eq) data ConsumeResult a- = Parsed (Message a)- | Error ParseError+ = Parsed (Message a)+ | Error ParseError data ParseError = ParseError String ByteString type Microseconds = Int - 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- }--+jsonMessage a =+ newMsg+ { AMQP.msgBody = Aeson.encode a+ , AMQP.msgContentType = Just "application/json"+ , AMQP.msgContentEncoding = Just "UTF-8"+ , AMQP.msgDeliveryMode = Just Persistent+ } -- | 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 conn rk msg =- liftIO $ withChannel conn $ \chan -> do- _ <- AMQP.publishMsg chan (exchange conn) (keyText rk) (jsonMessage msg)- return ()-+ liftIO $ withChannel conn $ \chan -> do+ _ <- AMQP.publishMsg chan (exchange conn) (keyText rk) (jsonMessage msg)+ return () -- | send a message to a queue. Enforces that the message type and queue name are correct at the type level --@@ -59,7 +57,6 @@ publish :: (ToJSON a, MonadIO m) => Connection -> Key Routing a -> a -> m () publish = publishToExchange - -- | Check for a message once and attempt to parse it -- -- > res <- consume conn queue@@ -69,25 +66,21 @@ -- > 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- liftIO $ AMQP.ackEnv env- let bd = AMQP.msgBody msg- case Aeson.eitherDecode bd of- Left err ->- return $ Just $ Error (ParseError err bd)-- Right v ->- return $ Just $ Parsed (Message bd v)-+ 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+ liftIO $ AMQP.ackEnv env+ let bd = AMQP.msgBody msg+ case Aeson.eitherDecode bd of+ Left err ->+ return $ Just $ Error (ParseError err bd)+ Right v ->+ return $ Just $ Parsed (Message bd v) -- | Block while checking for messages every N microseconds. Return once you find one. --
src/Network/AMQP/Worker/Poll.hs view
@@ -11,7 +11,7 @@ poll us action = untilJust $ do ma <- action case ma of- Just a -> return $ Just a- Nothing -> do- liftIO $ threadDelay us- return Nothing+ Just a -> return $ Just a+ Nothing -> do+ liftIO $ threadDelay us+ return Nothing
src/Network/AMQP/Worker/Queue.hs view
@@ -1,27 +1,22 @@ {-# LANGUAGE OverloadedStrings #-}+ module Network.AMQP.Worker.Queue where import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Text (Text)+import Network.AMQP (ExchangeOpts (..), QueueOpts (..)) import qualified Network.AMQP as AMQP-import Network.AMQP (QueueOpts(..), ExchangeOpts(..)) -import Network.AMQP.Worker.Key (Key(..), Routing, Binding, keyText, bindingKey, KeySegment)-import Network.AMQP.Worker.Connection (Connection, withChannel, exchange)----+import Network.AMQP.Worker.Connection (Connection, exchange, withChannel)+import Network.AMQP.Worker.Key (Binding, Key (..), KeySegment, Routing, bindingKey, keyText) -- | 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) - -- | Declare a topic queue, which will receive messages that match using wildcards -- -- > anyUsers :: Queue User@@ -29,14 +24,11 @@ topic :: KeySegment a => Key a msg -> QueueName -> Queue msg topic key name = Queue (bindingKey key) name - type QueueName = Text -data Queue msg =- Queue (Key Binding msg) QueueName- deriving (Show, Eq)--+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. --@@ -45,10 +37,10 @@ -- > 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 name (exchange conn) (keyText key)- return ()+ 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 name (exchange conn) (keyText key)+ return ()
src/Network/AMQP/Worker/Worker.hs view
@@ -1,20 +1,19 @@ {-# LANGUAGE FlexibleContexts #-}+ module Network.AMQP.Worker.Worker where import Control.Concurrent (threadDelay)-import Control.Exception (SomeException(..))-import Control.Monad.IO.Class (liftIO, MonadIO)-import Control.Monad.Catch (Exception(..), catch, MonadCatch)+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 Data.Default (Default (..)) import Network.AMQP.Worker.Connection (Connection)-import Network.AMQP.Worker.Queue (Queue(..))-import Network.AMQP.Worker.Message (Message(..), ConsumeResult(..), ParseError(..), Microseconds, consumeNext)--+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 --@@ -30,40 +29,39 @@ -- > 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)---+ 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)+ { 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- }+ def =+ WorkerOptions+ { pollDelay = 10 * 1000+ , loopDelay = 0+ } -- | Exceptions created while processing data WorkerException e- = MessageParseError ByteString String- | OtherException ByteString e- deriving (Show, Eq)+ = MessageParseError ByteString String+ | OtherException ByteString e+ deriving (Show, Eq) instance (Exception e) => Exception (WorkerException e)
− test/Example.hs
@@ -1,89 +0,0 @@-{-# 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
test/Spec.hs view
@@ -1,4 +1,4 @@-module Spec where+module Main where -- import Test.Tasty -- import Test.Tasty.HUnit