amqp-utils 0.6.4.0 → 0.6.6.0
raw patch · 10 files changed
+65/−5 lines, 10 files
Files
- ChangeLog.md +10/−0
- Network/AMQP/Utils/Connection.hs +1/−0
- Network/AMQP/Utils/Helpers.hs +8/−1
- Network/AMQP/Utils/Options.hs +26/−0
- README.md +2/−0
- agitprop.hs +3/−0
- amqp-utils.cabal +1/−3
- arbeite.hs +3/−0
- konsum.hs +9/−1
- plane.hs +2/−0
ChangeLog.md view
@@ -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
Network/AMQP/Utils/Connection.hs view
@@ -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 =
Network/AMQP/Utils/Helpers.hs view
@@ -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)
Network/AMQP/Utils/Options.hs view
@@ -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
README.md view
@@ -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
agitprop.hs view
@@ -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
amqp-utils.cabal view
@@ -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
arbeite.hs view
@@ -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")
konsum.hs view
@@ -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
plane.hs view
@@ -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")