plane.hs

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