amqp-utils 0.4.4.1 → 0.4.5.0
raw patch · 10 files changed
+245/−216 lines, 10 files
Files
- ChangeLog.md +8/−0
- Network/AMQP/Utils/Connection.hs +12/−12
- Network/AMQP/Utils/Helpers.hs +35/−34
- Network/AMQP/Utils/Options.hs +72/−56
- README.md +13/−11
- agitprop.hs +57/−49
- amqp-utils.cabal +1/−1
- arbeite.hs +17/−20
- konsum.hs +13/−16
- plane.hs +17/−17
ChangeLog.md view
@@ -1,5 +1,13 @@ # Revision history for haskell-amqp-utils +## 0.4.5.0 -- 2020-02-10++* format with stylish-haskell+* agitprop: rename variables in code+* new options -d (--dirscan) and -u (--remove) for agitprop hotfolder mode+* cwd to hotfolder directory in agitprop hotfolder mode+* change logging of boolean parameters+ ## 0.4.4.1 -- 2020-08-19 * change default connection timeout from 60 to 600s
Network/AMQP/Utils/Connection.hs view
@@ -2,17 +2,17 @@ module Network.AMQP.Utils.Connection where -import qualified Data.ByteString as B-import Data.Default.Class-import qualified Data.Text as T-import Network.AMQP-import Network.AMQP.Utils.Helpers-import Network.AMQP.Utils.Options-import qualified Network.Connection as N-import Network.TLS-import Network.TLS.Extra-import System.Timeout-import System.X509+import qualified Data.ByteString as B+import Data.Default.Class+import qualified Data.Text as T+import Network.AMQP+import Network.AMQP.Utils.Helpers+import Network.AMQP.Utils.Options+import qualified Network.Connection as N+import Network.TLS+import Network.TLS.Extra+import System.Timeout+import System.X509 -- | opens a connection and a channel connect :: Args -> IO (Connection, Channel)@@ -74,6 +74,6 @@ myCert (Just cert') (Just key') _ = do result <- credentialLoadX509 cert' key' case result of- Left x -> printparam "ERROR" x >> return Nothing+ Left x -> printparam "ERROR" x >> return Nothing Right x -> return $ Just x myCert _ _ _ = return Nothing
Network/AMQP/Utils/Helpers.hs view
@@ -2,28 +2,28 @@ module Network.AMQP.Utils.Helpers where -import Control.Concurrent-import qualified Control.Exception as X-import Control.Monad+import Control.Concurrent+import qualified Control.Exception as X+import Control.Monad import qualified Data.ByteString.Lazy.Char8 as BL-import qualified Data.ByteString.UTF8 as BS-import Data.Int (Int64)-import Data.List-import qualified Data.Map as M-import Data.Maybe-import qualified Data.Text as T-import Data.Time-import Data.Time.Clock.POSIX-import Data.Word (Word16)-import Network.AMQP-import Network.AMQP.Types-import Network.AMQP.Utils.Options-import Network.Socket (PortNumber)-import System.Directory (removeFile)-import System.Environment (getEnvironment)-import System.Exit-import System.IO-import System.Process+import qualified Data.ByteString.UTF8 as BS+import Data.Int (Int64)+import Data.List+import qualified Data.Map as M+import Data.Maybe+import qualified Data.Text as T+import Data.Time+import Data.Time.Clock.POSIX+import Data.Word (Word16)+import Network.AMQP+import Network.AMQP.Types+import Network.AMQP.Utils.Options+import Network.Socket (PortNumber)+import System.Directory (removeFile)+import System.Environment (getEnvironment)+import System.Exit+import System.IO+import System.Process -- | print config parameters class (Show a) =>@@ -44,7 +44,7 @@ instance (Flexprint a) => Flexprint (Maybe a) where empty = isNothing- printparam _ Nothing = return ()+ printparam _ Nothing = return () printparam x (Just y) = printparam x y instance Flexprint String where@@ -63,7 +63,8 @@ flexprint x = hPutStrLn stderr "" >> BL.hPut stderr x >> hPutStrLn stderr "" empty = BL.null -instance Flexprint Bool+instance Flexprint Bool where+ empty = not instance Flexprint Int @@ -119,13 +120,13 @@ -- | showing a FieldValue valueshow :: FieldValue -> String valueshow (FVString value) = BS.toString value-valueshow (FVInt8 value) = show value-valueshow (FVInt16 value) = show value-valueshow (FVInt32 value) = show value-valueshow (FVInt64 value) = show value-valueshow (FVFloat value) = show value+valueshow (FVInt8 value) = show value+valueshow (FVInt16 value) = show value+valueshow (FVInt32 value) = show value+valueshow (FVInt64 value) = show value+valueshow (FVFloat value) = show value valueshow (FVDouble value) = show value-valueshow value = show value+valueshow value = show value -- | skip showing body head if binary type isimage :: Maybe String -> Bool@@ -140,11 +141,11 @@ anriss' x = case x of Nothing -> id- Just y -> BL.take y+ Just y -> BL.take y -- | callback cmdline with optional parameters printopt :: (String, Maybe String) -> [String]-printopt (_, Nothing) = []+printopt (_, Nothing) = [] printopt (opt, Just s) = [opt, s] -- | prints header and head on stderr and returns@@ -196,7 +197,7 @@ return (cmdline, environment) where step (_, Nothing) xs = xs- step (k, Just v) xs = ("AMQP_" ++ k, v) : xs+ step (k, Just v) xs = ("AMQP_" ++ k, v) : xs cmdline = concat (map@@ -250,7 +251,7 @@ now' = case timediff of Just "now" -> Nothing- _ -> showtime tz $ Just nowutc+ _ -> showtime tz $ Just nowutc timestamp' = showtime tz msgtimeutc timestamp'' = liftM3@@ -349,7 +350,7 @@ if isJust action && isJust sout then ((fromJust action $ exitcode) (fromJust sout)) >> acke envi a else case exitcode of- ExitSuccess -> acke envi a+ ExitSuccess -> acke envi a ExitFailure _ -> reje envi a if (cleanupTmpFile a) then X.catch
Network/AMQP/Utils/Options.hs view
@@ -1,18 +1,18 @@ module Network.AMQP.Utils.Options where import qualified Data.ByteString.Char8 as BS-import Data.Default.Class-import Data.Int (Int64)-import qualified Data.Map as M-import Data.Maybe-import Data.Text (Text, pack)-import Data.Version (showVersion)-import Data.Word (Word16)-import Network.AMQP-import Network.AMQP.Types-import Network.Socket (PortNumber)-import Paths_amqp_utils (version)-import System.Console.GetOpt+import Data.Default.Class+import Data.Int (Int64)+import qualified Data.Map as M+import Data.Maybe+import Data.Text (Text, pack)+import Data.Version (showVersion)+import Data.Word (Word16)+import Network.AMQP+import Network.AMQP.Types+import Network.Socket (PortNumber)+import Paths_amqp_utils (version)+import System.Console.GetOpt portnumber :: Args -> PortNumber portnumber a@@ -23,52 +23,54 @@ -- | A data type for our options data Args = Args- { server :: String- , port :: Maybe PortNumber- , tls :: Bool- , vHost :: String+ { server :: String+ , port :: Maybe PortNumber+ , tls :: Bool+ , vHost :: String , currentExchange :: String- , bindings :: [(String, String)]- , rKey :: String- , anRiss :: Maybe Int64- , fileProcess :: Maybe String- , qName :: Maybe String- , cert :: Maybe String- , key :: Maybe String- , user :: String- , pass :: String- , preFetch :: Word16- , heartBeat :: Maybe Word16- , tempDir :: Maybe String- , additionalArgs :: [String]- , connectionName :: Maybe String- , tmpQName :: String- , inputFile :: String- , outputFile :: String- , lineMode :: Bool- , confirm :: Bool- , msgid :: Maybe Text- , msgtype :: Maybe Text- , userid :: Maybe Text- , appid :: Maybe Text- , clusterid :: Maybe Text- , contenttype :: Maybe Text+ , bindings :: [(String, String)]+ , rKey :: String+ , anRiss :: Maybe Int64+ , fileProcess :: Maybe String+ , qName :: Maybe String+ , cert :: Maybe String+ , key :: Maybe String+ , user :: String+ , pass :: String+ , preFetch :: Word16+ , heartBeat :: Maybe Word16+ , tempDir :: Maybe String+ , additionalArgs :: [String]+ , connectionName :: Maybe String+ , tmpQName :: String+ , inputFile :: String+ , outputFile :: String+ , lineMode :: Bool+ , confirm :: Bool+ , msgid :: Maybe Text+ , msgtype :: Maybe Text+ , userid :: Maybe Text+ , appid :: Maybe Text+ , clusterid :: Maybe Text+ , contenttype :: Maybe Text , contentencoding :: Maybe Text- , replyto :: Maybe Text- , prio :: Maybe Octet- , corrid :: Maybe Text- , msgexp :: Maybe Text- , msgheader :: Maybe FieldTable- , fnheader :: [String]- , suffix :: [String]- , magic :: Bool- , persistent :: Maybe DeliveryMode- , ack :: Bool- , requeuenack :: Bool- , rpc_timeout :: Double+ , replyto :: Maybe Text+ , prio :: Maybe Octet+ , corrid :: Maybe Text+ , msgexp :: Maybe Text+ , msgheader :: Maybe FieldTable+ , fnheader :: [String]+ , suffix :: [String]+ , magic :: Bool+ , persistent :: Maybe DeliveryMode+ , ack :: Bool+ , requeuenack :: Bool+ , rpc_timeout :: Double , connect_timeout :: Int- , simple :: Bool- , cleanupTmpFile :: Bool+ , simple :: Bool+ , cleanupTmpFile :: Bool+ , removeSentFile :: Bool+ , initialScan :: Bool } instance Default Args where@@ -120,6 +122,8 @@ 600 False False+ False+ False -- | all options allOptions :: [(String, OptDescr (Args -> Args))]@@ -302,6 +306,18 @@ "Allowed file suffixes in hotfolder mode") , ( "a" , Option+ ['u']+ ["remove"]+ (NoArg (\o -> o {removeSentFile = not (removeSentFile o)}))+ ("Toggle removal of sent file in hotfolder mode (default: " ++ show (removeSentFile def) ++ ")"))+ , ( "a"+ , Option+ ['d']+ ["dirscan"]+ (NoArg (\o -> o {initialScan = not (initialScan o)}))+ ("Toggle initial directory scan in hotfolder mode (default: " ++ show (initialScan def) ++ ")"))+ , ( "a"+ , Option ['m'] ["magic"] (NoArg (\o -> o {magic = not (magic o)}))@@ -473,4 +489,4 @@ longname 'k' = "konsum" longname 'r' = "arbeite" longname 'p' = "plane"-longname _ = "command"+longname _ = "command"
README.md view
@@ -44,7 +44,7 @@ -P PASSWORD --pass=PASSWORD Password for Auth -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: 60)+ -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 600) ### examples @@ -98,6 +98,8 @@ -h HEADER=VALUE --header=HEADER=VALUE Message Headers -F HEADERNAME --fnheader=HEADERNAME Put filename into this header -S SUFFIX --suffix=SUFFIX Allowed file suffixes in hotfolder mode+ -u --remove Toggle removal of sent file in hotfolder mode (default: False)+ -d --dirscan Toggle initial directory scan in hotfolder mode (default: False) -m --magic Toggle setting content-type and -encoding from file contents (default: False) -e --persistent Set persistent delivery -E --nonpersistent Set nonpersistent delivery@@ -112,15 +114,15 @@ -P PASSWORD --pass=PASSWORD Password for Auth -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: 60)+ -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 600) -If INPUTFILE is a file, the file is sent as a message and the program-exits. If INPUTFILE is a directory, the directory is watched via inotify-and every file, which is written and closed or moved in gets sent,-optionally only files which match one or several SUFFIXes. Optionally-the file name is written into a message header named HEADERNAME.-Optionally Content-Type and Content-Encoding headers are set via magic-retrieved from file contents.+If INPUTFILE is a file, the file is sent as a message and the program exits. If+INPUTFILE is a directory, the directory is watched via inotify ("hotfolder+mode") and every file, which is written and closed or moved in gets sent,+optionally only files which match one or several SUFFIXes. Optionally the file+name is written into a message header named HEADERNAME. Optionally+Content-Type and Content-Encoding headers are set via magic retrieved from file+contents. Line-by-line mode sends one message per INPUTFILE line. @@ -156,7 +158,7 @@ -P PASSWORD --pass=PASSWORD Password for Auth -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: 60)+ -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 600) ### examples @@ -191,7 +193,7 @@ -P PASSWORD --pass=PASSWORD Password for Auth -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: 60)+ -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 600) ### examples
agitprop.hs view
@@ -1,31 +1,33 @@-{-# LANGUAGE CPP #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-} -- generic AMQP publisher-import Control.Concurrent-import qualified Control.Exception as X-import Control.Monad (forever)-import qualified Data.ByteString.Lazy.Char8 as BL+import Control.Concurrent+import qualified Control.Exception as X+import Control.Monad (forever)+import qualified Data.ByteString.Lazy.Char8 as BL #if MIN_VERSION_hinotify(0,3,10)-import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Char8 as BS #endif-import Data.List (isSuffixOf)-import Data.Maybe-import qualified Data.Text as T-import Data.Time-import Data.Time.Clock.POSIX-import Data.Version (showVersion)-import Data.Word (Word64)-import Magic-import Network.AMQP-import Network.AMQP.Types-import Network.AMQP.Utils.Connection-import Network.AMQP.Utils.Helpers-import Network.AMQP.Utils.Options-import Paths_amqp_utils (version)-import System.Environment-import System.Exit-import System.INotify-import qualified System.Posix.Files as F+import Data.List (isSuffixOf)+import Data.Maybe+import qualified Data.Text as T+import Data.Time+import Data.Time.Clock.POSIX+import Data.Version (showVersion)+import Data.Word (Word64)+import Magic+import Network.AMQP+import Network.AMQP.Types+import Network.AMQP.Utils.Connection+import Network.AMQP.Utils.Helpers+import Network.AMQP.Utils.Options+import Paths_amqp_utils (version)+import System.Directory+import System.Environment+import System.Exit+import System.INotify+import qualified System.Posix.Files as F main :: IO () main = do@@ -40,7 +42,9 @@ then return False else F.getFileStatus (inputFile args) >>= return . F.isDirectory if isDir- then printparam "hotfolder" $ inputFile args+ then+ printparam "hotfolder" (inputFile args) >>+ printparam "initial scan" (initialScan args) else printparam "input file" [ inputFile args@@ -48,6 +52,7 @@ then "(line-by-line)" else "" ]+ printparam "remove sent file" (removeSentFile args && isDir) (conn, chan) <- connect args addChannelExceptionHandler chan (X.throwTo tid) printparam "confirm mode" $ confirm args@@ -56,21 +61,21 @@ confirmSelect chan False addConfirmationListener chan confirmCallback else return ()- let publishOneMsg = publishOneMsg' chan args+ let publishOneMsg = publishOneMsg' chan args {removeSentFile = removeSentFile args && isDir} X.catch (if isDir then do+ setCurrentDirectory (inputFile args)+ if (initialScan args)+ then getDirectoryContents "." >>= mapM_ (\fn -> handleFile publishOneMsg (suffix args) fn)+ else return() inotify <- initINotify wd <- addWatch inotify [CloseWrite, MoveIn]-#if MIN_VERSION_hinotify(0,3,10)- (BS.pack (inputFile args))-#else- (inputFile args)-#endif- (handleEvent publishOneMsg (suffix args) (inputFile args))+ "."+ (handleEvent publishOneMsg (suffix args)) hr $ "BEGIN watching " ++ (inputFile args) _ <- forever $ threadDelay 1000000 removeWatch wd@@ -116,37 +121,36 @@ handleEvent :: (Maybe String -> BL.ByteString -> IO ()) -> [String]- -> String -> Event -> IO () -- just handle closewrite and movedin events #if MIN_VERSION_hinotify(0,3,10)-handleEvent f s p (Closed False (Just x) True) =- handleFile f s (p ++ "/" ++ (BS.unpack x))-handleEvent f s p (MovedIn False x _) =- handleFile f s (p ++ "/" ++ (BS.unpack x))+handleEvent func suffixes (Closed False (Just fileName) True) =+ handleFile func suffixes (BS.unpack fileName)+handleEvent func suffixes (MovedIn False fileName _) =+ handleFile func suffixes (BS.unpack fileName) #else-handleEvent f s p (Closed False (Just x) True) = handleFile f s (p ++ "/" ++ x)-handleEvent f s p (MovedIn False x _) = handleFile f s (p ++ "/" ++ x)+handleEvent func suffixes (Closed False (Just fileName) True) = handleFile func suffixes fileName+handleEvent func suffixes (MovedIn False fileName _) = handleFile func suffixes fileName #endif-handleEvent _ _ _ _ = return ()+handleEvent _ _ _ = return () -- | Hotfolder file handler handleFile :: (Maybe String -> BL.ByteString -> IO ()) -> [String] -> FilePath -> IO () handleFile _ _ ('.':_) = return () -- ignore hidden files-handleFile f s@(_:_) x =- if any (flip isSuffixOf x) s- then handleFile f [] x+handleFile func suffixes@(_:_) fileName =+ if any (flip isSuffixOf fileName) suffixes+ then handleFile func [] fileName else return ()-handleFile f [] x =+handleFile func [] fileName = X.catch- (BL.readFile x >>= f (Just x))+ (BL.readFile fileName >>= func (Just fileName)) (\e -> printparam "exception in handleFile" (e :: X.SomeException)) -- | Publish one message with our settings-publishOneMsg' :: Channel -> Args -> Maybe String -> BL.ByteString -> IO ()-publishOneMsg' c a fn f = do+publishOneMsg' :: Channel -> Args -> Maybe FilePath -> BL.ByteString -> IO ()+publishOneMsg' chan a fn content = do printparam "sending" fn (mtype, mencoding) <- if (magic a) && isJust fn@@ -160,11 +164,11 @@ else return ((contenttype a), (contentencoding a)) now <- getCurrentTime >>= return . floor . utcTimeToPOSIXSeconds publishMsg- c+ chan (T.pack $ currentExchange a) (T.pack $ rKey a) newMsg- { msgBody = f+ { msgBody = content , msgDeliveryMode = persistent a , msgTimestamp = Just now , msgID = msgid a@@ -181,9 +185,13 @@ , msgHeaders = substheader (fnheader a) fn $ msgheader a } >>= printparam "sent"+ removeSentFileIfRequested (removeSentFile a) fn where substheader :: [String] -> Maybe String -> Maybe FieldTable -> Maybe FieldTable substheader (s:r) (Just fname) old = substheader r (Just fname) (addheader old (s ++ "=" ++ fname)) substheader _ _ old = old+ removeSentFileIfRequested False _ = return ()+ removeSentFileIfRequested True Nothing = return ()+ removeSentFileIfRequested True (Just fname) = printparam "removing" fname >> removeFile fname
amqp-utils.cabal view
@@ -1,6 +1,6 @@ name: amqp-utils -version: 0.4.4.1+version: 0.4.5.0 synopsis: AMQP toolset for the command line
arbeite.hs view
@@ -1,22 +1,22 @@ {-# LANGUAGE OverloadedStrings #-} -- generic AMQP rpc server-import Control.Concurrent-import qualified Control.Exception as X-import Control.Monad-import qualified Data.ByteString.Char8 as BS-import Data.Map (singleton)-import Data.Maybe-import qualified Data.Text as T-import Data.Time-import Data.Version (showVersion)-import Network.AMQP-import Network.AMQP.Types-import Network.AMQP.Utils.Connection-import Network.AMQP.Utils.Helpers-import Network.AMQP.Utils.Options-import Paths_amqp_utils (version)-import System.Environment+import Control.Concurrent+import qualified Control.Exception as X+import Control.Monad+import qualified Data.ByteString.Char8 as BS+import Data.Map (singleton)+import Data.Maybe+import qualified Data.Text as T+import Data.Time+import Data.Version (showVersion)+import Network.AMQP+import Network.AMQP.Types+import Network.AMQP.Utils.Connection+import Network.AMQP.Utils.Helpers+import Network.AMQP.Utils.Options+import Paths_amqp_utils (version)+import System.Environment main :: IO () main = do@@ -55,10 +55,7 @@ (rpcServerCallback tid args addiArgs chan) printparam "consumer tag" ctag printparam "send acks" $ ack args- printparam "requeue if rejected" $- if (ack args)- then Just (requeuenack args)- else Nothing+ printparam "requeue if rejected" $ (ack args) && (requeuenack args) hr "entering main loop" X.catch (forever $ threadDelay 5000000)
konsum.hs view
@@ -1,16 +1,16 @@ -- generic amqp consumer-import Control.Concurrent-import qualified Control.Exception as X-import Control.Monad-import qualified Data.Text as T-import Data.Time-import Data.Version (showVersion)-import Network.AMQP-import Network.AMQP.Utils.Connection-import Network.AMQP.Utils.Helpers-import Network.AMQP.Utils.Options-import Paths_amqp_utils (version)-import System.Environment+import Control.Concurrent+import qualified Control.Exception as X+import Control.Monad+import qualified Data.Text as T+import Data.Time+import Data.Version (showVersion)+import Network.AMQP+import Network.AMQP.Utils.Connection+import Network.AMQP.Utils.Helpers+import Network.AMQP.Utils.Options+import Paths_amqp_utils (version)+import System.Environment main :: IO () main = do@@ -48,10 +48,7 @@ (myCallback args addiArgs tid) printparam "consumer tag" ctag printparam "send acks" $ ack args- printparam "requeue if rejected" $- if (ack args)- then Just (show (requeuenack args))- else Nothing+ printparam "requeue if rejected" $ (ack args) && (requeuenack args) hr "entering main loop" X.catch (forever $ threadDelay 5000000)
plane.hs view
@@ -1,22 +1,22 @@ {-# LANGUAGE OverloadedStrings #-} -- generic AMQP rpc client-import Control.Concurrent-import qualified Control.Exception as X-import Control.Monad-import qualified Data.ByteString.Lazy.Char8 as BL-import qualified Data.Text as T-import Data.Time-import Data.Time.Clock.POSIX-import Data.Version (showVersion)-import Network.AMQP-import Network.AMQP.Utils.Connection-import Network.AMQP.Utils.Helpers-import Network.AMQP.Utils.Options-import Paths_amqp_utils (version)-import System.Environment-import System.Exit-import System.IO+import Control.Concurrent+import qualified Control.Exception as X+import Control.Monad+import qualified Data.ByteString.Lazy.Char8 as BL+import qualified Data.Text as T+import Data.Time+import Data.Time.Clock.POSIX+import Data.Version (showVersion)+import Network.AMQP+import Network.AMQP.Utils.Connection+import Network.AMQP.Utils.Helpers+import Network.AMQP.Utils.Options+import Paths_amqp_utils (version)+import System.Environment+import System.Exit+import System.IO main :: IO () main = do@@ -79,7 +79,7 @@ exceptionHandler :: RpcException -> IO (ExitCode) exceptionHandler ReceivedException = return ExitSuccess-exceptionHandler TimeoutException = return $ ExitFailure 1+exceptionHandler TimeoutException = return $ ExitFailure 1 rpcClientCallback :: Handle -> ThreadId -> Args -> (Message, Envelope) -> IO () rpcClientCallback h tid a m@(_, env) = do