diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,13 @@ # Revision history for haskell-amqp-utils +## 0.3.3.1 -- 2018-07-10 + +* fix debian builds +* konsum options ack and requeuenack +* fix hotfolder mode +* enable parallel build +* tls, hinotify, lts-12.0 compat + ## 0.3.2.0 -- 2018-07-04 * agitprop, a publisher diff --git a/Network/AMQP/Utils/Connection.hs b/Network/AMQP/Utils/Connection.hs --- a/Network/AMQP/Utils/Connection.hs +++ b/Network/AMQP/Utils/Connection.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE OverloadedStrings, CPP #-} module Network.AMQP.Utils.Connection where @@ -29,7 +29,15 @@ { sharedValidationCache = def , sharedCAStore = globalCertificateStore } - , clientSupported = def {supportedCiphers = ciphersuite_default} + , clientSupported = + def + { supportedCiphers = +#if MIN_VERSION_tls(1,3,9) + ciphersuite_default +#else + ciphersuite_all +#endif + } , clientHooks = def {onCertificateRequest = myCert (cert args) (key args)} } diff --git a/Network/AMQP/Utils/Options.hs b/Network/AMQP/Utils/Options.hs --- a/Network/AMQP/Utils/Options.hs +++ b/Network/AMQP/Utils/Options.hs @@ -51,6 +51,8 @@ , suffix :: [String] , magic :: Bool , persistent :: Maybe DeliveryMode + , ack :: Bool + , requeuenack :: Bool } instance Default Args where @@ -95,6 +97,8 @@ [] False Nothing + True + True -- | Common options cOptions :: [OptDescr (Args -> Args)] @@ -208,6 +212,17 @@ (ReqArg (\s o -> o {preFetch = read s}) "INT") ("Prefetch count. (0=unlimited, 1=off, default: " ++ show (preFetch def) ++ ")") + , Option + ['A'] + ["ack"] + (NoArg (\o -> o {ack = not (ack o)})) + ("Toggle ack messages (default: " ++ show (ack def) ++ ")") + , Option + ['R'] + ["requeuenack"] + (NoArg (\o -> o {requeuenack = not (requeuenack o)})) + ("Toggle requeue when rejected (default: " ++ + show (requeuenack def) ++ ")") ] -- | Options for agitprop diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -19,6 +19,8 @@ -l INT --charlimit=INT limit number of shown body chars (default: unlimited) -t[DIR] --tempdir[=DIR], --target[=DIR] tempdir (default: no file creation, -t without arg: /tmp) -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) -o SERVER --server=SERVER AMQP Server (default: localhost) -y VHOST --vhost=VHOST AMQP Virtual Host (default: /) -x EXCHANGE --exchange=EXCHANGE AMQP Exchange (default: default) diff --git a/agitprop.hs b/agitprop.hs --- a/agitprop.hs +++ b/agitprop.hs @@ -1,9 +1,12 @@ +{-# LANGUAGE CPP #-} -- generic AMQP publisher - import Control.Concurrent (threadDelay) 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 +#endif import Data.List (isSuffixOf) import Data.Maybe import qualified Data.Text as T @@ -52,23 +55,30 @@ addWatch inotify [CloseWrite, MoveIn] +#if MIN_VERSION_hinotify(0,3,10) + (BS.pack (inputFile args)) +#else (inputFile args) - (handleEvent publishOneMsg (suffix args)) - hr $ "watching " ++ (inputFile args) +#endif + (handleEvent publishOneMsg (suffix args) (inputFile args)) + hr $ "BEGIN watching " ++ (inputFile args) _ <- forever $ threadDelay 1000000 removeWatch wd + hr $ "END watching " ++ (inputFile args) else do - hr $ "sending " ++ (inputFile args) + hr $ "BEGIN sending" messageFile <- BL.readFile (inputFile args) if (lineMode args) then mapM_ (publishOneMsg Nothing) (BL.lines messageFile) - else publishOneMsg (Just (inputFile args)) messageFile) + else publishOneMsg (Just (inputFile args)) messageFile + hr "END sending") (\exception -> printparam' "exception" $ show (exception :: X.SomeException)) -- all done. wait and close. if (confirm args) - then waitForConfirms chan >>= return . show >> return () + then waitForConfirms chan >>= (printparam' "confirmed") . show else return () closeConnection conn + hr "connection closed" -- | The handler for publisher confirms confirmCallback :: (Word64, Bool, AckType) -> IO () @@ -83,11 +93,20 @@ -- | Hotfolder event handler handleEvent :: - (Maybe String -> BL.ByteString -> IO ()) -> [String] -> Event -> IO () + (Maybe String -> BL.ByteString -> IO ()) + -> [String] + -> String + -> Event + -> IO () -- just handle closewrite and movedin events -handleEvent f s (Closed False (Just x) True) = handleFile f s x -handleEvent f s (MovedIn False x _) = handleFile f s x -handleEvent _ _ _ = return () +#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)) +#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) +#endif +handleEvent _ _ _ _ = return () -- | Hotfolder file handler handleFile :: @@ -99,7 +118,7 @@ else return () handleFile f [] x = X.catch - (hr ("sending " ++ x) >> BL.readFile x >>= f (Just x)) + (BL.readFile x >>= f (Just x)) (\exception -> printparam' "exception in handleFile" $ show (exception :: X.SomeException)) @@ -107,6 +126,7 @@ -- | Publish one message with our settings publishOneMsg' :: Channel -> Args -> Maybe String -> BL.ByteString -> IO () publishOneMsg' c a fn f = do + printparam "sending" fn (mtype, mencoding) <- if (magic a) && isJust fn then do 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.3.2.1 +version: 0.3.3.1 synopsis: Generic Haskell AMQP Consumer @@ -29,7 +29,7 @@ cabal-version: >=1.10 -Tested-With: GHC == 7.10.2, GHC == 8.0.2, GHC == 8.2.2 +Tested-With: GHC == 7.10.2, GHC == 8.0.2, GHC == 8.2.2, GHC == 8.4.3 executable konsum main-is: konsum.hs @@ -46,7 +46,7 @@ amqp >=0.17 ghc-options: -threaded -Wall - + default-language: Haskell98 other-modules: Network.AMQP.Utils.Options, @@ -68,7 +68,7 @@ tls, amqp >=0.17, unix >= 2.7, - hinotify >= 0.3.8 && < 0.3.10, + hinotify >= 0.3.8, magic ghc-options: -threaded -Wall diff --git a/konsum.hs b/konsum.hs --- a/konsum.hs +++ b/konsum.hs @@ -1,5 +1,4 @@ -- generic amqp consumer - import Control.Concurrent import qualified Control.Exception as X import Control.Monad @@ -44,9 +43,16 @@ consumeMsgs chan queue - Ack - (myCallback (anRiss args) (fileProcess args) (tempDir args) addiArgs tid) + (if ack args + then Ack + else NoAck) + (myCallback args addiArgs tid) printparam' "consumer tag" $ T.unpack ctag + printparam' "send acks" $ show (ack args) + printparam "requeue if rejected" $ + if (ack args) + then Just (show (requeuenack args)) + else Nothing hr "entering main loop" X.catch (forever $ threadDelay 5000000) @@ -70,24 +76,16 @@ return q -- | process received message -myCallback :: - Maybe Int - -> Maybe String - -> Maybe String - -> [String] - -> ThreadId - -> (Message, Envelope) - -> IO () -myCallback anR filePr tempD addi tid m@(_, envi) = do +myCallback :: Args -> [String] -> ThreadId -> (Message, Envelope) -> IO () +myCallback a addi tid m@(_, envi) = do let numstring = show $ envDeliveryTag envi hr $ "BEGIN " ++ numstring now <- getZonedTime - callbackoptions <- printmsg m anR now + callbackoptions <- printmsg m (anRiss a) now either - (\e -> - printparam' "ERROR" (show (e :: X.SomeException)) >> rejectEnv envi True) + (\e -> printparam' "ERROR" (show (e :: X.SomeException)) >> reje envi a) return =<< - X.try (optionalFileStuff m callbackoptions addi numstring tempD filePr tid) + X.try (optionalFileStuff m callbackoptions addi numstring a tid) hr $ "END " ++ numstring -- | if the message is to be saved @@ -97,23 +95,22 @@ -> [String] -> [String] -> String - -> Maybe String - -> Maybe String + -> Args -> ThreadId -> IO () -optionalFileStuff (msg, envi) callbackoptions addi numstring tempD filePr tid = do - path <- saveFile tempD numstring (msgBody msg) +optionalFileStuff (msg, envi) callbackoptions addi numstring a tid = do + path <- saveFile (tempDir a) numstring (msgBody msg) printparam "saved to" path let callbackcmdline = liftM2 (constructCallbackCmdLine callbackoptions addi numstring) - filePr + (fileProcess a) path printparam "calling" $ fmap unwords callbackcmdline maybe - (ackEnv envi) + (acke envi a) (\c -> - forkFinally (doProc numstring envi c) (either (throwTo tid) return) >> + forkFinally (doProc a numstring envi c) (either (throwTo tid) return) >> return ()) callbackcmdline @@ -136,13 +133,25 @@ exe : "-f" : path : "-n" : num : opts ++ addi -- | call callback script -doProc :: String -> Envelope -> [String] -> IO () -doProc numstring envi (exe:args) = do +doProc :: Args -> String -> Envelope -> [String] -> IO () +doProc a numstring envi (exe:args) = do (_, _, _, processhandle) <- createProcess (proc exe args) {std_out = Inherit, std_err = Inherit} exitcode <- waitForProcess processhandle printparam' (numstring ++ " call returned") $ show exitcode case exitcode of - ExitSuccess -> ackEnv envi - ExitFailure _ -> rejectEnv envi True -doProc _ _ _ = return () + ExitSuccess -> acke envi a + ExitFailure _ -> reje envi a +doProc _ _ _ _ = return () + +-- | ack +acke :: Envelope -> Args -> IO () +acke envi a + | (ack a) = ackEnv envi + | otherwise = return () + +-- | reject +reje :: Envelope -> Args -> IO () +reje envi a + | (ack a) = rejectEnv envi (requeuenack a) + | otherwise = return ()