net-mqtt 0.8.7.2 → 0.8.7.3
raw patch · 3 files changed
+82/−14 lines, 3 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- net-mqtt.cabal +1/−1
- src/Network/MQTT/Client.hs +12/−12
- test/ClientSpec.hs +69/−1
net-mqtt.cabal view
@@ -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 <https://github.com/dustin/mqtt-hs#readme> category: Network
src/Network/MQTT/Client.hs view
@@ -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
test/ClientSpec.hs view
@@ -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 []))