amqp-utils 0.4.3.0 → 0.4.4.0

raw patch · 7 files changed

+99/−32 lines, 7 files

Files

• ChangeLog.md +4/−0
• Network/AMQP/Utils/Helpers.hs +69/−23
• Network/AMQP/Utils/Options.hs +1/−1
• amqp-utils.cabal +1/−1
• arbeite.hs +11/−3
• konsum.hs +12/−3
• plane.hs +1/−1

@@ -1,5 +1,9 @@ # Revision history for haskell-amqp-utils +## 0.4.4.0  -- 2020-02-18++* push callback options into environment variables+ ## 0.4.3.0  -- 2020-02-10  * specify -R YES option for callback in case of a redelivered message

@@ -20,6 +20,7 @@ import Network.AMQP.Utils.Options import Network.Socket (PortNumber) import System.Directory (removeFile)+import System.Environment (getEnvironment) import System.Exit import System.IO import System.Process@@ -87,17 +88,34 @@     hr' = take 72 $ (take 25 hr'') ++ " " ++ x ++ " " ++ hr''     hr'' = repeat '-' +-- | format headers for printing formatheaders :: ((T.Text, FieldValue) -> [a]) -> FieldTable -> [a] formatheaders f (FieldTable ll) = concat $ map f $ M.toList ll +-- | format headers for setting environment variables+formatheadersEnv ::+     ((Int, (T.Text, FieldValue)) -> [(String, String)])+  -> FieldTable+  -> [(String, String)]+formatheadersEnv f (FieldTable ll) = concat $ map f $ zip [0 ..] $ M.toList ll+ -- | log formatting fieldshow :: (T.Text, FieldValue) -> String fieldshow (k, v) = "\n        " ++ T.unpack k ++ ": " ++ valueshow v  -- | callback cmdline formatting-fieldshow' :: (T.Text, FieldValue) -> [String]-fieldshow' (k, v) = ["-h", T.unpack k ++ "=" ++ valueshow v]+fieldshowOpt :: (T.Text, FieldValue) -> [String]+fieldshowOpt (k, v) = ["-h", T.unpack k ++ "=" ++ valueshow v] +-- | environment variable formatting+fieldshowEnv :: (Int, (T.Text, FieldValue)) -> [(String, String)]+fieldshowEnv (n, (k, v)) =+  [ ("AMQP_HEADER_KEY_" ++ nn, T.unpack k)+  , ("AMQP_HEADER_VALUE_" ++ nn, valueshow v)+  ]+  where+    nn = show n+ -- | showing a FieldValue valueshow :: FieldValue -> String valueshow (FVString value) = BS.toString value@@ -129,13 +147,14 @@ printopt (_, Nothing) = [] printopt (opt, Just s) = [opt, s] --- | prints header and head on stderr and returns cmdline options to callback+-- | prints header and head on stderr and returns+-- cmdline options and environment variables to callback printmsg ::      Maybe Handle   -> (Message, Envelope)   -> Maybe Int64   -> ZonedTime-  -> IO [String]+  -> IO ([String], [(String, String)]) printmsg h (msg, envi) anR now = do   mapM_     (uncurry printparam)@@ -160,22 +179,43 @@     ]   printparam label anriss   mapM_ (\hdl -> BL.hPut hdl body >> hFlush hdl) h-  return $-    concat-      (map-         printopt-         [ ("-r", rkey)-         , ("-m", ctype)-         , ("-e", cenc)-         , ("-i", messageid)-         , ("-t", timestamp)-         , ("-p", pri)-         , ("-R", redeliv)-         ] ++-       maybeToList headers')+  oldenv <- getEnvironment+  let environment =+        foldr+          step+          oldenv+          [ ("ROUTINGKEY", rkey)+          , ("CONTENTTYPE", ctype)+          , ("ENCODING", cenc)+          , ("MSGID", messageid)+          , ("TIMESTAMP", timestamp)+          , ("PRIORITY", pri)+          , ("REDELIVERED", redeliv)+          ] +++        headersEnv+  return (cmdline, environment)   where+    step (_, Nothing) xs = xs+    step (k, Just v) xs = ("AMQP_" ++ k, v) : xs+    cmdline =+      concat+        (map+           printopt+           [ ("-r", rkey)+           , ("-m", ctype)+           , ("-e", cenc)+           , ("-i", messageid)+           , ("-t", timestamp)+           , ("-p", pri)+           , ("-R", redeliv)+           ] +++         headersOpt)     headers = fmap (formatheaders fieldshow) $ msgHeaders msg-    headers' = fmap (formatheaders fieldshow') $ msgHeaders msg+    headersOpt =+      maybeToList $ fmap (formatheaders fieldshowOpt) $ msgHeaders msg+    headersEnv =+      concat . maybeToList $+      fmap (formatheadersEnv fieldshowEnv) $ msgHeaders msg     body = msgBody msg     anriss =       if isimage ctype@@ -248,8 +288,9 @@   -> Args   -> ThreadId   -> Maybe (ExitCode -> BL.ByteString -> IO ())+  -> [(String, String)]   -> IO ()-optionalFileStuff (msg, envi) callbackoptions addi numstring a tid action = do+optionalFileStuff (msg, envi) callbackoptions addi numstring a tid action environment = do   path <- saveFile (tempDir a) numstring (msgBody msg)   printparam "saved to" path   let callbackcmdline =@@ -262,7 +303,7 @@     (acke envi a)     (\c ->        forkFinally-         (doProc a numstring envi c action path)+         (doProc a numstring envi c action path environment)          (either (throwTo tid) return) >>        return ())     callbackcmdline@@ -294,10 +335,13 @@   -> [String]   -> Maybe (ExitCode -> BL.ByteString -> IO ())   -> Maybe String+  -> [(String, String)]   -> IO ()-doProc a numstring envi (exe:args) action path = do+doProc a numstring envi (exe:args) action path environment = do   (_, h, _, processhandle) <--    createProcess (proc exe args) {std_out = out, std_err = Inherit}+    createProcess+      (proc exe args)+        {std_out = out, std_err = Inherit, env = Just environment'}   sout <- mapM BL.hGetContents h   exitcode <-     maybe 0 id (fmap BL.length sout) `seq` waitForProcess processhandle@@ -317,7 +361,9 @@       if isJust action         then CreatePipe         else Inherit-doProc _ _ _ _ _ _ = return ()+    environment' =+      ("AMQP_NUMBER", numstring) : ("AMQP_FILE", fromJust path) : environment+doProc _ _ _ _ _ _ _ = return ()  -- | ack acke :: Envelope -> Args -> IO ()

@@ -1,8 +1,8 @@ module Network.AMQP.Utils.Options where +import qualified Data.ByteString.Char8 as BS import Data.Default.Class import Data.Int (Int64)-import qualified Data.ByteString.Char8 as BS import qualified Data.Map as M import Data.Maybe import Data.Text (Text, pack)

@@ -1,6 +1,6 @@ name:                amqp-utils -version:             0.4.3.0+version:             0.4.4.0  synopsis:            Generic Haskell AMQP tools 

@@ -72,13 +72,21 @@   let numstring = show $ envDeliveryTag env   hr $ "BEGIN " ++ numstring   now <- getZonedTime-  callbackoptions <-+  (callbackoptions, callbackenv) <-     X.catch       (printmsg Nothing m (anRiss a) now)-      (\x -> X.throwTo tid (x :: X.SomeException) >> return [])+      (\x -> X.throwTo tid (x :: X.SomeException) >> return ([], []))   either (\e -> printparam "ERROR" (e :: X.SomeException)) return =<<     X.try-      (optionalFileStuff m callbackoptions addi numstring a tid (Just reply))+      (optionalFileStuff+         m+         callbackoptions+         addi+         numstring+         a+         tid+         (Just reply)+         callbackenv)   hr $ "END " ++ numstring   where     reply e contents = do

@@ -81,10 +81,19 @@   let numstring = show $ envDeliveryTag envi   hr $ "BEGIN " ++ numstring   now <- getZonedTime-  callbackoptions <-+  (callbackoptions, callbackenv) <-     X.catch       (printmsg Nothing m (anRiss a) now)-      (\x -> X.throwTo tid (x :: X.SomeException) >> return [])+      (\x -> X.throwTo tid (x :: X.SomeException) >> return ([], []))   either (\e -> printparam "ERROR" (e :: X.SomeException) >> reje envi a) return =<<-    X.try (optionalFileStuff m callbackoptions addi numstring a tid Nothing)+    X.try+      (optionalFileStuff+         m+         callbackoptions+         addi+         numstring+         a+         tid+         Nothing+         callbackenv)   hr $ "END " ++ numstring

@@ -89,7 +89,7 @@   _ <-     X.catch       (printmsg (Just h) m (anRiss a) now)-      (\x -> X.throwTo tid (x :: X.SomeException) >> return [])+      (\x -> X.throwTo tid (x :: X.SomeException) >> return ([], []))   throwTo tid ReceivedException  data RpcException