{-# LANGUAGE OverloadedStrings #-}
-- generic AMQP rpc client
import Control.Concurrent
import qualified Control.Exception as X
import Control.Monad
import qualified Data.ByteString.Lazy.Char8 as BL
import qualified Data.Text as T
import Data.Time
import Data.Time.Clock.POSIX
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
import System.Exit
import System.IO
main :: IO ()
main = do
hr "starting"
tid <- myThreadId
args <- getArgs >>= parseargs 'p'
X.onException
(printparam "rpc_timeout" [show (rpc_timeout args), "s"])
(error $ "invalid rpc_timeout")
printparam "client version" ["amqp-utils", showVersion version]
printparam "routing key" $ rKey args
(conn, chan) <- connect args
addChannelExceptionHandler chan (X.throwTo tid)
(q, _, _) <- declareQueue chan newQueue {queueExclusive = True}
if (currentExchange args /= "")
then do
printparam "exchange" $ currentExchange args
bindQueue chan q (T.pack $ currentExchange args) q
else return ()
let inputFile' = firstInputFile (inputFiles args)
printparam "input file" $ inputFile'
message <-
if inputFile' == "-"
then BL.getContents
else readFileRawLazy inputFile'
printparam "output file" $ outputFile args
h <-
if outputFile args == "-"
then return stdout
else openBinaryFile (outputFile args) WriteMode
ctag <- consumeMsgs chan q NoAck (rpcClientCallback h tid args)
printparam "consumer tag" ctag
now <- getCurrentTime >>= return . floor . utcTimeToPOSIXSeconds
hr "publishing request"
_ <-
publishMsg
chan
(T.pack $ currentExchange args)
(T.pack $ rKey args)
newMsg
{ msgBody = message
, msgReplyTo = Just q
, msgCorrelationID = corrid args
, msgExpiration = msgexp args
, msgTimestamp = Just now
, msgHeaders = msgheader args
}
hr "waiting for answer"
_ <-
forkIO
(threadDelay (floor (1000000 * rpc_timeout args)) >>
throwTo tid TimeoutException)
X.catch
(forever $ threadDelay 200000)
(\x -> do
ec <- exceptionHandler x
hr "closing connection"
closeConnection conn
printparam "exiting" ec
exitWith ec)
exceptionHandler :: RpcException -> IO (ExitCode)
exceptionHandler ReceivedException = hr "success" >> (return ExitSuccess)
exceptionHandler TimeoutException = hr "timeout" >> (return $ ExitFailure 1)
rpcClientCallback :: Handle -> ThreadId -> Args -> (Message, Envelope) -> IO ()
rpcClientCallback h tid a m@(_, env) = do
let numstring = show $ envDeliveryTag env
hr $ "received " ++ numstring
now <- getZonedTime
_ <-
X.catch
(printmsg (Just h) m (anRiss a) now)
(\x -> X.throwTo tid (x :: X.IOException) >> return ([], []))
throwTo tid ReceivedException
data RpcException
= ReceivedException
| TimeoutException
deriving (Show)
instance X.Exception RpcException