diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,24 @@ # Changelog for net-mqtt +## 0.5.0.0 + +Major release for MQTT version 5. + +The API is mostly the same, but a list of Property values is passed in +and returned from a few different fields. + +Subscribe responses are now more detailed in the error case, and also +return a `[Property]`. + +Connections default to `Protocol311` (3.1.1), and all behavior should +be backwards compatible in these cases. i.e., you can write code as +if it were destined for a v5 broker, but properties won't be sent and +responses will be inferred. If you specify your `_protocol` as +`Protocol50` in your `MQTTConfig`, the new features should all work. + +Various bugs were fixed along the path of making v5 compatibility, but +I'm pretty sure there's one left somewhere. + ## 0.2.4.2 Don't set a message ID of 0. diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -10,19 +10,35 @@ main :: IO () main = do - mc <- runClient mqttConfig{_hostname="localhost", _port=1883, _connID="hasqttl", + mc <- runClient mqttConfig{_hostname="test.mosquitto.org", _port=1883, _connID="hasq", -- _cleanSession=False, - _lwt=Just $ mkLWT "tmp/haskquit" "bye for now" False, - _msgCB=Just showme} + _lwt=Just $ (mkLWT "tmp/haskquit" "bye for now" False){ + _willProps=[PropUserProperty "lwt" "prop"]}, + _msgCB=Just showme, _protocol=Protocol50, + _connProps=[PropReceiveMaximum 65535, + PropTopicAliasMaximum 10, + PropRequestResponseInformation 1, + PropRequestProblemInformation 1]} putStrLn "connected!" - print =<< subscribe mc [("oro/#", QoS0), ("tmp/#", QoS2)] - p <- async $ forever $ publish mc "tmp/mqtths" "hi from haskell" False >> threadDelay 10000000 + print =<< svrProps mc + print =<< subscribe mc [("oro/#", defaultSubOptions), ("tmp/#", defaultSubOptions{_subQoS=QoS2})] + let pprops = [PropUserProperty "hello" "mqttv5"] + p <- async $ forever $ pubAliased mc "tmp/hi/from/haskell" "hi from haskell" False QoS1 pprops >> threadDelay 10000000 + putStrLn "Publishing at QoS > 0" - publishq mc "tmp/q" "this message is at QoS 2" False QoS2 + publishq mc "tmp/q1" "this message is at QoS 1" True QoS1 [PropUserProperty "hi" "there 1", + PropMessageExpiryInterval 30, + PropContentType "text/plain"] putStrLn "Published!" + putStrLn "Publishing at QoS > 1" + publishq mc "tmp/q2" "this message is at QoS 2" True QoS2 [PropUserProperty "hi" "there 2", + PropMessageExpiryInterval 30, + PropContentType "text/plain"] + putStrLn "Published!" + print =<< waitForClient mc cancel p - where showme _ t m = print (t,m) + where showme _ t m p = print (t,m,p) diff --git a/net-mqtt.cabal b/net-mqtt.cabal --- a/net-mqtt.cabal +++ b/net-mqtt.cabal @@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: 90cb428a2c77b19e3ed4d018bbbcb1db82904497e9599d297097926ab1bfe039 +-- hash: d7aba3d326ea418336a888f75d196bf0c3bd7c6d1ef718accc71709d2eeb8c3c name: net-mqtt -version: 0.2.4.2 +version: 0.5.0.0 synopsis: An MQTT Protocol Implementation. description: Please see the README on GitHub at category: Network @@ -40,6 +40,7 @@ build-depends: async >=2.2.1 && <2.3 , attoparsec >=0.13.2 && <0.14 + , attoparsec-binary >=0.2 && <1.0 , base >=4.7 && <5 , binary >=0.8.6 && <0.9 , bytestring >=0.10.8 && <0.11 @@ -62,6 +63,7 @@ build-depends: async >=2.2.1 && <2.3 , attoparsec >=0.13.2 && <0.14 + , attoparsec-binary >=0.2 && <1.0 , base >=4.7 && <5 , binary >=0.8.6 && <0.9 , bytestring >=0.10.8 && <0.11 @@ -88,6 +90,7 @@ , QuickCheck , async >=2.2.1 && <2.3 , attoparsec >=0.13.2 && <0.14 + , attoparsec-binary >=0.2 && <1.0 , base >=4.7 && <5 , binary >=0.8.6 && <0.9 , bytestring >=0.10.8 && <0.11 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 @@ -6,8 +6,13 @@ Maintainer : dustin@spy.net Stability : experimental -An MQTT protocol client, based on the 3.1.1 specification: - +An MQTT protocol client + +Both + +and + +are supported. -} {-# LANGUAGE OverloadedStrings #-} @@ -16,12 +21,13 @@ module Network.MQTT.Client ( -- * Configuring the client. MQTTConfig(..), MQTTClient, QoS(..), Topic, mqttConfig, mkLWT, LastWill(..), + ProtocolLevel(..), Property(..), SubOptions(..), defaultSubOptions, -- * Running and waiting for the client. runClient, runClientTLS, waitForClient, - connectURI, - disconnect, + connectURI, svrProps, + disconnect, normalDisconnect, -- * General client interactions. - subscribe, unsubscribe, publish, publishq + subscribe, unsubscribe, publish, publishq, pubAliased ) where import Control.Concurrent (threadDelay) @@ -34,7 +40,7 @@ readTVarIO, retry, writeTChan, writeTVar) import qualified Control.Exception as E -import Control.Monad (forever, void, when) +import Control.Monad (forever, guard, void, when) import Control.Monad.IO.Class (liftIO) import qualified Data.ByteString.Char8 as BCS import qualified Data.ByteString.Lazy as BL @@ -47,6 +53,7 @@ import Data.Conduit.Network.TLS (runTLSClient, tlsClientConfig) import Data.Map.Strict (Map) import qualified Data.Map.Strict as Map +import Data.Maybe (fromMaybe) import Data.Text (Text) import qualified Data.Text.Encoding as TE import Data.Word (Word16) @@ -59,7 +66,7 @@ import Network.MQTT.Topic (Filter, Topic) import Network.MQTT.Types as T -data ConnState = Starting | Connected | Disconnected deriving (Eq, Show) +data ConnState = Starting | Connected | Disconnected | DiscoErr DisconnectRequest deriving (Eq, Show) data DispatchType = DSubACK | DUnsubACK | DPubACK | DPubREC | DPubREL | DPubCOMP deriving (Eq, Show, Ord, Enum, Bounded) @@ -73,13 +80,16 @@ -- @ -- data MQTTClient = MQTTClient { - _ch :: TChan MQTTPkt - , _pktID :: TVar Word16 - , _cb :: Maybe (MQTTClient -> Topic -> BL.ByteString -> IO ()) - , _ts :: TVar [Async ()] - , _acks :: TVar (Map (DispatchType,Word16) (TChan MQTTPkt)) - , _st :: TVar ConnState - , _ct :: TVar (Async ()) + _ch :: TChan MQTTPkt + , _pktID :: TVar Word16 + , _cb :: Maybe (MQTTClient -> Topic -> BL.ByteString -> [Property] -> IO ()) + , _ts :: TVar [Async ()] + , _acks :: TVar (Map (DispatchType,Word16) (TChan MQTTPkt)) + , _st :: TVar ConnState + , _ct :: TVar (Async ()) + , _outA :: TVar (Map Topic Word16) + , _inA :: TVar (Map Word16 Topic) + , _svrProps :: TVar [Property] } -- | Configuration for setting up an MQTT client. @@ -91,16 +101,22 @@ , _password :: Maybe String -- ^ Optional password. , _cleanSession :: Bool -- ^ False if a session should be reused. , _lwt :: Maybe LastWill -- ^ LastWill message to be sent on client disconnect. - , _msgCB :: Maybe (MQTTClient -> Topic -> BL.ByteString -> IO ()) -- ^ Callback for incoming messages. + , _msgCB :: Maybe (MQTTClient -> Topic -> BL.ByteString -> [Property] -> IO ()) -- ^ Callback for incoming messages. + , _protocol :: ProtocolLevel -- ^ Protocol to use for the connection. + , _connProps :: [Property] -- ^ Properties to send to the broker in the CONNECT packet. } --- | A default MQTTConfig. A _connID /should/ be provided by the client in the returned config, --- but the defaults should work for testing. +-- | A default MQTTConfig. A _connID /should/ be provided by the +-- client in the returned config, but the defaults should work for +-- testing. In MQTTv5, an empty connection ID may be sent and the +-- server may assign an identifier for you and return it in the +-- 'PropAssignedClientIdentifier' property. mqttConfig :: MQTTConfig mqttConfig = MQTTConfig{_hostname="localhost", _port=1883, _connID="haskell-mqtt", _username=Nothing, _password=Nothing, _cleanSession=True, _lwt=Nothing, - _msgCB=Nothing} + _msgCB=Nothing, + _protocol=Protocol311, _connProps=mempty} -- | Connect to an MQTT server by URI. Currently only mqtt and mqtts @@ -144,20 +160,17 @@ -- | Set up and run a client from the given conduit AppData function. runClientAppData :: ((AppData -> IO ()) -> IO ()) -> MQTTConfig -> IO MQTTClient runClientAppData mkconn MQTTConfig{..} = do - ch <- newTChanIO - pid <- newTVarIO 1 - thr <- newTVarIO [] - acks <- newTVarIO mempty - st <- newTVarIO Starting - ct <- newTVarIO undefined - - let cli = MQTTClient{_ch=ch, - _cb=_msgCB, - _pktID=pid, - _ts=thr, - _acks=acks, - _st=st, - _ct=ct} + _ch <- newTChanIO + _pktID <- newTVarIO 1 + _ts <- newTVarIO [] + _acks <- newTVarIO mempty + _st <- newTVarIO Starting + _ct <- newTVarIO undefined + _outA <- newTVarIO mempty + _inA <- newTVarIO mempty + _svrProps <- newTVarIO mempty + let _cb = _msgCB + cli = MQTTClient{..} t <- async $ clientThread cli s <- atomically (waitForLaunch cli t) @@ -171,7 +184,10 @@ where connectAndRun = mkconn $ \ad -> E.bracket (start cli ad) cancelAll (run ad) - markDisco = atomically $ writeTVar (_st cli) Disconnected + markDisco = atomically $ do + st <- readTVar (_st cli) + guard $ st == Connected + writeTVar (_st cli) Disconnected start c@MQTTClient{..} ad = do runConduit $ do @@ -179,11 +195,12 @@ T._lastWill=_lwt, T._username=BC.pack <$> _username, T._password=BC.pack <$> _password, - T._cleanSession=_cleanSession} - yield (BL.toStrict $ toByteString req) .| appSink ad - (ConnACKPkt (ConnACKFlags _ val)) <- appSource ad .| sinkParser parsePacket + T._cleanSession=_cleanSession, + T._properties=_connProps} + yield (BL.toStrict $ toByteString _protocol req) .| appSink ad + (ConnACKPkt (ConnACKFlags _ val props)) <- appSource ad .| sinkParser (parsePacket _protocol) case val of - ConnAccepted -> pure () + ConnAccepted -> liftIO $ atomically $ writeTVar _svrProps props x -> fail (show x) pure c @@ -200,13 +217,13 @@ writeTVar _st Connected runConduit $ appSource ad - .| conduitParser parsePacket + .| conduitParser (parsePacket _protocol) .| C.mapM_ (\(_,x) -> liftIO (dispatch c pch x)) where processOut = runConduit $ - C.repeatM (liftIO (atomically $ readTChan _ch)) - .| C.map (BL.toStrict . toByteString) + C.repeatM (liftIO (atomically $ checkConnected c >> readTChan _ch)) + .| C.map (BL.toStrict . toByteString _protocol) .| appSink ad doPing = forever $ threadDelay pingPeriod >> sendPacketIO c PingPkt @@ -231,22 +248,23 @@ waitForClient :: MQTTClient -> IO (Either E.SomeException ()) waitForClient MQTTClient{..} = waitCatch =<< readTVarIO _ct -data MQTTException = Timeout | BadData deriving(Eq, Show) +data MQTTException = Timeout | BadData | Discod DisconnectRequest deriving(Eq, Show) instance E.Exception MQTTException dispatch :: MQTTClient -> TChan Bool -> MQTTPkt -> IO () dispatch c@MQTTClient{..} pch pkt = case pkt of - (PublishPkt p) -> pubMachine p - (SubACKPkt (SubscribeResponse i _)) -> delegate DSubACK i - (UnsubACKPkt (UnsubscribeResponse i)) -> delegate DUnsubACK i - (PubACKPkt (PubACK i)) -> delegate DPubACK i - (PubRECPkt (PubREC i)) -> delegate DPubREC i - (PubRELPkt (PubREL i)) -> delegate DPubREL i - (PubCOMPPkt (PubCOMP i)) -> delegate DPubCOMP i - PongPkt -> atomically . writeTChan pch $ True - x -> print x + (PublishPkt p) -> pubMachine p + (SubACKPkt (SubscribeResponse i _ _)) -> delegate DSubACK i + (UnsubACKPkt (UnsubscribeResponse i _)) -> delegate DUnsubACK i + (PubACKPkt (PubACK i _ _)) -> delegate DPubACK 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 + x -> print x where delegate dt pid = atomically $ do m <- readTVar _acks @@ -254,23 +272,39 @@ Nothing -> pure () Just ch -> writeTChan ch pkt + disco req = do + t <- readTVarIO _ct + atomically $ writeTVar _st (DiscoErr req) + cancelWith t (Discod req) + pubMachine PublishRequest{..} | _pubQoS == QoS2 = void $ async manageQoS2 >>= link - | _pubQoS == QoS1 = notify >> sendPacketIO c (PubACKPkt (PubACK _pubPktID)) + | _pubQoS == QoS1 = notify >> sendPacketIO c (PubACKPkt (PubACK _pubPktID 0 mempty)) | otherwise = notify where - notify = case _cb of - Nothing -> pure () - Just x -> x c (blToText _pubTopic) _pubBody + notify = do + topic <- resolveTopic (foldr aliasID Nothing _pubProps) + case _cb of + Nothing -> pure () + Just x -> x c topic _pubBody _pubProps + 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)) + sendPacketIO c (PubRECPkt (PubREC _pubPktID 0 mempty)) (PubRELPkt _) <- atomically $ readTChan ch pure () @@ -278,16 +312,21 @@ v <- timeout 10000000 (sendREC ch) case v of Nothing -> killConn c Timeout - _ -> notify >> sendPacketIO c (PubCOMPPkt (PubCOMP _pubPktID)) + _ -> 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 MQTTClient{..} = readTVar _st >>= check + where + check Starting = fail "not yet connected" + check Connected = pure () + check Disconnected = fail "disconnected" + check (DiscoErr req) = fail (show req) + sendPacket :: MQTTClient -> MQTTPkt -> STM () -sendPacket MQTTClient{..} p = do - st <- readTVar _st - when (st /= Connected) $ fail "not connected" - writeTChan _ch p +sendPacket c@MQTTClient{..} p = checkConnected c >> writeTChan _ch p sendPacketIO :: MQTTClient -> MQTTPkt -> IO () sendPacketIO c = atomically . sendPacket c @@ -299,7 +338,8 @@ blToText = TE.decodeUtf8 . BL.toStrict reservePktID :: MQTTClient -> [DispatchType] -> STM (TChan MQTTPkt, Word16) -reservePktID MQTTClient{..} dts = do +reservePktID c@MQTTClient{..} dts = do + checkConnected c ch <- newTChan pid <- readTVar _pktID modifyTVar' _pktID (\x -> case succ x of 0 -> 1; x' -> x') @@ -307,10 +347,10 @@ pure (ch,pid) releasePktID :: MQTTClient -> (DispatchType,Word16) -> STM () -releasePktID MQTTClient{..} k = modifyTVar' _acks (Map.delete k) +releasePktID c@MQTTClient{..} k = checkConnected c >> modifyTVar' _acks (Map.delete k) releasePktIDs :: MQTTClient -> [(DispatchType,Word16)] -> STM () -releasePktIDs MQTTClient{..} ks = modifyTVar' _acks deleteMany +releasePktIDs c@MQTTClient{..} ks = checkConnected c >> modifyTVar' _acks deleteMany where deleteMany m = foldr Map.delete m ks sendAndWait :: MQTTClient -> DispatchType -> (Word16 -> MQTTPkt) -> IO MQTTPkt @@ -329,18 +369,18 @@ -- | Subscribe to a list of topic filters with their respective QoSes. -- The accepted QoSes are returned in the same order as requested. -subscribe :: MQTTClient -> [(Filter, QoS)] -> IO [Maybe QoS] +subscribe :: MQTTClient -> [(Filter, SubOptions)] -> IO ([Either SubErr QoS], [Property]) subscribe c@MQTTClient{..} ls = do - r <- sendAndWait c DSubACK (\pid -> SubscribePkt $ SubscribeRequest pid ls') - let (SubACKPkt (SubscribeResponse _ rs)) = r - pure rs + r <- sendAndWait c DSubACK (\pid -> SubscribePkt $ SubscribeRequest pid ls' mempty) + let (SubACKPkt (SubscribeResponse _ rs props)) = r + pure (rs, props) where ls' = map (\(s, i) -> (textToBL s, i)) ls -- | Unsubscribe from a list of topic filters. -unsubscribe :: MQTTClient -> [Filter] -> IO () -unsubscribe c@MQTTClient{..} ls = - void $ sendAndWait c DUnsubACK (\pid -> UnsubscribePkt $ UnsubscribeRequest pid (map textToBL ls)) +unsubscribe :: MQTTClient -> [Filter] -> [Property] -> IO () +unsubscribe c@MQTTClient{..} ls props = + void $ sendAndWait c DUnsubACK (\pid -> UnsubscribePkt $ UnsubscribeRequest pid (map textToBL ls) props) -- | Publish a message (QoS 0). publish :: MQTTClient @@ -348,16 +388,17 @@ -> BL.ByteString -- ^ Message body -> Bool -- ^ Retain flag -> IO () -publish c t m r = void $ publishq c t m r QoS0 +publish c t m r = void $ publishq c t m r QoS0 mempty --- | Publish a message with the specified QoS. +-- | Publish a message with the specified QoS and Properties list. publishq :: MQTTClient -> Topic -- ^ Topic -> BL.ByteString -- ^ Message body -> Bool -- ^ Retain flag -> QoS -- ^ QoS + -> [Property] -- ^ Properties -> IO () -publishq c t m r q = do +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]) @@ -371,39 +412,94 @@ threadDelay 5000000 pub True pid - pkt dup pid = (PublishPkt $ PublishRequest { - _pubDup = dup, - _pubQoS = q, - _pubPktID = pid, - _pubRetain = r, - _pubTopic = textToBL t, - _pubBody = m}) + pkt dup pid = PublishPkt $ PublishRequest {_pubDup = dup, + _pubQoS = q, + _pubPktID = pid, + _pubRetain = r, + _pubTopic = textToBL t, + _pubBody = m, + _pubProps = props} satisfyQoS p ch pid | q == QoS0 = pure () - | q == QoS1 = void $ atomically $ readTChan ch + | q == QoS1 = void $ do + (PubACKPkt (PubACK _ st pprops)) <- atomically $ readTChan ch + when (st /= 0) $ fail ("qos 1 publish error: " <> show st <> " " <> show pprops) + pure () | q == QoS2 = waitRec | otherwise = error "invalid QoS" where waitRec = do - (PubRECPkt _) <- atomically $ readTChan ch - sendPacketIO c (PubRELPkt $ PubREL pid) + (PubRECPkt (PubREC _ st recprops)) <- atomically $ readTChan ch + when (st /= 0) $ fail ("qos 2 REC publish error: " <> show st <> " " <> show recprops) + sendPacketIO c (PubRELPkt $ PubREL pid 0 mempty) cancel p -- must not publish after rel - void $ atomically $ readTChan ch + (PubCOMPPkt (PubCOMP _ st' compprops)) <- atomically $ readTChan ch + when (st' /= 0) $ fail ("qos 2 COMP publish error: " <> show st' <> " " <> show compprops) + pure () -- | Disconnect from the MQTT server. -disconnect :: MQTTClient -> IO () -disconnect c@MQTTClient{..} = race_ getDisconnected orDieTrying +disconnect :: MQTTClient -> DiscoReason -> [Property] -> IO () +disconnect c@MQTTClient{..} reason props = race_ getDisconnected orDieTrying where - getDisconnected = sendPacketIO c DisconnectPkt >> waitForClient c + getDisconnected = sendPacketIO c (DisconnectPkt $ DisconnectRequest reason props) >> waitForClient c orDieTrying = threadDelay 10000000 >> killConn c Timeout +-- | Disconnect with 'DiscoNormalDisconnection' and no properties. +normalDisconnect :: MQTTClient -> IO () +normalDisconnect c = disconnect c DiscoNormalDisconnection mempty + -- | A convenience method for creating a LastWill. mkLWT :: Topic -> BL.ByteString -> Bool -> T.LastWill mkLWT t m r = T.LastWill{ T._willRetain=r, T._willQoS=QoS0, T._willTopic = textToBL t, - T._willMsg=m + T._willMsg=m, + T._willProps=mempty } + +-- | Get the list of properties that were sent from the broker at connect time. +svrProps :: MQTTClient -> IO [Property] +svrProps MQTTClient{..} = readTVarIO _svrProps + +maxAliases :: MQTTClient -> IO Word16 +maxAliases MQTTClient{..} = foldr f 0 <$> readTVarIO _svrProps + where + f (PropTopicAliasMaximum n) _ = n + f _ o = o + +-- | Publish a message with the specified QoS and Properties list. If +-- possible, use an alias to shorten the message length. The alias +-- list is managed by the client in a first-come, first-served basis, +-- so if you use this with more properties than the broker allows, +-- only the first N (up to TopicAliasMaximum, as specified by the +-- broker at connect time) will be aliased. +-- +-- This is safe to use as a general publish mechanism, as it will +-- default to not aliasing whenver there's not already an alias and we +-- can't create any more. +pubAliased :: MQTTClient + -> Topic -- ^ Topic + -> BL.ByteString -- ^ Message body + -> Bool -- ^ Retain flag + -> QoS -- ^ QoS + -> [Property] -- ^ Properties + -> 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 + + where + alias mv = atomically $ 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) 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 @@ -16,21 +16,28 @@ LastWill(..), MQTTPkt(..), QoS(..), ConnectRequest(..), connectRequest, ConnACKFlags(..), ConnACKRC(..), PublishRequest(..), PubACK(..), PubREC(..), PubREL(..), PubCOMP(..), - SubscribeRequest(..), SubscribeResponse(..), - UnsubscribeRequest(..), UnsubscribeResponse(..), + ProtocolLevel(..), Property(..), AuthRequest(..), + SubscribeRequest(..), SubOptions(..), defaultSubOptions, SubscribeResponse(..), SubErr(..), + RetainHandling(..), DisconnectRequest(..), + UnsubscribeRequest(..), UnsubscribeResponse(..), DiscoReason(..), parsePacket, ByteMe(toByteString), -- for testing - encodeLength, parseHdrLen, connACKRC + encodeLength, parseHdrLen, parseProperty, parseProperties, bsProps, + parseSubOptions, ByteSize(..) ) where import Control.Applicative (liftA2, (<|>)) -import Control.Monad (replicateM) +import Control.Monad (replicateM, when) +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 qualified Data.ByteString.Lazy as BL -import Data.Maybe (isJust) -import Data.Word (Word16, Word8) +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) @@ -48,17 +55,25 @@ (≪) = shiftL class ByteMe a where - toBytes :: a -> [Word8] - toBytes = BL.unpack . toByteString - toByteString :: a -> BL.ByteString - toByteString = BL.pack . toBytes + toBytes :: ProtocolLevel -> a -> [Word8] + toBytes p = BL.unpack . toByteString p + toByteString :: ProtocolLevel -> a -> BL.ByteString + toByteString p = BL.pack . toBytes p + +class ByteSize a where + toByte :: a -> Word8 + fromByte :: Word8 -> a + boolBit :: Bool -> Word8 boolBit False = 0 boolBit True = 1 parseHdrLen :: A.Parser Int -parseHdrLen = go 0 1 +parseHdrLen = decodeVarInt + +decodeVarInt :: A.Parser Int +decodeVarInt = go 0 1 where go :: Int -> Int -> A.Parser Int go v m = do @@ -69,7 +84,10 @@ else pure a encodeLength :: Int -> [Word8] -encodeLength n = go (n `quotRem` 128) +encodeLength = encodeVarInt + +encodeVarInt :: Int -> [Word8] +encodeVarInt n = go (n `quotRem` 128) where go (x,e) | x > 0 = en (e .|. 128) : go (x `quotRem` 128) @@ -78,30 +96,201 @@ en :: Int -> Word8 en = toEnum +encodeWord8 :: Word8 -> BL.ByteString +encodeWord8 = BL.singleton + encodeWord16 :: Word16 -> BL.ByteString encodeWord16 a = let (h,l) = a `quotRem` 256 in BL.pack [w h, w l] where w = toEnum.fromEnum +encodeWord32 :: Word32 -> BL.ByteString +encodeWord32 = runPut . putWord32be +encodeBytes :: BL.ByteString -> BL.ByteString +encodeBytes x = twoByteLen x <> x + +encodeUTF8 :: BL.ByteString -> BL.ByteString +encodeUTF8 = encodeBytes + +encodeUTF8Pair :: BL.ByteString -> BL.ByteString -> BL.ByteString +encodeUTF8Pair x y = encodeUTF8 x <> encodeUTF8 y + +twoByteLen :: BL.ByteString -> BL.ByteString +twoByteLen = encodeWord16 . toEnum . fromEnum . BL.length + blLength :: BL.ByteString -> BL.ByteString -blLength = BL.pack . encodeLength . fromEnum . BL.length +blLength = BL.pack . encodeVarInt . fromEnum . 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 . toEnum . fromEnum . BL.length) a <> a -data ProtocolLevel = Protocol311 deriving(Eq, Show) +data Property = PropPayloadFormatIndicator Word8 + | PropMessageExpiryInterval Word32 + | PropContentType BL.ByteString + | PropResponseTopic BL.ByteString + | PropCorrelationData BL.ByteString + | PropSubscriptionIdentifier Int + | PropSessionExpiryInterval Word32 + | PropAssignedClientIdentifier BL.ByteString + | PropServerKeepAlive Word16 + | PropAuthenticationMethod BL.ByteString + | PropAuthenticationData BL.ByteString + | PropRequestProblemInformation Word8 + | PropWillDelayInterval Word32 + | PropRequestResponseInformation Word8 + | PropResponseInformation BL.ByteString + | PropServerReference BL.ByteString + | PropReasonString BL.ByteString + | PropReceiveMaximum Word16 + | PropTopicAliasMaximum Word16 + | PropTopicAlias Word16 + | PropMaximumQoS Word8 + | PropRetainAvailable Word8 + | PropUserProperty BL.ByteString BL.ByteString + | PropMaximumPacketSize Word32 + | PropWildcardSubscriptionAvailable Word8 + | PropSubscriptionIdentifierAvailable Word8 + | PropSharedSubscriptionAvailable Word8 + deriving (Show, Eq) -instance ByteMe ProtocolLevel where toByteString _ = BL.singleton 4 +peW8 :: Word8 -> Word8 -> BL.ByteString +peW8 i x = BL.singleton i <> encodeWord8 x +peW16 :: Word8 -> Word16 -> BL.ByteString +peW16 i x = BL.singleton i <> encodeWord16 x + +peW32 :: Word8 -> Word32 -> BL.ByteString +peW32 i x = BL.singleton i <> encodeWord32 x + +peUTF8 :: Word8 -> BL.ByteString -> BL.ByteString +peUTF8 i x = BL.singleton i <> encodeUTF8 x + +peBin :: Word8 -> BL.ByteString -> BL.ByteString +peBin i x = BL.singleton i <> encodeBytes x + +peVarInt :: Word8 -> Int -> BL.ByteString +peVarInt i x = BL.singleton i <> (BL.pack . encodeVarInt) x + +instance ByteMe Property where + toByteString _ (PropPayloadFormatIndicator x) = peW8 0x01 x + + toByteString _ (PropMessageExpiryInterval x) = peW32 0x02 x + + toByteString _ (PropContentType x) = peUTF8 0x03 x + + toByteString _ (PropResponseTopic x) = peUTF8 0x08 x + + toByteString _ (PropCorrelationData x) = peBin 0x09 x + + toByteString _ (PropSubscriptionIdentifier x) = peVarInt 0x0b x + + toByteString _ (PropSessionExpiryInterval x) = peW32 0x11 x + + toByteString _ (PropAssignedClientIdentifier x) = peUTF8 0x12 x + + toByteString _ (PropServerKeepAlive x) = peW16 0x13 x + + toByteString _ (PropAuthenticationMethod x) = peUTF8 0x15 x + + toByteString _ (PropAuthenticationData x) = peBin 0x16 x + + toByteString _ (PropRequestProblemInformation x) = peW8 0x17 x + + toByteString _ (PropWillDelayInterval x) = peW32 0x18 x + + toByteString _ (PropRequestResponseInformation x) = peW8 0x19 x + + toByteString _ (PropResponseInformation x) = peUTF8 0x1a x + + toByteString _ (PropServerReference x) = peUTF8 0x1c x + + toByteString _ (PropReasonString x) = peUTF8 0x1f x + + toByteString _ (PropReceiveMaximum x) = peW16 0x21 x + + toByteString _ (PropTopicAliasMaximum x) = peW16 0x22 x + + toByteString _ (PropTopicAlias x) = peW16 0x23 x + + toByteString _ (PropMaximumQoS x) = peW8 0x24 x + + toByteString _ (PropRetainAvailable x) = peW8 0x25 x + + toByteString _ (PropUserProperty k v) = BL.singleton 0x26 <> encodeUTF8Pair k v + + toByteString _ (PropMaximumPacketSize x) = peW32 0x27 x + + toByteString _ (PropWildcardSubscriptionAvailable x) = peW8 0x28 x + + toByteString _ (PropSubscriptionIdentifierAvailable x) = peW8 0x29 x + + toByteString _ (PropSharedSubscriptionAvailable x) = peW8 0x2a x + +parseProperty :: A.Parser Property +parseProperty = (A.word8 0x01 >> PropPayloadFormatIndicator <$> A.anyWord8) + <|> (A.word8 0x02 >> PropMessageExpiryInterval <$> aWord32) + <|> (A.word8 0x03 >> PropContentType <$> aString) + <|> (A.word8 0x08 >> PropResponseTopic <$> aString) + <|> (A.word8 0x09 >> PropCorrelationData <$> aString) + <|> (A.word8 0x0b >> PropSubscriptionIdentifier <$> decodeVarInt) + <|> (A.word8 0x11 >> PropSessionExpiryInterval <$> aWord32) + <|> (A.word8 0x12 >> PropAssignedClientIdentifier <$> aString) + <|> (A.word8 0x13 >> PropServerKeepAlive <$> aWord16) + <|> (A.word8 0x15 >> PropAuthenticationMethod <$> aString) + <|> (A.word8 0x16 >> PropAuthenticationData <$> aString) + <|> (A.word8 0x17 >> PropRequestProblemInformation <$> A.anyWord8) + <|> (A.word8 0x18 >> PropWillDelayInterval <$> aWord32) + <|> (A.word8 0x19 >> PropRequestResponseInformation <$> A.anyWord8) + <|> (A.word8 0x1a >> PropResponseInformation <$> aString) + <|> (A.word8 0x1c >> PropServerReference <$> aString) + <|> (A.word8 0x1f >> PropReasonString <$> aString) + <|> (A.word8 0x21 >> PropReceiveMaximum <$> aWord16) + <|> (A.word8 0x22 >> PropTopicAliasMaximum <$> aWord16) + <|> (A.word8 0x23 >> PropTopicAlias <$> aWord16) + <|> (A.word8 0x24 >> PropMaximumQoS <$> A.anyWord8) + <|> (A.word8 0x25 >> PropRetainAvailable <$> A.anyWord8) + <|> (A.word8 0x26 >> PropUserProperty <$> aString <*> aString) + <|> (A.word8 0x27 >> PropMaximumPacketSize <$> aWord32) + <|> (A.word8 0x28 >> PropWildcardSubscriptionAvailable <$> A.anyWord8) + <|> (A.word8 0x29 >> PropSubscriptionIdentifierAvailable <$> A.anyWord8) + <|> (A.word8 0x2a >> PropSharedSubscriptionAvailable <$> A.anyWord8) + +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 + +parseProperties :: ProtocolLevel -> A.Parser [Property] +parseProperties Protocol311 = pure mempty +parseProperties Protocol50 = do + len <- decodeVarInt + props <- A.take len + subs props + + where + subs d = case A.parseOnly (A.many' parseProperty) d of + Left x -> fail x + Right x -> pure x + +-- | MQTT Protocol Levels +data ProtocolLevel = Protocol311 -- ^ MQTT 3.1.1 + | Protocol50 -- ^ MQTT 5.0 + deriving(Bounded, Enum, Eq, Show) + +instance ByteMe ProtocolLevel where + toByteString _ Protocol311 = BL.singleton 4 + toByteString _ Protocol50 = BL.singleton 5 + -- | An MQTT Will message. data LastWill = LastWill { _willRetain :: Bool , _willQoS :: QoS , _willTopic :: BL.ByteString , _willMsg :: BL.ByteString + , _willProps :: [Property] } deriving(Eq, Show) data ConnectRequest = ConnectRequest { @@ -111,24 +300,36 @@ , _cleanSession :: Bool , _keepAlive :: Word16 , _connID :: BL.ByteString + , _properties :: [Property] } deriving (Eq, Show) connectRequest :: ConnectRequest connectRequest = ConnectRequest{_username=Nothing, _password=Nothing, _lastWill=Nothing, - _cleanSession=True, _keepAlive=300, _connID=""} + _cleanSession=True, _keepAlive=300, _connID="", + _properties=mempty} instance ByteMe ConnectRequest where - toByteString ConnectRequest{..} = BL.singleton 0x10 - <> withLength val + toByteString prot ConnectRequest{..} = BL.singleton 0x10 + <> withLength (val prot) where - val :: BL.ByteString - val = "\NUL\EOTMQTT\EOT" -- MQTT + Protocol311 - <> BL.singleton connBits - <> encodeWord16 _keepAlive - <> toByteString _connID - <> lwt _lastWill - <> perhaps _username - <> perhaps _password + val :: ProtocolLevel -> BL.ByteString + val Protocol311 = "\NUL\EOTMQTT\EOT" -- MQTT + Protocol311 + <> BL.singleton connBits + <> encodeWord16 _keepAlive + <> toByteString prot _connID + <> lwt _lastWill + <> perhaps _username + <> perhaps _password + + val Protocol50 = "\NUL\EOTMQTT\ENQ" -- MQTT + Protocol50 + <> BL.singleton connBits + <> encodeWord16 _keepAlive + <> bsProps prot _properties + <> toByteString prot _connID + <> lwt _lastWill + <> perhaps _username + <> perhaps _password + connBits = hasu .|. hasp .|. willBits .|. clean where hasu = boolBit (isJust _username) ≪ 7 @@ -136,16 +337,19 @@ clean = boolBit _cleanSession ≪ 1 willBits = case _lastWill of Nothing -> 0 - Just LastWill{..} -> 4 .|. ((qosW _willQoS .&. 0x3) ≪ 4) .|. (boolBit _willRetain ≪ 5) + Just LastWill{..} -> 4 .|. ((qosW _willQoS .&. 0x3) ≪ 3) .|. (boolBit _willRetain ≪ 5) lwt :: Maybe LastWill -> BL.ByteString lwt Nothing = mempty - lwt (Just LastWill{..}) = toByteString _willTopic <> toByteString _willMsg + lwt (Just LastWill{..}) = bsProps prot _willProps + <> toByteString prot _willTopic + <> toByteString prot _willMsg perhaps :: Maybe BL.ByteString -> BL.ByteString perhaps Nothing = "" - perhaps (Just s) = toByteString s + perhaps (Just s) = toByteString prot s + data MQTTPkt = ConnPkt ConnectRequest | ConnACKPkt ConnACKFlags | PublishPkt PublishRequest @@ -159,38 +363,43 @@ | UnsubACKPkt UnsubscribeResponse | PingPkt | PongPkt - | DisconnectPkt + | DisconnectPkt DisconnectRequest + | AuthPkt AuthRequest deriving (Eq, Show) instance ByteMe MQTTPkt where - toByteString (ConnPkt x) = toByteString x - toByteString (ConnACKPkt x) = toByteString x - toByteString (PublishPkt x) = toByteString x - toByteString (PubACKPkt x) = toByteString x - toByteString (PubRELPkt x) = toByteString x - toByteString (PubRECPkt x) = toByteString x - toByteString (PubCOMPPkt x) = toByteString x - toByteString (SubscribePkt x) = toByteString x - toByteString (SubACKPkt x) = toByteString x - toByteString (UnsubscribePkt x) = toByteString x - toByteString (UnsubACKPkt x) = toByteString x - toByteString PingPkt = "\192\NUL" - toByteString PongPkt = "\208\NUL" - toByteString DisconnectPkt = "\224\NUL" + 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 + toByteString p (PubRELPkt x) = toByteString p x + toByteString p (PubRECPkt x) = toByteString p x + toByteString p (PubCOMPPkt x) = toByteString p x + toByteString p (SubscribePkt x) = toByteString p x + toByteString p (SubACKPkt x) = toByteString p x + toByteString p (UnsubscribePkt x) = toByteString p x + toByteString p (UnsubACKPkt x) = toByteString p x + toByteString _ PingPkt = "\192\NUL" + toByteString _ PongPkt = "\208\NUL" + toByteString p (DisconnectPkt x) = toByteString p x + toByteString p (AuthPkt x) = toByteString p x -parsePacket :: A.Parser MQTTPkt -parsePacket = parseConnect <|> parseConnectACK - <|> parsePublish <|> parsePubACK - <|> parsePubREC <|> parsePubREL <|> parsePubCOMP - <|> parseSubscribe <|> parseSubACK - <|> parseUnsubscribe <|> parseUnsubACK - <|> PingPkt <$ A.string "\192\NUL" <|> PongPkt <$ A.string "\208\NUL" - <|> DisconnectPkt <$ A.string "\224\NUL" +parsePacket :: ProtocolLevel -> A.Parser MQTTPkt +parsePacket p = parseConnect <|> parseConnectACK + <|> parsePublish p <|> parsePubACK + <|> parsePubREC <|> parsePubREL <|> parsePubCOMP + <|> parseSubscribe p <|> parseSubACK p + <|> parseUnsubscribe p <|> parseUnsubACK p + <|> PingPkt <$ A.string "\192\NUL" <|> PongPkt <$ A.string "\208\NUL" + <|> parseDisconnect p + <|> parseAuth aWord16 :: A.Parser Word16 -aWord16 = A.anyWord8 >>= \h -> A.anyWord8 >>= \l -> pure $ (c h ≪ 8) .|. c l - where c = toEnum . fromEnum +aWord16 = anyWord16be +aWord32 :: A.Parser Word32 +aWord32 = anyWord32be + aString :: A.Parser BL.ByteString aString = do n <- aWord16 @@ -201,66 +410,125 @@ parseConnect = do _ <- A.word8 0x10 _ <- parseHdrLen - _ <- A.string "\NUL\EOTMQTT\EOT" -- "MQTT" Protocol level = 4 + _ <- A.string "\NUL\EOTMQTT" -- "MQTT" + pl <- parseLevel + connFlagBits <- A.anyWord8 keepAlive <- aWord16 + props <- parseProperties pl cid <- aString - lwt <- parseLwt connFlagBits + lwt <- parseLwt pl connFlagBits u <- mstr (testBit connFlagBits 7) p <- mstr (testBit connFlagBits 6) + pure $ ConnPkt ConnectRequest{_connID=cid, _username=u, _password=p, _lastWill=lwt, _keepAlive=keepAlive, - _cleanSession=testBit connFlagBits 1} + _cleanSession=testBit connFlagBits 1, + _properties=props} where mstr :: Bool -> A.Parser (Maybe BL.ByteString) mstr False = pure Nothing mstr True = Just <$> aString - parseLwt bits + parseLevel = A.string "\EOT" $> Protocol311 + <|> A.string "\ENQ" $> Protocol50 + + parseLwt pl bits | testBit bits 2 = do + props <- parseProperties pl top <- aString msg <- aString pure $ Just LastWill{_willTopic=top, _willMsg=msg, _willRetain=testBit bits 5, - _willQoS=wQos $ (bits ≫ 3) .&. 0x3} + _willQoS=wQos $ (bits ≫ 3) .&. 0x3, + _willProps = props} | otherwise = pure Nothing -data ConnACKRC = ConnAccepted | UnacceptableProtocol - | IdentifierRejected | ServerUnavailable - | BadCredentials | NotAuthorized - | InvalidConnACKRC Word8 deriving(Eq, Show) +data ConnACKRC = ConnAccepted + -- 3.1.1 codes + | UnacceptableProtocol + | IdentifierRejected + | ServerUnavailable + | BadCredentials + | NotAuthorized + -- 5.0 codes + | ConnUnspecifiedError + | ConnMalformedPacket + | ConnProtocolError + | ConnImplementationSpecificError + | ConnUnsupportedProtocolVersion + | ConnClientIdentifierNotValid + | ConnBadUserNameOrPassword + | ConnNotAuthorized + | ConnServerUnavailable + | ConnServerBusy + | ConnBanned + | ConnBadAuthenticationMethod + | ConnTopicNameInvalid + | ConnPacketTooLarge + | ConnQuotaExceeded + | ConnPayloadFormatInvalid + | ConnRetainNotSupported + | ConnQosNotSupported + | ConnUseAnotherServer + | ConnServerMoved + | ConnConnectionRateExceeded + deriving(Eq, Show, Bounded, Enum) -connACKRC :: Word8 -> ConnACKRC -connACKRC 0 = ConnAccepted -connACKRC 1 = UnacceptableProtocol -connACKRC 2 = IdentifierRejected -connACKRC 3 = ServerUnavailable -connACKRC 4 = BadCredentials -connACKRC 5 = NotAuthorized -connACKRC x = InvalidConnACKRC x +instance ByteSize ConnACKRC where -connACKVal :: ConnACKRC -> Word8 -connACKVal ConnAccepted = 0 -connACKVal UnacceptableProtocol = 1 -connACKVal IdentifierRejected = 2 -connACKVal ServerUnavailable = 3 -connACKVal BadCredentials = 4 -connACKVal NotAuthorized = 5 -connACKVal (InvalidConnACKRC x) = x + toByte ConnAccepted = 0 + toByte UnacceptableProtocol = 1 + toByte IdentifierRejected = 2 + toByte ServerUnavailable = 3 + toByte BadCredentials = 4 + toByte NotAuthorized = 5 + toByte ConnUnspecifiedError = 0x80 + toByte ConnMalformedPacket = 0x81 + toByte ConnProtocolError = 0x82 + toByte ConnImplementationSpecificError = 0x83 + toByte ConnUnsupportedProtocolVersion = 0x84 + toByte ConnClientIdentifierNotValid = 0x85 + toByte ConnBadUserNameOrPassword = 0x86 + toByte ConnNotAuthorized = 0x87 + toByte ConnServerUnavailable = 0x88 + toByte ConnServerBusy = 0x89 + toByte ConnBanned = 0x8a + toByte ConnBadAuthenticationMethod = 0x8c + toByte ConnTopicNameInvalid = 0x90 + toByte ConnPacketTooLarge = 0x95 + toByte ConnQuotaExceeded = 0x97 + toByte ConnPayloadFormatInvalid = 0x99 + toByte ConnRetainNotSupported = 0x9a + toByte ConnQosNotSupported = 0x9b + toByte ConnUseAnotherServer = 0x9c + toByte ConnServerMoved = 0x9d + toByte ConnConnectionRateExceeded = 0x9f -data ConnACKFlags = ConnACKFlags Bool ConnACKRC deriving (Eq, Show) + fromByte b = fromMaybe ConnUnspecifiedError $ lookup b connACKRev +connACKRev :: [(Word8, ConnACKRC)] +connACKRev = map (\w -> (toByte w, w)) [minBound..] + + +data ConnACKFlags = ConnACKFlags Bool ConnACKRC [Property] deriving (Eq, Show) + instance ByteMe ConnACKFlags where - toBytes (ConnACKFlags sp rc) = [0x20, 2, boolBit sp, connACKVal rc] + toBytes prot (ConnACKFlags sp rc props) = let pbytes = BL.unpack $ bsProps prot props in + [0x20] + <> encodeVarInt (2 + length pbytes) + <>[ boolBit sp, toByte rc] <> pbytes parseConnectACK :: A.Parser MQTTPkt parseConnectACK = do _ <- A.word8 0x20 - _ <- A.word8 2 -- two bytes left + rl <- decodeVarInt -- remaining length + when (rl < 2) $ fail "conn ack packet too short" ackFlags <- A.anyWord8 rc <- A.anyWord8 - pure $ ConnACKPkt $ ConnACKFlags (testBit ackFlags 0) (connACKRC rc) + p <- parseProperties (if rl == 2 then Protocol311 else Protocol50) + pure $ ConnACKPkt $ ConnACKFlags (testBit ackFlags 0) (fromByte rc) p data PublishRequest = PublishRequest{ _pubDup :: Bool @@ -269,11 +537,12 @@ , _pubTopic :: BL.ByteString , _pubPktID :: Word16 , _pubBody :: BL.ByteString + , _pubProps :: [Property] } deriving(Eq, Show) instance ByteMe PublishRequest where - toByteString 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 @@ -281,10 +550,10 @@ pktid | _pubQoS == QoS0 = mempty | otherwise = encodeWord16 _pubPktID - val = toByteString _pubTopic <> pktid <> _pubBody + val = toByteString prot _pubTopic <> pktid <> bsProps prot _pubProps <> _pubBody -parsePublish :: A.Parser MQTTPkt -parsePublish = do +parsePublish :: ProtocolLevel -> A.Parser MQTTPkt +parsePublish prot = do w <- A.satisfy (\x -> x .&. 0xf0 == 0x30) plen <- parseHdrLen let _pubDup = w .&. 0x8 == 0x8 @@ -292,126 +561,337 @@ _pubRetain = w .&. 1 == 1 _pubTopic <- aString _pubPktID <- if _pubQoS == QoS0 then pure 0 else aWord16 - _pubBody <- BL.fromStrict <$> A.take (plen - (fromEnum $ BL.length _pubTopic) - 2 - qlen _pubQoS) + _pubProps <- parseProperties prot + _pubBody <- BL.fromStrict <$> A.take (plen - fromEnum (BL.length _pubTopic) - 2 + - qlen _pubQoS - propLen prot _pubProps ) pure $ PublishPkt PublishRequest{..} where qlen QoS0 = 0 qlen _ = 2 -data SubscribeRequest = SubscribeRequest Word16 [(BL.ByteString, QoS)] +data RetainHandling = SendOnSubscribe | SendOnSubscribeNew | DoNotSendOnSubscribe + deriving (Eq, Show, Bounded, Enum) + +data SubOptions = SubOptions{ + _retainHandling :: RetainHandling + , _retainAsPublished :: Bool + , _noLocal :: Bool + , _subQoS :: QoS + } deriving(Eq, Show) + +defaultSubOptions :: SubOptions +defaultSubOptions = SubOptions{_retainHandling=SendOnSubscribe, + _retainAsPublished=False, + _noLocal=False, + _subQoS=QoS0} + +instance ByteMe SubOptions where + toByteString _ SubOptions{..} = BL.singleton (rh .|. rap .|. nl .|. q) + + where + rh = case _retainHandling of + SendOnSubscribeNew -> 0x10 + DoNotSendOnSubscribe -> 0x20 + _ -> 0 + rap + | _retainAsPublished = 0x08 + | otherwise = 0 + nl + | _noLocal = 0x04 + | otherwise = 0 + q = qosW _subQoS + +parseSubOptions :: A.Parser SubOptions +parseSubOptions = do + w <- A.anyWord8 + let rh = case w ≫ 4 of + 1 -> SendOnSubscribeNew + 2 -> DoNotSendOnSubscribe + _ -> SendOnSubscribe + + pure $ SubOptions{ + _retainHandling=rh, + _retainAsPublished=testBit w 3, + _noLocal=testBit w 2, + _subQoS=wQos (w .&. 0x3)} + +subOptionsBytes :: ProtocolLevel -> [(BL.ByteString, SubOptions)] -> BL.ByteString +subOptionsBytes prot = BL.concat . map (\(bs,so) -> toByteString prot bs <> toByteString prot so) + +data SubscribeRequest = SubscribeRequest Word16 [(BL.ByteString, SubOptions)] [Property] deriving(Eq, Show) instance ByteMe SubscribeRequest where - toByteString (SubscribeRequest pid sreq) = - BL.singleton 0x82 - <> withLength (encodeWord16 pid <> reqs) - where reqs = (BL.concat . map (\(bs,q) -> toByteString bs <> BL.singleton (qosW q))) sreq + toByteString prot (SubscribeRequest pid sreq props) = + BL.singleton 0x82 <> withLength (encodeWord16 pid <> bsProps prot props <> subOptionsBytes prot sreq) -newtype PubACK = PubACK Word16 deriving(Eq, Show) +data PubACK = PubACK Word16 Word8 [Property] deriving(Eq, Show) +bsPubSeg :: ProtocolLevel -> Word8 -> Word16 -> Word8 -> [Property] -> BL.ByteString +bsPubSeg Protocol311 h pid _ _ = BL.singleton h <> withLength (encodeWord16 pid) +bsPubSeg Protocol50 h pid st props = BL.singleton h + <> withLength (encodeWord16 pid + <> BL.singleton st + <> mprop props) + where + mprop [] = mempty + mprop p = bsProps Protocol50 p + instance ByteMe PubACK where - toByteString (PubACK pid) = BL.singleton 0x40 <> withLength (encodeWord16 pid) + toByteString prot (PubACK pid st props) = bsPubSeg prot 0x40 pid st props +parsePubSeg :: A.Parser (Word16, Word8, [Property]) +parsePubSeg = do + rl <- parseHdrLen + mid <- aWord16 + st <- if rl > 2 then A.anyWord8 else pure 0 + props <- if rl > 4 then parseProperties Protocol50 else pure mempty + pure (mid, st, props) + parsePubACK :: A.Parser MQTTPkt parsePubACK = do _ <- A.word8 0x40 - _ <- parseHdrLen - PubACKPkt . PubACK <$> aWord16 + (mid, st, props) <- parsePubSeg + pure $ PubACKPkt (PubACK mid st props) -newtype PubREC = PubREC Word16 deriving(Eq, Show) +data PubREC = PubREC Word16 Word8 [Property] deriving(Eq, Show) instance ByteMe PubREC where - toByteString (PubREC pid) = BL.singleton 0x50 <> withLength (encodeWord16 pid) + toByteString prot (PubREC pid st props) = bsPubSeg prot 0x50 pid st props parsePubREC :: A.Parser MQTTPkt parsePubREC = do _ <- A.word8 0x50 - _ <- parseHdrLen - PubRECPkt . PubREC <$> aWord16 + (mid, st, props) <- parsePubSeg + pure $ PubRECPkt (PubREC mid st props) -newtype PubREL = PubREL Word16 deriving(Eq, Show) +data PubREL = PubREL Word16 Word8 [Property] deriving(Eq, Show) instance ByteMe PubREL where - toByteString (PubREL pid) = BL.singleton 0x62 <> withLength (encodeWord16 pid) + toByteString prot (PubREL pid st props) = bsPubSeg prot 0x62 pid st props parsePubREL :: A.Parser MQTTPkt parsePubREL = do _ <- A.word8 0x62 - _ <- parseHdrLen - PubRELPkt . PubREL <$> aWord16 + (mid, st, props) <- parsePubSeg + pure $ PubRELPkt (PubREL mid st props) -newtype PubCOMP = PubCOMP Word16 deriving(Eq, Show) +data PubCOMP = PubCOMP Word16 Word8 [Property] deriving(Eq, Show) instance ByteMe PubCOMP where - toByteString (PubCOMP pid) = BL.singleton 0x70 <> withLength (encodeWord16 pid) + toByteString prot (PubCOMP pid st props) = bsPubSeg prot 0x70 pid st props parsePubCOMP :: A.Parser MQTTPkt parsePubCOMP = do _ <- A.word8 0x70 - _ <- parseHdrLen - PubCOMPPkt . PubCOMP <$> aWord16 + (mid, st, props) <- parsePubSeg + pure $ PubCOMPPkt (PubCOMP mid st props) -parseSubscribe :: A.Parser MQTTPkt -parseSubscribe = do +parseSubscribe :: ProtocolLevel -> A.Parser MQTTPkt +parseSubscribe prot = do _ <- A.word8 0x82 hl <- parseHdrLen pid <- aWord16 - content <- A.take (fromEnum hl - 2) - SubscribePkt . SubscribeRequest pid <$> parseSubs content + props <- parseProperties prot + content <- A.take (fromEnum hl - 2 - propLen prot props) + subs <- parseSubs content + pure $ SubscribePkt (SubscribeRequest pid subs props) + where parseSubs l = case A.parseOnly (A.many1 parseSub) l of Left x -> fail x Right x -> pure x - parseSub = liftA2 (,) aString (wQos <$> A.anyWord8) + parseSub = liftA2 (,) aString parseSubOptions -data SubscribeResponse = SubscribeResponse Word16 [Maybe QoS] deriving (Eq, Show) +data SubscribeResponse = SubscribeResponse Word16 [Either SubErr QoS] [Property] deriving (Eq, Show) instance ByteMe SubscribeResponse where - toByteString (SubscribeResponse pid sres) = - BL.singleton 0x90 <> withLength (encodeWord16 pid <> BL.pack (b <$> sres)) + toByteString prot (SubscribeResponse pid sres props) = + BL.singleton 0x90 <> withLength (encodeWord16 pid <> bsProps prot props <> BL.pack (b <$> sres)) where - b Nothing = 0x80 - b (Just q) = qosW q + b (Left SubErrUnspecifiedError) = 0x80 + b (Left SubErrImplementationSpecificError) = 0x83 + b (Left SubErrNotAuthorized) = 0x87 + b (Left SubErrTopicFilterInvalid) = 0x8F + b (Left SubErrPacketIdentifierInUse) = 0x91 + b (Left SubErrQuotaExceeded) = 0x97 + b (Left SubErrSharedSubscriptionsNotSupported) = 0x9E + b (Left SubErrSubscriptionIdentifiersNotSupported) = 0xA1 + b (Left SubErrWildcardSubscriptionsNotSupported) = 0xA2 + b (Right q) = qosW q -parseSubACK :: A.Parser MQTTPkt -parseSubACK = do +propLen :: ProtocolLevel -> [Property] -> Int +propLen Protocol311 _ = 0 +propLen prot props = fromEnum $ BL.length (bsProps prot props) + +data SubErr = SubErrUnspecifiedError + | SubErrImplementationSpecificError + | SubErrNotAuthorized + | SubErrTopicFilterInvalid + | SubErrPacketIdentifierInUse + | SubErrQuotaExceeded + | SubErrSharedSubscriptionsNotSupported + | SubErrSubscriptionIdentifiersNotSupported + | SubErrWildcardSubscriptionsNotSupported + deriving (Eq, Show, Bounded, Enum) + +parseSubACK :: ProtocolLevel -> A.Parser MQTTPkt +parseSubACK prot = do _ <- A.word8 0x90 hl <- parseHdrLen pid <- aWord16 - SubACKPkt . SubscribeResponse pid <$> replicateM (hl-2) (p <$> A.anyWord8) + props <- parseProperties prot + res <- replicateM (hl-2 - propLen prot props) (p <$> A.anyWord8) + pure $ SubACKPkt (SubscribeResponse pid res props) where - p 0x80 = Nothing - p x = Just (wQos x) + p 0x80 = Left SubErrUnspecifiedError + p 0x83 = Left SubErrImplementationSpecificError + p 0x87 = Left SubErrNotAuthorized + p 0x8F = Left SubErrTopicFilterInvalid + p 0x91 = Left SubErrPacketIdentifierInUse + p 0x97 = Left SubErrQuotaExceeded + p 0x9E = Left SubErrSharedSubscriptionsNotSupported + p 0xA1 = Left SubErrSubscriptionIdentifiersNotSupported + p 0xA2 = Left SubErrWildcardSubscriptionsNotSupported + p x = Right (wQos x) -data UnsubscribeRequest = UnsubscribeRequest Word16 [BL.ByteString] +data UnsubscribeRequest = UnsubscribeRequest Word16 [BL.ByteString] [Property] deriving(Eq, Show) instance ByteMe UnsubscribeRequest where - toByteString (UnsubscribeRequest pid sreq) = + toByteString prot (UnsubscribeRequest pid sreq props) = BL.singleton 0xa2 - <> withLength (encodeWord16 pid <> mconcat (toByteString <$> sreq)) + <> withLength (encodeWord16 pid <> bsProps prot props <> mconcat (toByteString prot <$> sreq)) -parseUnsubscribe :: A.Parser MQTTPkt -parseUnsubscribe = do +parseUnsubscribe :: ProtocolLevel -> A.Parser MQTTPkt +parseUnsubscribe prot = do _ <- A.word8 0xa2 hl <- parseHdrLen pid <- aWord16 - content <- A.take (fromEnum hl - 2) - UnsubscribePkt . UnsubscribeRequest pid <$> parseSubs content + props <- parseProperties prot + content <- A.take (fromEnum hl - 2 - propLen prot props) + subs <- parseSubs content + pure $ UnsubscribePkt (UnsubscribeRequest pid subs props) where parseSubs l = case A.parseOnly (A.many1 aString) l of Left x -> fail x Right x -> pure x -newtype UnsubscribeResponse = UnsubscribeResponse Word16 deriving(Eq, Show) +data UnsubscribeResponse = UnsubscribeResponse Word16 [Property] deriving(Eq, Show) instance ByteMe UnsubscribeResponse where - toByteString (UnsubscribeResponse pid) = BL.singleton 0xb0 <> withLength (encodeWord16 pid) + toByteString prot (UnsubscribeResponse pid props) = BL.singleton 0xb0 + <> withLength (encodeWord16 pid <> bsProps prot props) -parseUnsubACK :: A.Parser MQTTPkt -parseUnsubACK = do +parseUnsubACK :: ProtocolLevel -> A.Parser MQTTPkt +parseUnsubACK prot = do _ <- A.word8 0xb0 _ <- parseHdrLen - UnsubACKPkt . UnsubscribeResponse <$> aWord16 + pid <- aWord16 + props <- parseProperties prot + pure $ UnsubACKPkt (UnsubscribeResponse pid props) + +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) + +parseAuth :: A.Parser MQTTPkt +parseAuth = do + _ <- A.word8 0xf0 + _ <- parseHdrLen + r <- AuthRequest <$> A.anyWord8 <*> parseProperties Protocol50 + pure $ AuthPkt r + +data DiscoReason = DiscoNormalDisconnection + | DiscoDisconnectWithWill + | DiscoUnspecifiedError + | DiscoMalformedPacket + | DiscoProtocolError + | DiscoImplementationSpecificError + | DiscoNotAuthorized + | DiscoServerBusy + | DiscoServershuttingDown + | DiscoKeepAliveTimeout + | DiscoSessiontakenOver + | DiscoTopicFilterInvalid + | DiscoTopicNameInvalid + | DiscoReceiveMaximumExceeded + | DiscoTopicAliasInvalid + | DiscoPacketTooLarge + | DiscoMessageRateTooHigh + | DiscoQuotaExceeded + | DiscoAdministrativeAction + | DiscoPayloadFormatInvalid + | DiscoRetainNotSupported + | DiscoQoSNotSupported + | DiscoUseAnotherServer + | DiscoServerMoved + | DiscoSharedSubscriptionsNotSupported + | DiscoConnectionRateExceeded + | DiscoMaximumConnectTime + | DiscoSubscriptionIdentifiersNotSupported + | DiscoWildcardSubscriptionsNotSupported + deriving (Show, Eq, Bounded, Enum) + +instance ByteSize DiscoReason where + + toByte DiscoNormalDisconnection = 0x00 + toByte DiscoDisconnectWithWill = 0x04 + toByte DiscoUnspecifiedError = 0x80 + toByte DiscoMalformedPacket = 0x81 + toByte DiscoProtocolError = 0x82 + toByte DiscoImplementationSpecificError = 0x83 + toByte DiscoNotAuthorized = 0x87 + toByte DiscoServerBusy = 0x89 + toByte DiscoServershuttingDown = 0x8B + toByte DiscoKeepAliveTimeout = 0x8D + toByte DiscoSessiontakenOver = 0x8e + toByte DiscoTopicFilterInvalid = 0x8f + toByte DiscoTopicNameInvalid = 0x90 + toByte DiscoReceiveMaximumExceeded = 0x93 + toByte DiscoTopicAliasInvalid = 0x94 + toByte DiscoPacketTooLarge = 0x95 + toByte DiscoMessageRateTooHigh = 0x96 + toByte DiscoQuotaExceeded = 0x97 + toByte DiscoAdministrativeAction = 0x98 + toByte DiscoPayloadFormatInvalid = 0x99 + toByte DiscoRetainNotSupported = 0x9a + toByte DiscoQoSNotSupported = 0x9b + toByte DiscoUseAnotherServer = 0x9c + toByte DiscoServerMoved = 0x9d + toByte DiscoSharedSubscriptionsNotSupported = 0x9e + toByte DiscoConnectionRateExceeded = 0x9f + toByte DiscoMaximumConnectTime = 0xa0 + toByte DiscoSubscriptionIdentifiersNotSupported = 0xa1 + toByte DiscoWildcardSubscriptionsNotSupported = 0xa2 + + fromByte w = fromMaybe DiscoMalformedPacket $ lookup w discoReasonRev + +discoReasonRev :: [(Word8, DiscoReason)] +discoReasonRev = map (\w -> (toByte w, w)) [minBound..] + +data DisconnectRequest = DisconnectRequest DiscoReason [Property] deriving (Eq, Show) + +instance ByteMe DisconnectRequest where + toByteString Protocol311 _ = "\224\NUL" + + toByteString Protocol50 (DisconnectRequest r props) = BL.singleton 0xe0 + <> withLength (BL.singleton (toByte r) + <> bsProps Protocol50 props) + +parseDisconnect :: ProtocolLevel -> A.Parser MQTTPkt +parseDisconnect Protocol311 = do + req <- DisconnectRequest DiscoNormalDisconnection mempty <$ A.string "\224\NUL" + pure $ DisconnectPkt req + +parseDisconnect Protocol50 = do + _ <- A.word8 0xe0 + rl <- parseHdrLen + r <- A.anyWord8 + props <- if rl > 1 then parseProperties Protocol50 else pure mempty + + pure $ DisconnectPkt (DisconnectRequest (fromByte r) props) diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -9,25 +9,32 @@ import qualified Data.ByteString.Lazy as L import Data.Word (Word8) import Numeric (showHex) -import Test.QuickCheck +import Test.QuickCheck as QC import Test.Tasty import Test.Tasty.HUnit import Test.Tasty.QuickCheck as QC import Network.MQTT.Topic -import Network.MQTT.Types +import Network.MQTT.Types as MT -prop_rtLengthParser :: NonNegative (Large Int) -> Property -prop_rtLengthParser (NonNegative (Large x)) = - x <= 268435455 ==> label (show (length e) <> "B") $ - cover 20 (length e > 1) "multi-byte" $ +newtype SizeT = SizeT Int deriving(Eq, Show) + +instance Arbitrary SizeT where + arbitrary = SizeT <$> oneof [ choose (0, 127), + choose (128, 16383), + choose (16384, 2097151), + choose (2097152, 268435455)] + +prop_rtLengthParser :: SizeT -> QC.Property +prop_rtLengthParser (SizeT x) = + label (show (length e) <> "B") $ d e == x where e = encodeLength x d :: [Word8] -> Int d l = case A.parse parseHdrLen (L.pack l) of - (A.Fail _ _ _) -> undefined - (A.Done _ v) -> v + A.Fail{} -> undefined + (A.Done _ v) -> v testPacketRT :: Assertion testPacketRT = mapM_ tryParse [ @@ -37,21 +44,27 @@ where tryParse s = do - let (A.Done _ x) = A.parse parsePacket s - case A.parse parsePacket (toByteString x) of - f@(A.Fail _ _ _) -> assertFailure (show f) - (A.Done _ x') -> assertEqual (show s) x x' + let (A.Done _ x) = A.parse (parsePacket Protocol311) s + case A.parse (parsePacket Protocol311) (toByteString Protocol311 x) of + f@A.Fail{} -> assertFailure (show f) + (A.Done _ x') -> assertEqual (show s) x x' +instance Arbitrary LastWill where + arbitrary = LastWill <$> arbitrary <*> arbitrary <*> astr <*> astr <*> arbitrary + +instance Arbitrary ProtocolLevel where arbitrary = arbitraryBoundedEnum + instance Arbitrary ConnectRequest where arbitrary = do - u <- mastr - p <- mastr - cid <- astr - cs <- arbitrary - ka <- arbitrary + _username <- mastr + _password <- mastr + _connID <- astr + _cleanSession <- arbitrary + _keepAlive <- arbitrary + _lastWill <- arbitrary + _properties <- arbitrary - pure ConnectRequest{_username=u, _password=p, _lastWill=Nothing, - _cleanSession=cs, _keepAlive=ka, _connID=cid} + pure ConnectRequest{..} mastr :: Gen (Maybe L.ByteString) mastr = fmap (L.fromStrict . BC.pack . getUnicodeString) <$> arbitrary @@ -62,7 +75,11 @@ instance Arbitrary QoS where arbitrary = arbitraryBoundedEnum -instance Arbitrary ConnACKFlags where arbitrary = ConnACKFlags <$> arbitrary <*> (connACKRC <$> choose (0,5)) +instance Arbitrary ConnACKFlags where + arbitrary = ConnACKFlags <$> arbitrary <*> arbitrary <*> arbitrary + shrink (ConnACKFlags b c pl) + | null pl = [] + | otherwise = ConnACKFlags b c <$> shrink pl instance Arbitrary PublishRequest where arbitrary = do @@ -72,40 +89,92 @@ _pubTopic <- astr _pubPktID <- if _pubQoS == QoS0 then pure 0 else arbitrary _pubBody <- astr + _pubProps <- arbitrary pure PublishRequest{..} instance Arbitrary PubACK where - arbitrary = PubACK <$> arbitrary + arbitrary = PubACK <$> arbitrary <*> arbitrary <*> arbitrary instance Arbitrary PubREL where - arbitrary = PubREL <$> arbitrary + arbitrary = PubREL <$> arbitrary <*> arbitrary <*> arbitrary instance Arbitrary PubREC where - arbitrary = PubREC <$> arbitrary + arbitrary = PubREC <$> arbitrary <*> arbitrary <*> arbitrary instance Arbitrary PubCOMP where - arbitrary = PubCOMP <$> arbitrary + arbitrary = PubCOMP <$> arbitrary <*> arbitrary <*> arbitrary instance Arbitrary SubscribeRequest where - arbitrary = arbitrary >>= \pid -> choose (1,11) >>= \n -> SubscribeRequest pid <$> vectorOf n sub + arbitrary = arbitrary >>= \pid -> choose (1,11) >>= \n -> SubscribeRequest pid <$> vectorOf n sub <*> arbitrary where sub = liftA2 (,) astr arbitrary + shrink (SubscribeRequest w s p) = + if length s < 2 then [] + else [SubscribeRequest w (take 1 s) p' | p' <- shrinkList (:[]) p, length p > 1] + +instance Arbitrary SubOptions where + arbitrary = SubOptions <$> arbitraryBoundedEnum <*> arbitrary <*> arbitrary <*> arbitrary + +instance Arbitrary SubErr where arbitrary = arbitraryBoundedEnum + instance Arbitrary SubscribeResponse where - arbitrary = arbitrary >>= \pid -> choose (1,11) >>= \n -> SubscribeResponse pid <$> vectorOf n sub - where sub = oneof (pure <$> [Just QoS0, Just QoS1, Just QoS2, Nothing]) - shrink (SubscribeResponse pid l) + arbitrary = arbitrary >>= \pid -> choose (1,11) >>= \n -> SubscribeResponse pid <$> vectorOf n arbitrary <*> arbitrary + + shrink (SubscribeResponse pid l props) | length l == 1 = [] - | otherwise = [SubscribeResponse pid sl | sl <- shrinkList (:[]) l, length sl > 0] + | otherwise = [SubscribeResponse pid sl props | sl <- shrinkList (:[]) l, not (null sl)] instance Arbitrary UnsubscribeRequest where - arbitrary = arbitrary >>= \pid -> choose (1,11) >>= \n -> UnsubscribeRequest pid <$> vectorOf n astr - shrink (UnsubscribeRequest p l) + arbitrary = arbitrary >>= \pid -> choose (1,11) >>= \n -> UnsubscribeRequest pid <$> vectorOf n astr <*> arbitrary + shrink (UnsubscribeRequest p l props) | length l == 1 = [] - | otherwise = [UnsubscribeRequest p sl | sl <- shrinkList (:[]) l, length sl > 0] + | otherwise = [UnsubscribeRequest p sl props | sl <- shrinkList (:[]) l, not (null sl)] instance Arbitrary UnsubscribeResponse where - arbitrary = UnsubscribeResponse <$> arbitrary + arbitrary = UnsubscribeResponse <$> arbitrary <*> arbitrary +instance Arbitrary MT.Property where + arbitrary = oneof [ + PropPayloadFormatIndicator <$> arbitrary, + PropMessageExpiryInterval <$> arbitrary, + PropMessageExpiryInterval <$> arbitrary, + PropContentType <$> astr, + PropResponseTopic <$> astr, + PropCorrelationData <$> astr, + PropSubscriptionIdentifier <$> arbitrary `suchThat` (>= 0), + PropSessionExpiryInterval <$> arbitrary, + PropAssignedClientIdentifier <$> astr, + PropServerKeepAlive <$> arbitrary, + PropAuthenticationMethod <$> astr, + PropAuthenticationData <$> astr, + PropRequestProblemInformation <$> arbitrary, + PropWillDelayInterval <$> arbitrary, + PropRequestResponseInformation <$> arbitrary, + PropResponseInformation <$> astr, + PropServerReference <$> astr, + PropReasonString <$> astr, + PropReceiveMaximum <$> arbitrary, + PropTopicAliasMaximum <$> arbitrary, + PropTopicAlias <$> arbitrary, + PropMaximumQoS <$> arbitrary, + PropRetainAvailable <$> arbitrary, + PropUserProperty <$> astr <*> astr, + PropMaximumPacketSize <$> arbitrary, + PropWildcardSubscriptionAvailable <$> arbitrary, + PropSubscriptionIdentifierAvailable <$> arbitrary, + PropSharedSubscriptionAvailable <$> arbitrary + ] + +instance Arbitrary AuthRequest where + arbitrary = AuthRequest <$> arbitrary <*> arbitrary + +instance Arbitrary ConnACKRC where arbitrary = arbitraryBoundedEnum + +instance Arbitrary DiscoReason where arbitrary = arbitraryBoundedEnum + +instance Arbitrary DisconnectRequest where + arbitrary = DisconnectRequest <$> arbitrary <*> arbitrary + instance Arbitrary MQTTPkt where arbitrary = oneof [ ConnPkt <$> arbitrary, @@ -119,19 +188,69 @@ SubACKPkt <$> arbitrary, UnsubscribePkt <$> arbitrary, UnsubACKPkt <$> arbitrary, - pure PingPkt, pure PongPkt, pure DisconnectPkt + pure PingPkt, pure PongPkt, + DisconnectPkt <$> arbitrary, + AuthPkt <$> arbitrary ] shrink (SubACKPkt x) = SubACKPkt <$> shrink x + shrink (ConnACKPkt x) = ConnACKPkt <$> shrink x shrink (UnsubscribePkt x) = UnsubscribePkt <$> shrink x + shrink (SubscribePkt x) = SubscribePkt <$> shrink x shrink _ = [] -prop_PacketRT :: MQTTPkt -> Property -prop_PacketRT p = label (lab p) $ case A.parse parsePacket (toByteString p) of - (A.Fail _ _ _) -> False - (A.Done _ r) -> r == p +prop_PacketRT50 :: MQTTPkt -> QC.Property +prop_PacketRT50 p = label (lab p) $ case A.parse (parsePacket Protocol50) (toByteString Protocol50 p) of + A.Fail{} -> False + (A.Done _ r) -> r == p where lab x = let (s,_) = break (== ' ') . show $ x in s +prop_PacketRT311 :: MQTTPkt -> QC.Property +prop_PacketRT311 p = available p ==> + let p' = v311mask p in + label (lab p') $ case A.parse (parsePacket Protocol311) (toByteString Protocol311 p') of + A.Fail{} -> False + (A.Done _ r) -> r == p' + + where + lab x = let (s,_) = break (== ' ') . show $ x in s + v311mask :: MQTTPkt -> MQTTPkt + v311mask (ConnPkt c) = ConnPkt (c{_properties=mempty, _lastWill=cl <$> _lastWill c}) + where cl lw = lw{_willProps=mempty} + v311mask (ConnACKPkt (ConnACKFlags a b _)) = ConnACKPkt (ConnACKFlags a b mempty) + v311mask (SubscribePkt (SubscribeRequest p s _)) = SubscribePkt (SubscribeRequest p c mempty) + where c = map (\(k,SubOptions{..}) -> (k,defaultSubOptions{_subQoS=_subQoS})) s + v311mask (SubACKPkt (SubscribeResponse p s _)) = SubACKPkt (SubscribeResponse p s mempty) + v311mask (UnsubscribePkt (UnsubscribeRequest p l _)) = UnsubscribePkt (UnsubscribeRequest p l mempty) + v311mask (UnsubACKPkt (UnsubscribeResponse p _)) = UnsubACKPkt (UnsubscribeResponse p mempty) + v311mask (PublishPkt req) = PublishPkt req{_pubProps=mempty} + v311mask (DisconnectPkt _) = DisconnectPkt (DisconnectRequest DiscoNormalDisconnection mempty) + v311mask (PubACKPkt (PubACK x _ _)) = PubACKPkt (PubACK x 0 mempty) + v311mask (PubRECPkt (PubREC x _ _)) = PubRECPkt (PubREC x 0 mempty) + v311mask (PubRELPkt (PubREL x _ _)) = PubRELPkt (PubREL x 0 mempty) + v311mask (PubCOMPPkt (PubCOMP x _ _)) = PubCOMPPkt (PubCOMP x 0 mempty) + v311mask x = x + + available (AuthPkt _) = False + available _ = True + +prop_PropertyRT :: MT.Property -> QC.Property +prop_PropertyRT p = label (lab p) $ case A.parse parseProperty (toByteString Protocol50 p) of + A.Fail{} -> False + (A.Done _ r) -> r == p + + where lab x = let (s,_) = break (== ' ') . show $ x in s + +prop_SubOptionsRT :: SubOptions -> Bool +prop_SubOptionsRT o = case A.parse parseSubOptions (toByteString Protocol50 o) of + A.Fail{} -> False + (A.Done _ r) -> r == o + +prop_PropertiesRT :: [MT.Property] -> Bool +prop_PropertiesRT p = case A.parse (parseProperties Protocol50) (bsProps Protocol50 p) of + A.Fail{} -> False + (A.Done _ r) -> r == p + testTopicMatching :: [TestTree] testTopicMatching = let allTopics = ["a", "a/b", "a/b/c/d", "b/a/c/d", "$SYS/a/b", "a/$SYS/b"] @@ -145,12 +264,22 @@ ("#/b", [])] in map (\(p,want) -> testCase (show p) $ assertEqual "" want (filter (match p) allTopics)) tsts +byteRT :: (ByteSize a, Show a, Eq a) => a -> Bool +byteRT x = x == (fromByte . toByte) x + tests :: [TestTree] tests = [ localOption (QC.QuickCheckTests 10000) $ testProperty "header length rt (parser)" prop_rtLengthParser, testCase "rt some packets" testPacketRT, - localOption (QC.QuickCheckTests 1000) $ testProperty "rt packets" prop_PacketRT, + localOption (QC.QuickCheckTests 1000) $ testProperty "rt packets 3.11" prop_PacketRT311, + localOption (QC.QuickCheckTests 1000) $ testProperty "rt packets 5.0" prop_PacketRT50, + localOption (QC.QuickCheckTests 1000) $ testProperty "rt property" prop_PropertyRT, + testProperty "rt properties" prop_PropertiesRT, + testProperty "sub options" prop_SubOptionsRT, + + testProperty "conn reasons" (byteRT :: ConnACKRC -> Bool), + testProperty "disco reasons" (byteRT :: DiscoReason -> Bool), testGroup "topic matching" testTopicMatching ]