diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,25 @@ # Changelog for net-mqtt +## 0.7.0.0 + +`ConnACKFlags` now has a `SessionReuse` type which makes it very clear +whether the session is resuming. I was affected by the boolean +blindness of the previous variant myself several times. + +Connection `_properties` is now called `_connProperties`. I developed +a separate `net-mqtt-lens` package that provides a lens into all +properties of all types that have properties, but a bit more +consistency here is good. + +`QoS` now has an `Ord` instance. + +A new `PktID` type alias makes it clear which `Word16` values were +meant to represent a packet ID. + +There are fewer threads in publish handling in both directions. This +mostly just simplified things, but it also helped prevent a few races +when a lot of values arrived at the same time. + ## 0.6.2.3 Remove a use of `fail` that prevents us from building under ghc 8.8. diff --git a/app/mqtt-watch/Main.hs b/app/mqtt-watch/Main.hs --- a/app/mqtt-watch/Main.hs +++ b/app/mqtt-watch/Main.hs @@ -6,8 +6,7 @@ import Control.Concurrent (threadDelay) import Control.Concurrent.Async (async, link) -import Control.Concurrent.STM (TChan, atomically, newTChanIO, - readTChan, writeTChan) +import Control.Concurrent.STM (TChan, atomically, newTChanIO, readTChan, writeTChan) import Control.Exception (Handler (..), IOException, catches) import Control.Monad (forever, when) import qualified Data.ByteString.Lazy as BL @@ -15,13 +14,11 @@ import qualified Data.Text.IO as TIO import Data.Word (Word32) import Network.MQTT.Client -import Network.MQTT.Types (ConnACKFlags (..)) +import Network.MQTT.Types (ConnACKFlags (..), SessionReuse (..)) import Network.URI -import Options.Applicative (Parser, argument, auto, execParser, - fullDesc, help, helper, info, long, - maybeReader, metavar, option, - progDesc, short, showDefault, some, - str, switch, value, (<**>)) +import Options.Applicative (Parser, argument, auto, execParser, fullDesc, help, helper, info, long, + maybeReader, metavar, option, progDesc, short, showDefault, some, str, + switch, value, (<**>)) import System.IO (stdout) data Msg = Msg Topic BL.ByteString [Property] @@ -31,6 +28,8 @@ , optHideProps :: Bool , optSessionTime :: Word32 , optVerbose :: Bool + , optQoS :: QoS + , optSubResume :: Bool , optTopics :: [Topic] } @@ -40,6 +39,8 @@ <*> switch (short 'p' <> help "hide properties") <*> option auto (long "session-timeout" <> showDefault <> value 0 <> help "mqtt session timeout (0 == clean)") <*> switch (short 'v' <> long "verbose" <> help "enable debug logging") + <*> option (toEnum <$> auto) (long "qos" <> short 'q' <> showDefault <> value QoS0 <> help "QoS level (0-2)") + <*> switch (long "always-subscribe" <> help "subscribe even when resuming a connection") <*> some (argument str (metavar "topics...")) printer :: TChan Msg -> Bool -> IO () @@ -63,17 +64,18 @@ mc <- connectURI mqttConfig{_msgCB=SimpleCallback (showme ch), _protocol=Protocol50, _cleanSession=optSessionTime == 0, _connProps=[PropReceiveMaximum 65535, - PropTopicAliasMaximum 10, + PropTopicAliasMaximum 10000, PropSessionExpiryInterval optSessionTime, PropRequestResponseInformation 1, PropRequestProblemInformation 1]} optUri (ConnACKFlags sp _ props) <- connACK mc - when optVerbose $ putStrLn (if sp then "" else "") + when optVerbose $ putStrLn (if sp == ExistingSession then "" else "") when optVerbose $ putStrLn ("Properties: " <> show props) - subres <- subscribe mc [(t, subOptions) | t <- optTopics] mempty - when optVerbose $ print subres + when (sp == NewSession || optSubResume) $ do + subres <- subscribe mc [(t, subOptions{_subQoS=optQoS}) | t <- optTopics] mempty + when optVerbose $ print subres print =<< waitForClient mc diff --git a/net-mqtt.cabal b/net-mqtt.cabal --- a/net-mqtt.cabal +++ b/net-mqtt.cabal @@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.31.2. +-- This file has been generated from package.yaml by hpack version 0.33.0. -- -- see: https://github.com/sol/hpack -- --- hash: a85ef0bb166997bc2d9f79955e997df1d959d52505a9539765fb63e1f9326742 +-- hash: 4fb6103336dacf66ccbff3745a76866a8f449a508a712b440117af7be4011cac name: net-mqtt -version: 0.6.2.3 +version: 0.7.0.0 synopsis: An MQTT Protocol Implementation. description: Please see the README on GitHub at category: Network diff --git a/src/Network/MQTT/Arbitrary.hs b/src/Network/MQTT/Arbitrary.hs --- a/src/Network/MQTT/Arbitrary.hs +++ b/src/Network/MQTT/Arbitrary.hs @@ -47,7 +47,7 @@ _cleanSession <- arbitrary _keepAlive <- arbitrary _lastWill <- arbitrary - _properties <- arbitrary + _connProperties <- arbitrary pure ConnectRequest{..} @@ -60,6 +60,8 @@ instance Arbitrary QoS where arbitrary = arbitraryBoundedEnum +instance Arbitrary SessionReuse where arbitrary = arbitraryBoundedEnum + instance Arbitrary ConnACKFlags where arbitrary = ConnACKFlags <$> arbitrary <*> arbitrary <*> arbitrary shrink (ConnACKFlags b c pl) @@ -95,7 +97,7 @@ shrink (SubscribeRequest w s p) = if length s < 2 then [] - else [SubscribeRequest w (take 1 s) p' | p' <- shrinkList (const []) p, not (null p)] + else [SubscribeRequest w (take 1 s) p' | not (null p), p' <- shrinkList (const []) p] instance Arbitrary SubOptions where arbitrary = SubOptions <$> arbitraryBoundedEnum <*> arbitrary <*> arbitrary <*> arbitrary @@ -164,7 +166,7 @@ instance Arbitrary MQTTPkt where arbitrary = oneof [ - ConnPkt <$> arbitrary, + ConnPkt <$> arbitrary <*> pure Protocol50, ConnACKPkt <$> arbitrary, PublishPkt <$> arbitrary, PubACKPkt <$> arbitrary, @@ -187,9 +189,9 @@ -- | v311mask strips all the v5 specific bits from an MQTTPkt. v311mask :: MQTTPkt -> MQTTPkt -v311mask (ConnPkt c@ConnectRequest{..}) = ConnPkt (c{_properties=mempty, - _password=mpw _username _password, - _lastWill=cl <$> _lastWill}) +v311mask (ConnPkt c@ConnectRequest{..} _) = ConnPkt (c{_connProperties=mempty, + _password=mpw _username _password, + _lastWill=cl <$> _lastWill}) Protocol311 where cl lw = lw{_willProps=mempty} mpw Nothing _ = Nothing mpw _ p = p 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 @@ -36,29 +36,23 @@ ) where import Control.Concurrent (myThreadId, threadDelay) -import Control.Concurrent.Async (Async, async, asyncThreadId, - cancel, cancelWith, link, race_, - wait, waitAnyCancel, withAsync) -import Control.Concurrent.STM (STM, TChan, TVar, atomically, - modifyTVar', newTChan, newTChanIO, - newTVarIO, readTChan, readTVar, - readTVarIO, retry, writeTChan, - writeTVar) +import Control.Concurrent.Async (Async, async, asyncThreadId, cancelWith, link, race_, wait, waitAnyCancel) +import Control.Concurrent.STM (STM, TChan, TVar, atomically, check, modifyTVar', newTChan, newTChanIO, + newTVarIO, orElse, readTChan, readTVar, readTVarIO, registerDelay, retry, + writeTChan, writeTVar) import Control.DeepSeq (force) import qualified Control.Exception as E -import Control.Monad (forever, guard, void, when) +import Control.Monad (forever, guard, unless, void, when) import Control.Monad.IO.Class (liftIO) +import Data.Bifunctor (first) import qualified Data.ByteString.Char8 as BCS import qualified Data.ByteString.Lazy as BL import qualified Data.ByteString.Lazy.Char8 as BC -import Data.Conduit (ConduitT, Void, await, runConduit, - yield, (.|)) +import Data.Conduit (ConduitT, Void, await, runConduit, yield, (.|)) import Data.Conduit.Attoparsec (conduitParser) import qualified Data.Conduit.Combinators as C -import Data.Conduit.Network (AppData, appSink, appSource, - clientSettings, runTCPClient) -import Data.Conduit.Network.TLS (runTLSClient, tlsClientConfig, - tlsClientTLSSettings) +import Data.Conduit.Network (AppData, appSink, appSource, clientSettings, runTCPClient) +import Data.Conduit.Network.TLS (runTLSClient, tlsClientConfig, tlsClientTLSSettings) import Data.Map.Strict (Map) import qualified Data.Map.Strict as Map import Data.Maybe (fromMaybe) @@ -66,13 +60,9 @@ import qualified Data.Text.Encoding as TE import Data.Word (Word16) import GHC.Conc (labelThread) -import Network.Connection (ConnectionParams (..), - TLSSettings (..), connectTo, - connectionClose, - connectionGetChunk, connectionPut, - initConnectionContext) -import Network.URI (URI (..), unEscapeString, uriPort, - uriRegName, uriUserInfo) +import Network.Connection (ConnectionParams (..), TLSSettings (..), connectTo, connectionClose, + connectionGetChunk, connectionPut, initConnectionContext) +import Network.URI (URI (..), unEscapeString, uriPort, uriRegName, uriUserInfo) import qualified Network.WebSockets as WS import Network.WebSockets.Stream (makeStream) import System.IO.Error (catchIOError, isEOFError) @@ -105,6 +95,7 @@ , _pktID :: TVar Word16 , _cb :: MessageCallback , _acks :: TVar (Map (DispatchType,Word16) (TChan MQTTPkt)) + , _inflight :: TVar (Map Word16 PublishRequest) , _st :: TVar ConnState , _ct :: TVar (Async ()) , _outA :: TVar (Map Topic Word16) @@ -222,15 +213,12 @@ cf True = runWSS wsSource :: WS.Connection -> ConduitT () BCS.ByteString IO () - wsSource ws = loop - where loop = do - bs <- liftIO $ WS.receiveData ws - if BCS.null bs then pure () else yield bs >> loop + wsSource ws = forever $ do + bs <- liftIO $ WS.receiveData ws + unless (BCS.null bs) $ yield bs wsSink :: WS.Connection -> ConduitT BCS.ByteString Void IO () - wsSink ws = loop - where - loop = await >>= maybe (pure ()) (\bs -> liftIO (WS.sendBinaryData ws bs) >> loop) + wsSink ws = maybe (pure ()) (\bs -> liftIO (WS.sendBinaryData ws bs) >> wsSink ws) =<< await runWSS :: String -> Int -> String -> WS.ConnectionOptions -> WS.Headers -> WS.ClientApp () -> IO () runWSS host port path options hdrs' app = do @@ -252,10 +240,7 @@ catchIOError (Just <$> connectionGetChunk conn) (\e -> if isEOFError e then pure Nothing else E.throwIO e) - writer conn maybeBytes = - case maybeBytes of - Nothing -> pure () - Just bytes -> connectionPut conn (BC.toStrict bytes) + writer conn = maybe (pure ()) (connectionPut conn . BC.toStrict) pingPeriod :: Int pingPeriod = 30000000 -- 30 seconds @@ -268,9 +253,6 @@ namedAsync :: String -> IO a -> IO (Async a) namedAsync s a = async a >>= \p -> labelThread (asyncThreadId p) s >> pure p -namedWithAsync :: String -> IO a -> (Async a -> IO b) -> IO b -namedWithAsync n a t = withAsync a (\p -> labelThread (asyncThreadId p) n >> t p) - namedTimeout :: String -> Int -> IO a -> IO (Maybe a) namedTimeout n to a = timeout to (myThreadId >>= \tid -> labelThread tid n >> a) @@ -289,11 +271,12 @@ _ch <- newTChanIO _pktID <- newTVarIO 1 _acks <- newTVarIO mempty + _inflight <- newTVarIO mempty _st <- newTVarIO Starting _ct <- newTVarIO undefined _outA <- newTVarIO mempty _inA <- newTVarIO mempty - _connACKFlags <- newTVarIO (ConnACKFlags False ConnUnspecifiedError mempty) + _connACKFlags <- newTVarIO (ConnACKFlags NewSession ConnUnspecifiedError mempty) _corr <- newTVarIO mempty let _cb = _msgCB cli = MQTTClient{..} @@ -310,7 +293,7 @@ where clientThread cli = E.finally connectAndRun markDisco where - connectAndRun = mkconn $ \ad -> (start cli ad) >>= (run ad) + connectAndRun = mkconn $ \ad -> start cli ad >>= run ad markDisco = atomically $ do st <- readTVar (_st cli) guard $ st == Starting || st == Connected @@ -323,7 +306,7 @@ T._username=BC.pack <$> _username, T._password=BC.pack <$> _password, T._cleanSession=_cleanSession, - T._properties=_connProps} + T._connProperties=_connProps} yield (BL.toStrict $ toByteString _protocol req) .| sink pure c @@ -342,9 +325,7 @@ .| conduitParser (parsePacket _protocol) .| C.mapM_ (\(_,x) -> liftIO (dispatch c pch x)) - onceConnected = atomically $ do - s <- readTVar _st - if s /= Connected then retry else pure () + onceConnected = atomically $ check . (== Connected) =<< readTVar _st processOut = runConduit $ C.repeatM (liftIO (atomically $ checkConnected c >> readTChan _ch)) @@ -353,13 +334,10 @@ doPing = forever $ threadDelay pingPeriod >> sendPacketIO c PingPkt - watchdog ch = do - r <- namedTimeout "MQTT ping timeout" (pingPeriod * 3) w - case r of - Nothing -> killConn c Timeout - Just _ -> watchdog ch - - where w = atomically . readTChan $ ch + watchdog ch = forever $ do + toch <- registerDelay (pingPeriod * 3) + timedOut <- atomically $ ((check =<< readTVar toch) >> pure True) `orElse` (readTChan ch >> pure False) + when timedOut $ killConn c Timeout waitForLaunch MQTTClient{..} t = do writeTVar _ct t @@ -398,12 +376,12 @@ dispatch c@MQTTClient{..} pch pkt = case pkt of (ConnACKPkt p) -> connACKd p - (PublishPkt p) -> void $ namedAsync "MQTT pub machine" (pubMachine p) >>= link + (PublishPkt p) -> pub p (SubACKPkt (SubscribeResponse i _ _)) -> delegate DSubACK i (UnsubACKPkt (UnsubscribeResponse i _ _)) -> delegate DUnsubACK i (PubACKPkt (PubACK i _ _)) -> delegate DPubACK i + (PubRELPkt (PubREL i _ _)) -> pubd i (PubRECPkt (PubREC i _ _)) -> delegate DPubREC i - (PubRELPkt (PubREL i _ _)) -> delegate DPubREL i (PubCOMPPkt (PubCOMP i _ _)) -> delegate DPubCOMP i (DisconnectPkt req) -> disco req PongPkt -> atomically . writeTChan pch $ True @@ -418,6 +396,54 @@ atomically $ writeTVar _st (ConnErr connr) cancelWith t (MQTTException $ show connr) + pub p@PublishRequest{_pubQoS=QoS0} = atomically (resolve p) >>= notify Nothing + pub p@PublishRequest{_pubQoS=QoS1, _pubPktID} = do + notify (Just (PubACKPkt (PubACK _pubPktID 0 mempty))) =<< atomically (resolve p) + pub p@PublishRequest{_pubQoS=QoS2} = atomically $ do + p'@PublishRequest{..} <- resolve p + modifyTVar' _inflight (Map.insert _pubPktID p') + sendPacket c (PubRECPkt (PubREC _pubPktID 0 mempty)) + + pubd i = do + mp <- atomically $ do + r <- Map.lookup i <$> readTVar _inflight + modifyTVar' _inflight (Map.delete i) + pure r + case mp of + Nothing -> sendPacketIO c (PubCOMPPkt (PubCOMP i 0x92 mempty)) + Just p -> notify (Just (PubCOMPPkt (PubCOMP i 0 mempty))) p + + notify rpkt p@PublishRequest{..} = do + atomically $ modifyTVar' _inflight (Map.delete _pubPktID) + corrs <- readTVarIO _corr + E.evaluate . force =<< case maybe _cb (\cd -> Map.findWithDefault _cb cd corrs) cdata of + NoCallback -> pure () + SimpleCallback f -> call (f c (blToText _pubTopic) _pubBody _pubProps) + LowLevelCallback f -> call (f c p) + + where + call a = link =<< namedAsync "notifier" (a >> respond) + respond = void $ traverse (sendPacketIO c) rpkt + cdata = foldr f Nothing _pubProps + where f (PropCorrelationData x) _ = Just x + f _ o = o + + resolve p@PublishRequest{..} = do + topic <- resolveTopic (foldr aliasID Nothing _pubProps) + pure p{_pubTopic=textToBL topic} + + where + aliasID (PropTopicAlias x) _ = Just x + aliasID _ o = o + + resolveTopic Nothing = pure (blToText _pubTopic) + resolveTopic (Just x) = do + when (_pubTopic /= "") $ modifyTVar' _inA (Map.insert x (blToText _pubTopic)) + m <- readTVar _inA + case Map.lookup x m of + Nothing -> mqttFail ("failed to lookup topic alias " <> show x) + Just t -> pure t + delegate dt pid = atomically $ do m <- readTVar _acks case Map.lookup (dt, pid) m of @@ -426,7 +452,6 @@ where nak DPubREC = sendPacket c (PubRELPkt (PubREL pid 0x92 mempty)) - nak DPubREL = sendPacket c (PubCOMPPkt (PubCOMP pid 0x92 mempty)) nak _ = pure () @@ -435,58 +460,11 @@ atomically $ writeTVar _st (DiscoErr req) cancelWith t (Discod req) - pubMachine pr@PublishRequest{..} - | _pubQoS == QoS2 = manageQoS2 - | _pubQoS == QoS1 = notify >> sendPacketIO c (PubACKPkt (PubACK _pubPktID 0 mempty)) - | otherwise = notify - - where - cdata = foldr f Nothing _pubProps - where f (PropCorrelationData x) _ = Just x - f _ o = o - - notify = do - topic <- resolveTopic (foldr aliasID Nothing _pubProps) - corrs <- readTVarIO _corr - E.evaluate . force =<< case maybe _cb (\cd -> Map.findWithDefault _cb cd corrs) cdata of - NoCallback -> pure () - SimpleCallback f -> f c topic _pubBody _pubProps - LowLevelCallback f -> f c pr{_pubTopic=textToBL topic} - - resolveTopic Nothing = pure (blToText _pubTopic) - resolveTopic (Just x) = do - when (_pubTopic /= "") $ atomically $ modifyTVar' _inA (Map.insert x (blToText _pubTopic)) - m <- readTVarIO _inA - pure (m Map.! x) - - aliasID (PropTopicAlias x) _ = Just x - aliasID _ o = o - - manageQoS2 = do - ch <- newTChanIO - atomically $ modifyTVar' _acks (Map.insert (DPubREL, _pubPktID) ch) - E.finally (manageQoS2' ch) (atomically $ releasePktID c (DPubREL, _pubPktID)) - where - sendREC ch = do - sendPacketIO c (PubRECPkt (PubREC _pubPktID 0 mempty)) - (PubRELPkt _) <- atomically $ readTChan ch - pure () - - manageQoS2' ch = do - v <- namedTimeout "QoS2 sendREC" 10000000 (sendREC ch) - case v of - Nothing -> killConn c Timeout - _ -> notify >> sendPacketIO c (PubCOMPPkt (PubCOMP _pubPktID 0 mempty)) - killConn :: E.Exception e => MQTTClient -> e -> IO () killConn MQTTClient{..} e = readTVarIO _ct >>= \t -> cancelWith t e checkConnected :: MQTTClient -> STM () -checkConnected mc = do - e <- stateX mc Connected - case e of - Nothing -> pure () - Just x -> E.throw x +checkConnected mc = maybe (pure ()) E.throw =<< stateX mc Connected -- | True if we're currently in a normally connected state (in the IO monad). isConnected :: MQTTClient -> IO Bool @@ -546,7 +524,7 @@ let (SubACKPkt (SubscribeResponse _ rs aprops)) = r pure (rs, aprops) - where ls' = map (\(s, i) -> (textToBL s, i)) ls + where ls' = map (first textToBL) ls -- | Unsubscribe from a list of topic filters. -- @@ -578,27 +556,23 @@ -> IO () publishq c t m r q props = do (ch,pid) <- atomically $ reservePktID c types - E.finally (publishAndWait ch pid q) (atomically $ releasePktIDs c [(t',pid) | t' <- types]) + E.finally (publishAndWait ch pid) (atomically $ releasePktIDs c [(t',pid) | t' <- types]) where types = [DPubACK, DPubREC, DPubCOMP] - publishAndWait _ pid QoS0 = sendPacketIO c (pkt False pid) - publishAndWait ch pid _ = namedWithAsync "MQTT satisfyQoS" (pub False pid) (\p -> satisfyQoS p ch pid) - - pub dup pid = do - sendPacketIO c (pkt dup pid) - threadDelay 5000000 - pub True pid + publishAndWait ch pid = do + sendPacketIO c (pkt pid) + when (q > QoS0) $ satisfyQoS ch pid - pkt dup pid = PublishPkt $ PublishRequest {_pubDup = dup, - _pubQoS = q, - _pubPktID = pid, - _pubRetain = r, - _pubTopic = textToBL t, - _pubBody = m, - _pubProps = props} + pkt pid = PublishPkt $ PublishRequest {_pubDup = False, + _pubQoS = q, + _pubPktID = pid, + _pubRetain = r, + _pubTopic = textToBL t, + _pubBody = m, + _pubProps = props} - satisfyQoS p ch pid + satisfyQoS ch pid | q == QoS0 = pure () | q == QoS1 = void $ do (PubACKPkt (PubACK _ st pprops)) <- atomically $ checkConnected c >> readTChan ch @@ -613,7 +587,6 @@ PubRECPkt (PubREC _ st recprops) -> do when (st /= 0) $ mqttFail ("qos 2 REC publish error: " <> show st <> " " <> show recprops) sendPacketIO c (PubRELPkt $ PubREL pid 0 mempty) - cancel p -- must not publish after rel 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) 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 @@ -14,13 +14,14 @@ module Network.MQTT.Types ( LastWill(..), MQTTPkt(..), QoS(..), - ConnectRequest(..), connectRequest, ConnACKFlags(..), ConnACKRC(..), + ConnectRequest(..), connectRequest, SessionReuse(..), ConnACKFlags(..), ConnACKRC(..), PublishRequest(..), PubACK(..), PubREC(..), PubREL(..), PubCOMP(..), ProtocolLevel(..), Property(..), AuthRequest(..), SubscribeRequest(..), SubOptions(..), subOptions, SubscribeResponse(..), SubErr(..), RetainHandling(..), DisconnectRequest(..), UnsubscribeRequest(..), UnsubscribeResponse(..), UnsubStatus(..), DiscoReason(..), - parsePacket, ByteMe(toByteString), + PktID, + parsePacket, ByteMe(toByteString), parseConnect, -- for testing encodeLength, parseHdrLen, parseProperty, parseProperties, bsProps, parseSubOptions, ByteSize(..) @@ -31,22 +32,20 @@ import Data.Attoparsec.Binary (anyWord16be, anyWord32be) import qualified Data.Attoparsec.ByteString.Lazy as A import Data.Binary.Put (putWord32be, runPut) -import Data.Bits (Bits (..), shiftL, testBit, - (.&.), (.|.)) +import Data.Bits (Bits (..), shiftL, testBit, (.&.), (.|.)) import qualified Data.ByteString.Lazy as BL import Data.Functor (($>)) -import Data.List (lookup) import Data.Maybe (fromMaybe, isJust) import Data.Word (Word16, Word32, Word8) -- | QoS values for publishing and subscribing. -data QoS = QoS0 | QoS1 | QoS2 deriving (Bounded, Enum, Eq, Show) +data QoS = QoS0 | QoS1 | QoS2 deriving (Bounded, Enum, Eq, Show, Ord) qosW :: QoS -> Word8 -qosW = toEnum.fromEnum +qosW = toEnum . fromEnum wQos :: Word8 -> QoS -wQos = toEnum.fromEnum +wQos = toEnum . fromIntegral (≫) :: Bits a => a -> Int -> a (≫) = shiftR @@ -78,7 +77,7 @@ go :: Int -> Int -> A.Parser Int go v m = do x <- A.anyWord8 - let a = fromEnum (x .&. 127) * m + v + let a = fromIntegral (x .&. 127) * m + v if x .&. 128 /= 0 then go a (m*128) else pure a @@ -101,7 +100,7 @@ encodeWord16 :: Word16 -> BL.ByteString encodeWord16 a = let (h,l) = a `quotRem` 256 in BL.pack [w h, w l] - where w = toEnum.fromEnum + where w = toEnum . fromIntegral encodeWord32 :: Word32 -> BL.ByteString encodeWord32 = runPut . putWord32be @@ -116,16 +115,16 @@ encodeUTF8Pair x y = encodeUTF8 x <> encodeUTF8 y twoByteLen :: BL.ByteString -> BL.ByteString -twoByteLen = encodeWord16 . toEnum . fromEnum . BL.length +twoByteLen = encodeWord16 . fromIntegral . BL.length blLength :: BL.ByteString -> BL.ByteString -blLength = BL.pack . encodeVarInt . fromEnum . BL.length +blLength = BL.pack . encodeVarInt . fromIntegral . BL.length withLength :: BL.ByteString -> BL.ByteString withLength a = blLength a <> a instance ByteMe BL.ByteString where - toByteString _ a = (encodeWord16 . toEnum . fromEnum . BL.length) a <> a + toByteString _ a = (encodeWord16 . fromIntegral . BL.length) a <> a -- | Property represents the various MQTT Properties that may sent or -- received along with packets in MQTT 5. For detailed use on when @@ -263,8 +262,8 @@ bsProps :: ProtocolLevel -> [Property] -> BL.ByteString bsProps Protocol311 _ = mempty -bsProps p l = let b = (mconcat . map (toByteString p)) l in - (BL.pack . encodeLength . fromEnum . BL.length) b <> b +bsProps p l = let b = foldMap (toByteString p) l in + (BL.pack . encodeLength . fromIntegral . BL.length) b <> b parseProperties :: ProtocolLevel -> A.Parser [Property] parseProperties Protocol311 = pure mempty @@ -291,23 +290,22 @@ } deriving(Eq, Show) data ConnectRequest = ConnectRequest { - _username :: Maybe BL.ByteString - , _password :: Maybe BL.ByteString - , _lastWill :: Maybe LastWill - , _cleanSession :: Bool - , _keepAlive :: Word16 - , _connID :: BL.ByteString - , _properties :: [Property] + _username :: Maybe BL.ByteString + , _password :: Maybe BL.ByteString + , _lastWill :: Maybe LastWill + , _cleanSession :: Bool + , _keepAlive :: Word16 + , _connID :: BL.ByteString + , _connProperties :: [Property] } deriving (Eq, Show) connectRequest :: ConnectRequest connectRequest = ConnectRequest{_username=Nothing, _password=Nothing, _lastWill=Nothing, _cleanSession=True, _keepAlive=300, _connID="", - _properties=mempty} + _connProperties=mempty} instance ByteMe ConnectRequest where - toByteString prot ConnectRequest{..} = BL.singleton 0x10 - <> withLength (val prot) + toByteString prot ConnectRequest{..} = BL.singleton 0x10 <> withLength (val prot) where val :: ProtocolLevel -> BL.ByteString val Protocol311 = "\NUL\EOTMQTT\EOT" -- MQTT + Protocol311 @@ -321,7 +319,7 @@ val Protocol50 = "\NUL\EOTMQTT\ENQ" -- MQTT + Protocol50 <> BL.singleton connBits <> encodeWord16 _keepAlive - <> bsProps prot _properties + <> bsProps prot _connProperties <> toByteString prot _connID <> lwt _lastWill <> perhaps _username @@ -333,7 +331,7 @@ hasp = boolBit ((prot == Protocol50 || isJust _username) && isJust _password) ≪ 6 clean = boolBit _cleanSession ≪ 1 willBits = case _lastWill of - Nothing -> 0 + Nothing -> 0 Just LastWill{..} -> 4 .|. ((qosW _willQoS .&. 0x3) ≪ 3) .|. (boolBit _willRetain ≪ 5) lwt :: Maybe LastWill -> BL.ByteString @@ -347,7 +345,7 @@ perhaps (Just s) = toByteString prot s -data MQTTPkt = ConnPkt ConnectRequest +data MQTTPkt = ConnPkt ConnectRequest ProtocolLevel | ConnACKPkt ConnACKFlags | PublishPkt PublishRequest | PubACKPkt PubACK @@ -365,7 +363,7 @@ deriving (Eq, Show) instance ByteMe MQTTPkt where - toByteString p (ConnPkt x) = toByteString p x + toByteString p (ConnPkt x _) = toByteString p x toByteString p (ConnACKPkt x) = toByteString p x toByteString p (PublishPkt x) = toByteString p x toByteString p (PubACKPkt x) = toByteString p x @@ -400,9 +398,12 @@ aString :: A.Parser BL.ByteString aString = do n <- aWord16 - s <- A.take (fromEnum n) + s <- A.take (fromIntegral n) pure $ BL.fromStrict s +-- | Parse a CONNect packet. This is useful when examining the +-- beginning of the stream as it allows you to determine the protocol +-- being used throughout the rest of the session. parseConnect :: A.Parser MQTTPkt parseConnect = do _ <- A.word8 0x10 @@ -421,7 +422,7 @@ pure $ ConnPkt ConnectRequest{_connID=cid, _username=u, _password=p, _lastWill=lwt, _keepAlive=keepAlive, _cleanSession=testBit connFlagBits 1, - _properties=props} + _connProperties=props} pl where mstr :: Bool -> A.Parser (Maybe BL.ByteString) @@ -508,14 +509,17 @@ connACKRev :: [(Word8, ConnACKRC)] connACKRev = map (\w -> (toByte w, w)) [minBound..] +data SessionReuse = NewSession | ExistingSession deriving (Show, Eq, Bounded, Enum) -data ConnACKFlags = ConnACKFlags Bool ConnACKRC [Property] deriving (Eq, Show) +-- | Connection acknowledgment details. +data ConnACKFlags = ConnACKFlags SessionReuse ConnACKRC [Property] deriving (Eq, Show) instance ByteMe ConnACKFlags where - toBytes prot (ConnACKFlags sp rc props) = let pbytes = BL.unpack $ bsProps prot props in - [0x20] - <> encodeVarInt (2 + length pbytes) - <>[ boolBit sp, toByte rc] <> pbytes + toBytes prot (ConnACKFlags sp rc props) = + let pbytes = BL.unpack $ bsProps prot props in + [0x20] + <> encodeVarInt (2 + length pbytes) + <>[ boolBit (sp /= NewSession), toByte rc] <> pbytes parseConnectACK :: A.Parser MQTTPkt parseConnectACK = do @@ -525,21 +529,27 @@ ackFlags <- A.anyWord8 rc <- A.anyWord8 p <- parseProperties (if rl == 2 then Protocol311 else Protocol50) - pure $ ConnACKPkt $ ConnACKFlags (testBit ackFlags 0) (fromByte rc) p + pure $ ConnACKPkt $ ConnACKFlags (sf $ testBit ackFlags 0) (fromByte rc) p + where sf False = NewSession + sf True = ExistingSession + +type PktID = Word16 + data PublishRequest = PublishRequest{ _pubDup :: Bool , _pubQoS :: QoS , _pubRetain :: Bool , _pubTopic :: BL.ByteString - , _pubPktID :: Word16 + , _pubPktID :: PktID , _pubBody :: BL.ByteString , _pubProps :: [Property] } deriving(Eq, Show) instance ByteMe PublishRequest where - toByteString prot PublishRequest{..} = BL.singleton (0x30 .|. f) - <> withLength val + toByteString prot PublishRequest{..} = + BL.singleton (0x30 .|. f) <> withLength val + where f = (db ≪ 3) .|. (qb ≪ 1) .|. rb db = boolBit _pubDup qb = qosW _pubQoS .&. 0x3 @@ -559,7 +569,7 @@ _pubTopic <- aString _pubPktID <- if _pubQoS == QoS0 then pure 0 else aWord16 _pubProps <- parseProperties prot - _pubBody <- BL.fromStrict <$> A.take (plen - fromEnum (BL.length _pubTopic) - 2 + _pubBody <- BL.fromStrict <$> A.take (plen - fromIntegral (BL.length _pubTopic) - 2 - qlen _pubQoS - propLen prot _pubProps ) pure $ PublishPkt PublishRequest{..} @@ -618,16 +628,16 @@ _subQoS=wQos (w .&. 0x3)} subOptionsBytes :: ProtocolLevel -> [(BL.ByteString, SubOptions)] -> BL.ByteString -subOptionsBytes prot = BL.concat . map (\(bs,so) -> toByteString prot bs <> toByteString prot so) +subOptionsBytes prot = foldMap (\(bs,so) -> toByteString prot bs <> toByteString prot so) -data SubscribeRequest = SubscribeRequest Word16 [(BL.ByteString, SubOptions)] [Property] +data SubscribeRequest = SubscribeRequest PktID [(BL.ByteString, SubOptions)] [Property] deriving(Eq, Show) instance ByteMe SubscribeRequest where toByteString prot (SubscribeRequest pid sreq props) = BL.singleton 0x82 <> withLength (encodeWord16 pid <> bsProps prot props <> subOptionsBytes prot sreq) -data PubACK = PubACK Word16 Word8 [Property] deriving(Eq, Show) +data PubACK = PubACK PktID Word8 [Property] deriving(Eq, Show) bsPubSeg :: ProtocolLevel -> Word8 -> Word16 -> Word8 -> [Property] -> BL.ByteString bsPubSeg Protocol311 h pid _ _ = BL.singleton h <> withLength (encodeWord16 pid) @@ -642,7 +652,7 @@ instance ByteMe PubACK where toByteString prot (PubACK pid st props) = bsPubSeg prot 0x40 pid st props -parsePubSeg :: A.Parser (Word16, Word8, [Property]) +parsePubSeg :: A.Parser (PktID, Word8, [Property]) parsePubSeg = do rl <- parseHdrLen mid <- aWord16 @@ -656,7 +666,7 @@ (mid, st, props) <- parsePubSeg pure $ PubACKPkt (PubACK mid st props) -data PubREC = PubREC Word16 Word8 [Property] deriving(Eq, Show) +data PubREC = PubREC PktID Word8 [Property] deriving(Eq, Show) instance ByteMe PubREC where toByteString prot (PubREC pid st props) = bsPubSeg prot 0x50 pid st props @@ -667,7 +677,7 @@ (mid, st, props) <- parsePubSeg pure $ PubRECPkt (PubREC mid st props) -data PubREL = PubREL Word16 Word8 [Property] deriving(Eq, Show) +data PubREL = PubREL PktID Word8 [Property] deriving(Eq, Show) instance ByteMe PubREL where toByteString prot (PubREL pid st props) = bsPubSeg prot 0x62 pid st props @@ -678,7 +688,7 @@ (mid, st, props) <- parsePubSeg pure $ PubRELPkt (PubREL mid st props) -data PubCOMP = PubCOMP Word16 Word8 [Property] deriving(Eq, Show) +data PubCOMP = PubCOMP PktID Word8 [Property] deriving(Eq, Show) instance ByteMe PubCOMP where toByteString prot (PubCOMP pid st props) = bsPubSeg prot 0x70 pid st props @@ -690,13 +700,13 @@ pure $ PubCOMPPkt (PubCOMP mid st props) -- Common header bits for subscribe, unsubscribe, and the sub acks. -parseSubHdr :: Word8 -> ProtocolLevel -> A.Parser a -> A.Parser (Word16, [Property], a) +parseSubHdr :: Word8 -> ProtocolLevel -> A.Parser a -> A.Parser (PktID, [Property], a) parseSubHdr b prot p = do _ <- A.word8 b hl <- parseHdrLen pid <- aWord16 props <- parseProperties prot - content <- A.take (fromEnum hl - 2 - propLen prot props) + content <- A.take (fromIntegral hl - 2 - propLen prot props) a <- subp content pure (pid, props, a) @@ -707,7 +717,7 @@ (pid, props, subs) <- parseSubHdr 0x82 prot $ A.many1 (liftA2 (,) aString parseSubOptions) pure $ SubscribePkt (SubscribeRequest pid subs props) -data SubscribeResponse = SubscribeResponse Word16 [Either SubErr QoS] [Property] deriving (Eq, Show) +data SubscribeResponse = SubscribeResponse PktID [Either SubErr QoS] [Property] deriving (Eq, Show) instance ByteMe SubscribeResponse where toByteString prot (SubscribeResponse pid sres props) = @@ -727,7 +737,7 @@ propLen :: ProtocolLevel -> [Property] -> Int propLen Protocol311 _ = 0 -propLen prot props = fromEnum $ BL.length (bsProps prot props) +propLen prot props = fromIntegral $ BL.length (bsProps prot props) data SubErr = SubErrUnspecifiedError | SubErrImplementationSpecificError @@ -757,7 +767,7 @@ p 0xA2 = Left SubErrWildcardSubscriptionsNotSupported p x = Right (wQos x) -data UnsubscribeRequest = UnsubscribeRequest Word16 [BL.ByteString] [Property] +data UnsubscribeRequest = UnsubscribeRequest PktID [BL.ByteString] [Property] deriving(Eq, Show) instance ByteMe UnsubscribeRequest where @@ -788,16 +798,16 @@ toByteString _ UnsubTopicFilterInvalid = BL.singleton 0x8F toByteString _ UnsubPacketIdentifierInUse = BL.singleton 0x91 -data UnsubscribeResponse = UnsubscribeResponse Word16 [Property] [UnsubStatus] deriving(Eq, Show) +data UnsubscribeResponse = UnsubscribeResponse PktID [Property] [UnsubStatus] deriving(Eq, Show) instance ByteMe UnsubscribeResponse where - toByteString Protocol311 (UnsubscribeResponse pid _ _) = BL.singleton 0xb0 - <> withLength (encodeWord16 pid) + toByteString Protocol311 (UnsubscribeResponse pid _ _) = + BL.singleton 0xb0 <> withLength (encodeWord16 pid) - toByteString Protocol50 (UnsubscribeResponse pid props res) = BL.singleton 0xb0 - <> withLength (encodeWord16 pid - <> bsProps Protocol50 props - <> mconcat (fmap (toByteString Protocol50) res)) + toByteString Protocol50 (UnsubscribeResponse pid props res) = + BL.singleton 0xb0 <> withLength (encodeWord16 pid + <> bsProps Protocol50 props + <> mconcat (fmap (toByteString Protocol50) res)) parseUnsubACK :: ProtocolLevel -> A.Parser MQTTPkt parseUnsubACK Protocol311 = do @@ -827,8 +837,8 @@ data AuthRequest = AuthRequest Word8 [Property] deriving (Eq, Show) instance ByteMe AuthRequest where - toByteString prot (AuthRequest i props) = BL.singleton 0xf0 - <> withLength (BL.singleton i <> bsProps prot props) + toByteString prot (AuthRequest i props) = + BL.singleton 0xf0 <> withLength (BL.singleton i <> bsProps prot props) parseAuth :: A.Parser MQTTPkt parseAuth = do @@ -910,9 +920,8 @@ instance ByteMe DisconnectRequest where toByteString Protocol311 _ = "\224\NUL" - toByteString Protocol50 (DisconnectRequest r props) = BL.singleton 0xe0 - <> withLength (BL.singleton (toByte r) - <> bsProps Protocol50 props) + toByteString Protocol50 (DisconnectRequest r props) = + BL.singleton 0xe0 <> withLength (BL.singleton (toByte r) <> bsProps Protocol50 props) parseDisconnect :: ProtocolLevel -> A.Parser MQTTPkt parseDisconnect Protocol311 = do