amqp 0.12.3 → 0.13.0

raw patch · 4 files changed

+135/−10 lines, 4 filesdep +stmPVP ok

version bump matches the API change (PVP)

Dependencies added: stm

API changes (from Hackage documentation)

Files

• Network/AMQP.hs +71/−1
• Network/AMQP/Internal.hs +54/−5
• Network/AMQP/Types.hs +5/−1
• amqp.cabal +5/−3

@@ -113,6 +113,13 @@     txCommit,
     txRollback,
 
+    -- * Confirms
+    confirmSelect,
+    waitForConfirms,
+    waitForConfirmsUntil,
+    addConfirmationListener,
+    ConfirmationResult(..),
+
     -- * Flow Control
     flow,
 
@@ -129,7 +136,11 @@     fromURI
 ) where
 
+import Control.Applicative
 import Control.Concurrent
+import Control.Concurrent.STM
+import Control.Monad(when)
+import Data.IORef
 import Data.List.Split (splitOn)
 import Data.Binary
 import Data.Binary.Put
@@ -139,6 +150,7 @@ 
 import qualified Data.ByteString.Lazy as BL
 import qualified Data.ByteString as BS
+import qualified Data.IntSet as IntSet
 import qualified Data.Text.Encoding as E
 import qualified Data.Map as M
 import qualified Data.Text as T
@@ -413,7 +425,10 @@             (fmap ShortString $ msgClusterID msg)
             )
             (msgBody msg))
-    return ()
+    nxtSeqNum <- readIORef (nextPublishSeqNum chan)
+    when (nxtSeqNum /= 0) $ do
+      writeIORef (nextPublishSeqNum chan) $ succ nxtSeqNum
+      atomically $ modifyTVar' (unconfirmedSet chan) $ \uSet -> IntSet.insert nxtSeqNum uSet
 
 -- | @getMsg chan ack queue@ gets a message from the specified queue. If @ack=='Ack'@, you have to call 'ackMsg' or 'ackEnv' for any message that you get, otherwise it might be delivered again in the future (by calling 'recoverMsgs')
 getMsg :: Channel -> Ack -> Text -> IO (Maybe (Message, Envelope))
@@ -491,6 +506,61 @@ txRollback chan = do
     (SimpleMethod Tx_rollback_ok) <- request chan $ SimpleMethod Tx_rollback
     return ()
+
+
+-------------------- PUBLISHER CONFIRMS ------------------------
+
+{- | @confirmSelect chan nowait@ puts the channel in publisher confirm mode. This mode is a RabbitMQ
+    extension where a producer receives confirmations when messages are successfully processed by
+    the broker. Publisher confirms are a relatively lightweight alternative to full transactional
+    mode. For details about the delivery guarantees and performace implications of this mode, see
+    https://www.rabbitmq.com/confirms.html. Note that on a single channel, publisher confirms and
+    transactions are mutually exclusive (you cannot select both at the same time).
+    When @nowait==True@ the server will not send back a response to this method.
+-}
+confirmSelect :: Channel -> Bool -> IO ()
+confirmSelect chan nowait = do
+    _ <- modifyIORef' (nextPublishSeqNum chan) $ \seqn -> if seqn == 0 then 1 else seqn
+    if nowait
+        then writeAssembly chan $ SimpleMethod (Confirm_select True)
+        else do
+            (SimpleMethod Confirm_select_ok) <- request chan $ SimpleMethod (Confirm_select False)
+            return ()
+
+{- | Calling this function will cause the invoking thread to block until all previously published
+messages have been acknowledged by the broker (positively or negatively). Returns a value of type
+@ConfirmationResult, holding a tuple of two @IntSet's @(acked, nacked), ontaining the delivery tags
+for the messages that have been confirmed by the broker.
+-}
+waitForConfirms :: Channel -> IO ConfirmationResult
+waitForConfirms chan = atomically $ Complete <$> waitForAllConfirms chan
+
+{- | Same as @waitForConfirms, but with a timeout in microseconds. Note that, since this operation
+may timeout before the server has acked or nacked all pending messages, the returned @ConfirmationResult
+should be pattern-matched for the constructors @Complete (acked, nacked) and @Partial (acked, nacked, pending)
+-}
+waitForConfirmsUntil :: Channel -> Int -> IO ConfirmationResult
+waitForConfirmsUntil chan timeout = do
+  delay <- registerDelay timeout
+  let partial = do
+        expired <- readTVar delay
+        if expired
+           then return . Partial =<< (,,)
+                <$> swapTVar (ackedSet chan) IntSet.empty
+                <*> swapTVar (nackedSet chan) IntSet.empty
+                <*> readTVar (unconfirmedSet chan)
+           else retry
+      complete = return . Complete =<< waitForAllConfirms chan
+  atomically $ complete `orElse` partial
+
+{- | Adds a handler which will be invoked each time the @Channel receives a confirmation from the broker.
+--The parameters passed to the the handler are the @deliveryTag for the message being confirmed, a flag
+indicating whether the confirmation refers to this message individually (@False) or all messages up to
+this one (@True) and an @AckType whose value can be either @BasicAck or @BasicNack.
+-}
+addConfirmationListener :: Channel -> ((Word64, Bool, AckType) -> IO ()) -> IO ()
+addConfirmationListener chan handler =
+    modifyMVar_ (confirmListeners chan) $ \listeners -> return $ handler:listeners
 
 --------------------- FLOW CONTROL ------------------------
 

