net-mqtt 0.8.7.0 → 0.8.7.1
raw patch · 5 files changed
+265/−31 lines, 5 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- net-mqtt.cabal +2/−1
- src/Network/MQTT/Client.hs +76/−30
- src/Network/MQTT/Types.hs +2/−0
- test/ClientSpec.hs +176/−0
- test/Spec.hs +9/−0
net-mqtt.cabal view
@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: net-mqtt-version: 0.8.7.0+version: 0.8.7.1 synopsis: An MQTT Protocol Implementation. description: Please see the README on GitHub at <https://github.com/dustin/mqtt-hs#readme> category: Network@@ -127,6 +127,7 @@ type: exitcode-stdio-1.0 main-is: Main.hs other-modules:+ ClientSpec DecayingSpec Example1 Example2
src/Network/MQTT/Client.hs view
@@ -15,10 +15,11 @@ are supported over plain TCP, TLS, WebSockets and Secure WebSockets. -} -{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-} module Network.MQTT.Client ( -- * Configuring the client.@@ -61,7 +62,7 @@ import Data.Map.Strict (Map) import qualified Data.Map.Strict as Map import qualified Data.Map.Strict.Decaying as Decaying-import Data.Maybe (fromJust, fromMaybe)+import Data.Maybe (fromJust, fromMaybe, isJust) import Data.Text (Text) import qualified Data.Text.Lazy as TL import qualified Data.Text.Encoding as TE@@ -131,6 +132,7 @@ , _corr :: Decaying.Map BL.ByteString MessageCallback , _cbM :: MVar (IO ()) , _cbHandle :: TVar (Maybe (Async ()))+ , _outFlight :: TVar Word16 -- ^ QoS 1 and 2 PUBLISHes in flight to the server } -- | Configuration for setting up an MQTT client.@@ -315,6 +317,7 @@ _corr <- Decaying.new 600 _cbM <- newEmptyMVar _cbHandle <- newTVarIO Nothing+ _outFlight <- newTVarIO 0 let _cb = _msgCB cli = MQTTClient{..} @@ -325,6 +328,10 @@ atomically $ checkConnected cli + -- Ordered callbacks get messages from the start, not only after a+ -- subscribe (a resumed session can deliver without one).+ runCallbackThread cli+ pure cli where@@ -369,7 +376,12 @@ .| C.map (BL.toStrict . toByteString _protocol) .| sink - doPing = forever $ threadDelay _pingPeriod >> sendPacketIO c PingPkt+ -- Ping at least twice per Server Keep Alive, if the server set+ -- one shorter than our period [MQTT-3.2.2-21].+ doPing = do+ ConnACKFlags _ _ props <- readTVarIO _connACKFlags+ let period = minimum (_pingPeriod : [fromIntegral k * 500000 | PropServerKeepAlive k <- props, k > 0])+ forever $ threadDelay period >> sendPacketIO c PingPkt watchdog ch = forever $ do toch <- registerDelay _pingPatience@@ -419,7 +431,7 @@ dispatch c@MQTTClient{..} pch pkt = case pkt of (ConnACKPkt p) -> connACKd p- (PublishPkt p) -> pub p+ (PublishPkt p) -> validTopic p >> pub p (SubACKPkt (SubscribeResponse i _ _)) -> delegate DSubACK i (UnsubACKPkt (UnsubscribeResponse i _ _)) -> delegate DUnsubACK i (PubACKPkt (PubACK i _ _)) -> delegate DPubACK i@@ -447,6 +459,16 @@ atomically $ writeTVar _st (ConnErr connr) maybeCancelWith (MQTTException $ show connr) t + -- A topic name can't have wildcards [MQTT-3.3.2-2]. One that does+ -- is a protocol error ending the connection, rather than an+ -- exception in whatever callback looks at the topic. (An empty one+ -- is a topic alias.)+ validTopic PublishRequest{_pubTopic}+ | BL.null _pubTopic || isJust (mkTopic (blToText _pubTopic)) = pure ()+ | otherwise = do+ sendPacketIO c (DisconnectPkt (DisconnectRequest DiscoProtocolError mempty))+ E.throwIO (MQTTException ("invalid topic in PUBLISH: " <> show _pubTopic))+ pub p@PublishRequest{_pubQoS=QoS0} = atomically (resolve p) >>= notify Nothing pub p@PublishRequest{_pubQoS=QoS1, _pubPktID} = notify (Just (PubACKPkt (PubACK _pubPktID 0 mempty))) =<< atomically (resolve p)@@ -505,7 +527,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 MQTTClient{_cb, _cbM, _cbHandle}+runCallbackThread c@MQTTClient{_cb, _cbM, _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@@ -526,8 +548,13 @@ isOrdered (OrderedLowLevelCallback _) = True isOrdered _ = False waitFor latch = atomically (check =<< readTVar latch)- -- Keep running callbacks from the MVar- runOrderedCallbacks = forever . join . takeMVar $ _cbM+ -- Keep running callbacks from the MVar. 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) ->+ case E.fromException e of+ Just (_ :: E.SomeAsyncException) -> E.throwIO e+ Nothing -> killConn c e >> E.throwIO e -- Stop the background thread for OrderedCallbacks. Does nothing for other callback types. stopCallbackThread :: MQTTClient -> IO ()@@ -640,24 +667,42 @@ -> QoS -- ^ QoS -> [Property] -- ^ Properties -> IO ()-publishq c t m r q props = do- (ch,pid) <- atomically $ reservePktID c types- E.finally (publishAndWait ch pid) (atomically $ releasePktIDs c [(t',pid) | t' <- types])+publishq c t m r q props = publishWith c (pure (t, props)) m r q +-- | Publish with the topic and properties from an STM action. The action+-- runs in the same transaction that queues the PUBLISH, so whatever it+-- decides (e.g. a topic alias) goes out in the order it was decided.+publishWith :: MQTTClient -> STM (Topic, [Property]) -> BL.ByteString -> Bool -> QoS -> IO ()+publishWith c@MQTTClient{_outFlight, _connACKFlags} mk m r q = do+ (ch,pid) <- atomically $ do+ (ch,pid) <- reservePktID c types+ when (q > QoS0) takeSlot+ (t, props) <- mk+ sendPacket c (pkt t props pid)+ pure (ch,pid)+ E.finally (when (q > QoS0) $ satisfyQoS ch pid) $ do+ when (q > QoS0) . atomically $ modifyTVar' _outFlight (subtract 1)+ atomically $ releasePktIDs c [(t',pid) | t' <- types]+ where types = [DPubACK, DPubREC, DPubCOMP]- publishAndWait ch pid = do- sendPacketIO c (pkt pid)- when (q > QoS0) $ satisfyQoS ch pid - pkt pid = PublishPkt $ PublishRequest {_pubDup = False,- _pubQoS = q,- _pubPktID = pid,- _pubRetain = r,- _pubTopic = topicToBL t,- _pubBody = m,- _pubProps = props}+ -- No more QoS 1 and 2 PUBLISHes in flight than the server's Receive+ -- Maximum [MQTT-3.3.4-7].+ takeSlot = do+ ConnACKFlags _ _ props <- readTVar _connACKFlags+ n <- readTVar _outFlight+ check (n < foldr (\p o -> case p of PropReceiveMaximum x -> x; _ -> o) maxBound props)+ writeTVar _outFlight (n + 1) + pkt t props pid = PublishPkt $ PublishRequest {_pubDup = False,+ _pubQoS = q,+ _pubPktID = pid,+ _pubRetain = r,+ _pubTopic = topicToBL t,+ _pubBody = m,+ _pubProps = props}+ satisfyQoS ch pid | q == QoS0 = pure () | q == QoS1 = void $ do@@ -671,10 +716,13 @@ isOK 16 = True -- It worked, but nobody cares (no matching subscribers) isOK _ = False + -- The packet ID is held until the PUBCOMP, since until then+ -- the server still has the exchange open [MQTT-4.3.3]. waitRec = atomically (checkConnected c >> readTChan ch) >>= \case PubRECPkt (PubREC _ st recprops) -> do unless (isOK st) $ mqttFail ("qos 2 REC publish error: " <> show st <> " " <> show recprops) sendPacketIO c (PubRELPkt $ PubREL pid 0 mempty)+ waitRec PubCOMPPkt (PubCOMP _ st' compprops) -> when (st' /= 0) $ mqttFail ("qos 2 COMP publish error: " <> show st' <> " " <> show compprops) wtf -> mqttFail ("unexpected packet received in QoS2 publish: " <> show wtf)@@ -741,20 +789,18 @@ -> IO () pubAliased c@MQTTClient{..} t m r q props = do x <- maxAliases c- (t', n) <- alias x- let np = props <> case n of- 0 -> mempty- _ -> [PropTopicAlias n]- publishq c t' m r q np+ -- The alias is chosen in the transaction that queues the PUBLISH, so+ -- the one that introduces it can't be overtaken by one that uses it.+ publishWith c (alias x) m r q where- alias mv = atomically $ do+ alias mv = do as <- readTVar _outA let n = toEnum (length as + 1) cur = Map.lookup t as v = fromMaybe (if n > mv then 0 else n) cur when (v > 0) $ writeTVar _outA (Map.insert t v as)- pure (maybe t (const "") cur, v)+ pure (maybe t (const "") cur, props <> [PropTopicAlias v | v > 0]) -- | Register a callback handler for a message with the given correlated data identifier. --
src/Network/MQTT/Types.hs view
@@ -627,6 +627,8 @@ _subQoS=QoS0} instance ByteMe SubOptions where+ -- 3.1.1 has only the QoS; the other bits are reserved [MQTT-3.8.3-4].+ toByteString Protocol311 SubOptions{..} = BL.singleton (qosW _subQoS) toByteString _ SubOptions{..} = BL.singleton (rh .|. rap .|. nl .|. q) where
+ test/ClientSpec.hs view
@@ -0,0 +1,176 @@+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++module ClientSpec where++import Control.Concurrent (threadDelay)+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)+import Control.Monad (forM_, forever, replicateM_)+import qualified Control.Exception as E+import Control.Monad.IO.Class (liftIO)+import qualified Data.Attoparsec.ByteString.Lazy as A+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BL+import Data.Conduit (awaitForever, yield)+import System.Timeout (timeout)+import Test.Tasty.HUnit++import Network.MQTT.Client+import Network.MQTT.Types as MT++-- | A fake broker for a client to connect to. It accepts the connection+-- with the given CONNACK properties and records everything the client+-- writes after the CONNECT. (The client writes each packet in one+-- chunk.)+fakeBroker :: [MT.Property] -> IO (TVar [BS.ByteString], (MQTTConduit -> IO ()) -> IO ())+fakeBroker props = (\(w, _, c) -> (w, c)) <$> scriptedBroker props++-- | A fake broker that also sends the client whatever packets it's given.+scriptedBroker :: [MT.Property] -> IO (TVar [BS.ByteString], MQTTPkt -> IO (), (MQTTConduit -> IO ()) -> IO ())+scriptedBroker props = do+ written <- newTVarIO []+ inject <- newTChanIO+ let enc = BL.toStrict . toByteString Protocol50+ src = yield (enc $ ConnACKPkt (ConnACKFlags NewSession ConnAccepted props))+ >> forever (liftIO (atomically (readTChan inject)) >>= yield . enc)+ sink = awaitForever $ \bs -> liftIO . atomically $ modifyTVar' written (<> [bs])+ pure (written, atomically . writeTChan inject, \f -> f (src, sink))++-- | Wait for the client to have written the nth packet after its CONNECT.+nthWritten :: Int -> TVar [BS.ByteString] -> IO MQTTPkt+nthWritten n written = do+ got <- timeout 5000000 . atomically $ do+ ws <- drop 1 <$> readTVar written+ check (length ws >= n)+ pure (ws !! (n - 1))+ bs <- maybe (assertFailure ("packet " <> show n <> " never arrived")) pure got+ either (assertFailure . ("unparseable packet: " <>)) pure . A.eitherResult . A.parse (parsePacket Protocol50) $ BL.fromStrict bs++-- | The PUBLISHes a fake broker received, in order, once there are n.+published :: Int -> TVar [BS.ByteString] -> IO [PublishRequest]+published n written = do+ got <- timeout 5000000 . atomically $ do+ ws <- drop 1 <$> readTVar written -- the CONNECT+ check (length ws >= n)+ pure ws+ ws <- maybe (assertFailure "publishes never arrived") pure got+ pure [p | Right (PublishPkt p) <- A.eitherResult . A.parse (parsePacket Protocol50) . BL.fromStrict <$> ws]++-- 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].+unit_concurrentAliasedPublishes :: Assertion+unit_concurrentAliasedPublishes = replicateM_ 200 $ do+ (written, conn) <- fakeBroker [PropTopicAliasMaximum 10]+ c <- runMQTTConduit conn mqttConfig{_protocol=Protocol50}+ mapConcurrently_ (\_ -> pubAliased c "a/b" "x" False QoS0 []) [1 .. 200 :: Int]+ ps <- published 200 written+ forM_ (take 1 ps) $ \p ->+ assertEqual ("first publish names the topic: " <> show p) "a/b" (_pubTopic p)+ stopClient c++-- A QoS 2 publish holds its packet ID until the PUBCOMP, since until then+-- the broker still has the exchange open, and a new PUBLISH reusing the+-- ID would be taken as a duplicate and dropped [MQTT-4.3.3].+unit_qos2HoldsIDUntilPubCOMP :: Assertion+unit_qos2HoldsIDUntilPubCOMP = do+ (written, inject, conn) <- scriptedBroker []+ c <- runMQTTConduit conn mqttConfig{_protocol=Protocol50}+ p <- async $ publishq c "a/b" "x" False QoS2 []+ PublishPkt PublishRequest{_pubPktID=pid} <- nthWritten 1 written+ inject (PubRECPkt (PubREC pid 0 []))+ nthWritten 2 written >>= assertEqual "PUBREL" (PubRELPkt (PubREL pid 0 []))+ threadDelay 100000+ poll p >>= maybe (pure ()) (const $ assertFailure "publish finished before PUBCOMP")+ inject (PubCOMPPkt (PubCOMP pid 0 []))+ timeout 2000000 (wait p) >>= maybe (assertFailure "publish didn't finish after PUBCOMP") pure+ stopClient c++-- The client pings often enough for a Server Keep Alive shorter than its+-- own ping period [MQTT-3.2.2-21].+unit_honorsServerKeepAlive :: Assertion+unit_honorsServerKeepAlive = do+ (written, _, conn) <- scriptedBroker [PropServerKeepAlive 1]+ c <- runMQTTConduit conn mqttConfig{_protocol=Protocol50}+ timeout 1500000 (nthWritten 1 written) >>= \case+ Just PingPkt -> pure ()+ x -> assertFailure ("expected a PINGREQ within 1.5s, got " <> show x)+ stopClient c++-- No more QoS 1 and 2 PUBLISHes are in flight than the server's Receive+-- Maximum [MQTT-3.3.4-7].+unit_honorsReceiveMaximum :: Assertion+unit_honorsReceiveMaximum = do+ (written, inject, conn) <- scriptedBroker [PropReceiveMaximum 1]+ c <- runMQTTConduit conn mqttConfig{_protocol=Protocol50}+ ps <- mapM (\b -> async $ publishq c "a/b" b False QoS1 []) ["one", "two"]+ PublishPkt PublishRequest{_pubPktID=pid} <- nthWritten 1 written+ threadDelay 200000+ n <- length . drop 1 <$> atomically (readTVar written)+ assertEqual "PUBLISHes in flight before the first ack" 1 n+ inject (PubACKPkt (PubACK pid 0 []))+ PublishPkt PublishRequest{_pubPktID=pid2} <- nthWritten 2 written+ inject (PubACKPkt (PubACK pid2 0 []))+ mapM_ (\p -> timeout 2000000 (wait p) >>= maybe (assertFailure "publish didn't finish") pure) ps+ stopClient c++-- An ordered callback gets messages even without a subscribe on this+-- connection (e.g. from a resumed session's subscriptions).+unit_orderedCallbackWithoutSubscribe :: Assertion+unit_orderedCallbackWithoutSubscribe = do+ (_, inject, conn) <- scriptedBroker []+ got <- newTVarIO Nothing+ c <- runMQTTConduit conn mqttConfig{_protocol=Protocol50,+ _msgCB=OrderedCallback (\_ t _ _ -> atomically $ modifyTVar' got (const (Just t)))}+ inject (PublishPkt PublishRequest{_pubDup=False, _pubQoS=QoS0, _pubRetain=False, _pubTopic="a/b",+ _pubPktID=0, _pubBody="x", _pubProps=[]})+ timeout 1000000 (atomically (readTVar got >>= maybe retry pure)) >>=+ assertEqual "callback called" (Just "a/b")+ stopClient c++-- | The client's connection ends (rather than wedging) within a couple of+-- seconds.+connectionEnds :: MQTTClient -> Assertion+connectionEnds c = E.try (timeout 2000000 (waitForClient c)) >>= \case+ Right Nothing -> assertFailure "the connection is still up"+ Left (_ :: E.SomeException) -> pure ()+ Right (Just ()) -> pure ()++-- A PUBLISH whose topic has wildcards is a protocol error [MQTT-3.3.2-2]+-- that ends the connection, rather than wedging it.+unit_wildcardPublishEndsConnection :: Assertion+unit_wildcardPublishEndsConnection = do+ (_, inject, conn) <- scriptedBroker []+ c <- runMQTTConduit conn mqttConfig{_protocol=Protocol50,+ _msgCB=OrderedLowLevelCallback (\_ p -> _pubTopic p `seq` pure ())}+ inject (PublishPkt PublishRequest{_pubDup=False, _pubQoS=QoS1, _pubRetain=False, _pubTopic="a/+",+ _pubPktID=1, _pubBody="x", _pubProps=[]})+ connectionEnds c++-- An ordered callback that fails ends the connection, as an unordered one+-- does, instead of leaving it up with nothing to handle messages.+unit_failingOrderedCallbackEndsConnection :: Assertion+unit_failingOrderedCallbackEndsConnection = do+ (_, inject, conn) <- scriptedBroker []+ c <- runMQTTConduit conn mqttConfig{_protocol=Protocol50,+ _msgCB=OrderedCallback (\_ _ _ _ -> E.throwIO (userError "callback failed"))}+ inject (PublishPkt PublishRequest{_pubDup=False, _pubQoS=QoS1, _pubRetain=False, _pubTopic="a/b",+ _pubPktID=1, _pubBody="x", _pubProps=[]})+ connectionEnds c++-- An ordered client that never subscribed (e.g. a bridge's destination,+-- getting a resumed session's messages) handles, and acks, more than one+-- message: the second used to block the reader for good.+unit_orderedWithoutSubscribeAcksAll :: Assertion+unit_orderedWithoutSubscribeAcksAll = do+ (written, inject, conn) <- scriptedBroker []+ c <- runMQTTConduit conn mqttConfig{_protocol=Protocol50, _msgCB=OrderedCallback (\_ _ _ _ -> pure ())}+ mapM_ (\i -> inject (PublishPkt PublishRequest{_pubDup=False, _pubQoS=QoS1, _pubRetain=False, _pubTopic="a/b",+ _pubPktID=i, _pubBody="x", _pubProps=[]})) [1, 2, 3]+ acks <- mapM (`nthWritten` written) [1, 2, 3]+ assertEqual "PUBACKs" [PubACKPkt (PubACK i 0 []) | i <- [1, 2, 3]] acks+ stopClient c
test/Spec.hs view
@@ -260,3 +260,12 @@ -- ...and bytes left over after its fields make it malformed. unit_connectTrailingBytes :: Assertion unit_connectTrailingBytes = rejects parseConnect (connectWithFlags 0x02 [0x00])++-- In 3.1.1, a SUBSCRIBE's options byte is just the QoS; the rest is+-- reserved [MQTT-3.8.3-4], so 5.0-only options aren't encoded.+prop_subscribe311OnlyQoS :: SubOptions -> QC.Property+prop_subscribe311OnlyQoS o =+ let pkt = SubscribePkt (SubscribeRequest 1 [("a", o)] [])+ in case A.parse (parsePacket Protocol311) (toByteString Protocol311 pkt) of+ A.Done "" got -> got === SubscribePkt (SubscribeRequest 1 [("a", subOptions{_subQoS=_subQoS o})] [])+ x -> counterexample (show x) False