amqp 0.15.0 → 0.15.1

raw patch · 4 files changed

+74/−2 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Network/AMQP.hs +2/−1
• amqp.cabal +1/−1
• changelog.md +4/−0
• test/BasicPublishSpec.hs +67/−0

@@ -119,7 +119,8 @@     waitForConfirmsUntil,
     addConfirmationListener,
     ConfirmationResult(..),
-
+    AckType(..),
+    
     -- * Flow Control
     flow,
 

@@ -1,5 +1,5 @@ Name:                amqp
-Version:             0.15.0
+Version:             0.15.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.15.1
+
+* export the `AckType` data-type and constructors
+
 ### 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.

@@ -6,8 +6,12 @@ import Network.AMQP
 
 import Data.ByteString.Lazy.Char8 as BL
+import Data.Map (Map)
+import qualified Data.Map as Map
+import Data.Word
 
 import Control.Concurrent (threadDelay)
+import Control.Concurrent.STM
 
 main :: IO ()
 main = hspec spec
@@ -60,3 +64,66 @@ 
                 _ <- deleteQueue ch q
                 closeConnection conn
+        context "confirmSelect" $ do
+            it "receives a confirmation message" $ do
+                let q = "haskell-amqp.queues.publish-over-fanout1"
+                    e = "haskell-amqp.fanout.d.na"
+                conn <- openConnection "127.0.0.1" "/" "guest" "guest"
+                ch   <- openChannel conn
+                confirmSelect ch True
+                (confirmMap, counter) <- atomically $ (,) <$> newTVar Map.empty <*> newTVar 0
+                addConfirmationListener ch (handleConfirms counter confirmMap)
+                _    <- declareExchange ch (newExchange {exchangeName = e,
+                                                         exchangeType = "fanout",
+                                                         exchangeDurable = True})
+
+                (_, _, _) <- declareQueue ch (newQueue {queueName = q, queueDurable = False})
+                _ <- purgeQueue ch q
+                bindQueue ch q e ""
+
+
+                _ <- traverse (\n -> do
+                                  sn' <- publishMsg ch e "" (newMsg {msgBody = (BL.pack "hello")})
+                                  case sn' of
+                                    Just sn -> atomically $ addSequenceNumber confirmMap (fromIntegral sn) n
+                                    Nothing -> return ()
+
+                              ) [1..5]
+
+
+                threadDelay (1000 * 100)
+
+                (_, n, _) <- declareQueue ch (newQueue {queueName = q, queueDurable = False})
+                n `shouldBe` 5
+
+                cMap' <- atomically $ readTVar confirmMap
+                cMap' `shouldBe` Map.empty
+
+                counter' <- atomically $ readTVar counter
+                counter' `shouldBe` 5
+
+                _ <- deleteQueue ch q
+                closeConnection conn
+
+
+addSequenceNumber :: TVar (Map Word64 Int) -> Word64 -> Int -> STM ()
+addSequenceNumber cMap sn n = modifyTVar' cMap (Map.insert sn n)
+
+removeSequenceNumber :: TVar (Map Word64 Int) -> Word64 -> STM ()
+removeSequenceNumber cMap sn = modifyTVar' cMap (Map.delete sn)
+
+increaseCounter :: TVar Int -> STM ()
+increaseCounter n = modifyTVar' n (+1)
+
+handleConfirms :: TVar Int -> TVar (Map Word64 Int) -> (Word64, Bool, AckType) -> IO ()
+handleConfirms c _ (_, False, BasicNack) = atomically $ increaseCounter c
+handleConfirms c _ (_, True, BasicNack) = atomically $ increaseCounter c
+handleConfirms c cMap (n, False, BasicAck) = atomically $ removeSequenceNumber cMap n >> increaseCounter c
+handleConfirms c cMap (n, True, BasicAck) = atomically $ do
+  cMap' <- readTVar cMap
+  let (lt, eq', _) = Map.splitLookup n cMap'
+  case eq' of
+    Just _ -> removeSequenceNumber cMap n >> increaseCounter c
+    Nothing -> return ()
+  _ <- traverse (\i -> removeSequenceNumber cMap i >> increaseCounter c) (Map.keys lt)
+  return ()