amqp-0.12: examples/ExampleConsumer.hs

{-# OPTIONS -XOverloadedStrings -XScopedTypeVariables #-}
import Network.AMQP
import Control.Concurrent
import qualified Control.Exception as CE
import qualified Data.ByteString.Lazy.Char8 as BL


main = do
    conn <- openConnection "127.0.0.1" "/" "guest" "guest"
    chan <- openChannel conn

    
    --declare queues, exchanges and bindings
    declareQueue chan newQueue {queueName = "myQueueDE"}
    declareQueue chan newQueue {queueName = "myQueueEN"}
    
    declareExchange chan newExchange {exchangeName = "topicExchg", exchangeType = "topic"}
    
    
    CE.catch (bindQueue chan "myQueueDE" "e" "de.*")
        (\(e::CE.SomeException) -> return ())
    print "B"
    -- bindQueue chan "myQueueEN" "topicExchg" "en.*"
    print "c"
    addChannelExceptionHandler chan $ \e -> do
        threadDelay 1000000
        print "OOOOOOOO"
    
    print "A"
    --subscribe to the queues
    consumeMsgs chan "myQueueDE" Ack myCallbackDE
    consumeMsgs chan "myQueueEN" Ack myCallbackEN
    
    
    getLine -- wait for keypress
    closeConnection conn
    putStrLn "connection closed"

    
    

myCallbackDE :: (Message,Envelope) -> IO ()
myCallbackDE (msg, env) = do
    putStrLn $ "received from DE: "++(BL.unpack $ msgBody msg)
    ackEnv env


myCallbackEN :: (Message,Envelope) -> IO ()
myCallbackEN (msg, env) = do
    putStrLn $ "received from EN: "++(BL.unpack $ msgBody msg)
    ackEnv env