amqp-conduit 0.1.0.1 → 0.2.0.0
raw patch · 3 files changed
+113/−209 lines, 3 filesdep ~base
Dependency ranges changed: base
Files
- amqp-conduit.cabal +5/−5
- src/Network/AMQP/Conduit.hs +102/−198
- test/test.hs +6/−6
amqp-conduit.cabal view
@@ -1,5 +1,5 @@ name: amqp-conduit-version: 0.1.0.1+version: 0.2.0.0 synopsis: Conduit bindings for AMQP (see amqp package) description: Conduit bindings for AMQP (see amqp package) Stability: alpha@@ -33,10 +33,10 @@ -- other-extensions: -- Other library packages from which modules are imported.- build-depends: base >=4.7 && <4.8+ build-depends: base >=4.7 && <4.9 , conduit >= 1.2 , amqp >= 0.10- , resourcet + , resourcet , transformers , lifted-base , exceptions@@ -47,7 +47,7 @@ -- Directories containing source files. hs-source-dirs: src - ghc-options: -Wall -fwarn-incomplete-patterns -O2+ ghc-options: -Wall -fwarn-incomplete-patterns test-suite test-amqp-conduit@@ -63,5 +63,5 @@ type: exitcode-stdio-1.0 main-is: test.hs hs-source-dirs: test- ghc-options: -Wall+ ghc-options: -Wall -fwarn-incomplete-patterns
src/Network/AMQP/Conduit.hs view
@@ -1,5 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TypeOperators #-} -- | -- Conduit bindings for AMQP (see amqp package) https://hackage.haskell.org/package/amqp@@ -22,27 +23,28 @@ -- >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 $ withAMQPChannel config $ \conn -> do--- > sendMsg str $$ amqpSendSink conn "myKey"--- > amqp <- createAMQPChannels config 10--- > amqp' <- createConsumers amqp $ \(msg,env) -> 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--- > _ <- deleteConsumers amqp'+-- > _ <- deleteConsumer amqp' -- > return () -- > -- >str :: String -- >str = "This is a test message" -- > -- >config :: AmqpConf--- >config = AmqpConf "amqp://guest:guest@192.168.59.103:5672/" queue exchange "myKey"+-- >config = AmqpConf "amqp://guest:guest@localhost:5672/" queue exchange "myKey" -- > where -- > exchange = newExchange {exchangeName = "myExchange", exchangeType = "direct"} -- > queue = newQueue {queueName = "myQueue"}@@ -72,53 +74,47 @@ , AmqpConf (..) , AmqpConn (..) , ExchangeKey+ , Exchange+ , QueueName , AmqpURI -- * Connection and Channel- , withAMQPChannel- , withAMQPChannels- , createAMQPChannels- , closeChannels+ , withChannel+ , createConnectionChannel , destoryConnection+ -- * Exchange and Queue utils+ , createQueue+ , createExchange+ , bindQueueExchange -- * Consumer utils- , createConsumers- , createConsumers' , createConsumer- , deleteConsumers , deleteConsumer- , pauseConsumers , pauseConsumer- , resumeConsumers , resumeConsumer-- , module Network.AMQP ) where++import Control.Exception (throwIO) import Control.Exception.Lifted (bracket)-import Control.Monad (replicateM) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.Trans.Control (MonadBaseControl) import Data.Conduit (Sink, Source, await, yield)-import Data.Maybe (isJust, isNothing) import Data.Text (Text) import Network.AMQP (AMQPException (ConnectionClosedException),- Ack (Ack, NoAck), Channel,- Connection, ConsumerTag, Envelope,+ Ack, Channel, Connection,+ ConsumerTag, Envelope, ExchangeOpts, Message, QueueOpts, addConnectionClosedHandler, bindQueue, cancelConsumer,- closeChannel, closeConnection,+ closeConnection, consumeMsgs, declareExchange, declareQueue, exchangeName, flow, fromURI,- getMsg, openChannel,- openConnection'', publishMsg,- queueName)-import Network.AMQP.Lifted (consumeMsgs)+ openChannel, openConnection'',+ publishMsg, queueName) -- | Amqp Connection and Channel data AmqpConn = AmqpConn { amqpConn :: Connection- , amqpChan :: [(Channel, Maybe ConsumerTag)]- , amqpConf :: AmqpConf- }+ , amqpChan :: (Channel, Maybe ConsumerTag)+ } -- | Amqp connection configuration. queue name, exchange name, exchange key name, and amqp URI. data AmqpConf = AmqpConf@@ -131,219 +127,130 @@ } type ExchangeKey = Text+type Exchange = Text+type QueueName = Text type AmqpURI = String --- |Create a AMQP connection and Channel(s) and run the given action. The connetion and channnels are properly released after the action finishes using it. Note that you should not use the given Connection, channels outside the action since it may be already been released.-withAMQPChannels :: (MonadIO m, MonadBaseControl IO m)+-- |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.- -> Int- -- ^ number of channels to be kept open in the connection.+ -- ^ Connection config to the AMQP server. -> (AmqpConn -> m a) -- ^ Action to be executed that uses the connection. -> m a-withAMQPChannels conf num f =+withChannel conf f = do bracket (do -- liftIO $ putStrLn "connecting.."- liftIO $ connect conf (numCheck num))+ liftIO $ connect (amqpUri conf)) (\conn -> do -- liftIO $ putStrLn "disconnecting.."- liftIO $liftIO $ disconnect conn)+ liftIO $ disconnect conn) (\conn -> do -- liftIO $ putStrLn "calling function" f conn) --- | Same as runAMQP, but only one channel is opened.-withAMQPChannel:: (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-withAMQPChannel conf = withAMQPChannels conf 1 --- | Create a connection and channels. Note that it's your responsability to properly close the connection and the channels when unneeded. Use withAMQPChannels for an automatic resource control.-createAMQPChannels :: (MonadIO m, MonadBaseControl IO m)- => AmqpConf+-- | 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.- -> Int- -- ^ number of channels to be kept open in the connection- -> m AmqpConn-createAMQPChannels conf num = liftIO $ connect conf (numCheck num)+ -> 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 :: (MonadIO m, MonadBaseControl IO m)- => (Channel, Maybe ConsumerTag)- -> Text+createConsumer :: AmqpConn+ -> QueueName -> Ack- -> ((Message, Envelope) -> m ())- -> m (Channel, Maybe ConsumerTag)-createConsumer (chan, tag) queue ack f =- case tag of- Nothing -> do- tag' <- consumeMsgs chan queue ack f- return (chan, Just tag')- Just _ -> return (chan, tag)+ -> ((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 :: MonadIO m- => (Channel, Maybe ConsumerTag)- -> m (Channel, Maybe ConsumerTag)-deleteConsumer (chan, tag) =+deleteConsumer :: AmqpConn -> IO AmqpConn+deleteConsumer conn = case tag of- Nothing -> return (chan, tag)+ Nothing -> return conn Just s -> do- liftIO $ putStrLn "cancel cunsumer channel."- liftIO $ cancelConsumer chan s- return (chan, Nothing)--createConsumersHelper :: (MonadIO m, MonadBaseControl IO m)- => AmqpConn- -> ((Message, Envelope) -> m ())- -> Ack- -> m AmqpConn-createConsumersHelper conn f ack = do- chan <- mapM (\chan -> createConsumer chan qName ack f) chanList- return $ conn {amqpChan = chan}- where- qName = queueName (amqpQueue (amqpConf conn))- chanList = filter (\(_, tag) -> isNothing tag) (amqpChan conn)---- | after processing for any message tha you get, automatically acknowledged (see "NoAck" in the "Network.AMQP" module)-createConsumers' :: (MonadIO m, MonadBaseControl IO m)- => AmqpConn- -> ((Message, Envelope) -> m ())- -> m AmqpConn-createConsumers' conn f = createConsumersHelper conn f NoAck---- | You have to call ackMsg or ackEnv after processing for any message that you get, otherwise it might be delivered again (see "ackMsg" and "ackEnv" in the "Network.AMQP" module)-createConsumers :: (MonadIO m, MonadBaseControl IO m)- => AmqpConn- -> ((Message, Envelope) -> m ())- -> m AmqpConn-createConsumers conn f = createConsumersHelper conn f Ack---- helper functions-deleteConsumers :: AmqpConn- -> IO AmqpConn-deleteConsumers conn = do- chan <- mapM deleteConsumer chanList- return $ conn {amqpChan = chan}+ putStrLn "cancel cunsumer channel."+ cancelConsumer chan s+ return $ conn {amqpChan = (chan, Nothing)} where- chanList = filter (\(_, tag) -> isJust tag) (amqpChan conn)+ (chan, tag) = amqpChan conn -pauseConsumers :: AmqpConn+pauseConsumer :: AmqpConn -> IO AmqpConn-pauseConsumers conn = do- chan <- mapM pauseConsumer chanList- return $ conn {amqpChan = chan}- where- chanList = filter (\(_, tag) -> isJust tag) (amqpChan conn)--pauseConsumer :: (Channel, Maybe ConsumerTag )- -> IO (Channel, Maybe ConsumerTag) pauseConsumer chan = flowConsumer chan False -resumeConsumers :: AmqpConn+resumeConsumer :: AmqpConn -> IO AmqpConn-resumeConsumers conn = do- chan <- mapM resumeConsumer chanList- return $ conn {amqpChan = chan}- where- chanList = filter (\(_, tag) -> isJust tag) (amqpChan conn)--resumeConsumer :: (Channel, Maybe ConsumerTag)- -> IO (Channel, Maybe ConsumerTag) resumeConsumer chan = flowConsumer chan True -flowConsumer :: (Channel, Maybe ConsumerTag)+flowConsumer :: AmqpConn -> Bool- -> IO (Channel, Maybe ConsumerTag)-flowConsumer (chan, tag) flag=+ -> IO AmqpConn+flowConsumer conn flag = case tag of- Nothing -> return (chan, tag)+ Nothing -> return conn Just _ -> do flow chan flag- return (chan, tag)+ return conn+ where+ (chan, tag) = amqpChan conn -closeChannels :: AmqpConn- -> IO AmqpConn-closeChannels conn = do- mapM_ (\(chan, _) -> closeChannel chan ) (amqpChan conn)- return $ conn {amqpChan =[]}+-- utils+--+createQueue :: AmqpConf -> AmqpConn -> IO (QueueName, Int, Int)+createQueue conf conn = declareQueue (fst $ amqpChan conn) (amqpQueue conf) -destoryConnection :: AmqpConn- -> IO ()-destoryConnection conn = do- addConnectionClosedHandler (amqpConn conn) False (return ())- closeConnection (amqpConn conn)+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 :: AmqpConf- -> Int+connect :: AmqpURI -> IO AmqpConn-connect conf num = do+connect uri = do -- make a connection and a channel of the connection.- conn <- openConnection'' uri- chan <- replicateM num (initChan conf conn)-- -- set a excetion when closeing the connection- -- addConnectionClosedHandler conn True (throwIO (ConnectionClosedException "Connection Closed."))- return $ AmqpConn conn chan conf- where- uri = fromURI (amqpUri conf)--initChan :: AmqpConf- -> Connection- -> IO (Channel, Maybe ConsumerTag)-initChan conf conn = do+ conn <- openConnection'' $ fromURI uri chan <- openChannel conn - -- debug- -- liftIO $ putStrLn ("URI = " ++ (show (amqpUri conf)))- -- liftIO $ putStrLn ("exchange name = " ++ (show eName))- -- liftIO $ putStrLn ("ecchange key = " ++ (show (key)))- -- liftIO $ putStrLn ("queue name= = " ++ (show (qName)))-- -- declare exchange, a queue and binding- declareExchange chan (amqpExchange conf)- _ <- declareQueue chan (amqpQueue conf)- bindQueue chan qName eName key-- return (chan, Nothing)- where- qName = queueName (amqpQueue conf)- key = amqpExchanKey conf- eName = exchangeName (amqpExchange conf)-+ -- 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- -- mapM_ (\(chan, _) -> closeChannel chan) (amqpChan conn) closeConnection (amqpConn conn) -send :: AmqpConn- -> ExchangeKey- -> Message- -> IO ()-send conn key msg = do- -- liftIO $ putStrLn ("key = " ++ show key ++ ", msg = " ++ show msg)- publishMsg- channel- (exchangeName (amqpExchange (amqpConf conn))) key msg+sendMsg :: AmqpConn+ -> Exchange+ -> ExchangeKey+ -> Message+ -> IO ()+sendMsg conn exchange key msg = do+ publishMsg chan exchange key msg where- channel = fst $ head $ (amqpChan conn)--numCheck :: Int -> Int-numCheck int- | int <= 0 = 1- | int > 0 = int- | otherwise = 1+ chan = fst (amqpChan conn) --- | this is for consumer, if you usea it with createConsumers, You have to call ackMsg or ackEnv after processing for any message that you get, otherwise it might be delivered again (see "ackMsg" and "ackEnv" in the "Network.AMQP" module)-amqpReceiveSource :: (Monad m, MonadIO m)+-- | Source as consuming data pushed.+amqpReceiveSource :: MonadIO m => (Message, Envelope) -> Source m (Message, Envelope) amqpReceiveSource (msg, env) = loop@@ -352,15 +259,12 @@ yield (msg, env) loop --- | a sink as sending msgs.-amqpSendSink :: (Monad m, MonadIO m)+-- | Sink as sending data.+amqpSendSink :: MonadIO m => AmqpConn+ -> Exchange -> ExchangeKey -> Sink Message m ()-amqpSendSink conn key = loop+amqpSendSink conn exchange key = loop where- loop = await >>= maybe (return ()) (\v -> (liftIO $ send conn key v) >> loop)----+ loop = await >>= maybe (return ()) (\v -> (liftIO $ sendMsg conn exchange key v) >> loop)
test/test.hs view
@@ -13,23 +13,23 @@ main = hspec $ do describe "produce and consume test" $ do it "send a message and recieve the message" $ do- runResourceT $ withAMQPChannel config $ \conn -> do- sendMsg str $$ amqpSendSink conn "myKey"- amqp <- createAMQPChannels config 1- amqp' <- createConsumers amqp $ \(msg,env) -> do+ runResourceT $ withChannel config $ \conn -> do+ sendMsg str $$ amqpSendSink conn "myExchange" "myKey"+ amqp <- createConnectionChannel 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- _ <- deleteConsumers amqp'+ _ <- deleteConsumer amqp' return () str :: String str = "This is a test message" config :: AmqpConf-config = AmqpConf "amqp://guest:guest@localhost:5672/" queue exchange "myKey"+config = AmqpConf "amqp://guest:guest@192.168.59.103:5672/" queue exchange "myKey" where exchange = newExchange {exchangeName = "myExchange", exchangeType = "direct"} queue = newQueue {queueName = "myQueue"}