amqp-utils 0.3.4.0 → 0.3.6.0

raw patch · 7 files changed

+51/−30 lines, 7 files

Files

• ChangeLog.md +6/−0
• Network/AMQP/Utils/Helpers.hs +18/−16
• Network/AMQP/Utils/Options.hs +10/−2
• amqp-utils.cabal +2/−2
• arbeite.hs +3/−4
• konsum.hs +2/−1
• plane.hs +10/−5

@@ -1,5 +1,11 @@ # Revision history for haskell-amqp-utils +## 0.3.6.0  -- 2019-02-15++* bug-fix: avoid deadlock in arbeite+* plane: add --header option+* rpc: log to stderr, result to stdout+ ## 0.3.4.0  -- 2018-07-21  * bug fix: re-add exception handler

@@ -25,26 +25,27 @@ printwithlabel :: String -> Maybe (IO ()) -> IO () printwithlabel _ Nothing = return () printwithlabel labl (Just i) = do-  mapM_ putStr [" --- ", labl, ": "]+  mapM_ (hPutStr stderr) [" --- ", labl, ": "]   i-  hFlush stdout+  hFlush stderr  -- | optional parameters printparam :: String -> Maybe String -> IO ()-printparam labl ms = printwithlabel labl $ fmap putStrLn ms+printparam labl ms = printwithlabel labl $ fmap (hPutStrLn stderr) ms  -- | required parameters printparam' :: String -> String -> IO () printparam' d s = printparam d (Just s)  -- | head chars of body-printbody :: (String, Maybe BL.ByteString) -> IO ()-printbody (labl, ms) =-  printwithlabel labl $ fmap (\s -> putStrLn "" >> BL.putStrLn s) ms+printbody :: Handle -> (String, Maybe BL.ByteString) -> IO ()+printbody h (labl, ms) =+  printwithlabel labl $+  fmap (\s -> hPutStrLn stderr "" >> BL.hPutStr h s >> hPutStrLn stderr "") ms  -- | log marker hr :: String -> IO ()-hr x = putStrLn hr' >> hFlush stdout+hr x = hPutStrLn stderr hr' >> hFlush stderr   where     hr' = take 72 $ (take 25 hr'') ++ " " ++ x ++ " " ++ hr''     hr'' = repeat '-'@@ -86,9 +87,9 @@ printopt (_, Nothing) = [] printopt (opt, Just s) = [opt, s] --- | prints header and head on STDOUT and returns cmdline options to callback-printmsg :: (Message, Envelope) -> Maybe Int -> ZonedTime -> IO [String]-printmsg (msg, envi) anR now = do+-- | prints header and head on stderr and returns cmdline options to callback+printmsg :: Handle -> (Message, Envelope) -> Maybe Int -> ZonedTime -> IO [String]+printmsg h (msg, envi) anR now = do   mapM_     (uncurry printparam)     [ ("routing key", rkey)@@ -110,7 +111,7 @@     , ("expiration", mexp)     , ("delivery mode", mdelivmode)     ]-  printbody (label, anriss)+  printbody h (label, anriss)   return $     concat       (map@@ -197,7 +198,7 @@   -> String   -> Args   -> ThreadId-  -> Maybe (ExitCode -> Handle -> IO ())+  -> Maybe (ExitCode -> BL.ByteString -> IO ())   -> IO () optionalFileStuff (msg, envi) callbackoptions addi numstring a tid action = do   path <- saveFile (tempDir a) numstring (msgBody msg)@@ -241,15 +242,16 @@   -> String   -> Envelope   -> [String]-  -> Maybe (ExitCode -> Handle -> IO ())+  -> Maybe (ExitCode -> BL.ByteString -> IO ())   -> IO () doProc a numstring envi (exe:args) action = do   (_, h, _, processhandle) <-     createProcess (proc exe args) {std_out = out, std_err = Inherit}-  exitcode <- waitForProcess processhandle+  sout <- mapM BL.hGetContents h+  exitcode <- maybe 0 id (fmap BL.length sout) `seq` waitForProcess processhandle   printparam' (numstring ++ " call returned") $ show exitcode-  if isJust action-    then ((fromJust action $ exitcode) (fromJust h)) >> acke envi a+  if isJust action && isJust sout+    then ((fromJust action $ exitcode) (fromJust sout)) >> acke envi a     else case exitcode of            ExitSuccess -> acke envi a            ExitFailure _ -> reje envi a

