diff --git a/net-mqtt.cabal b/net-mqtt.cabal --- a/net-mqtt.cabal +++ b/net-mqtt.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: net-mqtt -version: 0.8.7.2 +version: 0.8.7.3 synopsis: An MQTT Protocol Implementation. description: Please see the README on GitHub at category: Network diff --git a/src/Network/MQTT/Client.hs b/src/Network/MQTT/Client.hs --- a/src/Network/MQTT/Client.hs +++ b/src/Network/MQTT/Client.hs @@ -40,10 +40,9 @@ import Control.Concurrent (myThreadId, threadDelay) import Control.Concurrent.Async (Async, async, asyncThreadId, cancel, cancelWith, link, race_, wait, waitAnyCancel) -import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar) -import Control.Concurrent.STM (STM, TChan, TVar, atomically, check, modifyTVar', newTChan, newTChanIO, - newTVarIO, orElse, readTChan, readTVar, readTVarIO, registerDelay, retry, - writeTChan, writeTVar) +import Control.Concurrent.STM (STM, TChan, TQueue, TVar, atomically, check, modifyTVar', newTChan, + newTChanIO, newTQueueIO, newTVarIO, orElse, readTChan, readTQueue, readTVar, + readTVarIO, registerDelay, retry, writeTChan, writeTQueue, writeTVar) import Control.DeepSeq (force) import qualified Control.Exception as E import Control.Monad (forever, guard, join, unless, void, when) @@ -103,8 +102,9 @@ -- Use a simple 'System.Timeout.timeout' to restrain unbounded thread runtime. | SimpleCallback (MQTTClient -> Topic -> BL.ByteString -> [Property] -> IO ()) -- | Callbacks are guaranteed to be invoked in the same order messages are received. - -- In high throughput scenarios, slow callbacks may cause the underlying TCP connection to block, - -- potentially being terminated by the broker. + -- Slow callbacks do not block the underlying TCP connection. Pending QoS 0 messages + -- can accumulate while a callback is running. With MQTT 5, pending QoS 1 and 2 + -- messages are bounded by the Receive Maximum. -- Typically slower than `SimpleCallback`. | OrderedCallback (MQTTClient -> Topic -> BL.ByteString -> [Property] -> IO ()) -- | A LowLevelCallback receives the client and the entire publish request, providing @@ -130,7 +130,7 @@ , _inA :: TVar (Map Word16 Topic) , _connACKFlags :: TVar ConnACKFlags , _corr :: Decaying.Map BL.ByteString MessageCallback - , _cbM :: MVar (IO ()) + , _cbQ :: TQueue (IO ()) , _cbHandle :: TVar (Maybe (Async ())) , _outFlight :: TVar Word16 -- ^ QoS 1 and 2 PUBLISHes in flight to the server } @@ -315,7 +315,7 @@ _inA <- newTVarIO mempty _connACKFlags <- newTVarIO (ConnACKFlags NewSession ConnUnspecifiedError mempty) _corr <- Decaying.new 600 - _cbM <- newEmptyMVar + _cbQ <- newTQueueIO _cbHandle <- newTVarIO Nothing _outFlight <- newTVarIO 0 let _cb = _msgCB @@ -494,7 +494,7 @@ where call a = link =<< namedAsync "notifier" (a >> respond) - callOrd a = putMVar _cbM $ a >> respond + callOrd a = atomically $ writeTQueue _cbQ (a *> respond) respond = traverse_ (sendPacketIO c) rpkt cdata = foldr f Nothing _pubProps where f (PropCorrelationData x) _ = Just x @@ -528,7 +528,7 @@ -- Run OrderedCallbacks in a background thread. Does nothing for other callback types. -- We keep the async handle in a TVar and make sure only one of these threads is running. runCallbackThread :: MQTTClient -> IO () -runCallbackThread c@MQTTClient{_cb, _cbM, _cbHandle} +runCallbackThread c@MQTTClient{_cb, _cbQ, _cbHandle} | isOrdered _cb = do -- We always spawn a thread, but may kill it if we already have -- one. The new thread won't start until we atomically confirm @@ -549,10 +549,10 @@ isOrdered (OrderedLowLevelCallback _) = True isOrdered _ = False waitFor latch = atomically (check =<< readTVar latch) - -- Keep running callbacks from the MVar. One that fails ends the + -- Keep running callbacks from the queue. One that fails ends the -- connection, as with unordered callbacks, rather than leaving it -- up with nothing to run the rest. - runOrderedCallbacks = forever (join (takeMVar _cbM)) `E.catch` \(e :: E.SomeException) -> + runOrderedCallbacks = forever (join (atomically (readTQueue _cbQ))) `E.catch` \(e :: E.SomeException) -> case E.fromException e of Just (_ :: E.SomeAsyncException) -> E.throwIO e Nothing -> killConn c e >> E.throwIO e diff --git a/test/ClientSpec.hs b/test/ClientSpec.hs --- a/test/ClientSpec.hs +++ b/test/ClientSpec.hs @@ -8,7 +8,7 @@ import Control.Concurrent.Async (mapConcurrently_) import Control.Concurrent.Async (async, poll, wait) import Control.Concurrent.STM (TVar, atomically, check, modifyTVar', newTChanIO, newTVarIO, - readTChan, readTVar, retry, writeTChan) + readTChan, readTVar, retry, writeTChan, writeTVar) import Control.Monad (forM_, forever, replicateM_) import qualified Control.Exception as E import Control.Monad.IO.Class (liftIO) @@ -16,6 +16,7 @@ import qualified Data.ByteString as BS import qualified Data.ByteString.Lazy as BL import Data.Conduit (awaitForever, yield) +import Data.Word (Word16) import System.Timeout (timeout) import Test.Tasty.HUnit @@ -60,6 +61,10 @@ ws <- maybe (assertFailure "publishes never arrived") pure got pure [p | Right (PublishPkt p) <- A.eitherResult . A.parse (parsePacket Protocol50) . BL.fromStrict <$> ws] +incomingPublish :: QoS -> Word16 -> MQTTPkt +incomingPublish q i = PublishPkt PublishRequest{_pubDup=False, _pubQoS=q, _pubRetain=False, _pubTopic="a/b", + _pubPktID=i, _pubBody="x", _pubProps=[]} + -- Concurrent aliased publishes to one topic: the first PUBLISH using the -- alias has to carry the topic too, or the broker can't know what the -- alias means [5.0 ยง3.3.2.3.4]. @@ -174,3 +179,66 @@ acks <- mapM (`nthWritten` written) [1, 2, 3] assertEqual "PUBACKs" [PubACKPkt (PubACK i 0 []) | i <- [1, 2, 3]] acks stopClient c + +-- A blocked ordered callback must not stop the reader from processing later +-- packets. Three PUBLISHes used to fill the one-slot MVar and block the third. +unit_orderedCallbackDoesNotBlockReader :: Assertion +unit_orderedCallbackDoesNotBlockReader = do + (_, inject, conn) <- scriptedBroker [] + started <- newTVarIO False + release <- newTVarIO False + c <- runMQTTConduit conn mqttConfig{_protocol=Protocol50, + _msgCB=OrderedCallback (\_ _ _ _ -> do + atomically $ writeTVar started True + atomically $ check =<< readTVar release)} + flip E.finally (stopClient c) $ do + mapM_ (inject . incomingPublish QoS1) [1, 2, 3] + timeout 1000000 (atomically $ readTVar started >>= check) >>= + maybe (assertFailure "the first callback never started") pure + inject (DisconnectPkt (DisconnectRequest DiscoNormalDisconnection [])) + connectionEnds c + +-- Ordered QoS 1 callbacks acknowledge in receive order, but only after their +-- callbacks have completed. +unit_orderedCallbacksAckAfterCompletionInOrder :: Assertion +unit_orderedCallbacksAckAfterCompletionInOrder = do + (written, inject, conn) <- scriptedBroker [] + started <- newTVarIO [] + completed <- newTVarIO [] + release <- newTVarIO False + c <- runMQTTConduit conn mqttConfig{_protocol=Protocol50, + _msgCB=OrderedLowLevelCallback (\_ p -> do + atomically $ modifyTVar' started (<> [_pubPktID p]) + atomically $ check =<< readTVar release + atomically $ modifyTVar' completed (<> [_pubPktID p]))} + flip E.finally (stopClient c) $ do + mapM_ (inject . incomingPublish QoS1) [1, 2, 3] + timeout 1000000 (atomically $ readTVar started >>= check . (== [1])) >>= + maybe (assertFailure "the first callback never started") pure + beforeRelease <- atomically $ drop 1 <$> readTVar written + assertEqual "PUBACKs before the callback completes" [] beforeRelease + atomically $ writeTVar release True + acks <- mapM (`nthWritten` written) [1, 2, 3] + assertEqual "PUBACKs" [PubACKPkt (PubACK i 0 []) | i <- [1, 2, 3]] acks + atomically (readTVar completed) >>= assertEqual "callback completion order" [1, 2, 3] + +-- QoS 2 still sends PUBREC immediately, while PUBCOMP waits for the callback. +unit_orderedQoS2CompletesAfterCallback :: Assertion +unit_orderedQoS2CompletesAfterCallback = do + (written, inject, conn) <- scriptedBroker [] + started <- newTVarIO False + release <- newTVarIO False + c <- runMQTTConduit conn mqttConfig{_protocol=Protocol50, + _msgCB=OrderedCallback (\_ _ _ _ -> do + atomically $ writeTVar started True + atomically $ check =<< readTVar release)} + flip E.finally (stopClient c) $ do + inject (incomingPublish QoS2 1) + nthWritten 1 written >>= assertEqual "PUBREC" (PubRECPkt (PubREC 1 0 [])) + inject (PubRELPkt (PubREL 1 0 [])) + timeout 1000000 (atomically $ readTVar started >>= check) >>= + maybe (assertFailure "the callback never started") pure + beforeRelease <- atomically $ drop 2 <$> readTVar written + assertEqual "PUBCOMP before the callback completes" [] beforeRelease + atomically $ writeTVar release True + nthWritten 2 written >>= assertEqual "PUBCOMP" (PubCOMPPkt (PubCOMP 1 0 []))