amqp-worker

[ bsd3, library, program, unclassified ] [ Propose Tags ]

Please see the README on GitHub at https://github.com/seanhess/server#readme


[Skip to Readme]
Versions [faq] 0.2.0, 0.2.1, 0.2.2, 0.2.3, 0.2.4, 0.2.5, 0.3.2
Dependencies aeson, amqp, amqp-worker, base (>=4.9 && <5), bytestring, data-default, exceptions, monad-control, monad-loops, mtl, resource-pool, split, text, transformers-base [details]
License BSD-3-Clause
Copyright Orbital Labs
Author Sean Hess
Maintainer seanhess@gmail.com
Home page https://github.com/seanhess/amqp-worker#readme
Bug tracker https://github.com/seanhess/amqp-worker/issues
Source repo head: git clone https://github.com/seanhess/amqp-worker
Uploaded by seanhess at Fri Apr 12 16:28:05 UTC 2019
Distributions NixOS:0.3.2
Executables example
Downloads 2139 total (112 in the last 30 days)
Rating (no votes yet) [estimated by rule of succession]
Your Rating
  • λ
  • λ
  • λ
Status Hackage Matrix CI
Docs available [build log]
Last success reported on 2019-04-12 [all 1 reports]

Modules

[Index] [Quick Jump]

Downloads

Maintainer's Corner

For package maintainers and hackage trustees


Readme for amqp-worker-0.3.2

[back to package description]

AMQP Worker

Type-safe AMQP workers. Compatible with RabbitMQ

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