konsum.hs

   1 -- generic amqp consumer
   2 import           Control.Concurrent
   3 import qualified Control.Exception             as X
   4 import qualified Data.Text                     as T
   5 import           Data.Time
   6 import           Data.Version                  (showVersion)
   7 import           Network.AMQP
   8 import           Network.AMQP.Utils.Connection
   9 import           Network.AMQP.Utils.Helpers
  10 import           Network.AMQP.Utils.Options
  11 import           Paths_amqp_utils              (version)
  12 import           System.Environment
  13
  14 main :: IO ()
  15 main = do
  16   hr "starting"
  17   tid <- myThreadId
  18   args <- getArgs >>= parseargs 'k'
  19   let addiArgs = reverse $ additionalArgs args
  20   printparam "client version" ["amqp-utils", showVersion version]
  21   (conn, chan) <- connect args
  22   addChannelExceptionHandler chan (X.throwTo tid)
  23   -- set prefetch
  24   printparam "prefetch" $ preFetch args
  25   qos chan 0 (preFetch args) False
  26   -- attach to given queue? or build exclusive queue and bind it?
  27   queue <-
  28     maybe
  29       (tempQueue chan (tmpQName args) (bindings args))
  30       (return)
  31       (fmap T.pack (qName args))
  32   printparam "queue name" queue
  33   printparam "shown body chars" $ anRiss args
  34   printparam "temp dir" $ tempDir args
  35   printparam "callback" $ fileProcess args
  36   printparam "callback args" $ addiArgs
  37   printparam "cleanup temp file" $
  38     maybe Nothing (\_ -> Just (cleanupTmpFile args)) (fileProcess args)
  39   -- subscribe to the queue
  40   ctag <-
  41     consumeMsgs
  42       chan
  43       queue
  44       (if ack args
  45          then Ack
  46          else NoAck)
  47       (myCallback args addiArgs tid)
  48   printparam "consumer tag" ctag
  49   printparam "send acks" $ ack args
  50   printparam "requeue if rejected" $ (ack args) && (requeuenack args)
  51   hr "entering main loop"
  52   sleepingBeauty >>= printparam "exception"
  53   closeConnection conn
  54   hr "connection closed"
  55
  56 -- | exclusive temp queue
  57 tempQueue :: Channel -> String -> [(String, String)] -> IO T.Text
  58 tempQueue chan tmpqname bindlist = do
  59   (q, _, _) <-
  60     declareQueue
  61       chan
  62       newQueue {queueExclusive = True, queueName = T.pack tmpqname}
  63   mapM_
  64     (\(xchange, bkey) ->
  65        bindQueue chan q (T.pack xchange) (T.pack bkey) >>
  66        printparam "binding" [xchange, bkey])
  67     bindlist
  68   return q
  69
  70 -- | process received message
  71 myCallback :: Args -> [String] -> ThreadId -> (Message, Envelope) -> IO ()
  72 myCallback a addi tid m@(_, envi) = do
  73   let numstring = show $ envDeliveryTag envi
  74   hr $ "BEGIN " ++ numstring
  75   now <- getZonedTime
  76   (callbackoptions, callbackenv) <-
  77     X.catch
  78       (printmsg Nothing m (anRiss a) now)
  79       (\x -> X.throwTo tid (x :: X.IOException) >> return ([], []))
  80   either (\e -> printparam "ERROR" (e :: X.IOException) >> reje envi a) return =<<
  81     X.try
  82       (optionalFileStuff
  83          m
  84          callbackoptions
  85          addi
  86          numstring
  87          a
  88          tid
  89          Nothing
  90          callbackenv)
  91   hr $ "END " ++ numstring