diff --git a/Network/AMQP.hs b/Network/AMQP.hs --- a/Network/AMQP.hs +++ b/Network/AMQP.hs @@ -62,6 +62,7 @@ Channel, openChannel, addReturnListener, + addChannelExceptionHandler, qos, -- * Exchanges diff --git a/Network/AMQP/Helpers.hs b/Network/AMQP/Helpers.hs --- a/Network/AMQP/Helpers.hs +++ b/Network/AMQP/Helpers.hs @@ -2,6 +2,7 @@ module Network.AMQP.Helpers where import Control.Concurrent +import Control.Exception import Control.Monad import Data.Int (Int64) import System.Clock @@ -51,4 +52,10 @@ scheduleAtFixedRate :: Int -> IO () -> IO ThreadId scheduleAtFixedRate interval_µs action = forkIO $ forever $ do action - threadDelay interval_µs + threadDelay interval_µs + +-- | Copy of base's 'forkFinally', to support GHC < 7.6.x +forkFinally' :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId +forkFinally' action and_then = + mask $ \restore -> + forkIO $ try (restore action) >>= and_then diff --git a/Network/AMQP/Internal.hs b/Network/AMQP/Internal.hs --- a/Network/AMQP/Internal.hs +++ b/Network/AMQP/Internal.hs @@ -183,13 +183,16 @@ updateLastReceived conn forwardToChannel chanID payload ) - (\(e :: CE.IOException) -> myThreadId >>= killConnection conn (show e)) + (\(e :: CE.IOException) -> myThreadId >>= killConnection conn (CE.toException e)) connectionReceiver conn where - forwardToChannel 0 (MethodPayload Connection_close_ok) = myThreadId >>= killConnection conn "closed by user" + + closedByUserEx = ConnectionClosedException "closed by user" + + forwardToChannel 0 (MethodPayload Connection_close_ok) = myThreadId >>= killConnection conn (CE.toException closedByUserEx) forwardToChannel 0 (MethodPayload (Connection_close _ (ShortString errorMsg) _ _)) = do writeFrame (connHandle conn) $ Frame 0 $ MethodPayload Connection_close_ok - myThreadId >>= killConnection conn (T.unpack errorMsg) + myThreadId >>= killConnection conn (CE.toException . ConnectionClosedException . T.unpack $ errorMsg) forwardToChannel 0 HeartbeatPayload = return () forwardToChannel 0 payload = putStrLn $ "Got unexpected msg on channel zero: " ++ show payload forwardToChannel chanID payload = do @@ -253,23 +256,30 @@ let conn = Connection handle cChanAllocator cChannels (fromIntegral maxFrameSize) cClosed ccl writeLock cClosedHandlers cLastReceived cLastSent --spawn the connectionReceiver - connThread <- forkIO $ CE.finally (connectionReceiver conn) $ do + connThread <- forkFinally' (connectionReceiver conn) $ \res -> do -- try closing socket CE.catch (Conn.connectionClose handle) (\(_ :: CE.SomeException) -> return ()) -- mark as closed modifyMVar_ cClosed $ return . Just . maybe "unknown reason" id - --kill all channel-threads + -- kill all channel-threads, making sure the channel threads will + -- be killed taking into account the overall state of the + -- connection: if the thread died for an unexpected exception, + -- inform the channel threads downstream accordingly. Otherwise + -- just use a normal 'killThread' finaliser. + let finaliser = case res of + Left ex -> ex + Right _ -> CE.toException CE.ThreadKilled modifyMVar_ cChannels $ \x -> do - mapM_ (killThread . snd) $ IM.elems x + mapM_ (flip CE.throwTo finaliser . snd) $ IM.elems x return IM.empty -- mark connection as closed, so all pending calls to 'closeConnection' can now return void $ tryPutMVar ccl () -- notify connection-close-handlers - withMVar cClosedHandlers sequence + withMVar cClosedHandlers sequence_ case heartbeatTimeout of Nothing -> return () @@ -344,8 +354,10 @@ receiveTimeout = (fromIntegral rate) * 4 * 2 -- 2*timeout in µs sendTimeout = (fromIntegral rate) * 2 -- timeout/2 in µs + skippedBeatEx = ConnectionClosedException "killed connection after missing 2 consecutive heartbeats" + checkReceiveTimeout = check (connLastReceived conn) receiveTimeout - (killConnection conn "killed connection after missing 2 consecutive heartbeats" connThread) + (killConnection conn (CE.toException skippedBeatEx) connThread) checkSendTimeout = check (connLastSent conn) sendTimeout (writeFrame (connHandle conn) (Frame 0 HeartbeatPayload)) @@ -362,10 +374,10 @@ updateLastReceived conn = modifyMVar_ (connLastReceived conn) (const getTimestamp) -- | kill the connection thread abruptly -killConnection :: Connection -> String -> ThreadId -> IO () -killConnection conn msg connThread = do - modifyMVar_ (connClosed conn) $ const $ return $ Just msg - killThread connThread +killConnection :: Connection -> CE.SomeException -> ThreadId -> IO () +killConnection conn ex connThread = do + modifyMVar_ (connClosed conn) $ const $ return $ Just (show ex) + throwTo connThread ex -- | closes a connection -- @@ -453,7 +465,8 @@ chanActive :: Lock, -- used for flow-control. if lock is closed, no content methods will be sent chanClosed :: MVar (Maybe String), consumers :: MVar (M.Map Text ((Message, Envelope) -> IO ())), -- who is consumer of a queue? (consumerTag => callback) - returnListeners :: MVar ([(Message, PublishError) -> IO ()]) + returnListeners :: MVar ([(Message, PublishError) -> IO ()]), + chanExceptionHandlers :: MVar [CE.SomeException -> IO ()] } msgFromContentHeaderProperties :: ContentHeaderProperties -> BL.ByteString -> Message @@ -518,7 +531,7 @@ ) handleAsync (SimpleMethod (Channel_close _ (ShortString errorMsg) _ _)) = do closeChannel' chan errorMsg - killThread =<< myThreadId + myThreadId >>= flip CE.throwTo (ConnectionClosedException . T.unpack $ errorMsg) handleAsync (SimpleMethod (Channel_flow active)) = do if active then openLock $ chanActive chan @@ -549,6 +562,11 @@ addReturnListener chan listener = do modifyMVar_ (returnListeners chan) $ \listeners -> return $ listener:listeners +-- | registers a callback function that is called whenever a channel is closed by an exception. +addChannelExceptionHandler :: Channel -> (CE.SomeException -> IO ()) -> IO () +addChannelExceptionHandler chan handler = do + modifyMVar_ (chanExceptionHandlers chan) $ \handlers -> return $ handler:handlers + -- closes the channel internally; but doesn't tell the server closeChannel' :: Channel -> Text -> IO () closeChannel' c reason = do @@ -581,13 +599,17 @@ closed <- newMVar Nothing conss <- newMVar M.empty listeners <- newMVar [] + handlers <- newMVar [] --add new channel to connection's channel map newChannel <- modifyMVar (connChannels c) $ \mp -> do newChannelID <- allocateChannel (connChanAllocator c) - let newChannel = Channel c newInQueue outRes (fromIntegral newChannelID) lastConsTag ca closed conss listeners - thrID <- forkIO $ CE.finally (channelReceiver newChannel) - (closeChannel' newChannel "closed") + let newChannel = Channel c newInQueue outRes (fromIntegral newChannelID) lastConsTag ca closed conss listeners handlers + thrID <- forkFinally' (channelReceiver newChannel) $ \res -> do + closeChannel' newChannel "closed" + case res of + Right _ -> return () + Left ex -> readMVar handlers >>= mapM_ ($ ex) when (IM.member newChannelID mp) $ CE.throwIO $ userError "openChannel fail: channel already open" return (IM.insert newChannelID (newChannel, thrID) mp, newChannel) @@ -598,7 +620,10 @@ closeChannel :: Channel -> IO () closeChannel c = do SimpleMethod Channel_close_ok <- request c $ SimpleMethod $ Channel_close 0 (ShortString "") 0 0 - closeChannel' c "user called closeChannel" + withMVar (connChannels $ connection c) $ \chans -> do + case IM.lookup (fromIntegral $ channelID c) chans of + Just (_, thrID) -> killThread thrID + Nothing -> return () -- | writes multiple frames to the channel atomically writeFrames :: Channel -> [FramePayload] -> IO () diff --git a/amqp.cabal b/amqp.cabal --- a/amqp.cabal +++ b/amqp.cabal @@ -1,5 +1,5 @@ Name: amqp -Version: 0.11 +Version: 0.12 Synopsis: Client library for AMQP servers (currently only RabbitMQ) Description: Client library for AMQP servers (currently only RabbitMQ) . diff --git a/examples/ExampleConsumer.hs b/examples/ExampleConsumer.hs --- a/examples/ExampleConsumer.hs +++ b/examples/ExampleConsumer.hs @@ -1,6 +1,7 @@ -{-# OPTIONS -XOverloadedStrings #-} +{-# OPTIONS -XOverloadedStrings -XScopedTypeVariables #-} import Network.AMQP - +import Control.Concurrent +import qualified Control.Exception as CE import qualified Data.ByteString.Lazy.Char8 as BL @@ -14,10 +15,18 @@ declareQueue chan newQueue {queueName = "myQueueEN"} declareExchange chan newExchange {exchangeName = "topicExchg", exchangeType = "topic"} - bindQueue chan "myQueueDE" "topicExchg" "de.*" - bindQueue chan "myQueueEN" "topicExchg" "en.*" - + + CE.catch (bindQueue chan "myQueueDE" "e" "de.*") + (\(e::CE.SomeException) -> return ()) + print "B" + -- bindQueue chan "myQueueEN" "topicExchg" "en.*" + print "c" + addChannelExceptionHandler chan $ \e -> do + threadDelay 1000000 + print "OOOOOOOO" + + print "A" --subscribe to the queues consumeMsgs chan "myQueueDE" Ack myCallbackDE consumeMsgs chan "myQueueEN" Ack myCallbackEN diff --git a/examples/ExampleProducer.hs b/examples/ExampleProducer.hs --- a/examples/ExampleProducer.hs +++ b/examples/ExampleProducer.hs @@ -1,6 +1,6 @@ {-# OPTIONS -XOverloadedStrings #-} import Network.AMQP - +import Control.Concurrent import qualified Data.ByteString.Lazy.Char8 as BL @@ -8,13 +8,16 @@ conn <- openConnection "127.0.0.1" "/" "guest" "guest" chan <- openChannel conn + addChannelExceptionHandler chan $ \e -> do + threadDelay 1000000 + print e --declare queues, exchanges and bindings declareQueue chan newQueue {queueName = "myQueueDE"} declareQueue chan newQueue {queueName = "myQueueEN"} declareExchange chan newExchange {exchangeName = "topicExchg", exchangeType = "topic"} - bindQueue chan "myQueueDE" "topicExchg" "de.*" + -- bindQueue chan "myQueueDE" "topicExcheg" "de.*" bindQueue chan "myQueueEN" "topicExchg" "en.*" --publish messages @@ -26,8 +29,9 @@ (newMsg {msgBody = (BL.pack "hello world"), msgDeliveryMode = Just NonPersistent} ) - - + + -- closeChannel chan + getLine closeConnection conn