@@ -4,12 +4,15 @@ import Paths_amqp(version)
 import Data.Version(showVersion)
 
+import Control.Applicative
 import Control.Concurrent
+import Control.Concurrent.STM
 import Control.Monad
 import Data.Binary
 import Data.Binary.Get
 import Data.Binary.Put as BPut
 import Data.Int (Int64)
+import Data.IORef
 import Data.Maybe
 import Data.Text (Text)
 import Network
@@ -22,6 +25,7 @@ import qualified Data.Map as M
 import qualified Data.Foldable as F
 import qualified Data.IntMap as IM
+import qualified Data.IntSet as IntSet
 import qualified Data.Sequence as Seq
 import qualified Data.Text as T
 import qualified Data.Text.Encoding as E
@@ -33,6 +37,7 @@ import Network.AMQP.Generated
 import Network.AMQP.ChannelAllocator
 
+data AckType = BasicAck | BasicNack deriving Show
 
 data DeliveryMode = Persistent -- ^ the message will survive server restarts (if the queue is durable)
                   | NonPersistent -- ^ the message may be lost after server restarts
@@ -362,13 +367,13 @@ 
     skippedBeatEx = ConnectionClosedException "killed connection after missing 2 consecutive heartbeats"
 
-    checkReceiveTimeout = check (connLastReceived conn) receiveTimeout
+    checkReceiveTimeout = doCheck (connLastReceived conn) receiveTimeout
         (killConnection conn (CE.toException skippedBeatEx) connThread)
 
-    checkSendTimeout = check (connLastSent conn) sendTimeout
+    checkSendTimeout = doCheck (connLastSent conn) sendTimeout
         (writeFrame (connHandle conn) (Frame 0 HeartbeatPayload))
 
-    check var timeout_µs action = withMVar var $ \lastFrameTime -> do
+    doCheck var timeout_µs action = withMVar var $ \lastFrameTime -> do
         time <- getTimestamp
         when (time >= lastFrameTime + timeout_µs) $ do
             action
@@ -468,10 +473,16 @@                     channelID :: Word16,
                     lastConsumerTag :: MVar Int,
 
+                    nextPublishSeqNum :: IORef Int,
+                    unconfirmedSet :: TVar IntSet.IntSet,
+                    ackedSet :: TVar IntSet.IntSet,  --delivery tags
+                    nackedSet :: TVar IntSet.IntSet, --accumulate here.
+
                     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 ()]),
+                    confirmListeners :: MVar ([(Word64, Bool, AckType) -> IO ()]),
                     chanExceptionHandlers :: MVar [CE.SomeException -> IO ()]
                 }
 
@@ -516,6 +527,8 @@     isResponse (ContentMethod (Basic_return _ _ _ _) _ _) = False
     isResponse (SimpleMethod (Channel_flow _)) = False
     isResponse (SimpleMethod (Channel_close _ _ _ _)) = False
+    isResponse (SimpleMethod (Basic_ack _ _)) = False
+    isResponse (SimpleMethod (Basic_nack _ _ _)) = False
     isResponse _ = True
 
     --Basic.Deliver: forward msg to registered consumer
@@ -551,8 +564,31 @@         withMVar (returnListeners chan) $ \listeners ->
             forM_ listeners $ \l -> CE.catch (l (msg, pubError)) $ \(ex :: CE.SomeException) ->
                 hPutStrLn stderr $ "return listener on channel ["++(show $ channelID chan)++"] handling error ["++show pubError++"] threw exception: "++show ex
