amqp-utils 0.3.2.1 → 0.3.3.1
raw patch · 7 files changed
+107/−45 lines, 7 filesdep ~hinotify
Dependency ranges changed: hinotify
Files
- ChangeLog.md +8/−0
- Network/AMQP/Utils/Connection.hs +10/−2
- Network/AMQP/Utils/Options.hs +15/−0
- README.md +2/−0
- agitprop.hs +31/−11
- amqp-utils.cabal +4/−4
- konsum.hs +37/−28
ChangeLog.md view
@@ -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
Network/AMQP/Utils/Connection.hs view
@@ -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)} }
Network/AMQP/Utils/Options.hs view
@@ -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
README.md view
@@ -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)
agitprop.hs view
@@ -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
amqp-utils.cabal view
@@ -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
konsum.hs view
@@ -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 ()