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.0 +version: 0.8.7.1 synopsis: An MQTT Protocol Implementation. description: Please see the README on GitHub at category: Network @@ -127,6 +127,7 @@ type: exitcode-stdio-1.0 main-is: Main.hs other-modules: + ClientSpec DecayingSpec Example1 Example2 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 @@ -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. -- diff --git a/src/Network/MQTT/Types.hs b/src/Network/MQTT/Types.hs --- a/src/Network/MQTT/Types.hs +++ b/src/Network/MQTT/Types.hs @@ -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 diff --git a/test/ClientSpec.hs b/test/ClientSpec.hs new file mode 100644 --- /dev/null +++ b/test/ClientSpec.hs @@ -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 diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -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