amqp 0.11 → 0.12

raw patch · 6 files changed

+75/−29 lines, 6 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Network/AMQP.hs +1/−0
• Network/AMQP/Helpers.hs +8/−1
• Network/AMQP/Internal.hs +43/−18
• amqp.cabal +1/−1
• examples/ExampleConsumer.hs +14/−5
• examples/ExampleProducer.hs +8/−4

@@ -62,6 +62,7 @@     Channel,
     openChannel,
     addReturnListener,
+    addChannelExceptionHandler,
     qos,
 
     -- * Exchanges

@@ -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

@@ -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 ()

@@ -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)
                      .

@@ -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

@@ -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