net-mqtt (empty) → 0.1.0.0
raw patch · 8 files changed
+1097/−0 lines, 8 filesdep +HUnitdep +QuickCheckdep +asyncsetup-changed
Dependencies added: HUnit, QuickCheck, async, attoparsec, base, binary, bytestring, conduit, conduit-extra, containers, net-mqtt, network-conduit-tls, stm, tasty, tasty-hunit, tasty-quickcheck, text
Files
- LICENSE +30/−0
- README.md +29/−0
- Setup.hs +2/−0
- app/Main.hs +28/−0
- net-mqtt.cabal +100/−0
- src/Network/MQTT/Client.hs +348/−0
- src/Network/MQTT/Types.hs +417/−0
- test/Spec.hs +143/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Dustin Sallings (c) 2018++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Dustin Sallings nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,29 @@+# mqtt++An [MQTT][mqtt] protocol implementation for Haskell.++## Client Examples++### Publish++```haskell+main :: IO+main = do+ mc <- runClient mqttConfig{}+ publish mc "tmp/topic" "hello!" False+```++### Subscribe++```haskell+main :: IO+main = do+ mc <- runClient mqttConfig{_msgCB=Just msgReceived}+ print =<< subscribe mc [("tmp/topic1", QoS0), ("tmp/topic2", QoS0)]+ print =<< waitForClient mc -- wait for the the client to disconnect++ where+ msgReceived t m = print (t,m)+```++[mqtt]: http://mqtt.org/
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ app/Main.hs view
@@ -0,0 +1,28 @@+{-# LANGUAGE OverloadedStrings #-}++module Main where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (async, cancel)+import Control.Monad (forever)++import Network.MQTT.Client++main :: IO ()+main = do+ mc <- runClient mqttConfig{_hostname="localhost", _port=1883, _connID="hasqttl",+ -- _cleanSession=False,+ _lwt=Just $ mkLWT "tmp/haskquit" "bye for now" False,+ _msgCB=Just showme}+ putStrLn "connected!"+ print =<< subscribe mc [("oro/#", QoS0), ("tmp/#", QoS2)]+ p <- async $ forever $ publish mc "tmp/mqtths" "hi from haskell" False >> threadDelay 10000000++ putStrLn "Publishing at QoS > 0"+ publishq mc "tmp/q" "this message is at QoS 2" False QoS2+ putStrLn "Published!"++ print =<< waitForClient mc+ cancel p++ where showme t m = print (t,m)
+ net-mqtt.cabal view
@@ -0,0 +1,100 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.31.1.+--+-- see: https://github.com/sol/hpack+--+-- hash: 2e4a3950e20e2a35101f270d00e002870b13f22dc165ced24aa166762c501924++name: net-mqtt+version: 0.1.0.0+synopsis: An MQTT Protocol Implementation.+description: Please see the README on GitHub at <https://github.com/dustin/mqtt#readme>+category: Network+homepage: https://github.com/dustin/mqtt#readme+bug-reports: https://github.com/dustin/mqtt/issues+author: Dustin Sallings+maintainer: dustin@spy.net+copyright: BSD3+license: BSD3+license-file: LICENSE+build-type: Simple+extra-source-files:+ README.md++source-repository head+ type: git+ location: https://github.com/dustin/mqtt++library+ exposed-modules:+ Network.MQTT.Client+ Network.MQTT.Types+ other-modules:+ Paths_net_mqtt+ hs-source-dirs:+ src+ ghc-options: -Wall+ build-depends:+ async >=2.2.1 && <2.3+ , attoparsec >=0.13.2 && <0.14+ , base >=4.7 && <5+ , binary >=0.8.6 && <0.9+ , bytestring >=0.10.8 && <0.11+ , conduit >=1.3.1 && <1.4+ , conduit-extra >=1.3.0 && <1.4+ , containers >=0.6.0 && <0.7+ , network-conduit-tls >=1.3.2 && <1.4+ , stm >=2.5.0 && <2.6+ , text >=1.2.3 && <1.3+ default-language: Haskell2010++executable mqtt+ main-is: Main.hs+ other-modules:+ Paths_net_mqtt+ hs-source-dirs:+ app+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N+ build-depends:+ async >=2.2.1 && <2.3+ , attoparsec >=0.13.2 && <0.14+ , base >=4.7 && <5+ , binary >=0.8.6 && <0.9+ , bytestring >=0.10.8 && <0.11+ , conduit >=1.3.1 && <1.4+ , conduit-extra >=1.3.0 && <1.4+ , containers >=0.6.0 && <0.7+ , net-mqtt+ , network-conduit-tls >=1.3.2 && <1.4+ , stm >=2.5.0 && <2.6+ , text >=1.2.3 && <1.3+ default-language: Haskell2010++test-suite mqtt-test+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ other-modules:+ Paths_net_mqtt+ hs-source-dirs:+ test+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends:+ HUnit+ , QuickCheck+ , async >=2.2.1 && <2.3+ , attoparsec >=0.13.2 && <0.14+ , base >=4.7 && <5+ , binary >=0.8.6 && <0.9+ , bytestring >=0.10.8 && <0.11+ , conduit >=1.3.1 && <1.4+ , conduit-extra >=1.3.0 && <1.4+ , containers >=0.6.0 && <0.7+ , net-mqtt+ , network-conduit-tls >=1.3.2 && <1.4+ , stm >=2.5.0 && <2.6+ , tasty+ , tasty-hunit+ , tasty-quickcheck+ , text >=1.2.3 && <1.3+ default-language: Haskell2010
+ src/Network/MQTT/Client.hs view
@@ -0,0 +1,348 @@+{-|+Module : Network.MQTT.Client.+Description : An MQTT client.+Copyright : (c) Dustin Sallings, 2019+License : BSD3+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>+-}++{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Network.MQTT.Client (+ -- * Configuring the client.+ MQTTConfig(..), MQTTClient, QoS(..), Topic, mqttConfig, mkLWT, LastWill(..),+ -- * Running and waiting for the client.+ runClient, runClientTLS, waitForClient,+ disconnect,+ -- * General client interactions.+ subscribe, unsubscribe, publish, publishq+ ) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (Async, async, cancel, cancelWith,+ race_, wait, waitCatch, withAsync)+import Control.Concurrent.STM (STM, TChan, TVar, atomically,+ modifyTVar', newTChan, newTChanIO,+ newTVarIO, readTChan, readTVar,+ readTVarIO, retry, writeTChan,+ writeTVar)+import qualified Control.Exception as E+import Control.Monad (forever, void, when)+import Control.Monad.IO.Class (liftIO)+import qualified Data.ByteString.Char8 as BCS+import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Lazy.Char8 as BC+import Data.Conduit (runConduit, yield, (.|))+import Data.Conduit.Attoparsec (conduitParser, sinkParser)+import qualified Data.Conduit.Combinators as C+import Data.Conduit.Network (AppData, appSink, appSource,+ clientSettings, runTCPClient)+import Data.Conduit.Network.TLS (runTLSClient, tlsClientConfig)+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map+import Data.Text (Text)+import qualified Data.Text.Encoding as TE+import Data.Word (Word16)+++import Network.MQTT.Types as T++-- | Topic is a type alias for topic values.+type Topic = Text++data ConnState = Starting | Connected | Disconnected deriving (Eq, Show)++data DispatchType = DSubACK | DUnsubACK | DPubACK | DPubREC | DPubREL | DPubCOMP+ deriving (Eq, Show, Ord, Enum, Bounded)++-- | The MQTT client.+-- A client may be built using either runClient or runClientTLS. For example:+--+-- @+-- mc <- runClient mqttConfig{}+-- publish mc "some/topic" "some message" False+-- @+--+data MQTTClient = MQTTClient {+ _ch :: TChan MQTTPkt+ , _pktID :: TVar Word16+ , _cb :: Maybe (Topic -> BL.ByteString -> IO ())+ , _ts :: TVar [Async ()]+ , _acks :: TVar (Map (DispatchType,Word16) (TChan MQTTPkt))+ , _st :: TVar ConnState+ , _ct :: TVar (Async ())+ }++-- | Configuration for setting up an MQTT client.+data MQTTConfig = MQTTConfig{+ _hostname :: String -- ^ Host to connect to.+ , _port :: Int -- ^ Port number.+ , _connID :: String -- ^ Unique connection ID (required).+ , _username :: Maybe String -- ^ Optional username.+ , _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 (Topic -> BL.ByteString -> IO ()) -- ^ Callback for incoming messages.+ }++-- | A default MQTTConfig. A _connID /should/ be provided by the client in the returned config,+-- but the defaults should work for testing.+mqttConfig :: MQTTConfig+mqttConfig = MQTTConfig{_hostname="localhost", _port=1883, _connID="haskell-mqtt",+ _username=Nothing, _password=Nothing,+ _cleanSession=True, _lwt=Nothing,+ _msgCB=Nothing}+++-- | Set up and run a client from the given config.+runClient :: MQTTConfig -> IO MQTTClient+runClient cfg@MQTTConfig{..} = runClientAppData (runTCPClient (clientSettings _port (BCS.pack _hostname))) cfg++-- | Set up and run a client connected via TLS.+runClientTLS :: MQTTConfig -> IO MQTTClient+runClientTLS cfg@MQTTConfig{..} = runClientAppData (runTLSClient (tlsClientConfig _port (BCS.pack _hostname))) cfg++-- | 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 0+ 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}++ t <- async $ clientThread cli+ s <- atomically (waitForLaunch cli t)++ when (s /= Connected) $ wait t++ pure cli++ where+ clientThread cli = E.finally connectAndRun markDisco+ where+ connectAndRun = mkconn $ \ad ->+ E.bracket (start cli ad) cancelAll (run ad)+ markDisco = atomically $ writeTVar (_st cli) Disconnected++ start c@MQTTClient{..} ad = do+ runConduit $ do+ let req = connectRequest{T._connID=BC.pack _connID,+ 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+ case val of+ ConnAccepted -> pure ()+ x -> fail (show x)++ pure c++ run ad c@MQTTClient{..} = do+ o <- async processOut+ p <- async doPing++ atomically $ do+ modifyTVar' _ts (\l -> o:p:l)+ writeTVar _st Connected++ runConduit $ appSource ad+ .| conduitParser parsePacket+ .| C.mapM_ (\(_,x) -> liftIO (dispatch c x))++ where+ processOut = runConduit $+ C.repeatM (liftIO (atomically $ readTChan _ch))+ .| C.map (BL.toStrict . toByteString)+ .| appSink ad++ doPing = forever $ threadDelay 30000000 >> sendPacketIO c PingPkt++ waitForLaunch MQTTClient{..} t = do+ writeTVar _ct t+ c <- readTVar _st+ if c == Starting then retry else pure c++ cancelAll MQTTClient{..} = mapM_ cancel =<< readTVarIO _ts++-- | Wait for a client to terminate its connection.+waitForClient :: MQTTClient -> IO (Either E.SomeException ())+waitForClient MQTTClient{..} = waitCatch =<< readTVarIO _ct++data MQTTException = Timeout | BadData deriving(Eq, Show)++instance E.Exception MQTTException++dispatch :: MQTTClient -> MQTTPkt -> IO ()+dispatch c@MQTTClient{..} 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 -> pure ()+ x -> print x++ where delegate dt pid = atomically $ do+ m <- readTVar _acks+ case Map.lookup (dt, pid) m of+ Nothing -> pure ()+ Just ch -> writeTChan ch pkt++ pubMachine PublishRequest{..}+ | _pubQoS == QoS2 = void $ async manageQoS2+ | _pubQoS == QoS1 = notify >> sendPacketIO c (PubACKPkt (PubACK _pubPktID))+ | otherwise = notify++ where+ notify = case _cb of+ Nothing -> pure ()+ Just x -> x (blToText _pubTopic) _pubBody++ manageQoS2 = do+ ch <- newTChanIO+ atomically $ modifyTVar' _acks (Map.insert (DPubREL, _pubPktID) ch)+ E.finally (manageQoS2' ch) (atomically $ releasePktID c (DPubREL, _pubPktID))+ where+ manageQoS2' ch = do+ sendPacketIO c (PubRECPkt (PubREC _pubPktID))+ (PubRELPkt _) <- atomically $ readTChan ch+ notify+ sendPacketIO c (PubCOMPPkt (PubCOMP _pubPktID))++sendPacket :: MQTTClient -> MQTTPkt -> STM ()+sendPacket MQTTClient{..} p = do+ st <- readTVar _st+ when (st /= Connected) $ fail "not connected"+ writeTChan _ch p++sendPacketIO :: MQTTClient -> MQTTPkt -> IO ()+sendPacketIO c = atomically . sendPacket c++textToBL :: Text -> BL.ByteString+textToBL = BL.fromStrict . TE.encodeUtf8++blToText :: BL.ByteString -> Text+blToText = TE.decodeUtf8 . BL.toStrict++reservePktID :: MQTTClient -> [DispatchType] -> STM (TChan MQTTPkt, Word16)+reservePktID MQTTClient{..} dts = do+ ch <- newTChan+ pid <- readTVar _pktID+ modifyTVar' _pktID succ+ modifyTVar' _acks (Map.union (Map.fromList [((t, pid), ch) | t <- dts]))+ pure (ch,pid)++releasePktID :: MQTTClient -> (DispatchType,Word16) -> STM ()+releasePktID MQTTClient{..} k = modifyTVar' _acks (Map.delete k)++releasePktIDs :: MQTTClient -> [(DispatchType,Word16)] -> STM ()+releasePktIDs MQTTClient{..} ks = modifyTVar' _acks deleteMany+ where deleteMany m = foldr Map.delete m ks++sendAndWait :: MQTTClient -> DispatchType -> (Word16 -> MQTTPkt) -> IO MQTTPkt+sendAndWait c@MQTTClient{..} dt f = do+ (ch,pid) <- atomically $ do+ (ch,pid) <- reservePktID c [dt]+ sendPacket c (f pid)+ pure (ch,pid)++ -- Wait for the response in a separate transaction.+ atomically $ releasePktID c (dt,pid) >> readTChan ch++-- | Subscribe to a list of topics with their respective QoSes. The+-- accepted QoSes are returned in the same order as requested.+subscribe :: MQTTClient -> [(Topic, QoS)] -> IO [Maybe QoS]+subscribe c@MQTTClient{..} ls = do+ r <- sendAndWait c DSubACK (\pid -> SubscribePkt $ SubscribeRequest pid ls')+ let (SubACKPkt (SubscribeResponse _ rs)) = r+ pure rs++ where ls' = map (\(s, i) -> (textToBL s, i)) ls++-- | Unsubscribe from a list of topics.+unsubscribe :: MQTTClient -> [Topic] -> IO ()+unsubscribe c@MQTTClient{..} ls =+ void $ sendAndWait c DUnsubACK (\pid -> UnsubscribePkt $ UnsubscribeRequest pid (map textToBL ls))++-- | Publish a message (QoS 0).+publish :: MQTTClient+ -> Topic -- ^ Topic+ -> BL.ByteString -- ^ Message body+ -> Bool -- ^ Retain flag+ -> IO ()+publish c t m r = void $ publishq c t m r QoS0++-- | Publish a message with the specified QoS.+publishq :: MQTTClient+ -> Topic -- ^ Topic+ -> BL.ByteString -- ^ Message body+ -> Bool -- ^ Retain flag+ -> QoS -- ^ QoS+ -> IO ()+publishq c t m r q = do+ (ch,pid) <- atomically $ reservePktID c types+ E.finally (publishAndWait ch pid) (atomically $ releasePktIDs c [(t',pid) | t' <- types])++ where+ types = [DPubREC, DPubCOMP]+ publishAndWait ch pid = withAsync (pub False pid) (\p -> satisfyQoS p ch pid)++ pub dup pid = do+ sendPacketIO c (PublishPkt $ PublishRequest {+ _pubDup = dup,+ _pubQoS = q,+ _pubPktID = pid,+ _pubRetain = r,+ _pubTopic = textToBL t,+ _pubBody = m})+ threadDelay 5000000+ pub True pid++ satisfyQoS p ch pid+ | q == QoS0 = pure ()+ | q == QoS1 = void $ atomically $ readTChan ch+ | q == QoS2 = waitRec+ | otherwise = error "invalid QoS"++ where+ waitRec = do+ (PubRECPkt _) <- atomically $ readTChan ch+ sendPacketIO c (PubRELPkt $ PubREL pid)+ cancel p -- must not publish after rel+ void $ atomically $ readTChan ch++-- | Disconnect from the MQTT server.+disconnect :: MQTTClient -> IO ()+disconnect c@MQTTClient{..} = race_ getDisconnected orDieTrying+ where+ getDisconnected = sendPacketIO c DisconnectPkt >> waitForClient c+ orDieTrying = threadDelay 10000000 >> readTVarIO _ct >>= \t -> cancelWith t Timeout++-- | 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+ }
+ src/Network/MQTT/Types.hs view
@@ -0,0 +1,417 @@+{-|+Module : Network.MQTT.Types+Description : Parsers and serializers for MQTT.+Copyright : (c) Dustin Sallings, 2019+License : BSD3+Maintainer : dustin@spy.net+Stability : experimental++MQTT Types.+-}++{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Network.MQTT.Types (+ LastWill(..), MQTTPkt(..), QoS(..),+ ConnectRequest(..), connectRequest, ConnACKFlags(..), ConnACKRC(..),+ PublishRequest(..), PubACK(..), PubREC(..), PubREL(..), PubCOMP(..),+ SubscribeRequest(..), SubscribeResponse(..),+ UnsubscribeRequest(..), UnsubscribeResponse(..),+ parsePacket, ByteMe(toByteString),+ -- for testing+ encodeLength, parseHdrLen, connACKRC+ ) where++import Control.Applicative (liftA2, (<|>))+import Control.Monad (replicateM)+import qualified Data.Attoparsec.ByteString.Lazy as A+import Data.Bits (Bits (..), shiftL, testBit,+ (.&.), (.|.))+import qualified Data.ByteString.Lazy as BL+import Data.Maybe (isJust)+import Data.Word (Word16, Word8)++-- | QoS values for publishing and subscribing.+data QoS = QoS0 | QoS1 | QoS2 deriving (Bounded, Enum, Eq, Show)++qosW :: QoS -> Word8+qosW = toEnum.fromEnum++wQos :: Word8 -> QoS+wQos = toEnum.fromEnum++(≫) :: Bits a => a -> Int -> a+(≫) = shiftR++(≪) :: Bits a => a -> Int -> a+(≪) = shiftL++class ByteMe a where+ toBytes :: a -> [Word8]+ toBytes = BL.unpack . toByteString+ toByteString :: a -> BL.ByteString+ toByteString = BL.pack . toBytes++boolBit :: Bool -> Word8+boolBit False = 0+boolBit True = 1++parseHdrLen :: A.Parser Int+parseHdrLen = go 0 1+ where+ go :: Int -> Int -> A.Parser Int+ go v m = do+ x <- A.anyWord8+ let a = fromEnum (x .&. 127) * m + v+ if x .&. 128 /= 0+ then go a (m*128)+ else pure a++encodeLength :: Int -> [Word8]+encodeLength n = go (n `quotRem` 128)+ where+ go (x,e)+ | x > 0 = en (e .|. 128) : go (x `quotRem` 128)+ | otherwise = [en e]++ en :: Int -> Word8+ en = toEnum++encodeWord16 :: Word16 -> BL.ByteString+encodeWord16 a = let (h,l) = a `quotRem` 256 in BL.pack [w h, w l]+ where w = toEnum.fromEnum+++blLength :: BL.ByteString -> BL.ByteString+blLength = BL.pack . encodeLength . 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++data ProtocolLevel = Protocol311 deriving(Eq, Show)++instance ByteMe ProtocolLevel where toByteString _ = BL.singleton 4++-- | An MQTT Will message.+data LastWill = LastWill {+ _willRetain :: Bool+ , _willQoS :: QoS+ , _willTopic :: BL.ByteString+ , _willMsg :: BL.ByteString+ } deriving(Eq, Show)++data ConnectRequest = ConnectRequest {+ _username :: Maybe BL.ByteString+ , _password :: Maybe BL.ByteString+ , _lastWill :: Maybe LastWill+ , _cleanSession :: Bool+ , _keepAlive :: Word16+ , _connID :: BL.ByteString+ } deriving (Eq, Show)++connectRequest :: ConnectRequest+connectRequest = ConnectRequest{_username=Nothing, _password=Nothing, _lastWill=Nothing,+ _cleanSession=True, _keepAlive=300, _connID=""}++instance ByteMe ConnectRequest where+ toByteString ConnectRequest{..} = BL.singleton 0x10+ <> withLength val+ where+ val :: BL.ByteString+ val = "\NUL\EOTMQTT\EOT" -- MQTT + Protocol311+ <> BL.singleton connBits+ <> encodeWord16 _keepAlive+ <> toByteString _connID+ <> lwt _lastWill+ <> perhaps _username+ <> perhaps _password+ connBits = hasu .|. hasp .|. willBits .|. clean+ where+ hasu = boolBit (isJust _username) ≪ 7+ hasp = boolBit (isJust _password) ≪ 6+ clean = boolBit _cleanSession ≪ 1+ willBits = case _lastWill of+ Nothing -> 0+ Just LastWill{..} -> 4 .|. ((qosW _willQoS .&. 0x3) ≪ 4) .|. (boolBit _willRetain ≪ 5)++ lwt :: Maybe LastWill -> BL.ByteString+ lwt Nothing = mempty+ lwt (Just LastWill{..}) = toByteString _willTopic <> toByteString _willMsg++ perhaps :: Maybe BL.ByteString -> BL.ByteString+ perhaps Nothing = ""+ perhaps (Just s) = toByteString s++data MQTTPkt = ConnPkt ConnectRequest+ | ConnACKPkt ConnACKFlags+ | PublishPkt PublishRequest+ | PubACKPkt PubACK+ | PubRECPkt PubREC+ | PubRELPkt PubREL+ | PubCOMPPkt PubCOMP+ | SubscribePkt SubscribeRequest+ | SubACKPkt SubscribeResponse+ | UnsubscribePkt UnsubscribeRequest+ | UnsubACKPkt UnsubscribeResponse+ | PingPkt+ | PongPkt+ | DisconnectPkt+ 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"++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"++aWord16 :: A.Parser Word16+aWord16 = A.anyWord8 >>= \h -> A.anyWord8 >>= \l -> pure $ (c h ≪ 8) .|. c l+ where c = toEnum . fromEnum++aString :: A.Parser BL.ByteString+aString = do+ n <- aWord16+ s <- A.take (fromEnum n)+ pure $ BL.fromStrict s++parseConnect :: A.Parser MQTTPkt+parseConnect = do+ _ <- A.word8 0x10+ _ <- parseHdrLen+ _ <- A.string "\NUL\EOTMQTT\EOT" -- "MQTT" Protocol level = 4+ connFlagBits <- A.anyWord8+ keepAlive <- aWord16+ cid <- aString+ lwt <- parseLwt 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}++ where+ mstr :: Bool -> A.Parser (Maybe BL.ByteString)+ mstr False = pure Nothing+ mstr True = Just <$> aString++ parseLwt bits+ | testBit bits 2 = do+ top <- aString+ msg <- aString+ pure $ Just LastWill{_willTopic=top, _willMsg=msg,+ _willRetain=testBit bits 5,+ _willQoS=wQos $ (bits ≫ 3) .&. 0x3}+ | otherwise = pure Nothing++data ConnACKRC = ConnAccepted | UnacceptableProtocol+ | IdentifierRejected | ServerUnavailable+ | BadCredentials | NotAuthorized+ | InvalidConnACKRC Word8 deriving(Eq, Show)++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++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++data ConnACKFlags = ConnACKFlags Bool ConnACKRC deriving (Eq, Show)++instance ByteMe ConnACKFlags where+ toBytes (ConnACKFlags sp rc) = [0x20, 2, boolBit sp, connACKVal rc]++parseConnectACK :: A.Parser MQTTPkt+parseConnectACK = do+ _ <- A.word8 0x20+ _ <- A.word8 2 -- two bytes left+ ackFlags <- A.anyWord8+ rc <- A.anyWord8+ pure $ ConnACKPkt $ ConnACKFlags (testBit ackFlags 0) (connACKRC rc)++data PublishRequest = PublishRequest{+ _pubDup :: Bool+ , _pubQoS :: QoS+ , _pubRetain :: Bool+ , _pubTopic :: BL.ByteString+ , _pubPktID :: Word16+ , _pubBody :: BL.ByteString+ } deriving(Eq, Show)++instance ByteMe PublishRequest where+ toByteString PublishRequest{..} = BL.singleton (0x30 .|. f)+ <> withLength val+ where f = (db ≪ 3) .|. (qb ≪ 1) .|. rb+ db = boolBit _pubDup+ qb = qosW _pubQoS .&. 0x3+ rb = boolBit _pubRetain+ pktid+ | _pubQoS == QoS0 = mempty+ | otherwise = encodeWord16 _pubPktID+ val = toByteString _pubTopic <> pktid <> _pubBody++parsePublish :: A.Parser MQTTPkt+parsePublish = do+ w <- A.satisfy (\x -> x .&. 0xf0 == 0x30)+ plen <- parseHdrLen+ let _pubDup = w .&. 0x8 == 0x8+ _pubQoS = wQos $ (w ≫ 1) .&. 3+ _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)+ pure $ PublishPkt PublishRequest{..}++ where qlen QoS0 = 0+ qlen _ = 2++data SubscribeRequest = SubscribeRequest Word16 [(BL.ByteString, QoS)]+ 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++newtype PubACK = PubACK Word16 deriving(Eq, Show)++instance ByteMe PubACK where+ toByteString (PubACK pid) = BL.singleton 0x40 <> withLength (encodeWord16 pid)++parsePubACK :: A.Parser MQTTPkt+parsePubACK = do+ _ <- A.word8 0x40+ _ <- parseHdrLen+ PubACKPkt . PubACK <$> aWord16++newtype PubREC = PubREC Word16 deriving(Eq, Show)++instance ByteMe PubREC where+ toByteString (PubREC pid) = BL.singleton 0x50 <> withLength (encodeWord16 pid)++parsePubREC :: A.Parser MQTTPkt+parsePubREC = do+ _ <- A.word8 0x50+ _ <- parseHdrLen+ PubRECPkt . PubREC <$> aWord16++newtype PubREL = PubREL Word16 deriving(Eq, Show)++instance ByteMe PubREL where+ toByteString (PubREL pid) = BL.singleton 0x62 <> withLength (encodeWord16 pid)++parsePubREL :: A.Parser MQTTPkt+parsePubREL = do+ _ <- A.word8 0x62+ _ <- parseHdrLen+ PubRELPkt . PubREL <$> aWord16++newtype PubCOMP = PubCOMP Word16 deriving(Eq, Show)++instance ByteMe PubCOMP where+ toByteString (PubCOMP pid) = BL.singleton 0x70 <> withLength (encodeWord16 pid)++parsePubCOMP :: A.Parser MQTTPkt+parsePubCOMP = do+ _ <- A.word8 0x70+ _ <- parseHdrLen+ PubCOMPPkt . PubCOMP <$> aWord16++parseSubscribe :: A.Parser MQTTPkt+parseSubscribe = do+ _ <- A.word8 0x82+ hl <- parseHdrLen+ pid <- aWord16+ content <- A.take (fromEnum hl - 2)+ SubscribePkt . SubscribeRequest pid <$> parseSubs content+ 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)++data SubscribeResponse = SubscribeResponse Word16 [Maybe QoS] deriving (Eq, Show)++instance ByteMe SubscribeResponse where+ toByteString (SubscribeResponse pid sres) =+ BL.singleton 0x90 <> withLength (encodeWord16 pid <> BL.pack (b <$> sres))++ where+ b Nothing = 0x80+ b (Just q) = qosW q++parseSubACK :: A.Parser MQTTPkt+parseSubACK = do+ _ <- A.word8 0x90+ hl <- parseHdrLen+ pid <- aWord16+ SubACKPkt . SubscribeResponse pid <$> replicateM (hl-2) (p <$> A.anyWord8)++ where+ p 0x80 = Nothing+ p x = Just (wQos x)++data UnsubscribeRequest = UnsubscribeRequest Word16 [BL.ByteString]+ deriving(Eq, Show)++instance ByteMe UnsubscribeRequest where+ toByteString (UnsubscribeRequest pid sreq) =+ BL.singleton 0xa2+ <> withLength (encodeWord16 pid <> mconcat (toByteString <$> sreq))++parseUnsubscribe :: A.Parser MQTTPkt+parseUnsubscribe = do+ _ <- A.word8 0xa2+ hl <- parseHdrLen+ pid <- aWord16+ content <- A.take (fromEnum hl - 2)+ UnsubscribePkt . UnsubscribeRequest pid <$> parseSubs content+ 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)++instance ByteMe UnsubscribeResponse where+ toByteString (UnsubscribeResponse pid) = BL.singleton 0xb0 <> withLength (encodeWord16 pid)++parseUnsubACK :: A.Parser MQTTPkt+parseUnsubACK = do+ _ <- A.word8 0xb0+ _ <- parseHdrLen+ UnsubACKPkt . UnsubscribeResponse <$> aWord16
+ test/Spec.hs view
@@ -0,0 +1,143 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++import Control.Applicative (liftA2)+import Control.Monad (mapM_)+import qualified Data.Attoparsec.ByteString.Lazy as A+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as BC+import qualified Data.ByteString.Lazy as L+import Data.Word (Word8)+import Numeric (showHex)+import Test.QuickCheck+import Test.Tasty+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck as QC++import Network.MQTT.Types++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" $+ 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++testPacketRT :: Assertion+testPacketRT = mapM_ tryParse [+ "\DLE0\NUL\EOTMQTT\EOT\198\SOH,\NUL\ACKsomeid\NUL\btmp/test\NUL\STXhi\NUL\ACKdustin\NUL\ACKpasswd",+ " \STX\SOH\NUL"+ ]++ 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'++instance Arbitrary ConnectRequest where+ arbitrary = do+ u <- mastr+ p <- mastr+ cid <- astr+ cs <- arbitrary+ ka <- arbitrary++ pure ConnectRequest{_username=u, _password=p, _lastWill=Nothing,+ _cleanSession=cs, _keepAlive=ka, _connID=cid}++mastr :: Gen (Maybe L.ByteString)+mastr = fmap (L.fromStrict . BC.pack . getUnicodeString) <$> arbitrary++astr :: Gen L.ByteString+astr = L.fromStrict . BC.pack . getUnicodeString <$> arbitrary++instance Arbitrary QoS where+ arbitrary = arbitraryBoundedEnum++instance Arbitrary ConnACKFlags where arbitrary = ConnACKFlags <$> arbitrary <*> (connACKRC <$> choose (0,5))++instance Arbitrary PublishRequest where+ arbitrary = do+ _pubDup <- arbitrary+ _pubQoS <- arbitrary+ _pubRetain <- arbitrary+ _pubTopic <- astr+ _pubPktID <- if _pubQoS == QoS0 then pure 0 else arbitrary+ _pubBody <- astr+ pure PublishRequest{..}++instance Arbitrary PubACK where+ arbitrary = PubACK <$> arbitrary++instance Arbitrary PubREL where+ arbitrary = PubREL <$> arbitrary++instance Arbitrary PubREC where+ arbitrary = PubREC <$> arbitrary++instance Arbitrary PubCOMP where+ arbitrary = PubCOMP <$> arbitrary++instance Arbitrary SubscribeRequest where+ arbitrary = arbitrary >>= \pid -> choose (1,11) >>= \n -> SubscribeRequest pid <$> vectorOf n sub+ where sub = liftA2 (,) astr arbitrary++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)+ | length l == 1 = []+ | otherwise = [SubscribeResponse pid sl | sl <- shrinkList (:[]) l, length sl > 0]++instance Arbitrary UnsubscribeRequest where+ arbitrary = arbitrary >>= \pid -> choose (1,11) >>= \n -> UnsubscribeRequest pid <$> vectorOf n astr+ shrink (UnsubscribeRequest p l)+ | length l == 1 = []+ | otherwise = [UnsubscribeRequest p sl | sl <- shrinkList (:[]) l, length sl > 0]++instance Arbitrary UnsubscribeResponse where+ arbitrary = UnsubscribeResponse <$> arbitrary++instance Arbitrary MQTTPkt where+ arbitrary = oneof [+ ConnPkt <$> arbitrary,+ ConnACKPkt <$> arbitrary,+ PublishPkt <$> arbitrary,+ PubACKPkt <$> arbitrary,+ PubRELPkt <$> arbitrary,+ PubRECPkt <$> arbitrary,+ PubCOMPPkt <$> arbitrary,+ SubscribePkt <$> arbitrary,+ SubACKPkt <$> arbitrary,+ UnsubscribePkt <$> arbitrary,+ UnsubACKPkt <$> arbitrary,+ pure PingPkt, pure PongPkt, pure DisconnectPkt+ ]+ shrink (SubACKPkt x) = SubACKPkt <$> shrink x+ shrink (UnsubscribePkt x) = UnsubscribePkt <$> 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++ where lab x = let (s,_) = break (== ' ') . show $ x in s++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+ ]++main :: IO ()+main = defaultMain $ testGroup "All Tests" tests