@@ -33,6 +33,7 @@   , connectionName :: Maybe String   , tmpQName :: String   , inputFile :: String+  , outputFile :: String   , lineMode :: Bool   , confirm :: Bool   , msgid :: Maybe Text@@ -80,6 +81,7 @@       Nothing       ""       "/dev/stdin"+      "/dev/stdout"       False       False       Nothing@@ -164,12 +166,18 @@         ["routingkey"]         (ReqArg (\s o -> o {rKey = s}) "ROUTINGKEY")         "AMQP routing key")-  , ( "a"+  , ( "ap"     , Option         ['f']         ["inputfile"]         (ReqArg (\s o -> o {inputFile = s}) "INPUTFILE")         ("Message input file (default: " ++ (inputFile def) ++ ")"))+  , ( "p"+    , Option+        ['O']+        ["outputfile"]+        (ReqArg (\s o -> o {outputFile = s}) "OUTPUTFILE")+        ("Message output file (default: " ++ (outputFile def) ++ ")"))   , ( "a"     , Option         ['l']@@ -254,7 +262,7 @@         ["exp"]         (ReqArg (\s o -> o {msgexp = Just $ pack s}) "EXP")         "Message Expiration")-  , ( "a"+  , ( "ap"     , Option         ['h']         ["header"]

@@ -1,6 +1,6 @@ name:                amqp-utils -version:             0.3.4.0+version:             0.3.6.0  synopsis:            Generic Haskell AMQP Consumer @@ -29,7 +29,7 @@  cabal-version:       >=1.10 -Tested-With: GHC == 7.10.2, GHC == 8.0.2, GHC == 8.2.2, GHC == 8.4.3+Tested-With: GHC == 7.10.2, GHC == 8.0.2, GHC == 8.2.2, GHC == 8.4.4, GHC == 8.6.3  executable konsum   main-is:             konsum.hs

@@ -3,7 +3,6 @@ import Control.Concurrent import qualified Control.Exception as X import Control.Monad-import qualified Data.ByteString.Lazy.Char8 as BL import Data.Map (singleton) import Data.Maybe import qualified Data.Text as T@@ -16,6 +15,7 @@ import Network.AMQP.Utils.Options import Paths_amqp_utils (version) import System.Environment+import System.IO  main :: IO () main = do@@ -67,15 +67,14 @@   now <- getZonedTime   callbackoptions <-     X.catch-      (printmsg m (anRiss a) now)+      (printmsg stderr m (anRiss a) now)       (\x -> X.throwTo tid (x :: X.SomeException) >> return [])   either (\e -> printparam' "ERROR" (show (e :: X.SomeException))) return =<<     X.try       (optionalFileStuff m callbackoptions addi numstring a tid (Just reply))   hr $ "END " ++ numstring   where-    reply e h = do-      contents <- BL.hGetContents h+    reply e contents = do       void $         publishMsg           c

@@ -11,6 +11,7 @@ import Network.AMQP.Utils.Options import Paths_amqp_utils (version) import System.Environment+import System.IO  main :: IO () main = do@@ -81,7 +82,7 @@   now <- getZonedTime   callbackoptions <-     X.catch-      (printmsg m (anRiss a) now)+      (printmsg stderr m (anRiss a) now)       (\x -> X.throwTo tid (x :: X.SomeException) >> return [])   either     (\e -> printparam' "ERROR" (show (e :: X.SomeException)) >> reje envi a)

@@ -16,6 +16,7 @@ import Paths_amqp_utils (version) import System.Environment import System.Exit+import System.IO  main :: IO () main = do@@ -30,9 +31,12 @@   (conn, chan) <- connect args   addChannelExceptionHandler chan (X.throwTo tid)   (q, _, _) <- declareQueue chan newQueue {queueExclusive = True}-  ctag <- consumeMsgs chan q NoAck (rpcClientCallback tid args)-  printparam' "consumer tag" $ T.unpack ctag+  printparam' "input file" $ inputFile args   message <- BL.readFile (inputFile args)+  printparam' "output file" $ outputFile args+  h <- openBinaryFile (outputFile args) WriteMode+  ctag <- consumeMsgs chan q NoAck (rpcClientCallback h tid args)+  printparam' "consumer tag" $ T.unpack ctag   now <- getCurrentTime >>= return . floor . utcTimeToPOSIXSeconds   hr "publishing request"   _ <- publishMsg@@ -45,6 +49,7 @@       , msgCorrelationID = corrid args       , msgExpiration = msgexp args       , msgTimestamp = Just now+      , msgHeaders = msgheader args       }   hr "waiting for answer"   _ <- forkIO@@ -63,14 +68,14 @@ exceptionHandler ReceivedException = return ExitSuccess exceptionHandler TimeoutException = return $ ExitFailure 1 -rpcClientCallback :: ThreadId -> Args -> (Message, Envelope) -> IO ()-rpcClientCallback tid a m@(_, env) = do+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 m (anRiss a) now)+      (printmsg h m (anRiss a) now)       (\x -> X.throwTo tid (x :: X.SomeException) >> return [])   throwTo tid ReceivedException