amqp-utils 0.3.7.1 → 0.4.0.0
raw patch · 8 files changed
+49/−19 lines, 8 files
Files
- ChangeLog.md +9/−0
- Network/AMQP/Utils/Helpers.hs +9/−5
- Network/AMQP/Utils/Options.hs +3/−3
- agitprop.hs +8/−2
- amqp-utils.cabal +2/−2
- arbeite.hs +6/−2
- konsum.hs +1/−2
- plane.hs +11/−3
ChangeLog.md view
@@ -1,5 +1,14 @@ # Revision history for haskell-amqp-utils +## 0.4.0.0 -- 2019-11-11+* agitprop: show exchange arg+* agitprop does not need -q or -Q+* allow RPC with dedicated exchange+* add missing hFlush+* stdin / stdout handling without using /dev/+* cleanup plane answer file handling+* duplicate plane body to outfile and stderr+ ## 0.3.7.1 -- 2019-09-10 * connect timeout
Network/AMQP/Utils/Helpers.hs view
@@ -38,10 +38,13 @@ printparam' d s = printparam d (Just s) -- | head chars of body-printbody :: Handle -> (String, Maybe BL.ByteString) -> IO ()-printbody h (labl, ms) =+printbody :: String -> Maybe BL.ByteString -> IO ()+printbody labl ms = do printwithlabel labl $- fmap (\s -> hPutStrLn stderr "" >> BL.hPutStr h s >> hPutStrLn stderr "") ms+ fmap+ (\s -> hPutStrLn stderr "" >> BL.hPut stderr s >> hPutStrLn stderr "")+ ms+ hFlush stderr -- | log marker hr :: String -> IO ()@@ -88,7 +91,7 @@ printopt (opt, Just s) = [opt, s] -- | prints header and head on stderr and returns cmdline options to callback-printmsg :: Handle -> (Message, Envelope) -> Maybe Int -> ZonedTime -> IO [String]+printmsg :: Maybe Handle -> (Message, Envelope) -> Maybe Int -> ZonedTime -> IO [String] printmsg h (msg, envi) anR now = do mapM_ (uncurry printparam)@@ -111,7 +114,8 @@ , ("expiration", mexp) , ("delivery mode", mdelivmode) ]- printbody h (label, anriss)+ printbody label anriss+ mapM_ (\hdl -> BL.hPut hdl body >> hFlush hdl) h return $ concat (map
Network/AMQP/Utils/Options.hs view
@@ -81,8 +81,8 @@ [] Nothing ""- "/dev/stdin"- "/dev/stdout"+ "-"+ "-" False False Nothing@@ -321,7 +321,7 @@ ["qname"] (ReqArg (\s o -> o {tmpQName = s}) "TEMPQNAME") "Name for temporary exclusive Queue")- , ( "akp"+ , ( "akrp" , Option ['x'] ["exchange"]
agitprop.hs view
@@ -33,7 +33,10 @@ printparam' "client version" $ "amqp-utils " ++ (showVersion version) printparam' "routing key" $ rKey args printparam' "exchange" $ currentExchange args- isDir <- F.getFileStatus (inputFile args) >>= return . F.isDirectory+ isDir <-+ if inputFile args == "-"+ then return False+ else F.getFileStatus (inputFile args) >>= return . F.isDirectory if isDir then printparam' "hotfolder" $ inputFile args else printparam' "input file" $@@ -70,7 +73,10 @@ hr $ "END watching " ++ (inputFile args) else do hr $ "BEGIN sending"- messageFile <- BL.readFile (inputFile args)+ messageFile <-+ if inputFile args == "-"+ then BL.getContents+ else BL.readFile (inputFile args) if (lineMode args) then mapM_ (publishOneMsg Nothing) (BL.lines messageFile) else publishOneMsg (Just (inputFile args)) messageFile
amqp-utils.cabal view
@@ -1,6 +1,6 @@ name: amqp-utils -version: 0.3.7.1+version: 0.4.0.0 synopsis: Generic Haskell AMQP tools @@ -31,7 +31,7 @@ cabal-version: >=1.10 -Tested-With: GHC ==7.10.2 || ==8.0.2 || ==8.2.2 || ==8.4.4 || >= 8.6.4+Tested-With: GHC ==7.10.2 || ==8.0.2 || ==8.2.2 || ==8.4.4 || >= 8.6.5 executable konsum main-is: konsum.hs
arbeite.hs view
@@ -15,7 +15,6 @@ import Network.AMQP.Utils.Options import Paths_amqp_utils (version) import System.Environment-import System.IO main :: IO () main = do@@ -38,6 +37,11 @@ (return) (fmap T.pack (qName args)) printparam' "queue name" $ T.unpack queue+ if (currentExchange args /= "")+ then do+ printparam' "exchange" $ currentExchange args+ bindQueue chan queue (T.pack $ currentExchange args) queue+ else return () ctag <- consumeMsgs chan@@ -67,7 +71,7 @@ now <- getZonedTime callbackoptions <- X.catch- (printmsg stderr m (anRiss a) now)+ (printmsg Nothing m (anRiss a) now) (\x -> X.throwTo tid (x :: X.SomeException) >> return []) either (\e -> printparam' "ERROR" (show (e :: X.SomeException))) return =<< X.try
konsum.hs view
@@ -11,7 +11,6 @@ import Network.AMQP.Utils.Options import Paths_amqp_utils (version) import System.Environment-import System.IO main :: IO () main = do@@ -82,7 +81,7 @@ now <- getZonedTime callbackoptions <- X.catch- (printmsg stderr m (anRiss a) now)+ (printmsg Nothing m (anRiss a) now) (\x -> X.throwTo tid (x :: X.SomeException) >> return []) either (\e -> printparam' "ERROR" (show (e :: X.SomeException)) >> reje envi a)
plane.hs view
@@ -31,10 +31,18 @@ (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 () printparam' "input file" $ inputFile args- message <- BL.readFile (inputFile args)+ message <-+ if inputFile args == "-"+ then BL.getContents+ else BL.readFile (inputFile args) printparam' "output file" $ outputFile args- h <- openBinaryFile (outputFile args) WriteMode+ h <- if outputFile args == "-" then return stdout else openBinaryFile (outputFile args) WriteMode ctag <- consumeMsgs chan q NoAck (rpcClientCallback h tid args) printparam' "consumer tag" $ T.unpack ctag now <- getCurrentTime >>= return . floor . utcTimeToPOSIXSeconds@@ -75,7 +83,7 @@ now <- getZonedTime _ <- X.catch- (printmsg h m (anRiss a) now)+ (printmsg (Just h) m (anRiss a) now) (\x -> X.throwTo tid (x :: X.SomeException) >> return []) throwTo tid ReceivedException