diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,24 @@ # Revision history for haskell-amqp-utils -## 0.2.1.4 -- YYYY-mm-dd +## 0.3.0.0 -- 2017-11-21 + +* add nix with `amqp_0_18_1` +* multiple bindings per queue +* remove cool smart special options + +## 0.2.2.0 -- 2017-11-20 + +* option -Q (set queue name for temporary exclusive queue) + +## 0.2.1.5 -- 2017-09-25 + +* travis +* debian +* repair debian jessie +* callback: pass-through timestamp, ignore non-existing sha +* reduce load +* amqp 0.17, use coName + +## 0.2.1.4 -- 2017-06-01 * First version. Released on an unsuspecting world. diff --git a/amqp-utils.cabal b/amqp-utils.cabal --- a/amqp-utils.cabal +++ b/amqp-utils.cabal @@ -1,6 +1,6 @@ name: amqp-utils -version: 0.2.1.5 +version: 0.3.0.0 synopsis: Generic Haskell AMQP Consumer diff --git a/konsum.hs b/konsum.hs --- a/konsum.hs +++ b/konsum.hs @@ -5,7 +5,7 @@ -- -- run: -- ./konsum --- ./konsum -o amqp.example.com -p 5673 -T -k amqp-key.pem -c amqp-crt.pem -y vhost -x exchange -X./callback.sh -a -c -a callback.config.sh -f 2 routing.key.# 500 +-- ./konsum -o amqp.example.com -p 5673 -T -k amqp-key.pem -c amqp-crt.pem -y vhost -x exchange -X./callback.sh -a -c -a callback.config.sh -f 2 -r routing.key.# -l 500 -- ./konsum -o amqp.example.com -U user -P pass -q queue -t -- -- custom CA cert via enviroment: @@ -85,12 +85,12 @@ qos chan 0 (fromIntegral $ preFetch args) False -- attach to given queue? or build exclusive queue and bind it? - queue <- maybe (tempQueue chan (exChange args) (bindingKey args)) + queue <- maybe (tempQueue chan (tmpQName args) (bindings args) (currentExchange args)) (return) (fmap T.pack (qName args)) printparam' "queue name" $ T.unpack queue - printparam "head chars" $ fmap show $ anRiss args + printparam "shown body chars" $ fmap show $ anRiss args printparam "temp dir" $ tempDir args printparam "callback" $ fileProcess args printparam "callback args" $ listToMaybeUnwords addiArgs @@ -128,12 +128,15 @@ myCert _ _ _ = return Nothing -- exclusive temp queue -tempQueue :: Channel -> String -> String -> IO T.Text -tempQueue chan xchange bkey = do - (q, _, _) <- declareQueue chan newQueue { queueExclusive = True } - bindQueue chan q (T.pack xchange) (T.pack bkey) - printparam' "exchange" xchange - printparam' "binding key" bkey +tempQueue :: Channel -> String -> [(String, String)] -> String -> IO T.Text +tempQueue chan tmpqname bindlist x = do + (q, _, _) <- declareQueue chan + newQueue { queueExclusive = True + , queueName = T.pack tmpqname + } + mapM_ (\(xchange, bkey) -> bindQueue chan q (T.pack xchange) (T.pack bkey) >> + printparam' "binding" (xchange ++ ":" ++ bkey)) + (if null bindlist then [ (x, "#") ] else bindlist) return q -- log cmdline options @@ -165,24 +168,25 @@ fmap (\s -> putStrLn "" >> BL.putStrLn s) ms -- options for everybody -data Args = Args { server :: String - , port :: Int - , tls :: Bool - , vHost :: String - , exChange :: String - , bindingKey :: String - , anRiss :: Maybe Int - , fileProcess :: Maybe String - , qName :: Maybe String - , cert :: Maybe String - , key :: Maybe String - , user :: String - , pass :: String - , preFetch :: Int - , heartBeat :: Maybe Int - , tempDir :: Maybe String - , additionalArgs :: [String] - , connectionName :: Maybe String +data Args = Args { server :: String + , port :: Int + , tls :: Bool + , vHost :: String + , currentExchange :: String + , bindings :: [(String, String)] + , anRiss :: Maybe Int + , fileProcess :: Maybe String + , qName :: Maybe String + , cert :: Maybe String + , key :: Maybe String + , user :: String + , pass :: String + , preFetch :: Int + , heartBeat :: Maybe Int + , tempDir :: Maybe String + , additionalArgs :: [String] + , connectionName :: Maybe String + , tmpQName :: String } instance Default Args where @@ -191,7 +195,7 @@ False "/" "default" - "#" + [] Nothing Nothing Nothing @@ -204,6 +208,7 @@ Nothing [] Nothing + "" callback :: String callback = "/usr/lib/haskell-amqp-utils/callback" @@ -219,8 +224,19 @@ ("AMQP Virtual Host (default: " ++ vHost def ++ ")") , Option [ 'x' ] [ "exchange" ] - (ReqArg (\s o -> o { exChange = s }) "EXCHANGE") - ("AMQP Exchange (default: " ++ exChange def ++ ")") + (ReqArg (\s o -> o { currentExchange = s }) "EXCHANGE") + ("AMQP Exchange (default: default)") + , Option [ 'r' ] + [ "bindingkey" ] + (ReqArg (\s o -> o { bindings = (currentExchange o, s) : + (bindings o) + }) + "BINDINGKEY") + ("AMQP binding key (default: #)") + , Option [ 'Q' ] + [ "qname" ] + (ReqArg (\s o -> o { tmpQName = s }) "TEMPQNAME") + "Name for temporary exclusive Queue" , Option [ 'X' ] [ "execute" ] (OptArg (\s o -> o { fileProcess = Just (fromMaybe callback s) @@ -263,10 +279,6 @@ (ReqArg (\s o -> o { preFetch = read s }) "INT") ("Prefetch count. (0=unlimited, 1=off, default: " ++ show (preFetch def) ++ ")") - , Option [ 'r' ] - [ "bindingkey" ] - (ReqArg (\s o -> o { bindingKey = s }) "BINDINGKEY") - ("AMQP binding key (default: " ++ bindingKey def ++ ")") , Option [ 's' ] [ "heartbeats" ] (ReqArg (\s o -> o { heartBeat = (Just (read s)) }) "INT") @@ -286,6 +298,10 @@ [ "name" ] (ReqArg (\s o -> o { connectionName = Just s }) "NAME") "connection name, will be shown in RabbitMQ web interface" + , Option [ 'l' ] + [ "charlimit" ] + (ReqArg (\s o -> o { anRiss = Just (read s :: Int) }) "INT") + "limit number of shown body chars (default: unlimited)" ] usage :: String @@ -297,22 +313,10 @@ \konsum [options] [bindingkey] [body byte count to show]\n\n\ \Options:" --- special options -parsearg :: Args -> String -> Args -parsearg args arg - -- binding keys - | elem '.' arg = args { bindingKey = arg } - | elem '#' arg = args { bindingKey = arg } - | elem '*' arg = args { bindingKey = arg } - -- number -> show how many bytes - | all (flip elem ['0' .. '9']) arg = - args { anRiss = Just (read arg :: Int) } - | otherwise = args - --- apply both options and parsearg onto argstring +-- apply options onto argstring parseargs :: [String] -> IO Args parseargs argstring = case getOpt Permute options argstring of - (o, n, []) -> return $ foldl parsearg (foldl (flip id) def o) n + (o, [], []) -> return $ foldl (flip id) def o (_, _, errs) -> ioError $ userError $ concat errs ++ usageInfo usage options -- process received message