net-mqtt 0.6.2.3 → 0.7.0.0
raw patch · 6 files changed
+210/−204 lines, 6 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
- Network.MQTT.Types: [_properties] :: ConnectRequest -> [Property]
+ Network.MQTT.Arbitrary: instance Test.QuickCheck.Arbitrary.Arbitrary Network.MQTT.Types.SessionReuse
+ Network.MQTT.Types: ExistingSession :: SessionReuse
+ Network.MQTT.Types: NewSession :: SessionReuse
+ Network.MQTT.Types: [_connProperties] :: ConnectRequest -> [Property]
+ Network.MQTT.Types: data SessionReuse
+ Network.MQTT.Types: instance GHC.Classes.Eq Network.MQTT.Types.SessionReuse
+ Network.MQTT.Types: instance GHC.Classes.Ord Network.MQTT.Types.QoS
+ Network.MQTT.Types: instance GHC.Enum.Bounded Network.MQTT.Types.SessionReuse
+ Network.MQTT.Types: instance GHC.Enum.Enum Network.MQTT.Types.SessionReuse
+ Network.MQTT.Types: instance GHC.Show.Show Network.MQTT.Types.SessionReuse
+ Network.MQTT.Types: parseConnect :: Parser MQTTPkt
+ Network.MQTT.Types: type PktID = Word16
- Network.MQTT.Types: ConnACKFlags :: Bool -> ConnACKRC -> [Property] -> ConnACKFlags
+ Network.MQTT.Types: ConnACKFlags :: SessionReuse -> ConnACKRC -> [Property] -> ConnACKFlags
- Network.MQTT.Types: ConnPkt :: ConnectRequest -> MQTTPkt
+ Network.MQTT.Types: ConnPkt :: ConnectRequest -> ProtocolLevel -> MQTTPkt
- Network.MQTT.Types: PubACK :: Word16 -> Word8 -> [Property] -> PubACK
+ Network.MQTT.Types: PubACK :: PktID -> Word8 -> [Property] -> PubACK
- Network.MQTT.Types: PubCOMP :: Word16 -> Word8 -> [Property] -> PubCOMP
+ Network.MQTT.Types: PubCOMP :: PktID -> Word8 -> [Property] -> PubCOMP
- Network.MQTT.Types: PubREC :: Word16 -> Word8 -> [Property] -> PubREC
+ Network.MQTT.Types: PubREC :: PktID -> Word8 -> [Property] -> PubREC
- Network.MQTT.Types: PubREL :: Word16 -> Word8 -> [Property] -> PubREL
+ Network.MQTT.Types: PubREL :: PktID -> Word8 -> [Property] -> PubREL
- Network.MQTT.Types: PublishRequest :: Bool -> QoS -> Bool -> ByteString -> Word16 -> ByteString -> [Property] -> PublishRequest
+ Network.MQTT.Types: PublishRequest :: Bool -> QoS -> Bool -> ByteString -> PktID -> ByteString -> [Property] -> PublishRequest
- Network.MQTT.Types: SubscribeRequest :: Word16 -> [(ByteString, SubOptions)] -> [Property] -> SubscribeRequest
+ Network.MQTT.Types: SubscribeRequest :: PktID -> [(ByteString, SubOptions)] -> [Property] -> SubscribeRequest
- Network.MQTT.Types: SubscribeResponse :: Word16 -> [Either SubErr QoS] -> [Property] -> SubscribeResponse
+ Network.MQTT.Types: SubscribeResponse :: PktID -> [Either SubErr QoS] -> [Property] -> SubscribeResponse
- Network.MQTT.Types: UnsubscribeRequest :: Word16 -> [ByteString] -> [Property] -> UnsubscribeRequest
+ Network.MQTT.Types: UnsubscribeRequest :: PktID -> [ByteString] -> [Property] -> UnsubscribeRequest
- Network.MQTT.Types: UnsubscribeResponse :: Word16 -> [Property] -> [UnsubStatus] -> UnsubscribeResponse
+ Network.MQTT.Types: UnsubscribeResponse :: PktID -> [Property] -> [UnsubStatus] -> UnsubscribeResponse
- Network.MQTT.Types: [_pubPktID] :: PublishRequest -> Word16
+ Network.MQTT.Types: [_pubPktID] :: PublishRequest -> PktID
Files
- Changelog.md +20/−0
- app/mqtt-watch/Main.hs +14/−12
- net-mqtt.cabal +3/−3
- src/Network/MQTT/Arbitrary.hs +8/−6
- src/Network/MQTT/Client.hs +91/−118
- src/Network/MQTT/Types.hs +74/−65
Changelog.md view
@@ -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.
app/mqtt-watch/Main.hs view
@@ -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 "<resuming session>" else "<new session>")+ when optVerbose $ putStrLn (if sp == ExistingSession then "<resuming session>" else "<new session>") 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
net-mqtt.cabal view
@@ -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 <https://github.com/dustin/mqtt-hs#readme> category: Network
src/Network/MQTT/Arbitrary.hs view
@@ -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
src/Network/MQTT/Client.hs view
@@ -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)
src/Network/MQTT/Types.hs view
@@ -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