amqp 0.22.0 → 0.22.1

raw patch · 3 files changed

+21/−10 lines, 3 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

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

@@ -148,7 +148,8 @@     connBlockedHandlers :: MVar [(Text -> IO (), IO ())],
     connLastReceived :: MVar Int64, -- the timestamp from a monotonic clock when the last frame was received
     connLastSent :: MVar Int64, -- the timestamp from a monotonic clock when the last frame was written
-    connServerProperties :: FieldTable -- the server properties sent in Connection_start
+    connServerProperties :: FieldTable, -- the server properties sent in Connection_start
+    connThread :: MVar (Maybe ThreadId)
 }
 
 -- | Represents the parameters to connect to a broker or a cluster of brokers.
@@ -278,10 +279,12 @@     cBlockedHandlers <- newMVar []
     cLastReceived <- getTimestamp >>= newMVar
     cLastSent <- getTimestamp >>= newMVar
-    let conn = Connection handle cChanAllocator cChannels (fromIntegral maxFrameSize) cClosed ccl writeLock cClosedHandlers cBlockedHandlers cLastReceived cLastSent serverProps
+    cThread <- newMVar Nothing
 
+    let conn = Connection handle cChanAllocator cChannels (fromIntegral maxFrameSize) cClosed ccl writeLock cClosedHandlers cBlockedHandlers cLastReceived cLastSent serverProps cThread
+
     --spawn the connectionReceiver
-    connThread <- forkFinally' (connectionReceiver conn) $ \res -> do
+    connThreadId <- forkFinally' (connectionReceiver conn) $ \res -> do
         -- try closing socket
         CE.catch (Conn.connectionClose handle) (\(_ :: CE.SomeException) -> return ())
 
@@ -309,9 +312,10 @@     case heartbeatTimeout of
         Nothing      -> return ()
         Just timeout -> do
-            heartbeatThread <- watchHeartbeats conn (fromIntegral timeout) connThread
+            heartbeatThread <- watchHeartbeats conn (fromIntegral timeout) connThreadId
             addConnectionClosedHandler conn True (killThread heartbeatThread)
 
+    modifyMVar_ cThread $ \_ -> return $ Just connThreadId
     return conn
   where
     connect excs ((host, port) : rest) = do
@@ -408,9 +412,9 @@ 
 -- | kill the connection thread abruptly
 killConnection :: Connection -> CloseType -> CE.SomeException -> ThreadId -> IO ()
-killConnection conn closeType ex connThread = do
+killConnection conn closeType ex connThreadId = do
     modifyMVar_ (connClosed conn) $ const $ return $ Just (closeType, show ex)
-    throwTo connThread ex
+    throwTo connThreadId ex
 
 -- | closes a connection
 --
@@ -425,9 +429,12 @@             0 -- class_id
             0 -- method_id
         )
-        (\ (_ :: CE.IOException) ->
-            --do nothing if connection is already closed
-            return ()
+        (\ (e :: CE.IOException) -> do
+            -- Make sure the connection is closed.
+            -- (This pattern match can't fail, since openConnection'' will always fill
+            --  the variable in, and there's no other way to get a connection.)
+            Just thrID <- readMVar (connThread c)
+            killConnection c Abnormal (CE.toException e) thrID
         )
 
     -- wait for connection_close_ok by the server; this MVar gets filled in the CE.finally handler in openConnection'

@@ -1,5 +1,5 @@ Name:                amqp
-Version:             0.22.0
+Version:             0.22.1
 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.22.1
+
+* fix a potential deadlock in `closeConnection`
+
 ### Version 0.22.0
 
 * `closeChannel` now uses a `ChannelClosedException` to kill the channel-thread instead of `ThreadKilled`. This fixes a bug in 0.21.0 that led to errors being printed on the screen when `closeChannel` was called.