diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,37 @@ # Changelog for net-mqtt +## 0.6.0.0 + +Many changes went into this release. + +### New Features + +* WebSocket support (ws:// and wss://) +* Added the `mqtt-watch` CLI tool (which I use a lot) +* Lots of work on correctness WRT connection and callback failures. +* Low-Level callbacks (providing all the details of the published + message) +* Added `runMQTTConduit` to allow running the client over any Conduit + provider. + +### Incompatibilities + +As part of adding features and improving correctness, I've made a few +API changes. + +* `waitForClient` now throws an exception on failure. This greatly + simplified usage and has made a variety of my applications more + reliable when networks and computers fail. +* Callbacks are now `MessageCallback`. This primarily allowed me to + separate `SimpleCallback` (the thing most apps want) and + `LowLevelCallback` (a thing I needed when building an MQTT bridge). +* Removed `runClient` and `runClientTLS` from the API. They don't + provide any value over `connectURI` or `runMQTTConduit`. +* All callbacks are now asynchronous. Before, QoS2 would be by + necessity, but a bad callback could cause problems with the + machinery, so they're all independent now. This may not be + noticeable in most applications, but it's something to consider. + ## 0.5.1.0 The QuickCheck Arbitrary instances are exported in a module now, diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -9,7 +9,8 @@ ```haskell main :: IO main = do - mc <- runClient mqttConfig{} + let (Just uri) = parseURI "mqtt://test.mosquitto.org" + mc <- connectURI mqttConfig{} uri publish mc "tmp/topic" "hello!" False ``` @@ -18,9 +19,10 @@ ```haskell main :: IO main = do - mc <- runClient mqttConfig{_msgCB=Just msgReceived} + let (Just uri) = parseURI "mqtt://test.mosquitto.org" + mc <- connectURI mqttConfig{_msgCB=Just msgReceived} uri print =<< subscribe mc [("tmp/topic1", subOptions), ("tmp/topic2", subOptions)] [] - print =<< waitForClient mc -- wait for the the client to disconnect + waitForClient mc -- wait for the the client to disconnect where msgReceived _ t m p = print (t,m,p) diff --git a/app/Main.hs b/app/Main.hs deleted file mode 100644 --- a/app/Main.hs +++ /dev/null @@ -1,46 +0,0 @@ -{-# 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="test.mosquitto.org", _port=1883, _connID="hasq", - -- _cleanSession=False, - _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 =<< svrProps mc - print =<< subscribe mc [("oro/#", subOptions), ("tmp/#", subOptions{_subQoS=QoS2})] mempty - - print =<< unsubscribe mc ["oro/#"] mempty - - 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/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 p = print (t,m,p) diff --git a/app/example/Main.hs b/app/example/Main.hs new file mode 100644 --- /dev/null +++ b/app/example/Main.hs @@ -0,0 +1,50 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Main where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (async, cancel, link) +import Control.Monad (forever) +import Network.URI (parseURI) + +import Network.MQTT.Client + +main :: IO () +main = do + let (Just uri) = parseURI "mqtt://test.mosquitto.org" + + mc <- connectURI mqttConfig{_lwt=Just $ (mkLWT "tmp/haskquit" "bye for now" False){ + _willProps=[PropUserProperty "lwt" "prop"]}, + _msgCB=SimpleCallback showme, _protocol=Protocol50, + _connProps=[PropReceiveMaximum 65535, + PropTopicAliasMaximum 10, + PropRequestResponseInformation 1, + PropRequestProblemInformation 1]} + uri + + putStrLn "connected!" + print =<< svrProps mc + print =<< subscribe mc [("oro/#", subOptions), ("tmp/#", subOptions{_subQoS=QoS2})] mempty + + print =<< unsubscribe mc ["oro/#"] mempty + + let pprops = [PropUserProperty "hello" "mqttv5"] + p <- async $ forever $ pubAliased mc "tmp/hi/from/haskell" "hi from haskell" False QoS1 pprops >> threadDelay 10000000 + link p + + putStrLn "Publishing at QoS > 0" + 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 p = print (t,m,p) diff --git a/app/mqtt-watch/Main.hs b/app/mqtt-watch/Main.hs new file mode 100644 --- /dev/null +++ b/app/mqtt-watch/Main.hs @@ -0,0 +1,66 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} + +module Main where + +import Control.Concurrent.Async (async, link) +import Control.Concurrent.STM (TChan, atomically, newTChanIO, + readTChan, writeTChan) +import Control.Monad (forever, void, when) +import qualified Data.ByteString.Lazy as BL +import Data.Maybe (fromJust) +import qualified Data.Text.IO as TIO +import Network.MQTT.Client +import Network.URI +import Options.Applicative (Parser, argument, execParser, + fullDesc, help, helper, info, long, + maybeReader, metavar, option, + progDesc, short, showDefault, some, + str, switch, value, (<**>)) +import System.IO (stdout) + +data Msg = Msg Topic BL.ByteString [Property] + +data Options = Options { + optUri :: URI + , optHideProps :: Bool + , optTopics :: [Topic] + } + +options :: Parser Options +options = Options + <$> option (maybeReader parseURI) (long "mqtt-uri" <> short 'u' <> showDefault <> value (fromJust $ parseURI "mqtt://localhost/") <> help "mqtt broker URI") + <*> switch (short 'p' <> help "hide properties") + <*> some (argument str (metavar "topics...")) + +printer :: TChan Msg -> Bool -> IO () +printer ch showProps = forever $ do + (Msg t m props) <- atomically $ readTChan ch + TIO.putStr $ mconcat [t, " → "] + BL.hPut stdout m + putStrLn "" + when showProps $ mapM_ (putStrLn . (" " <>) . drop 4 . show) props + +run :: Options -> IO () +run Options{..} = do + ch <- newTChanIO + async (printer ch (not optHideProps)) >>= link + + mc <- connectURI mqttConfig{_msgCB=SimpleCallback (showme ch), _protocol=Protocol50, + _connProps=[PropReceiveMaximum 65535, + PropTopicAliasMaximum 10, + PropRequestResponseInformation 1, + PropRequestProblemInformation 1]} + optUri + + void $ subscribe mc [(t, subOptions) | t <- optTopics] mempty + + print =<< waitForClient mc + + where showme ch _ t m props = atomically $ writeTChan ch $ Msg t m props + +main :: IO () +main = run =<< execParser opts + + where opts = info (options <**> helper) + ( fullDesc <> progDesc "Watch stuff.") diff --git a/net-mqtt.cabal b/net-mqtt.cabal --- a/net-mqtt.cabal +++ b/net-mqtt.cabal @@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: ae5e3dfe2f39a8729bd34aa29df254e2761a1537f932d801ec64c90aee241705 +-- hash: ae5feb10e5396e515bcfcd831e587293224b3acb60a3ce8e80e34c9eef4f9a66 name: net-mqtt -version: 0.5.1.0 +version: 0.6.0.0 synopsis: An MQTT Protocol Implementation. description: Please see the README on GitHub at category: Network @@ -39,7 +39,7 @@ src ghc-options: -Wall build-depends: - QuickCheck + QuickCheck >=2.12.6.1 && <2.13 , async >=2.2.1 && <2.3 , attoparsec >=0.13.2 && <0.14 , attoparsec-binary >=0.2 && <1.0 @@ -49,21 +49,24 @@ , conduit >=1.3.1 && <1.4 , conduit-extra >=1.3.0 && <1.4 , containers >=0.5.0 && <0.7 + , deepseq >=1.4.4.0 && <1.5 , network-conduit-tls >=1.3.2 && <1.4 , network-uri >=2.6.1 && <2.7 , stm >=2.4.0 && <2.6 , text >=1.2.3 && <1.3 + , websockets >=0.12.5.3 && <0.13 + , wuss >=1.1.12 && <1.2 default-language: Haskell2010 -executable mqtt +executable mqtt-example main-is: Main.hs other-modules: Paths_net_mqtt hs-source-dirs: - app + app/example ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N build-depends: - QuickCheck + QuickCheck >=2.12.6.1 && <2.13 , async >=2.2.1 && <2.3 , attoparsec >=0.13.2 && <0.14 , attoparsec-binary >=0.2 && <1.0 @@ -73,13 +76,45 @@ , conduit >=1.3.1 && <1.4 , conduit-extra >=1.3.0 && <1.4 , containers >=0.5.0 && <0.7 + , deepseq >=1.4.4.0 && <1.5 , net-mqtt , network-conduit-tls >=1.3.2 && <1.4 , network-uri >=2.6.1 && <2.7 , stm >=2.4.0 && <2.6 , text >=1.2.3 && <1.3 + , websockets >=0.12.5.3 && <0.13 + , wuss >=1.1.12 && <1.2 default-language: Haskell2010 +executable mqtt-watch + main-is: Main.hs + other-modules: + Paths_net_mqtt + hs-source-dirs: + app/mqtt-watch + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N + build-depends: + QuickCheck >=2.12.6.1 && <2.13 + , 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.5 && <0.9 + , bytestring + , conduit >=1.3.1 && <1.4 + , conduit-extra >=1.3.0 && <1.4 + , containers >=0.5.0 && <0.7 + , deepseq >=1.4.4.0 && <1.5 + , net-mqtt + , network-conduit-tls >=1.3.2 && <1.4 + , network-uri >=2.6.1 && <2.7 + , optparse-applicative + , stm >=2.4.0 && <2.6 + , text >=1.2.3 && <1.3 + , websockets >=0.12.5.3 && <0.13 + , wuss >=1.1.12 && <1.2 + default-language: Haskell2010 + test-suite mqtt-test type: exitcode-stdio-1.0 main-is: Spec.hs @@ -90,7 +125,7 @@ ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: HUnit - , QuickCheck + , QuickCheck >=2.12.6.1 && <2.13 , async >=2.2.1 && <2.3 , attoparsec >=0.13.2 && <0.14 , attoparsec-binary >=0.2 && <1.0 @@ -100,6 +135,7 @@ , conduit >=1.3.1 && <1.4 , conduit-extra >=1.3.0 && <1.4 , containers >=0.5.0 && <0.7 + , deepseq >=1.4.4.0 && <1.5 , net-mqtt , network-conduit-tls >=1.3.2 && <1.4 , network-uri >=2.6.1 && <2.7 @@ -108,4 +144,6 @@ , tasty-hunit , tasty-quickcheck , text >=1.2.3 && <1.3 + , websockets >=0.12.5.3 && <0.13 + , wuss >=1.1.12 && <1.2 default-language: Haskell2010 diff --git a/src/Network/MQTT/Arbitrary.hs b/src/Network/MQTT/Arbitrary.hs --- a/src/Network/MQTT/Arbitrary.hs +++ b/src/Network/MQTT/Arbitrary.hs @@ -9,8 +9,7 @@ Arbitrary instances for QuickCheck. -} -{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE RecordWildCards #-} {-# OPTIONS_GHC -Wno-orphans #-} module Network.MQTT.Arbitrary ( @@ -96,7 +95,7 @@ shrink (SubscribeRequest w s p) = if length s < 2 then [] - else [SubscribeRequest w (take 1 s) p' | p' <- shrinkList (:[]) p, length p > 1] + else [SubscribeRequest w (take 1 s) p' | p' <- shrinkList (const []) p, not (null p)] instance Arbitrary SubOptions where arbitrary = SubOptions <$> arbitraryBoundedEnum <*> arbitrary <*> arbitrary <*> arbitrary @@ -108,13 +107,13 @@ shrink (SubscribeResponse pid l props) | length l == 1 = [] - | otherwise = [SubscribeResponse pid sl props | sl <- shrinkList (:[]) l, not (null sl)] + | otherwise = [SubscribeResponse pid sl props | sl <- shrinkList (const []) l, not (null sl)] instance Arbitrary UnsubscribeRequest where 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 props | sl <- shrinkList (:[]) l, not (null sl)] + | otherwise = [UnsubscribeRequest p sl props | sl <- shrinkList (const []) l, not (null sl)] instance Arbitrary UnsubStatus where arbitrary = arbitraryBoundedEnum diff --git a/src/Network/MQTT/Client.hs b/src/Network/MQTT/Client.hs --- a/src/Network/MQTT/Client.hs +++ b/src/Network/MQTT/Client.hs @@ -12,7 +12,7 @@ and -are supported. +are supported over plain TCP, TLS, WebSockets and Secure WebSockets. -} {-# LANGUAGE OverloadedStrings #-} @@ -21,33 +21,36 @@ module Network.MQTT.Client ( -- * Configuring the client. MQTTConfig(..), MQTTClient, QoS(..), Topic, mqttConfig, mkLWT, LastWill(..), - ProtocolLevel(..), Property(..), SubOptions(..), subOptions, + ProtocolLevel(..), Property(..), SubOptions(..), subOptions, MessageCallback(..), -- * Running and waiting for the client. - runClient, runClientTLS, waitForClient, - connectURI, + waitForClient, + connectURI, isConnected, disconnect, normalDisconnect, -- * General client interactions. subscribe, unsubscribe, publish, publishq, pubAliased, - svrProps + svrProps, + -- * Low-level bits + runMQTTConduit, MQTTConduit ) where import Control.Concurrent (threadDelay) import Control.Concurrent.Async (Async, async, cancel, cancelWith, - link, race_, wait, waitCatch, - withAsync) + link, race_, wait, withAsync) import Control.Concurrent.STM (STM, TChan, TVar, atomically, modifyTVar', newTChan, newTChanIO, newTVarIO, readTChan, readTVar, readTVarIO, retry, writeTChan, writeTVar) +import Control.DeepSeq (force) import qualified Control.Exception as E 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 import qualified Data.ByteString.Lazy.Char8 as BC -import Data.Conduit (runConduit, yield, (.|)) -import Data.Conduit.Attoparsec (conduitParser, sinkParser) +import Data.Conduit (ConduitT, Void, await, runConduit, + yield, (.|)) +import Data.Conduit.Attoparsec (conduitParser) import qualified Data.Conduit.Combinators as C import Data.Conduit.Network (AppData, appSink, appSource, clientSettings, runTCPClient) @@ -60,30 +63,36 @@ import Data.Word (Word16) import Network.URI (URI (..), unEscapeString, uriPort, uriRegName, uriUserInfo) +import qualified Network.WebSockets as WS import System.Timeout (timeout) - - +import qualified Wuss as WSS import Network.MQTT.Topic (Filter, Topic) import Network.MQTT.Types as T -data ConnState = Starting | Connected | Disconnected | DiscoErr DisconnectRequest | ConnErr ConnACKFlags deriving (Eq, Show) +data ConnState = Starting + | Connected + | Stopped + | Disconnected + | DiscoErr DisconnectRequest + | ConnErr ConnACKFlags deriving (Eq, Show) data DispatchType = DSubACK | DUnsubACK | DPubACK | DPubREC | DPubREL | DPubCOMP deriving (Eq, Show, Ord, Enum, Bounded) + +-- | Callback invoked on each incoming subscribed message. +data MessageCallback = NoCallback + | SimpleCallback (MQTTClient -> Topic -> BL.ByteString -> [Property] -> IO ()) + | LowLevelCallback (MQTTClient -> PublishRequest -> IO ()) + -- | 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 --- @ --- +-- See 'connectURI' for the most straightforward example. data MQTTClient = MQTTClient { _ch :: TChan MQTTPkt , _pktID :: TVar Word16 - , _cb :: Maybe (MQTTClient -> Topic -> BL.ByteString -> [Property] -> IO ()) + , _cb :: MessageCallback , _ts :: TVar [Async ()] , _acks :: TVar (Map (DispatchType,Word16) (TChan MQTTPkt)) , _st :: TVar ConnState @@ -95,53 +104,68 @@ -- | 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 (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. + _cleanSession :: Bool -- ^ False if a session should be reused. + , _lwt :: Maybe LastWill -- ^ LastWill message to be sent on client disconnect. + , _msgCB :: MessageCallback -- ^ Callback for incoming messages. + , _protocol :: ProtocolLevel -- ^ Protocol to use for the connection. + , _connProps :: [Property] -- ^ Properties to send to the broker in the CONNECT packet. + , _hostname :: String -- ^ Host to connect to (parsed from the URI) + , _port :: Int -- ^ Port number (parsed from the URI) + , _connID :: String -- ^ Unique connection ID (parsed from the URI) + , _username :: Maybe String -- ^ Optional username (parsed from the URI) + , _password :: Maybe String -- ^ Optional password (parsed from the URI) } --- | 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 +-- | A default 'MQTTConfig'. A '_connID' /may/ be required depending on +-- your broker (or if you just want an identifiable/resumable +-- connection). 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. +-- 'PropAssignedClientIdentifier' 'Property'. mqttConfig :: MQTTConfig -mqttConfig = MQTTConfig{_hostname="localhost", _port=1883, _connID="haskell-mqtt", +mqttConfig = MQTTConfig{_hostname="", _port=1883, _connID="", _username=Nothing, _password=Nothing, _cleanSession=True, _lwt=Nothing, - _msgCB=Nothing, + _msgCB=NoCallback, _protocol=Protocol311, _connProps=mempty} - --- | Connect to an MQTT server by URI. Currently only mqtt and mqtts --- URLs are supported. The host, port, username, and password will be --- derived from the URI and the values supplied in the config will be --- ignored. +-- | Connect to an MQTT server by URI. +-- +-- @mqtt://@, @mqtts://@, @ws://@, and @wss://@ URLs are supported. +-- The host, port, username, and password will be derived from the URI +-- and the values supplied in the config will be ignored. +-- +-- > main :: IO +-- > main = do +-- > let (Just uri) = parseURI "mqtt://test.mosquitto.org" +-- > mc <- connectURI mqttConfig{} uri +-- > publish mc "tmp/topic" "hello!" False connectURI :: MQTTConfig -> URI -> IO MQTTClient -connectURI cfg uri = do +connectURI cfg@(MQTTConfig{..}) uri = do let cf = case uriScheme uri of "mqtt:" -> runClient "mqtts:" -> runClientTLS + "ws:" -> runWS uri False + "wss:" -> runWS uri True us -> fail $ "invalid URI scheme: " <> us (Just a) = uriAuthority uri (u,p) = up (uriUserInfo a) - cf cfg{_hostname=uriRegName a, _port=port (uriPort a) (uriScheme uri), - Network.MQTT.Client._username=u, Network.MQTT.Client._password=p} + cf cfg{Network.MQTT.Client._connID=cid _protocol (uriFragment uri), + _hostname=uriRegName a, _port=port (uriPort a) (uriScheme uri), + Network.MQTT.Client._username=u, Network.MQTT.Client._password=p} where port "" "mqtt:" = 1883 port "" "mqtts:" = 8883 - port x _ = read x + port "" "ws:" = 80 + port "" "wss:" = 443 + port x _ = (read . tail) x + cid _ ['#'] = "" + cid _ ('#':xs) = xs + cid _ _ = "" + up "" = (Nothing, Nothing) up x = let (u,r) = break (== ':') (init x) in (Just (unEscapeString u), if r == "" then Nothing else Just (unEscapeString $ tail r)) @@ -149,18 +173,58 @@ -- | 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 +runClient cfg@MQTTConfig{..} = tcpCompat (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 +runClientTLS cfg@MQTTConfig{..} = tcpCompat (runTLSClient (tlsClientConfig _port (BCS.pack _hostname))) cfg +-- Compatibility mechanisms for TCP Conduit bits. +tcpCompat :: ((AppData -> IO ()) -> IO ()) -> MQTTConfig -> IO MQTTClient +tcpCompat mkconn cfg = runMQTTConduit (adapt mkconn) cfg + where adapt mk f = mk (f . adaptor) + adaptor ad = (appSource ad, appSink ad) + +runWS :: URI -> Bool -> MQTTConfig -> IO MQTTClient +runWS uri secure cfg@MQTTConfig{..} = runMQTTConduit (adapt $ (cf secure) _hostname _port (uriPath uri) WS.defaultConnectionOptions hdrs) cfg + where + hdrs = [("Sec-WebSocket-Protocol", "mqtt")] + adapt mk f = mk (f . adaptor) + adaptor s = (wsSource s, wsSink s) + + cf False = WS.runClientWith + cf True = \h p -> WSS.runSecureClientWith h ((fromInteger . toInteger) p) + + wsSource :: WS.Connection -> ConduitT () BCS.ByteString IO () + wsSource ws = loop + where loop = do + bs <- liftIO $ WS.receiveData ws + if BCS.null bs then pure () else yield bs >> loop + + wsSink :: WS.Connection -> ConduitT BCS.ByteString Void IO () + wsSink ws = loop + where + loop = await >>= maybe (pure ()) (\bs -> liftIO (WS.sendBinaryData ws bs) >> loop) + + pingPeriod :: Int pingPeriod = 30000000 -- 30 seconds --- | Set up and run a client from the given conduit AppData function. -runClientAppData :: ((AppData -> IO ()) -> IO ()) -> MQTTConfig -> IO MQTTClient -runClientAppData mkconn MQTTConfig{..} = do +mqttFail :: String -> a +mqttFail = E.throw . MQTTException + +-- | MQTTConduit provides a source and sink for data as used by 'runMQTTConduit'. +type MQTTConduit = (ConduitT () BCS.ByteString IO (), ConduitT BCS.ByteString Void IO ()) + +-- | Set up and run a client with a conduit context function. +-- +-- The provided action calls another IO action with a 'MQTTConduit' as a +-- parameter. It is expected that this action will manage the +-- lifecycle of the conduit source/sink on behalf of the client. +runMQTTConduit :: ((MQTTConduit -> IO ()) -> IO ()) -- ^ an action providing an 'MQTTConduit' in an execution context + -> MQTTConfig -- ^ the 'MQTTConfig' + -> IO MQTTClient +runMQTTConduit mkconn MQTTConfig{..} = do _ch <- newTChanIO _pktID <- newTVarIO 1 _ts <- newTVarIO [] @@ -192,52 +256,47 @@ guard $ st == Starting || st == Connected writeTVar (_st cli) Disconnected - start c@MQTTClient{..} ad = do - runConduit $ do + start c@MQTTClient{..} (_,sink) = do + void . 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, T._properties=_connProps} - yield (BL.toStrict $ toByteString _protocol req) .| appSink ad - (ConnACKPkt connr@(ConnACKFlags _ val props)) <- appSource ad .| sinkParser (parsePacket _protocol) - liftIO $ atomically $ do - writeTVar _svrProps props - writeTVar _st $ if val == ConnAccepted then Connected else (ConnErr connr) - - when (val /= ConnAccepted) $ fail (show val) + yield (BL.toStrict $ toByteString _protocol req) .| sink pure c - run ad c@MQTTClient{..} = do - o <- async processOut + run (src,sink) c@MQTTClient{..} = do + o <- async $ whenConnected >> processOut pch <- newTChanIO - p <- async doPing + p <- async $ whenConnected >> doPing to <- async (watchdog pch) link to - atomically $ do - modifyTVar' _ts (\l -> o:p:to:l) - writeTVar _st Connected + atomically $ modifyTVar' _ts (\l -> o:p:to:l) - runConduit $ appSource ad + runConduit $ src .| conduitParser (parsePacket _protocol) .| C.mapM_ (\(_,x) -> liftIO (dispatch c pch x)) where + whenConnected = atomically $ do + s <- readTVar _st + if s /= Connected then retry else pure () + processOut = runConduit $ C.repeatM (liftIO (atomically $ checkConnected c >> readTChan _ch)) .| C.map (BL.toStrict . toByteString _protocol) - .| appSink ad + .| sink doPing = forever $ threadDelay pingPeriod >> sendPacketIO c PingPkt - watchdog ch = do r <- timeout (pingPeriod * 3) w case r of - Nothing -> E.throwIO Timeout + Nothing -> killConn c Timeout Just _ -> watchdog ch where w = atomically . readTChan $ ch @@ -250,17 +309,38 @@ 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 +-- An exception is thrown if the client didn't terminate expectedly. +waitForClient :: MQTTClient -> IO () +waitForClient c@MQTTClient{..} = do + wait =<< readTVarIO _ct + e <- atomically $ stateX c Stopped + case e of + Nothing -> pure () + Just x -> E.throwIO x -data MQTTException = Timeout | BadData | Discod DisconnectRequest deriving(Eq, Show) +stateX :: MQTTClient -> ConnState -> STM (Maybe E.SomeException) +stateX MQTTClient{..} want = f <$> readTVar _st + where + je = Just . E.toException . MQTTException + + f :: ConnState -> Maybe E.SomeException + f Connected = if want == Connected then Nothing else je "unexpectedly connected" + f Stopped = if want == Stopped then Nothing else je "unexpectedly stopped" + f Disconnected = je "disconnected" + f Starting = je "died while starting" + f (DiscoErr x) = Just . E.toException . Discod $ x + f (ConnErr e) = je (show e) + +data MQTTException = Timeout | BadData | Discod DisconnectRequest | MQTTException String 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 + (ConnACKPkt p) -> connACKd p + (PublishPkt p) -> void $ async (pubMachine p) >>= link (SubACKPkt (SubscribeResponse i _ _)) -> delegate DSubACK i (UnsubACKPkt (UnsubscribeResponse i _ _)) -> delegate DUnsubACK i (PubACKPkt (PubACK i _ _)) -> delegate DPubACK i @@ -271,28 +351,44 @@ PongPkt -> atomically . writeTChan pch $ True x -> print x - where delegate dt pid = atomically $ do + where connACKd connr@(ConnACKFlags _ val props) = case val of + ConnAccepted -> atomically $ do + writeTVar _svrProps props + writeTVar _st Connected + _ -> do + t <- readTVarIO _ct + atomically $ writeTVar _st (ConnErr connr) + cancelWith t (MQTTException $ show connr) + + delegate dt pid = atomically $ do m <- readTVar _acks case Map.lookup (dt, pid) m of - Nothing -> pure () + Nothing -> nak dt Just ch -> writeTChan ch pkt + where + nak DPubREC = sendPacket c (PubRELPkt (PubREL pid 0x92 mempty)) + nak DPubREL = sendPacket c (PubCOMPPkt (PubCOMP pid 0x92 mempty)) + nak _ = pure () + + disco req = do t <- readTVarIO _ct atomically $ writeTVar _st (DiscoErr req) cancelWith t (Discod req) - pubMachine PublishRequest{..} - | _pubQoS == QoS2 = void $ async manageQoS2 >>= link + pubMachine pr@PublishRequest{..} + | _pubQoS == QoS2 = manageQoS2 | _pubQoS == QoS1 = notify >> sendPacketIO c (PubACKPkt (PubACK _pubPktID 0 mempty)) | otherwise = notify where notify = do topic <- resolveTopic (foldr aliasID Nothing _pubProps) - case _cb of - Nothing -> pure () - Just x -> x c topic _pubBody _pubProps + E.evaluate . force =<< case _cb of + NoCallback -> pure () + SimpleCallback f -> f c topic _pubBody _pubProps + LowLevelCallback f -> f c pr{_pubTopic=textToBL topic} resolveTopic Nothing = pure (blToText _pubTopic) resolveTopic (Just x) = do @@ -317,20 +413,22 @@ v <- timeout 10000000 (sendREC ch) case v of Nothing -> killConn c Timeout - _ -> notify >> sendPacketIO c (PubCOMPPkt (PubCOMP _pubPktID 0 mempty)) + _ -> 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) - check (ConnErr req) = fail (show req) +checkConnected mc = do + e <- stateX mc Connected + case e of + Nothing -> pure () + Just x -> E.throw x +-- | True if we're currently in a normally connected state. +isConnected :: MQTTClient -> IO Bool +isConnected MQTTClient{..} = (Connected ==) <$> readTVarIO _st + sendPacket :: MQTTClient -> MQTTPkt -> STM () sendPacket c@MQTTClient{..} p = checkConnected c >> writeTChan _ch p @@ -369,12 +467,12 @@ -- Wait for the response in a separate transaction. atomically $ do st <- readTVar _st - when (st /= Connected) $ fail "disconnected waiting for response" + when (st /= Connected) $ mqttFail "disconnected waiting for response" releasePktID c (dt,pid) readTChan ch --- | Subscribe to a list of topic filters with their respective QoSes. --- The accepted QoSes are returned in the same order as requested. +-- | Subscribe to a list of topic filters with their respective 'QoS'es. +-- The accepted 'QoS'es are returned in the same order as requested. subscribe :: MQTTClient -> [(Filter, SubOptions)] -> [Property] -> IO ([Either SubErr QoS], [Property]) subscribe c@MQTTClient{..} ls props = do r <- sendAndWait c DSubACK (\pid -> SubscribePkt $ SubscribeRequest pid ls' props) @@ -437,31 +535,37 @@ | q == QoS0 = pure () | q == QoS1 = void $ do (PubACKPkt (PubACK _ st pprops)) <- atomically $ readTChan ch - when (st /= 0) $ fail ("qos 1 publish error: " <> show st <> " " <> show pprops) + when (st /= 0) $ mqttFail ("qos 1 publish error: " <> show st <> " " <> show pprops) | q == QoS2 = waitRec | otherwise = error "invalid QoS" where waitRec = do - (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 - (PubCOMPPkt (PubCOMP _ st' compprops)) <- atomically $ readTChan ch - when (st' /= 0) $ fail ("qos 2 COMP publish error: " <> show st' <> " " <> show compprops) + rpkt <- atomically $ readTChan ch + case rpkt of + PubRECPkt (PubREC _ st recprops) -> do + when (st /= 0) $ mqttFail ("qos 2 REC publish error: " <> show st <> " " <> show recprops) + sendPacketIO c (PubRELPkt $ PubREL pid 0 mempty) + cancel p -- must not publish after rel + PubCOMPPkt (PubCOMP _ st' compprops) -> do + when (st' /= 0) $ mqttFail ("qos 2 COMP publish error: " <> show st' <> " " <> show compprops) + wtf -> mqttFail ("unexpected packet received in QoS2 publish: " <> show wtf) -- | Disconnect from the MQTT server. disconnect :: MQTTClient -> DiscoReason -> [Property] -> IO () disconnect c@MQTTClient{..} reason props = race_ getDisconnected orDieTrying where - getDisconnected = sendPacketIO c (DisconnectPkt $ DisconnectRequest reason props) >> waitForClient c + getDisconnected = do + sendPacketIO c (DisconnectPkt $ DisconnectRequest reason props) + wait =<< readTVarIO _ct + atomically $ writeTVar _st Stopped 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. +-- | A convenience method for creating a 'LastWill'. mkLWT :: Topic -> BL.ByteString -> Bool -> T.LastWill mkLWT t m r = T.LastWill{ T._willRetain=r, @@ -481,7 +585,7 @@ f (PropTopicAliasMaximum n) _ = n f _ o = o --- | Publish a message with the specified QoS and Properties list. If +-- | Publish a message with the specified 'QoS' and 'Property' 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, diff --git a/src/Network/MQTT/Types.hs b/src/Network/MQTT/Types.hs --- a/src/Network/MQTT/Types.hs +++ b/src/Network/MQTT/Types.hs @@ -127,6 +127,9 @@ instance ByteMe BL.ByteString where toByteString _ a = (encodeWord16 . toEnum . fromEnum . BL.length) a <> a +-- | Property represents the various MQTT Properties that may sent or +-- received along with packets in MQTT 5. For detailed use on when +-- and where to use them, consult with the MQTT 5.0 spec. data Property = PropPayloadFormatIndicator Word8 | PropMessageExpiryInterval Word32 | PropContentType BL.ByteString @@ -267,13 +270,7 @@ 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 + either fail pure . A.parseOnly (A.many' parseProperty) =<< A.take len -- | MQTT Protocol Levels data ProtocolLevel = Protocol311 -- ^ MQTT 3.1.1 @@ -583,7 +580,7 @@ , _subQoS :: QoS -- ^ Maximum QoS to use for this subscription. } deriving(Eq, Show) --- | Reasonable subscription option defaults at QoS0. +-- | Reasonable subscription option defaults at 'QoS0'. subOptions :: SubOptions subOptions = SubOptions{_retainHandling=SendOnSubscribe, _retainAsPublished=False, @@ -692,22 +689,24 @@ (mid, st, props) <- parsePubSeg pure $ PubCOMPPkt (PubCOMP mid st props) -parseSubscribe :: ProtocolLevel -> A.Parser MQTTPkt -parseSubscribe prot = do - _ <- A.word8 0x82 +-- Common header bits for subscribe, unsubscribe, and the sub acks. +parseSubHdr :: Word8 -> ProtocolLevel -> A.Parser a -> A.Parser (Word16, [Property], a) +parseSubHdr b prot p = do + _ <- A.word8 b hl <- parseHdrLen pid <- aWord16 props <- parseProperties prot content <- A.take (fromEnum hl - 2 - propLen prot props) - subs <- parseSubs content - pure $ SubscribePkt (SubscribeRequest pid subs props) + a <- subp content + pure (pid, props, a) - where - parseSubs l = case A.parseOnly (A.many1 parseSub) l of - Left x -> fail x - Right x -> pure x - parseSub = liftA2 (,) aString parseSubOptions + where subp = either fail pure . A.parseOnly p +parseSubscribe :: ProtocolLevel -> A.Parser MQTTPkt +parseSubscribe prot = do + (pid, props, subs) <- parseSubHdr 0x82 prot $ A.many1 (liftA2 (,) aString parseSubOptions) + pure $ SubscribePkt (SubscribeRequest pid subs props) + data SubscribeResponse = SubscribeResponse Word16 [Either SubErr QoS] [Property] deriving (Eq, Show) instance ByteMe SubscribeResponse where @@ -743,11 +742,7 @@ parseSubACK :: ProtocolLevel -> A.Parser MQTTPkt parseSubACK prot = do - _ <- A.word8 0x90 - hl <- parseHdrLen - pid <- aWord16 - props <- parseProperties prot - res <- replicateM (hl-2 - propLen prot props) (p <$> A.anyWord8) + (pid, props, res) <- parseSubHdr 0x90 prot $ A.many1 (p <$> A.anyWord8) pure $ SubACKPkt (SubscribeResponse pid res props) where @@ -772,17 +767,8 @@ parseUnsubscribe :: ProtocolLevel -> A.Parser MQTTPkt parseUnsubscribe prot = do - _ <- A.word8 0xa2 - hl <- parseHdrLen - pid <- aWord16 - props <- parseProperties prot - content <- A.take (fromEnum hl - 2 - propLen prot props) - subs <- parseSubs content + (pid, props, subs) <- parseSubHdr 0xa2 prot $ A.many1 aString 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 data UnsubStatus = UnsubSuccess | UnsubNoSubscriptionExisted diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -1,5 +1,4 @@ {-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE RecordWildCards #-} import Control.Monad (mapM_) import qualified Data.Attoparsec.ByteString.Lazy as A