optionally delay negative acknowledgements

option --delaynack / -D

diff --git a/Network/AMQP/Utils/Helpers.hs b/Network/AMQP/Utils/Helpers.hs
index f2f214326907ad4da4d189eeeaea160424526f4e..fad8aafefb22988fa433f773cb15cded0d1f987b 100644 (file)
--- a/Network/AMQP/Utils/Helpers.hs
+++ b/Network/AMQP/Utils/Helpers.hs
@@ -415,9 +415,16 @@ acke envi a
 -- | 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)
 sleepingBeauty =

diff --git a/Network/AMQP/Utils/Options.hs b/Network/AMQP/Utils/Options.hs
index 22eacf459dbfe13e4db812ede529ef2337c4e1dc..fc733f89a2ac4316ec62505c0d7318a136227cb6 100644 (file)
--- a/Network/AMQP/Utils/Options.hs
+++ b/Network/AMQP/Utils/Options.hs
@@ -81,6 +81,7 @@ data Args =
     , moveSentFileTo  :: Maybe RawFilePath
     , initialScan     :: Bool
     , streamoffset    :: FieldTable
+    , delaynack       :: Int
     }
 
 instance Default Args where
@@ -136,6 +137,7 @@ instance Default Args where
       Nothing
       False
       (FieldTable M.empty)
+      0
 
 -- | all options
 allOptions :: [(String, OptDescr (Args -> Args))]
@@ -466,6 +468,13 @@ allOptions =
         (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

diff --git a/README.md b/README.md
index 547642b92c1f589830343a6279dea0d780823a0b..cd73909cb92c962c29e4d43573cfa1b2e68213d6 100644 (file)
--- a/README.md
+++ b/README.md
@@ -43,6 +43,7 @@ generic AMQP commandline tools for use with RabbitMQ
       -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/doc/konsum.md b/doc/konsum.md
index 563b9283da645314ea86fc99023746ec90dfa200..a8b79a2be1287ceee6026c993b7d0ec8259af75e 100644 (file)
--- a/doc/konsum.md
+++ b/doc/konsum.md
@@ -47,6 +47,7 @@ konsum - an AMQP consumer.
       -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)
 <!--
 SPDX-FileCopyrightText: 2022 Frank Doepper
 

diff --git a/konsum.hs b/konsum.hs
index 5264373a8499de7e12da810004eb3497cfddeeac..711d56509b6622c084baf5c547863c41d164c0d0 100644 (file)
--- a/konsum.hs
+++ b/konsum.hs
@@ -58,6 +58,8 @@ main = do
   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