amqp-utils 0.2.1.5 → 0.3.0.0
raw patch · 3 files changed
+73/−50 lines, 3 files
Files
- ChangeLog.md +20/−1
- amqp-utils.cabal +1/−1
- konsum.hs +52/−48
ChangeLog.md view
@@ -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.
amqp-utils.cabal view
@@ -1,6 +1,6 @@ name: amqp-utils -version: 0.2.1.5+version: 0.3.0.0 synopsis: Generic Haskell AMQP Consumer
konsum.hs view
@@ -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