-- generic amqp consumer
import Control.Concurrent
import qualified Control.Exception as X
import Control.Monad
import qualified Data.Text as T
import Data.Time
import Data.Version (showVersion)
import Network.AMQP
import Network.AMQP.Utils.Connection
import Network.AMQP.Utils.Helpers
import Network.AMQP.Utils.Options
import Paths_amqp_utils (version)
import System.Environment
main :: IO ()
main = do
hr "starting"
tid <- myThreadId
args <- getArgs >>= parseargs 'k'
let addiArgs = reverse $ additionalArgs args
printparam "client version" ["amqp-utils", showVersion version]
(conn, chan) <- connect args
addChannelExceptionHandler chan (X.throwTo tid)
-- set prefetch
printparam "prefetch" $ preFetch args
qos chan 0 (preFetch args) False
-- attach to given queue? or build exclusive queue and bind it?
queue <-
maybe
(tempQueue chan (tmpQName args) (bindings args) (currentExchange args))
(return)
(fmap T.pack (qName args))
printparam "queue name" queue
printparam "shown body chars" $ anRiss args
printparam "temp dir" $ tempDir args
printparam "callback" $ fileProcess args
printparam "callback args" $ addiArgs
printparam "cleanup temp file" $
maybe Nothing (\_ -> Just (cleanupTmpFile args)) (fileProcess args)
-- subscribe to the queue
ctag <-
consumeMsgs
chan
queue
(if ack args
then Ack
else NoAck)
(myCallback args addiArgs tid)
printparam "consumer tag" ctag
printparam "send acks" $ ack args
printparam "requeue if rejected" $
if (ack args)
then Just (show (requeuenack args))
else Nothing
hr "entering main loop"
X.catch
(forever $ threadDelay 5000000)
(\e -> printparam "exception" (e :: X.SomeException))
closeConnection conn
hr "connection closed"
-- | exclusive temp queue
tempQueue :: Channel -> String -> [(String, String)] -> String -> IO T.Text
tempQueue chan tmpqname bindlist x = do
(q, _, _) <-
declareQueue
chan
newQueue {queueExclusive = True, queueName = T.pack tmpqname}
mapM_
(\(xchange, bkey) ->
bindQueue chan q (T.pack xchange) (T.pack bkey) >>
printparam "binding" [xchange, bkey])
(if null bindlist
then [(x, "#")]
else bindlist)
return q
-- | process received message
myCallback :: Args -> [String] -> ThreadId -> (Message, Envelope) -> IO ()
myCallback a addi tid m@(_, envi) = do
let numstring = show $ envDeliveryTag envi
hr $ "BEGIN " ++ numstring
now <- getZonedTime
callbackoptions <-
X.catch
(printmsg Nothing m (anRiss a) now)
(\x -> X.throwTo tid (x :: X.SomeException) >> return [])
either (\e -> printparam "ERROR" (e :: X.SomeException) >> reje envi a) return =<<
X.try (optionalFileStuff m callbackoptions addi numstring a tid Nothing)
hr $ "END " ++ numstring