diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,8 @@ # Revision history for haskell-amqp-utils +## 0.3.7.1 -- 2019-09-10 +* connect timeout + ## 0.3.6.0 -- 2019-02-15 * bug-fix: avoid deadlock in arbeite 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 @@ -11,6 +11,7 @@ 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 @@ -20,6 +21,7 @@ printparam' "port" $ show $ port args printparam' "vhost" $ vHost args printparam "connection_name" $ connectionName args + printparam' "connect timeout" $ (show (connect_timeout args)) ++ "s" globalCertificateStore <- getSystemCertificateStore let myTLS = N.TLSSettings @@ -33,7 +35,8 @@ , clientHooks = def {onCertificateRequest = myCert (cert args) (key args)} } - conn <- + Just conn <- + timeout to $ openConnection'' defaultConnectionOpts { coAuth = @@ -49,8 +52,10 @@ , coHeartbeatDelay = fmap fromIntegral $ heartBeat args , coName = fmap T.pack $ connectionName args } - chan <- openChannel conn + Just chan <- timeout to $ openChannel conn return (conn, chan) + where + to = connect_timeout args * 1000000 -- addChannelExceptionHandler chan -- (\exception -> closeConnection conn >> @@ -62,6 +67,8 @@ -- noValidation = ValidationCache -- (\_ _ _ -> return ValidationCachePass) -- (\_ _ _ -> return ()) +-- +-- -- | provides the TLS client certificate myCert :: Maybe FilePath -> Maybe FilePath -> t -> IO (Maybe Credential) myCert (Just cert') (Just key') _ = do 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 @@ -54,7 +54,8 @@ , persistent :: Maybe DeliveryMode , ack :: Bool , requeuenack :: Bool - , timeout :: Double + , rpc_timeout :: Double + , connect_timeout :: Int } instance Default Args where @@ -103,6 +104,7 @@ True True 5 + 60 -- | all options allOptions :: [(String, OptDescr (Args -> Args))] @@ -241,9 +243,9 @@ , ( "p" , Option ['t'] - ["timeout"] - (ReqArg (\s o -> o {timeout = read s}) "SECONDS") - ("How long to wait for reply (default: " ++ show (timeout def) ++ ")")) + ["rpc_timeout"] + (ReqArg (\s o -> o {rpc_timeout = read s}) "SECONDS") + ("How long to wait for reply (default: " ++ show (rpc_timeout def) ++ ")")) , ( "a" , Option [] @@ -307,13 +309,13 @@ ["charlimit"] (ReqArg (\s o -> o {anRiss = Just (read s :: Int)}) "INT") "limit number of shown body chars (default: unlimited)") - , ( "akr" + , ( "kr" , Option ['q'] ["queue"] (ReqArg (\s o -> o {qName = Just s}) "QUEUENAME") "Ignore Exchange and bind to existing Queue") - , ( "akrp" + , ( "krp" , Option ['Q'] ["qname"] @@ -385,6 +387,12 @@ ["name"] (ReqArg (\s o -> o {connectionName = Just s}) "NAME") "connection name, will be shown in RabbitMQ web interface") + , ( "akrp" + , Option + ['w'] + ["connect_timeout"] + (ReqArg (\s o -> o {connect_timeout = read s}) "SECONDS") + ("timeout for establishing initial connection (default: " ++ show (connect_timeout def) ++ ")")) ] -- | Options for the executables diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -23,8 +23,6 @@ -A --ack Toggle ack messages (default: True) -R --requeuenack Toggle requeue when rejected (default: True) -l INT --charlimit=INT limit number of shown body chars (default: unlimited) - -q QUEUENAME --queue=QUEUENAME Ignore Exchange and bind to existing Queue - -Q TEMPQNAME --qname=TEMPQNAME Name for temporary exclusive Queue -x EXCHANGE --exchange=EXCHANGE AMQP Exchange (default: "") -o SERVER --server=SERVER AMQP Server (default: localhost) -y VHOST --vhost=VHOST AMQP Virtual Host (default: /) diff --git a/agitprop.hs b/agitprop.hs --- a/agitprop.hs +++ b/agitprop.hs @@ -32,6 +32,7 @@ args <- getArgs >>= parseargs 'a' printparam' "client version" $ "amqp-utils " ++ (showVersion version) printparam' "routing key" $ rKey args + printparam' "exchange" $ currentExchange args isDir <- F.getFileStatus (inputFile args) >>= return . F.isDirectory if isDir then printparam' "hotfolder" $ inputFile args diff --git a/amqp-utils.cabal b/amqp-utils.cabal --- a/amqp-utils.cabal +++ b/amqp-utils.cabal @@ -1,10 +1,11 @@ name: amqp-utils -version: 0.3.6.0 +version: 0.3.7.1 -synopsis: Generic Haskell AMQP Consumer +synopsis: Generic Haskell AMQP tools -description: AMQP consumer which can +description: AMQP tools consisting of: + AMQP consumer which can create a temporary queue and attach it to an exchange, or attach to an existing queue; display header and body info; @@ -12,6 +13,7 @@ call a callback script. AMQP publisher with file, line-by-line and hotfolder capabilities. + AMQP rpc client and server. license: GPL-3 @@ -29,7 +31,7 @@ cabal-version: >=1.10 -Tested-With: GHC == 7.10.2, GHC == 8.0.2, GHC == 8.2.2, GHC == 8.4.4, GHC == 8.6.3 +Tested-With: GHC ==7.10.2 || ==8.0.2 || ==8.2.2 || ==8.4.4 || >= 8.6.4 executable konsum main-is: konsum.hs diff --git a/plane.hs b/plane.hs --- a/plane.hs +++ b/plane.hs @@ -24,8 +24,8 @@ tid <- myThreadId args <- getArgs >>= parseargs 'p' X.onException - (printparam' "timeout" $ show $ timeout args) - (error $ "invalid timeout") + (printparam' "rpc_timeout" $ show (rpc_timeout args) ++ "s") + (error $ "invalid rpc_timeout") printparam' "client version" $ "amqp-utils " ++ (showVersion version) printparam' "destination queue" $ tmpQName args (conn, chan) <- connect args @@ -53,7 +53,7 @@ } hr "waiting for answer" _ <- forkIO - (threadDelay (floor (1000000 * timeout args)) >> + (threadDelay (floor (1000000 * rpc_timeout args)) >> throwTo tid TimeoutException) X.catch (forever $ threadDelay 200000)