amqp 0.14.1 → 0.15.0

raw patch · 4 files changed

+25/−5 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Network/AMQP.hs +3/−0
• Network/AMQP/Internal.hs +17/−4
• amqp.cabal +1/−1
• changelog.md +4/−0

@@ -131,6 +131,7 @@ 
     -- * Exceptions
     AMQPException(..),
+    ChanThreadKilledException,
 
     -- * URI parsing
     fromURI
@@ -352,6 +353,8 @@ ackToBool NoAck = True
 
 -- | @consumeMsgs chan queue ack callback@ subscribes to the given queue and returns a consumerTag. For any incoming message, the callback will be run. If @ack == 'Ack'@ you will have to acknowledge all incoming messages (see 'ackMsg' and 'ackEnv')
+--
+-- If you do any exception handling inside the callback, you should make sure not to catch 'ChanThreadKilledException', or re-throw it if you did catch it, since it is used internally by the library to close channels.
 --
 -- NOTE: The callback will be run on the same thread as the channel thread (every channel spawns its own thread to listen for incoming data) so DO NOT perform any request on @chan@ inside the callback (however, you CAN perform requests on other open channels inside the callback, though I wouldn't recommend it).
 -- Functions that can safely be called on @chan@ are 'ackMsg', 'ackEnv', 'rejectMsg', 'recoverMsgs'. If you want to perform anything more complex, it's a good idea to wrap it inside 'forkIO'.

@@ -278,7 +278,7 @@                 -- 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
+                let finaliser = ChanThreadKilledException $ case res of
                         Left ex -> ex
                         Right _ -> CE.toException CE.ThreadKilled
                 modifyMVar_ cChannels $ \x -> do
@@ -486,6 +486,16 @@                     chanExceptionHandlers :: MVar [CE.SomeException -> IO ()]
                 }
 
+-- | Thrown in the channel thread when the connection gets closed.
+-- When handling exceptions in a subscription callback, make sure to re-throw this so the channel thread can be stopped.
+data ChanThreadKilledException = ChanThreadKilledException { cause :: CE.SomeException }
+  deriving (Show)
+
+instance CE.Exception ChanThreadKilledException
+
+unwrapChanThreadKilledException :: CE.SomeException -> CE.SomeException
+unwrapChanThreadKilledException e = maybe e cause $ CE.fromException e
+
 msgFromContentHeaderProperties :: ContentHeaderProperties -> BL.ByteString -> Message
 msgFromContentHeaderProperties (CHBasic content_type content_encoding headers delivery_mode priority correlation_id reply_to expiration message_id timestamp message_type user_id application_id cluster_id) body =
     let msgId = fromShortString message_id
@@ -542,8 +552,11 @@                     let env = Envelope {envDeliveryTag = deliveryTag, envRedelivered = redelivered,
                                     envExchangeName = exchange, envRoutingKey = routingKey, envChannel = chan}
 
-                    CE.catch (subscriber (msg, env))
-                        (\(e::CE.SomeException) -> hPutStrLn stderr $ "AMQP callback threw exception: " ++ show e)
+                    CE.catches (subscriber (msg, env))
+                        [
+                          CE.Handler (\(e::ChanThreadKilledException) -> CE.throwIO $ cause e),
+                          CE.Handler (\(e::CE.SomeException) -> hPutStrLn stderr $ "AMQP callback threw exception: " ++ show e)
+                        ]
                 Nothing ->
                     -- got a message, but have no registered subscriber; so drop it
                     return ()
@@ -656,7 +669,7 @@                    closeChannel' newChannel "closed"
                    case res of
                      Right _ -> return ()
-                     Left ex -> readMVar handlers >>= mapM_ ($ ex)
+                     Left ex -> readMVar handlers >>= mapM_ ($ unwrapChanThreadKilledException ex)
         when (IM.member newChannelID mp) $ CE.throwIO $ userError "openChannel fail: channel already open"
         return (IM.insert newChannelID (newChannel, thrID) mp, newChannel)
 

@@ -1,5 +1,5 @@ Name:                amqp
-Version:             0.14.1
+Version:             0.15.0
 Synopsis:            Client library for AMQP servers (currently only RabbitMQ)
 Description:         Client library for AMQP servers (currently only RabbitMQ)
 License:             BSD3

@@ -1,3 +1,7 @@+### Version 0.15.0
+
+* The way channels are closed internally was changed. This may affect you if you have installed an exception handler inside the callback passed to `consumeMsgs`. Specifically, the exceptions used internally to close channels are now wrapped inside `ChanThreadKilledException`. You should make sure to re-throw this exception if you did catch it.
+
 ### Version 0.14.1
 
 * show all exceptions if no host can be connected to