amqp-conduit-0.2.0.0: src/Network/AMQP/Conduit.hs
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE TypeOperators #-}
-- |
-- Conduit bindings for AMQP (see amqp package) https://hackage.haskell.org/package/amqp
--
-- Create a AMQP connection, a channel, declare a queue and an exchange
-- and run the given action.
--
-- /Example/:
--
-- Connect to a server, declare a queue and an exchange and setup a callback for messages coming in on the queue. Then publish a single message to our new exchange
--
-- >{-# LANGUAGE OverloadedStrings #-}
-- >
-- >import Control.Concurrent (threadDelay)
-- >import Control.Monad.IO.Class (MonadIO, liftIO)
-- >import Control.Monad.Trans.Resource (runResourceT)
-- >import qualified Data.ByteString.Lazy.Char8 as BL
-- >import Data.Conduit
-- >import Network.AMQP
-- >import Network.AMQP.Conduit
-- >import Test.Hspec
-- >
-- >
-- >main :: IO ()
-- >main = hspec $ do
-- > describe "produce and consume test" $ do
-- > it "send a message and recieve the message" $ do
-- > runResourceT $ withChannel config $ \conn -> do
-- > sendMsg str $$ amqpSendSink conn "myExchange" "myKey"
-- > amqp <- createChannel config
-- > amqp' <- createConsumer amqp "myQueue" Ack $ \(msg,env) -> do
-- > amqpReceiveSource (msg,env) $$ printMsg
-- > -- | NOTE: RabbitMQ 1.7 doesn't implement this command.
-- > -- amqp'' <- pauseConsumers amqp'
-- > -- amqp''' <- resumeConsumers amqp''
-- > threadDelay $ 15 * 1000000
-- > _ <- deleteConsumer amqp'
-- > return ()
-- >
-- >str :: String
-- >str = "This is a test message"
-- >
-- >config :: AmqpConf
-- >config = AmqpConf "amqp://guest:guest@localhost:5672/" queue exchange "myKey"
-- > where
-- > exchange = newExchange {exchangeName = "myExchange", exchangeType = "direct"}
-- > queue = newQueue {queueName = "myQueue"}
-- >
-- >sendMsg :: (Monad m, MonadIO m) => String -> Source m Message
-- >sendMsg msg = do
-- > yield (newMsg {msgBody = (BL.pack msg),msgDeliveryMode = Just Persistent} )
-- >
-- >printMsg :: (Monad m, MonadIO m) => Sink (Message, Envelope) m ()
-- >printMsg = do
-- > m <- await
-- > case m of
-- > Nothing -> printMsg
-- > Just (msg,env) -> do
-- > liftIO $ ackEnv env
-- > liftIO $ (BL.unpack $ msgBody msg) `shouldBe` str
-- > liftIO $ putStrLn $ "received message: " ++ (BL.unpack $ msgBody msg)
-- > -
-- >
module Network.AMQP.Conduit (
-- * Conduit
amqpReceiveSource
, amqpSendSink
-- * Data type
, AmqpConf (..)
, AmqpConn (..)
, ExchangeKey
, Exchange
, QueueName
, AmqpURI
-- * Connection and Channel
, withChannel
, createConnectionChannel
, destoryConnection
-- * Exchange and Queue utils
, createQueue
, createExchange
, bindQueueExchange
-- * Consumer utils
, createConsumer
, deleteConsumer
, pauseConsumer
, resumeConsumer
) where
import Control.Exception (throwIO)
import Control.Exception.Lifted (bracket)
import Control.Monad.IO.Class (MonadIO, liftIO)
import Control.Monad.Trans.Control (MonadBaseControl)
import Data.Conduit (Sink, Source, await, yield)
import Data.Text (Text)
import Network.AMQP (AMQPException (ConnectionClosedException),
Ack, Channel, Connection,
ConsumerTag, Envelope,
ExchangeOpts, Message, QueueOpts,
addConnectionClosedHandler,
bindQueue, cancelConsumer,
closeConnection, consumeMsgs,
declareExchange, declareQueue,
exchangeName, flow, fromURI,
openChannel, openConnection'',
publishMsg, queueName)
-- | Amqp Connection and Channel
data AmqpConn = AmqpConn
{ amqpConn :: Connection
, amqpChan :: (Channel, Maybe ConsumerTag)
}
-- | Amqp connection configuration. queue name, exchange name, exchange key name, and amqp URI.
data AmqpConf = AmqpConf
{
-- | Connection string to the database.
amqpUri :: AmqpURI
, amqpQueue :: QueueOpts
, amqpExchange :: ExchangeOpts
, amqpExchanKey :: ExchangeKey
}
type ExchangeKey = Text
type Exchange = Text
type QueueName = Text
type AmqpURI = String
-- |Create a AMQP connection and a channel and run the given action. The connetion and channnel are properly released after the action finishes using it. Note that you should not use the given Connection, channel outside the action since it may be already been released.
withChannel:: (MonadIO m, MonadBaseControl IO m)
=> AmqpConf
-- ^ Connection config to the AMQP server.
-> (AmqpConn -> m a)
-- ^ Action to be executed that uses the connection.
-> m a
withChannel conf f = do
bracket
(do
-- liftIO $ putStrLn "connecting.."
liftIO $ connect (amqpUri conf))
(\conn -> do
-- liftIO $ putStrLn "disconnecting.."
liftIO $ disconnect conn)
(\conn -> do
-- liftIO $ putStrLn "calling function"
f conn)
-- | Create a connection and a channel. Note that it's your responsability to properly close the connection and the channels when unneeded. Use withAMQPChannels for an automatic resource control.
createConnectionChannel :: AmqpConf
-- ^ Connection config to the AMQP server.
-> IO AmqpConn
createConnectionChannel conf = connect $ amqpUri conf
-- | Close a connection
destoryConnection :: AmqpConn
-> IO ()
destoryConnection conn = do
addConnectionClosedHandler (amqpConn conn) False (return ())
closeConnection (amqpConn conn)
-- Consumer utils
createConsumer :: AmqpConn
-> QueueName
-> Ack
-> ((Message, Envelope) -> IO ())
-> IO AmqpConn
createConsumer conn queue ack f = do
tag' <- getTag chan
return $ conn {amqpChan =(chan, Just tag')}
where
getTag chan' = consumeMsgs chan' queue ack f
chan = fst $ amqpChan conn
deleteConsumer :: AmqpConn -> IO AmqpConn
deleteConsumer conn =
case tag of
Nothing -> return conn
Just s -> do
putStrLn "cancel cunsumer channel."
cancelConsumer chan s
return $ conn {amqpChan = (chan, Nothing)}
where
(chan, tag) = amqpChan conn
pauseConsumer :: AmqpConn
-> IO AmqpConn
pauseConsumer chan = flowConsumer chan False
resumeConsumer :: AmqpConn
-> IO AmqpConn
resumeConsumer chan = flowConsumer chan True
flowConsumer :: AmqpConn
-> Bool
-> IO AmqpConn
flowConsumer conn flag =
case tag of
Nothing -> return conn
Just _ -> do
flow chan flag
return conn
where
(chan, tag) = amqpChan conn
-- utils
--
createQueue :: AmqpConf -> AmqpConn -> IO (QueueName, Int, Int)
createQueue conf conn = declareQueue (fst $ amqpChan conn) (amqpQueue conf)
createExchange :: AmqpConf -> AmqpConn -> IO ()
createExchange conf conn = declareExchange (fst $ amqpChan conn) (amqpExchange conf)
bindQueueExchange :: AmqpConf -> AmqpConn -> IO ()
bindQueueExchange conf conn =
bindQueue (fst $ amqpChan conn) (queueName (amqpQueue conf)) (exchangeName (amqpExchange conf)) (amqpExchanKey conf)
-- internal
connect :: AmqpURI
-> IO AmqpConn
connect uri = do
-- make a connection and a channel of the connection.
conn <- openConnection'' $ fromURI uri
chan <- openChannel conn
-- set a excetion when closeing the connection
addConnectionClosedHandler conn True (throwIO (ConnectionClosedException "Connection Closed."))
return $ AmqpConn conn (chan, Nothing)
disconnect :: AmqpConn
-> IO ()
disconnect conn = do
closeConnection (amqpConn conn)
sendMsg :: AmqpConn
-> Exchange
-> ExchangeKey
-> Message
-> IO ()
sendMsg conn exchange key msg = do
publishMsg chan exchange key msg
where
chan = fst (amqpChan conn)
-- | Source as consuming data pushed.
amqpReceiveSource :: MonadIO m
=> (Message, Envelope)
-> Source m (Message, Envelope)
amqpReceiveSource (msg, env) = loop
where
loop = do
yield (msg, env)
loop
-- | Sink as sending data.
amqpSendSink :: MonadIO m
=> AmqpConn
-> Exchange
-> ExchangeKey
-> Sink Message m ()
amqpSendSink conn exchange key = loop
where
loop = await >>= maybe (return ()) (\v -> (liftIO $ sendMsg conn exchange key v) >> loop)