diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -6,6 +6,16 @@ # Revision history for haskell-amqp-utils +## 0.6.6.0 -- 2024-12-06 + +* LineBuffering +* print heartbeat parameter +* optionally delay negative acknowledgements (--delaynack/-D) + +## 0.6.5.0 -- 2023-11-14 + +* konsum: new --stream\_offset option + ## 0.6.4.0 -- 2023-07-20 * fix exception handling and avoid hang diff --git a/Network/AMQP/Utils/Connection.hs b/Network/AMQP/Utils/Connection.hs --- a/Network/AMQP/Utils/Connection.hs +++ b/Network/AMQP/Utils/Connection.hs @@ -25,6 +25,7 @@ printparam "port" $ portnumber args printparam "vhost" $ vHost args printparam "connection_name" $ connectionName args + printparam "heartbeat" $ liftA2 (\ n m -> show n ++ m) (heartBeat args) (Just " s") printparam "connect timeout" $ [show (connect_timeout args), "s"] globalCertificateStore <- getSystemCertificateStore let myTLS = diff --git a/Network/AMQP/Utils/Helpers.hs b/Network/AMQP/Utils/Helpers.hs --- a/Network/AMQP/Utils/Helpers.hs +++ b/Network/AMQP/Utils/Helpers.hs @@ -415,8 +415,15 @@ -- | reject reje :: Envelope -> Args -> IO () reje envi a - | (ack a) = rejectEnv envi (requeuenack a) + | (ack a) = delayreject (delaynack a) >> rejectEnv envi (requeuenack a) | otherwise = return () + +-- | optionally delay reject +delayreject :: Int -> IO () +delayreject 0 = return () +delayreject delay = do + printparam "delaying negative acknowledgement for" [(show delay),"s"] + threadDelay (delay * 1000000) -- | main loop: sleep forever or wait for an exception sleepingBeauty :: IO (X.SomeException) diff --git a/Network/AMQP/Utils/Options.hs b/Network/AMQP/Utils/Options.hs --- a/Network/AMQP/Utils/Options.hs +++ b/Network/AMQP/Utils/Options.hs @@ -20,6 +20,7 @@ import Paths_amqp_utils (version) import System.Console.GetOpt import System.FilePath.Posix.ByteString (RawFilePath) +import Text.Read (readMaybe) portnumber :: Args -> PortNumber portnumber a @@ -79,6 +80,8 @@ , removeSentFile :: Bool , moveSentFileTo :: Maybe RawFilePath , initialScan :: Bool + , streamoffset :: FieldTable + , delaynack :: Int } instance Default Args where @@ -133,6 +136,8 @@ False Nothing False + (FieldTable M.empty) + 0 -- | all options allOptions :: [(String, OptDescr (Args -> Args))] @@ -307,6 +312,14 @@ (\s o -> o {msgheader = addheader (msgheader o) s}) "HEADER=VALUE") "Message Headers") + , ( "k" + , Option + [] + ["stream_offset"] + (ReqArg + (\s o -> o {streamoffset = mkStreamOffset s}) + "OFFSET") + "x-stream-offset consumer argument") , ( "a" , Option ['F'] @@ -455,6 +468,13 @@ (ReqArg (\s o -> o {connect_timeout = read s}) "SECONDS") ("timeout for establishing initial connection (default: " ++ show (connect_timeout def) ++ ")")) + , ( "k" + , Option + ['D'] + ["delaynack"] + (ReqArg (\s o -> o {delaynack = read s}) "SECONDS") + ("delay negative acknowledgements (default: " ++ + show (delaynack def) ++ ")")) ] -- | Options for the executables @@ -473,6 +493,12 @@ getval :: String -> FieldValue getval s = FVString $ BS.pack $ tail $ dropWhile (/= '=') s + +-- | Parse streamoffset argument as number or string +mkStreamOffset :: String -> FieldTable +mkStreamOffset s = FieldTable $ M.singleton (pack "x-stream-offset") value + where + value = maybe (FVString $ BS.pack s) FVInt64 $ readMaybe s -- | 'parseargs' exename argstring -- applies options onto argstring diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -25,6 +25,7 @@ -f INT --prefetch=INT Prefetch count. (0=unlimited, 1=off, default: 1) -A --ack Toggle ack messages (default: True) -R --requeuenack Toggle requeue when rejected (default: True) + --stream_offset=OFFSET x-stream-offset consumer argument -l INT --charlimit=INT limit number of shown body chars (default: unlimited) -q QUEUENAME --queue=QUEUENAME Ignore Exchange and bind to existing Queue -i --simple call callback with one arg (filename) only @@ -42,6 +43,7 @@ -s INT --heartbeats=INT heartbeat interval (0=disable, default: set by server) -n NAME --name=NAME connection name, will be shown in RabbitMQ web interface -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 600) + -D SECONDS --delaynack=SECONDS delay negative acknowledgements (default: 0) ### EXAMPLES diff --git a/agitprop.hs b/agitprop.hs --- a/agitprop.hs +++ b/agitprop.hs @@ -33,12 +33,15 @@ import System.INotify #endif import qualified System.Posix.Files.ByteString as F +import System.IO main :: IO () main = do hr "starting" tid <- myThreadId args <- getArgs >>= parseargs 'a' + hSetBuffering stdout LineBuffering + hSetBuffering stderr LineBuffering printparam "client version" ["amqp-utils", showVersion version] printparam "routing key" $ rKey args printparam "exchange" $ currentExchange args diff --git a/amqp-utils.cabal b/amqp-utils.cabal --- a/amqp-utils.cabal +++ b/amqp-utils.cabal @@ -4,7 +4,7 @@ name: amqp-utils -version: 0.6.4.0 +version: 0.6.6.0 synopsis: AMQP toolset for the command line @@ -34,8 +34,6 @@ extra-source-files: ChangeLog.md, README.md cabal-version: >=1.10 - -Tested-With: GHC ==7.10.2 || ==8.0.2 || ==8.2.2 || ==8.4.4 || ==8.6.5 || >=8.8.4 executable konsum main-is: konsum.hs diff --git a/arbeite.hs b/arbeite.hs --- a/arbeite.hs +++ b/arbeite.hs @@ -21,12 +21,15 @@ import Network.AMQP.Utils.Options import Paths_amqp_utils (version) import System.Environment +import System.IO main :: IO () main = do hr "starting" tid <- myThreadId args <- getArgs >>= parseargs 'r' + hSetBuffering stdout LineBuffering + hSetBuffering stderr LineBuffering X.onException (printparam "worker" $ fromJust $ fileProcess args) (error "-X option required") diff --git a/konsum.hs b/konsum.hs --- a/konsum.hs +++ b/konsum.hs @@ -14,12 +14,15 @@ import Network.AMQP.Utils.Options import Paths_amqp_utils (version) import System.Environment +import System.IO main :: IO () main = do hr "starting" tid <- myThreadId args <- getArgs >>= parseargs 'k' + hSetBuffering stdout LineBuffering + hSetBuffering stderr LineBuffering let addiArgs = reverse $ additionalArgs args printparam "client version" ["amqp-utils", showVersion version] (conn, chan) <- connect args @@ -34,6 +37,7 @@ (return) (fmap T.pack (qName args)) printparam "queue name" queue + printparam "consumer args" $ formatheaders fieldshow $ streamoffset args printparam "shown body chars" $ anRiss args printparam "temp dir" $ tempDir args printparam "callback" $ fileProcess args @@ -42,16 +46,20 @@ maybe Nothing (\_ -> Just (cleanupTmpFile args)) (fileProcess args) -- subscribe to the queue ctag <- - consumeMsgs + consumeMsgs' chan queue (if ack args then Ack else NoAck) (myCallback args addiArgs tid) + (\_ -> return ()) + (streamoffset args) printparam "consumer tag" ctag printparam "send acks" $ ack args printparam "requeue if rejected" $ (ack args) && (requeuenack args) + printparam "delay negative acknowledgements for" $ if delaynack args == 0 then Nothing else Just [(show (delaynack args)),"s"] + hr "entering main loop" sleepingBeauty >>= (\x -> do diff --git a/plane.hs b/plane.hs --- a/plane.hs +++ b/plane.hs @@ -27,6 +27,8 @@ hr "starting" tid <- myThreadId args <- getArgs >>= parseargs 'p' + hSetBuffering stdout LineBuffering + hSetBuffering stderr LineBuffering X.onException (printparam "rpc_timeout" [show (rpc_timeout args), "s"]) (error $ "invalid rpc_timeout")