amqp 0.13.1 → 0.14.0

raw patch · 4 files changed

+32/−24 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

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

@@ -140,7 +140,6 @@ 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
@@ -394,14 +393,17 @@ 
 -- | @publishMsg chan exchange routingKey msg@ publishes @msg@ to the exchange with the provided @exchange@. The effect of @routingKey@ depends on the type of the exchange.
 --
+-- Returns the sequence-number of the message (only if the channel is in publisher confirm mode; see 'confirmSelect').
+--
 -- NOTE: This method may temporarily block if the AMQP server requested us to stop sending content data (using the flow control mechanism). So don't rely on this method returning immediately.
-publishMsg :: Channel -> Text -> Text -> Message -> IO ()
+publishMsg :: Channel -> Text -> Text -> Message -> IO (Maybe Int)
 publishMsg chan exchange routingKey msg = publishMsg' chan exchange routingKey False msg
 
 -- | Like 'publishMsg', but additionally allows you to specify whether the 'mandatory' flag should be set.
-publishMsg' :: Channel -> Text -> Text -> Bool -> Message -> IO ()
+publishMsg' :: Channel -> Text -> Text -> Bool -> Message -> IO (Maybe Int)
 publishMsg' chan exchange routingKey mandatory msg = do
-    writeAssembly chan (ContentMethod (Basic_publish
+    modifyMVar (nextPublishSeqNum chan) $ \nxtSeqNum -> do
+        writeAssembly chan (ContentMethod (Basic_publish
             1 -- ticket; ignored by rabbitMQ
             (ShortString exchange)
             (ShortString routingKey)
@@ -412,7 +414,7 @@             (fmap ShortString $ msgContentType msg)
             (fmap ShortString $ msgContentEncoding msg)
             (msgHeaders msg)
-            (fmap deliveryModeToInt $ msgDeliveryMode msg) -- delivery_mode
+            (fmap deliveryModeToInt $ msgDeliveryMode msg)
             (msgPriority msg)
             (fmap ShortString $ msgCorrelationID msg)
             (fmap ShortString $ msgReplyTo msg)
@@ -425,11 +427,13 @@             (fmap ShortString $ msgClusterID msg)
             )
             (msgBody msg))
-    nxtSeqNum <- readIORef (nextPublishSeqNum chan)
-    when (nxtSeqNum /= 0) $ do
-      writeIORef (nextPublishSeqNum chan) $ succ nxtSeqNum
-      atomically $ modifyTVar' (unconfirmedSet chan) $ \uSet -> IntSet.insert nxtSeqNum uSet
 
+        if nxtSeqNum /= 0
+            then do
+                atomically $ modifyTVar' (unconfirmedSet chan) $ \uSet -> IntSet.insert nxtSeqNum uSet
+                return (succ nxtSeqNum, Just nxtSeqNum)
+           else return (0, Nothing)
+
 -- | @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))
 getMsg chan ack queue = do
@@ -520,7 +524,7 @@ -}
 confirmSelect :: Channel -> Bool -> IO ()
 confirmSelect chan nowait = do
-    _ <- modifyIORef' (nextPublishSeqNum chan) $ \seqn -> if seqn == 0 then 1 else seqn
+    modifyMVar_ (nextPublishSeqNum chan) $ \seqn -> return $ if seqn == 0 then 1 else seqn
     if nowait
         then writeAssembly chan $ SimpleMethod (Confirm_select True)
         else do
@@ -529,15 +533,15 @@ 
 {- | 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
+'ConfirmationResult', holding a tuple of two IntSets @(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)
+{- | 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
@@ -553,10 +557,10 @@       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.
+{- | 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 =

@@ -12,7 +12,6 @@ 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
@@ -169,12 +168,14 @@ data TLSSettings =
     TLSTrusted   -- ^ Require trusted certificates (Recommended).
   | TLSUntrusted -- ^ Allow untrusted certificates (Discouraged. Vulnerable to man-in-the-middle attacks)
+  | TLSCustom Conn.TLSSettings -- ^ Provide your own custom TLS settings
 
 connectionTLSSettings :: TLSSettings -> Maybe Conn.TLSSettings
 connectionTLSSettings tlsSettings =
   Just $ case tlsSettings of
     TLSTrusted -> Conn.TLSSettingsSimple False False False
     TLSUntrusted -> Conn.TLSSettingsSimple True False False
+    TLSCustom x -> x
 
 -- | A 'SASLMechanism' is described by its name ('saslName'), its initial response ('saslInitialResponse'), and an optional function ('saslChallengeFunc') that
 -- transforms a security challenge provided by the server into response, which is then sent back to the server for verification.
@@ -472,7 +473,7 @@                     channelID :: Word16,
                     lastConsumerTag :: MVar Int,
 
-                    nextPublishSeqNum :: IORef Int,
+                    nextPublishSeqNum :: MVar Int,
                     unconfirmedSet :: TVar IntSet.IntSet,
                     ackedSet :: TVar IntSet.IntSet,  --delivery tags
                     nackedSet :: TVar IntSet.IntSet, --accumulate here.
@@ -642,7 +643,7 @@     retListeners <- newMVar []
     aSet <- newTVarIO IntSet.empty
     nSet <- newTVarIO IntSet.empty
-    nxtSeq <- newIORef 0
+    nxtSeq <- newMVar 0
     unconfSet <- newTVarIO IntSet.empty
     cnfListeners <- newMVar []
     handlers <- newMVar []

@@ -1,9 +1,7 @@ Name:                amqp
-Version:             0.13.1
+Version:             0.14.0
 Synopsis:            Client library for AMQP servers (currently only RabbitMQ)
 Description:         Client library for AMQP servers (currently only RabbitMQ)
-                     .
-                     Changelog: <https://github.com/hreinhardt/amqp/blob/master/README.md>
 License:             BSD3
 Stability:           alpha
 License-file:        LICENSE

@@ -1,3 +1,8 @@+### Version 0.14.0
+
+* publishMsg now returns the message sequence-number
+* new `TLSCustom` field in `TLSSettings`
+
 ### Version 0.13.1
 
 * don't print to stderr when `openConnection` fails to connect to a broker