net-mqtt 0.2.4.2 → 0.5.0.0

raw patch · 6 files changed

+1017/−274 lines, 6 filesdep +attoparsec-binaryPVP ok

version bump matches the API change (PVP)

Dependencies added: attoparsec-binary

API changes (from Hackage documentation)

Files

• Changelog.md +19/−0
• app/Main.hs +23/−7
• net-mqtt.cabal +5/−2
• src/Network/MQTT/Client.hs +184/−88
• src/Network/MQTT/Types.hs +618/−138
• test/Spec.hs +168/−39

@@ -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.

@@ -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)

@@ -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 <https://github.com/dustin/mqtt-hs#readme> 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

@@ -6,8 +6,13 @@ Maintainer  : dustin@spy.net Stability   : experimental -An MQTT protocol client, based on the 3.1.1 specification:-<http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html>+An MQTT protocol client++Both+<http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html MQTT 3.1.1>+and+<https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html MQTT 5.0>+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)

@@ -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)

@@ -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   ]