+    handleAsync (SimpleMethod (Basic_ack deliveryTag multiple)) = handleConfirm deliveryTag multiple BasicAck
+    handleAsync (SimpleMethod (Basic_nack deliveryTag multiple _)) = handleConfirm deliveryTag multiple BasicNack
     handleAsync m = error ("Unknown method: " ++ show m)
 
+    handleConfirm deliveryTag multiple k = do
+        withMVar (confirmListeners chan) $ \listeners ->
+            forM_ listeners $ \l -> CE.catch (l (deliveryTag, multiple, k)) $ \(ex :: CE.SomeException) ->
+                hPutStrLn stderr $ "confirm listener on channel ["++(show $ channelID chan)++"] handling method "++(show k)++" threw exception: "++ show ex
+
+        let seqNum = fromIntegral deliveryTag
+        let targetSet = case k of
+              BasicAck  -> (ackedSet chan)
+              BasicNack -> (nackedSet chan)
+        atomically $ do
+          unconfSet <- readTVar (unconfirmedSet chan)
+          let (merge, pending) = if multiple
+                                     then (IntSet.union confs, pending')
+                                     else (IntSet.insert seqNum, IntSet.delete seqNum unconfSet)
+                                  where
+                                    confs = fst parts
+                                    pending' = snd parts
+                                    parts = IntSet.partition (\n -> n <= seqNum) unconfSet
+          modifyTVar' targetSet (\ts -> merge ts)
+          writeTVar (unconfirmedSet chan) pending
+
     basicReturnToPublishError (Basic_return code (ShortString errText) (ShortString exchange) (ShortString routingKey)) =
         let replyError = case code of
                 312 -> Unroutable errText
@@ -604,13 +640,18 @@     ca <- newLock
     closed <- newMVar Nothing
     conss <- newMVar M.empty
-    listeners <- newMVar []
+    retListeners <- newMVar []
+    aSet <- newTVarIO IntSet.empty
+    nSet <- newTVarIO IntSet.empty
+    nxtSeq <- newIORef 0
+    unconfSet <- newTVarIO IntSet.empty
+    cnfListeners <- 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 handlers
+        let newChannel = Channel c newInQueue outRes (fromIntegral newChannelID) lastConsTag nxtSeq unconfSet aSet nSet ca closed conss retListeners cnfListeners handlers
         thrID <- forkFinally' (channelReceiver newChannel) $ \res -> do
                    closeChannel' newChannel "closed"
                    case res of
@@ -719,3 +760,11 @@             case chc of
                 Just r -> CE.throwIO $ ChannelClosedException r
                 Nothing -> CE.throwIO $ ConnectionClosedException "unknown reason"
+
+waitForAllConfirms :: Channel -> STM (IntSet.IntSet, IntSet.IntSet)
+waitForAllConfirms chan = do
+  pending <- readTVar $ (unconfirmedSet chan)
+  check (IntSet.null pending)
+  return =<< (,)
+    <$> swapTVar (ackedSet chan) IntSet.empty
+    <*> swapTVar (nackedSet chan) IntSet.empty

@@ -17,11 +17,13 @@     FieldTable(..),
     FieldValue(..),
     Decimals,
-    DecimalValue(..)
+    DecimalValue(..),
+    ConfirmationResult(..),
 ) where
 
 import Control.Applicative
 import Data.Int
+import Data.IntSet (IntSet)
 import Data.Binary
 import Data.Binary.Get
 import Data.Binary.IEEE754
@@ -209,3 +211,5 @@     put (DecimalValue a b) = put a >> put b
 
 type Decimals = Octet
+
+data ConfirmationResult = Complete (IntSet, IntSet) | Partial (IntSet, IntSet, IntSet) deriving Show

@@ -1,5 +1,5 @@ Name:                amqp
-Version:             0.12.3
+Version:             0.13.0
 Synopsis:            Client library for AMQP servers (currently only RabbitMQ)
 Description:         Client library for AMQP servers (currently only RabbitMQ)
                      .
@@ -33,7 +33,8 @@     clock >= 0.4.0.1,
     monad-control >= 0.3,
     connection == 0.2.*,
-    vector
+    vector,
+    stm >= 2.4.0
   if flag(network-uri)
      build-depends: network-uri >= 2.6, network > 2.6
   else
@@ -85,7 +86,8 @@     , hspec-expectations >= 0.3.3
     , connection == 0.2.*
     , vector
+    , stm >= 2.4.0
   if flag(network-uri)
      build-depends: network-uri >= 2.6, network > 2.6
   else
-     build-depends: network < 2.6+     build-depends: network < 2.6