plane.hs

   1 {-# LANGUAGE OverloadedStrings #-}
   2
   3 -- generic AMQP rpc client
   4 import Control.Concurrent
   5 import qualified Control.Exception as X
   6 import Control.Monad
   7 import qualified Data.ByteString.Lazy.Char8 as BL
   8 import qualified Data.Text as T
   9 import Data.Time
  10 import Data.Time.Clock.POSIX
  11 import Data.Version (showVersion)
  12 import Network.AMQP
  13 import Network.AMQP.Utils.Connection
  14 import Network.AMQP.Utils.Helpers
  15 import Network.AMQP.Utils.Options
  16 import Paths_amqp_utils (version)
  17 import System.Environment
  18 import System.Exit
  19 import System.IO
  20
  21 main :: IO ()
  22 main = do
  23   hr "starting"
  24   tid <- myThreadId
  25   args <- getArgs >>= parseargs 'p'
  26   X.onException
  27     (printparam "rpc_timeout" $ show (rpc_timeout args) ++ "s")
  28     (error $ "invalid rpc_timeout")
  29   printparam "client version" $ "amqp-utils " ++ (showVersion version)
  30   printparam "destination queue" $ tmpQName args
  31   (conn, chan) <- connect args
  32   addChannelExceptionHandler chan (X.throwTo tid)
  33   (q, _, _) <- declareQueue chan newQueue {queueExclusive = True}
  34   if (currentExchange args /= "")
  35     then do
  36       printparam "exchange" $ currentExchange args
  37       bindQueue chan q (T.pack $ currentExchange args) q
  38     else return ()
  39   printparam "input file" $ inputFile args
  40   message <-
  41     if inputFile args == "-"
  42       then BL.getContents
  43       else BL.readFile (inputFile args)
  44   printparam "output file" $ outputFile args
  45   h <- if outputFile args == "-" then return stdout else openBinaryFile (outputFile args) WriteMode
  46   ctag <- consumeMsgs chan q NoAck (rpcClientCallback h tid args)
  47   printparam "consumer tag" ctag
  48   now <- getCurrentTime >>= return . floor . utcTimeToPOSIXSeconds
  49   hr "publishing request"
  50   _ <- publishMsg
  51     chan
  52     (T.pack $ currentExchange args)
  53     (T.pack $ tmpQName args)
  54     newMsg
  55       { msgBody = message
  56       , msgReplyTo = Just q
  57       , msgCorrelationID = corrid args
  58       , msgExpiration = msgexp args
  59       , msgTimestamp = Just now
  60       , msgHeaders = msgheader args
  61       }
  62   hr "waiting for answer"
  63   _ <- forkIO
  64     (threadDelay (floor (1000000 * rpc_timeout args)) >>
  65      throwTo tid TimeoutException)
  66   X.catch
  67     (forever $ threadDelay 200000)
  68     (\x -> do
  69        ec <- exceptionHandler x
  70        hr "closing connection"
  71        closeConnection conn
  72        printparam "exiting" ec
  73        exitWith ec)
  74
  75 exceptionHandler :: RpcException -> IO (ExitCode)
  76 exceptionHandler ReceivedException = return ExitSuccess
  77 exceptionHandler TimeoutException = return $ ExitFailure 1
  78
  79 rpcClientCallback :: Handle -> ThreadId -> Args -> (Message, Envelope) -> IO ()
  80 rpcClientCallback h tid a m@(_, env) = do
  81   let numstring = show $ envDeliveryTag env
  82   hr $ "received " ++ numstring
  83   now <- getZonedTime
  84   _ <-
  85     X.catch
  86       (printmsg (Just h) m (anRiss a) now)
  87       (\x -> X.throwTo tid (x :: X.SomeException) >> return [])
  88   throwTo tid ReceivedException
  89
  90 data RpcException
  91   = ReceivedException
  92   | TimeoutException
  93   deriving (Show)
  94
  95 instance X.Exception RpcException