net-mqtt 0.5.1.0 → 0.6.0.0

raw patch · 10 files changed

+425/−195 lines, 10 filesdep +deepseqdep +optparse-applicativedep +websocketsdep ~QuickCheckdep ~bytestringnew-component:exe:mqtt-examplenew-component:exe:mqtt-watchPVP ok

version bump matches the API change (PVP)

Dependencies added: deepseq, optparse-applicative, websockets, wuss

Dependency ranges changed: QuickCheck, bytestring

API changes (from Hackage documentation)

Files

• Changelog.md +32/−0
• README.md +5/−3
• app/Main.hs +0/−46
• app/example/Main.hs +50/−0
• app/mqtt-watch/Main.hs +66/−0
• net-mqtt.cabal +45/−7
• src/Network/MQTT/Arbitrary.hs +4/−5
• src/Network/MQTT/Client.hs +204/−100
• src/Network/MQTT/Types.hs +19/−33
• test/Spec.hs +0/−1

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

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

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

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

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

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

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

@@ -12,7 +12,7 @@ <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.+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,

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

@@ -1,5 +1,4 @@ {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards   #-}  import           Control.Monad                   (mapM_) import qualified Data.Attoparsec.ByteString.Lazy as A