Network.AMQP.Worker

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.

{-# 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

Routing Keys

newtype Key a msg

Keys describe routing and binding info for a message

Constructors

Key [a]

Instances

Monoid (Key a msg)

Defined in Network.AMQP.Worker.Key

Methods

mempty :: Key a msg

mappend :: Key a msg -> Key a msg -> Key a msg

mconcat :: [Key a msg] -> Key a msg

Semigroup (Key a msg)

Defined in Network.AMQP.Worker.Key

Methods

(<>) :: Key a msg -> Key a msg -> Key a msg

sconcat :: NonEmpty (Key a msg) -> Key a msg

stimes :: Integral b => b -> Key a msg -> Key a msg

Show a => Show (Key a msg)

Defined in Network.AMQP.Worker.Key

Methods

showsPrec :: Int -> Key a msg -> ShowS

show :: Key a msg -> String

showList :: [Key a msg] -> ShowS

Eq a => Eq (Key a msg)

Defined in Network.AMQP.Worker.Key

Methods

(==) :: Key a msg -> Key a msg -> Bool

(/=) :: Key a msg -> Key a msg -> Bool

data Routing

Every message is sent with a specific routing key

newCommentKey :: Key Routing Comment
newCommentKey = key "posts" & word "1" & word "comments" & word "new"

Instances

KeySegment Routing

Defined in Network.AMQP.Worker.Key

Methods

toText :: Routing -> Text

fromText :: Text -> Routing

toBind :: Routing -> Binding

Show Routing

Defined in Network.AMQP.Worker.Key

Methods

showsPrec :: Int -> Routing -> ShowS

show :: Routing -> String

showList :: [Routing] -> ShowS

Eq Routing

Defined in Network.AMQP.Worker.Key

Methods

(==) :: Routing -> Routing -> Bool

(/=) :: Routing -> Routing -> Bool

data Binding

A dynamic binding address for topic queues

commentsKey :: Key Binding Comment
commentsKey = key "posts" & star & word "comments" & hash

Instances

KeySegment Binding

Defined in Network.AMQP.Worker.Key

Methods

toText :: Binding -> Text

fromText :: Text -> Binding

toBind :: Binding -> Binding

Show Binding

Defined in Network.AMQP.Worker.Key

Methods

showsPrec :: Int -> Binding -> ShowS

show :: Binding -> String

showList :: [Binding] -> ShowS

Eq Binding

Defined in Network.AMQP.Worker.Key

Methods

(==) :: Binding -> Binding -> Bool

(/=) :: Binding -> Binding -> Bool

word :: KeySegment a => Text -> Key a msg -> Key a msg

Match a specific word

key :: Text -> Key Routing msg

Create a new key

star :: KeySegment a => Key a msg -> Key Binding msg

Match any one word

hash :: KeySegment a => Key a msg -> Key Binding msg

Match any words

Connecting

data Connection

Internal connection details

connect :: MonadIO m => ConnectionOpts -> m Connection

Connect to the AMQP server.

>>> conn <- connect (fromURI "amqp://guest:guest@localhost:5672")

disconnect :: MonadIO m => Connection -> m ()

exchange :: Connection -> ExchangeName

fromURI :: String -> ConnectionOpts

Parses amqp standard URI of the form amqp://user:passwordhost:port/vhost and returns a ConnectionOpts for use with openConnection'' | Any of these fields may be empty and will be replaced with defaults from amqp:/guest:guest@localhost:5672@

Initializing queues

data Queue msg

Constructors

Queue (Key Binding msg) QueueName

Instances

Show (Queue msg)

Defined in Network.AMQP.Worker.Queue

Methods

showsPrec :: Int -> Queue msg -> ShowS

show :: Queue msg -> String

showList :: [Queue msg] -> ShowS

Eq (Queue msg)

Defined in Network.AMQP.Worker.Queue

Methods

(==) :: Queue msg -> Queue msg -> Bool

(/=) :: Queue msg -> Queue msg -> Bool

direct :: Key Routing msg -> Queue msg

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")

topic :: KeySegment a => Key a msg -> QueueName -> Queue msg

Declare a topic queue, which will receive messages that match using wildcards

anyUsers :: Queue User
anyUsers = Worker.topic "anyUsers" (key "users" & star)

bindQueue :: MonadIO m => Connection -> Queue msg -> m ()

Queues must be bound before you publish messages to them, or the messages will not be saved.

let queue = Worker.direct (key "users" & word "new") :: Queue User
conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")
Worker.bindQueue conn queue

Sending Messages

publish :: (ToJSON a, MonadIO m) => Connection -> Key Routing a -> a -> m ()

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")

publishToExchange :: (ToJSON a, MonadIO m) => Connection -> Key Routing a -> a -> m ()

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")

Reading Messages

consume :: (FromJSON msg, MonadIO m) => Connection -> Queue msg -> m (Maybe (ConsumeResult msg))

Check for a message once and attempt to parse it

res <- consume conn queue
case res of
  Just (Parsed m) -> print m
  Just (Error e) -> putStrLn "could not parse message"
  Nothing -> putStrLn "No messages on the queue"

consumeNext :: (FromJSON msg, MonadIO m) => Microseconds -> Connection -> Queue msg -> m (ConsumeResult msg)

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"

data ConsumeResult a

Constructors

Parsed (Message a)

Error ParseError

data ParseError

Constructors

ParseError String ByteString

data Message a

a parsed message from the queue

Constructors

Message

Fields

• body :: ByteString
• value :: a

Instances

Show a => Show (Message a)

Defined in Network.AMQP.Worker.Message

Methods

showsPrec :: Int -> Message a -> ShowS

show :: Message a -> String

showList :: [Message a] -> ShowS

Eq a => Eq (Message a)

Defined in Network.AMQP.Worker.Message

Methods

(==) :: Message a -> Message a -> Bool

(/=) :: Message a -> Message a -> Bool

Worker

worker :: (FromJSON a, MonadIO m, MonadCatch m) => Connection -> WorkerOptions -> Queue a -> (WorkerException SomeException -> m ()) -> (Message a -> m ()) -> m ()

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"

data WorkerException e

Exceptions created while processing

Constructors

MessageParseError ByteString String

OtherException ByteString e

Instances

Exception e => Exception (WorkerException e)

Defined in Network.AMQP.Worker.Worker

Methods

toException :: WorkerException e -> SomeException

fromException :: SomeException -> Maybe (WorkerException e)

displayException :: WorkerException e -> String

Show e => Show (WorkerException e)

Defined in Network.AMQP.Worker.Worker

Methods

showsPrec :: Int -> WorkerException e -> ShowS

show :: WorkerException e -> String

showList :: [WorkerException e] -> ShowS

Eq e => Eq (WorkerException e)

Defined in Network.AMQP.Worker.Worker

Methods

(==) :: WorkerException e -> WorkerException e -> Bool

(/=) :: WorkerException e -> WorkerException e -> Bool

data WorkerOptions

Options for worker

Constructors

WorkerOptions

Fields

• pollDelay :: Microseconds

  Delay between checks to consume. Defaults to 10ms

• loopDelay :: Microseconds

  Delay between calls to job. Defaults to 0

Instances

Show WorkerOptions

Defined in Network.AMQP.Worker.Worker

Methods

showsPrec :: Int -> WorkerOptions -> ShowS

show :: WorkerOptions -> String

showList :: [WorkerOptions] -> ShowS

Default WorkerOptions

Defined in Network.AMQP.Worker.Worker

Methods

def :: WorkerOptions

Eq WorkerOptions

Defined in Network.AMQP.Worker.Worker

Methods

(==) :: WorkerOptions -> WorkerOptions -> Bool

(/=) :: WorkerOptions -> WorkerOptions -> Bool

type Microseconds = Int

def :: Default a => a

The default value for this type.