Network.AMQP.Worker

Type safe and simplified message queues with AMQP

How to use this library

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)

Binding and Routing Keys

newtype Key a msg

Messages are published with a specific identifier called a Routing key. Queues can use Binding Keys to control which messages are delivered to them.

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" & any1 & word "comments" & many

Constructors

Key [Binding]

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 (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 (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 Binding

Instances

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

data Routing

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

A specific word. Can be used to chain Routing keys or Binding keys

key :: Text -> Key Routing msg

Start a new routing key (can also be used for bindings)

any1 :: Key a msg -> Key Binding msg

Match any one word. Equivalent to *. Converts to a Binding key and can no longer be used to publish messaages

many :: Key a msg -> Key Binding msg

Match zero or more words. Equivalient to #. Converts to a Binding key and can no longer be used to publish messages

Connecting

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

Connect to the AMQP server.

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

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@

data Connection

Sending Messages

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

send a message to a queue. Enforces that the message type and queue name are correct at the type level

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

Initializing queues

queue :: MonadIO m => Connection -> QueuePrefix -> Key a msg -> m (Queue msg)

Create a queue to receive messages matching the Key with a name prefixed via queueName.

q <- Worker.queue conn "main" $ key "messages" & any1
Worker.worker conn def q onError onMessage

queueNamed :: MonadIO m => Connection -> QueueName -> Key a msg -> m (Queue msg)

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

data Queue msg

A queue is an inbox for messages to be delivered

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

queueName :: QueuePrefix -> Key a msg -> QueueName

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

type QueueName = Text

newtype QueuePrefix

Constructors

QueuePrefix Text

Instances

IsString QueuePrefix

Defined in Network.AMQP.Worker.Queue

Methods

fromString :: String -> QueuePrefix

Show QueuePrefix

Defined in Network.AMQP.Worker.Queue

Methods

showsPrec :: Int -> QueuePrefix -> ShowS

show :: QueuePrefix -> String

showList :: [QueuePrefix] -> ShowS

Default QueuePrefix

Defined in Network.AMQP.Worker.Queue

Methods

def :: QueuePrefix

Eq QueuePrefix

Defined in Network.AMQP.Worker.Queue

Methods

(==) :: QueuePrefix -> QueuePrefix -> Bool

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

Messages

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.