Network/AMQP/Utils/Helpers.hs

   1 {-# LANGUAGE FlexibleInstances #-}
   2
   3 module Network.AMQP.Utils.Helpers where
   4
   5 import Control.Concurrent
   6 import qualified Control.Exception as X
   7 import Control.Monad
   8 import qualified Data.ByteString.Lazy.Char8 as BL
   9 import qualified Data.ByteString.UTF8 as BS
  10 import Data.Int (Int64)
  11 import Data.List
  12 import qualified Data.Map as M
  13 import Data.Maybe
  14 import qualified Data.Text as T
  15 import Data.Time
  16 import Data.Time.Clock.POSIX
  17 import Data.Word (Word16)
  18 import Network.AMQP
  19 import Network.AMQP.Types
  20 import Network.AMQP.Utils.Options
  21 import Network.Socket (PortNumber)
  22 import System.Directory (removeFile)
  23 import System.Environment (getEnvironment)
  24 import System.Exit
  25 import System.IO
  26 import System.Process
  27
  28 -- | print config parameters
  29 class (Show a) =>
  30       Flexprint a
  31   where
  32   flexprint :: a -> IO ()
  33   flexprint = (hPutStrLn stderr) . show
  34   empty :: a -> Bool
  35   empty _ = False
  36   printparam :: String -> a -> IO ()
  37   printparam label x =
  38     if empty x
  39       then return ()
  40       else do
  41         mapM_ (hPutStr stderr) [" --- ", label, ": "]
  42         flexprint x
  43         hFlush stderr
  44
  45 instance (Flexprint a) => Flexprint (Maybe a) where
  46   empty = isNothing
  47   printparam _ Nothing = return ()
  48   printparam x (Just y) = printparam x y
  49
  50 instance Flexprint String where
  51   flexprint = hPutStrLn stderr
  52   empty = null
  53
  54 instance Flexprint [String] where
  55   flexprint = flexprint . unwords
  56   empty = null
  57
  58 instance Flexprint T.Text where
  59   flexprint = flexprint . T.unpack
  60   empty = T.null
  61
  62 instance Flexprint BL.ByteString where
  63   flexprint x = hPutStrLn stderr "" >> BL.hPut stderr x >> hPutStrLn stderr ""
  64   empty = BL.null
  65
  66 instance Flexprint Bool
  67
  68 instance Flexprint Int
  69
  70 instance Flexprint Int64
  71
  72 instance Flexprint Word16
  73
  74 instance Flexprint ExitCode
  75
  76 instance Flexprint X.SomeException
  77
  78 instance Flexprint AMQPException
  79
  80 instance Flexprint ConfirmationResult
  81
  82 instance Flexprint PortNumber
  83
  84 -- | log marker
  85 hr :: String -> IO ()
  86 hr x = hPutStrLn stderr hr' >> hFlush stderr
  87   where
  88     hr' = take 72 $ (take 25 hr'') ++ " " ++ x ++ " " ++ hr''
  89     hr'' = repeat '-'
  90
  91 -- | format headers for printing
  92 formatheaders :: ((T.Text, FieldValue) -> [a]) -> FieldTable -> [a]
  93 formatheaders f (FieldTable ll) = concat $ map f $ M.toList ll
  94
  95 -- | format headers for setting environment variables
  96 formatheaders' ::
  97      ((Int, (T.Text, FieldValue)) -> [(String, String)]) -> FieldTable -> [(String,String)]
  98 formatheaders' f (FieldTable ll) = concat $ map f $ zip [0 ..] $ M.toList ll
  99
 100 -- | log formatting
 101 fieldshow :: (T.Text, FieldValue) -> String
 102 fieldshow (k, v) = "\n        " ++ T.unpack k ++ ": " ++ valueshow v
 103
 104 -- | callback cmdline formatting
 105 fieldshow' :: (T.Text, FieldValue) -> [String]
 106 fieldshow' (k, v) = ["-h", T.unpack k ++ "=" ++ valueshow v]
 107
 108 -- | environment variable formatting
 109 fieldshow'' :: (Int, (T.Text, FieldValue)) -> [(String, String)]
 110 fieldshow'' (n, (k, v)) =
 111   [("AMQP_HEADER_KEY_" ++ nn, T.unpack k), ("AMQP_HEADER_VALUE_" ++ nn, valueshow v)]
 112   where
 113     nn = show n
 114
 115 -- | showing a FieldValue
 116 valueshow :: FieldValue -> String
 117 valueshow (FVString value) = BS.toString value
 118 valueshow (FVInt8 value) = show value
 119 valueshow (FVInt16 value) = show value
 120 valueshow (FVInt32 value) = show value
 121 valueshow (FVInt64 value) = show value
 122 valueshow (FVFloat value) = show value
 123 valueshow (FVDouble value) = show value
 124 valueshow value = show value
 125
 126 -- | skip showing body head if binary type
 127 isimage :: Maybe String -> Bool
 128 isimage Nothing = False
 129 isimage (Just ctype)
 130   | isPrefixOf "application/xml" ctype = False
 131   | isPrefixOf "application/json" ctype = False
 132   | otherwise = any (flip isPrefixOf ctype) ["application", "image"]
 133
 134 -- | show the first bytes of message body
 135 anriss' :: Maybe Int64 -> BL.ByteString -> BL.ByteString
 136 anriss' x =
 137   case x of
 138     Nothing -> id
 139     Just y -> BL.take y
 140
 141 -- | callback cmdline with optional parameters
 142 printopt :: (String, Maybe String) -> [String]
 143 printopt (_, Nothing) = []
 144 printopt (opt, Just s) = [opt, s]
 145
 146 -- | prints header and head on stderr and returns
 147 -- cmdline options and environment variables to callback
 148 printmsg ::
 149      Maybe Handle
 150   -> (Message, Envelope)
 151   -> Maybe Int64
 152   -> ZonedTime
 153   -> IO ([String], [(String, String)])
 154 printmsg h (msg, envi) anR now = do
 155   mapM_
 156     (uncurry printparam)
 157     [ ("routing key", rkey)
 158     , ("message-id", messageid)
 159     , ("headers", headers)
 160     , ("content-type", ctype)
 161     , ("content-encoding", cenc)
 162     , ("redelivered", redeliv)
 163     , ("timestamp", timestamp'')
 164     , ("time now", now')
 165     , ("size", size)
 166     , ("priority", pri)
 167     , ("type", mtype)
 168     , ("user id", muserid)
 169     , ("application id", mappid)
 170     , ("cluster id", mclusterid)
 171     , ("reply to", mreplyto)
 172     , ("correlation id", mcorrid)
 173     , ("expiration", mexp)
 174     , ("delivery mode", mdelivmode)
 175     ]
 176   printparam label anriss
 177   mapM_ (\hdl -> BL.hPut hdl body >> hFlush hdl) h
 178   oldenv <- getEnvironment
 179   let environment =
 180         foldr
 181           step
 182           oldenv
 183           [ ("ROUTINGKEY", rkey)
 184           , ("CONTENTTYPE", ctype)
 185           , ("ENCODING", cenc)
 186           , ("MSGID", messageid)
 187           , ("TIMESTAMP", timestamp)
 188           , ("PRIORITY", pri)
 189           , ("REDELIVERED", redeliv)
 190           ] ++
 191         headersEnv
 192   return (cmdline, environment)
 193   where
 194     step (_, Nothing) xs = xs
 195     step (k, Just v) xs = ("AMQP_" ++ k, v) : xs
 196     cmdline =
 197       concat
 198         (map
 199            printopt
 200            [ ("-r", rkey)
 201            , ("-m", ctype)
 202            , ("-e", cenc)
 203            , ("-i", messageid)
 204            , ("-t", timestamp)
 205            , ("-p", pri)
 206            , ("-R", redeliv)
 207            ] ++
 208          headersOpt)
 209     headers = fmap (formatheaders fieldshow) $ msgHeaders msg
 210     headersOpt = maybeToList $ fmap (formatheaders fieldshow') $ msgHeaders msg
 211     headersEnv =
 212       concat . maybeToList $ fmap (formatheaders' fieldshow'') $ msgHeaders msg
 213     body = msgBody msg
 214     anriss =
 215       if isimage ctype
 216         then Nothing
 217         else Just (anriss' anR body) :: Maybe BL.ByteString
 218     anriss'' = maybe "" (\a -> "first " ++ (show a) ++ " bytes of ") anR
 219     label = anriss'' ++ "body"
 220     ctype = fmap T.unpack $ msgContentType msg
 221     cenc = fmap T.unpack $ msgContentEncoding msg
 222     rkey = Just . T.unpack $ envRoutingKey envi
 223     messageid = fmap T.unpack $ msgID msg
 224     pri = fmap show $ msgPriority msg
 225     mtype = fmap show $ msgType msg
 226     muserid = fmap show $ msgUserID msg
 227     mappid = fmap show $ msgApplicationID msg
 228     mclusterid = fmap show $ msgClusterID msg
 229     mreplyto = fmap show $ msgReplyTo msg
 230     mcorrid = fmap show $ msgCorrelationID msg
 231     mexp = fmap show $ msgExpiration msg
 232     mdelivmode = fmap show $ msgDeliveryMode msg
 233     size = Just . show $ BL.length body
 234     redeliv =
 235       if envRedelivered envi
 236         then Just "YES"
 237         else Nothing
 238     tz = zonedTimeZone now
 239     nowutc = zonedTimeToUTCFLoor now
 240     msgtime = msgTimestamp msg
 241     msgtimeutc = fmap (posixSecondsToUTCTime . realToFrac) msgtime
 242     timestamp = fmap show msgtime
 243     timediff = fmap (difftime nowutc) msgtimeutc
 244     now' =
 245       case timediff of
 246         Just "now" -> Nothing
 247         _ -> showtime tz $ Just nowutc
 248     timestamp' = showtime tz msgtimeutc
 249     timestamp'' =
 250       liftM3
 251         (\a b c -> a ++ " (" ++ b ++ ") (" ++ c ++ ")")
 252         timestamp
 253         timestamp'
 254         timediff
 255
 256 -- | timestamp conversion
 257 zonedTimeToUTCFLoor :: ZonedTime -> UTCTime
 258 zonedTimeToUTCFLoor x =
 259   posixSecondsToUTCTime $
 260   realToFrac ((floor . utcTimeToPOSIXSeconds . zonedTimeToUTC) x :: Timestamp)
 261
 262 -- | show the timestamp
 263 showtime :: TimeZone -> Maybe UTCTime -> Maybe String
 264 showtime tz = fmap (show . (utcToZonedTime tz))
 265
 266 -- | show difference between two timestamps
 267 difftime :: UTCTime -> UTCTime -> String
 268 difftime now msg
 269   | now == msg = "now"
 270   | now > msg = diff ++ " ago"
 271   | otherwise = diff ++ " in the future"
 272   where
 273     diff = show (diffUTCTime now msg)
 274
 275 -- | if the message is to be saved
 276 -- and maybe processed further
 277 optionalFileStuff ::
 278      (Message, Envelope)
 279   -> [String]
 280   -> [String]
 281   -> String
 282   -> Args
 283   -> ThreadId
 284   -> Maybe (ExitCode -> BL.ByteString -> IO ())
 285   -> [(String,String)]
 286   -> IO ()
 287 optionalFileStuff (msg, envi) callbackoptions addi numstring a tid action environment = do
 288   path <- saveFile (tempDir a) numstring (msgBody msg)
 289   printparam "saved to" path
 290   let callbackcmdline =
 291         liftM2
 292           (constructCallbackCmdLine (simple a) callbackoptions addi numstring)
 293           (fileProcess a)
 294           path
 295   printparam "calling" callbackcmdline
 296   maybe
 297     (acke envi a)
 298     (\c ->
 299        forkFinally
 300          (doProc a numstring envi c action path environment)
 301          (either (throwTo tid) return) >>
 302        return ())
 303     callbackcmdline
 304
 305 -- | save message into temp file
 306 saveFile :: Maybe String -> String -> BL.ByteString -> IO (Maybe String)
 307 saveFile Nothing _ _ = return Nothing
 308 saveFile (Just tempD) numstring body = do
 309   (p, h) <-
 310     openBinaryTempFileWithDefaultPermissions
 311       tempD
 312       ("amqp-utils-" ++ numstring ++ "-.tmp")
 313   BL.hPut h body
 314   hClose h
 315   return $ Just p
 316
 317 -- | construct cmdline for callback script
 318 constructCallbackCmdLine ::
 319      Bool -> [String] -> [String] -> String -> String -> String -> [String]
 320 constructCallbackCmdLine True _ addi _ exe path = exe : addi ++ path : []
 321 constructCallbackCmdLine False opts addi num exe path =
 322   exe : "-f" : path : "-n" : num : opts ++ addi
 323
 324 -- | call callback script
 325 doProc ::
 326      Args
 327   -> String
 328   -> Envelope
 329   -> [String]
 330   -> Maybe (ExitCode -> BL.ByteString -> IO ())
 331   -> Maybe String
 332   -> [(String,String)]
 333   -> IO ()
 334 doProc a numstring envi (exe:args) action path environment = do
 335   (_, h, _, processhandle) <-
 336     createProcess
 337       (proc exe args)
 338         {std_out = out, std_err = Inherit, env = Just environment'}
 339   sout <- mapM BL.hGetContents h
 340   exitcode <-
 341     maybe 0 id (fmap BL.length sout) `seq` waitForProcess processhandle
 342   printparam (numstring ++ " call returned") exitcode
 343   if isJust action && isJust sout
 344     then ((fromJust action $ exitcode) (fromJust sout)) >> acke envi a
 345     else case exitcode of
 346            ExitSuccess -> acke envi a
 347            ExitFailure _ -> reje envi a
 348   if (cleanupTmpFile a)
 349     then X.catch
 350            (maybe (return ()) removeFile path)
 351            (\e -> printparam "error removing temp file" (e :: X.SomeException))
 352     else return ()
 353   where
 354     out =
 355       if isJust action
 356         then CreatePipe
 357         else Inherit
 358     environment' =
 359       ("AMQP_NUMBER",numstring):("AMQP_FILE",fromJust path):environment
 360 doProc _ _ _ _ _ _ _ = return ()
 361
 362 -- | ack
 363 acke :: Envelope -> Args -> IO ()
 364 acke envi a
 365   | (ack a) = ackEnv envi
 366   | otherwise = return ()
 367
 368 -- | reject
 369 reje :: Envelope -> Args -> IO ()
 370 reje envi a
 371   | (ack a) = rejectEnv envi (requeuenack a)
 372   | otherwise = return ()