net-mqtt 0.6.0.2 → 0.6.1.0

raw patch · 3 files changed

+91/−37 lines, 3 filesdep +connectiondep −wussPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: connection

Dependencies removed: wuss

API changes (from Hackage documentation)

Files

• Changelog.md +11/−0
• net-mqtt.cabal +7/−7
• src/Network/MQTT/Client.hs +73/−30

@@ -1,5 +1,16 @@ # Changelog for net-mqtt +## 0.6.1.0++Users can now specify TLSSettings for mqtts:// and wss:// connections.++Small bit of refactoring of main threads used by the client.  It's a+bit easier to reason about their lifecycle now.++All (at least most) of the threads in use by the client are named so+when you're looking at an eventlog, you can see what's coming and+going.+ ## 0.6.0.2  Export MQTTException (thrown from various internal bits).

@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 0cfcfaf1554a40ae7f8ade026cc7d416ad0f198dd2a780919b6f4a6789ccc83d+-- hash: 37d707924711ee323b90898ae6a53dfe919a58be4f6484ee60ce0028ce6b2bb2  name:           net-mqtt-version:        0.6.0.2+version:        0.6.1.0 synopsis:       An MQTT Protocol Implementation. description:    Please see the README on GitHub at <https://github.com/dustin/mqtt-hs#readme> category:       Network@@ -48,6 +48,7 @@     , bytestring >=0.10.8 && <0.11     , conduit >=1.3.1 && <1.4     , conduit-extra >=1.3.0 && <1.4+    , connection >=0.2.0 && <0.4     , containers >=0.5.0 && <0.7     , deepseq >=1.4.4.0 && <1.5     , network-conduit-tls >=1.3.2 && <1.4@@ -55,7 +56,6 @@     , 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-example@@ -75,6 +75,7 @@     , bytestring >=0.10.8 && <0.11     , conduit >=1.3.1 && <1.4     , conduit-extra >=1.3.0 && <1.4+    , connection >=0.2.0 && <0.4     , containers >=0.5.0 && <0.7     , deepseq >=1.4.4.0 && <1.5     , net-mqtt@@ -83,7 +84,6 @@     , 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@@ -92,7 +92,7 @@       Paths_net_mqtt   hs-source-dirs:       app/mqtt-watch-  ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N+  ghc-options: -Wall -threaded -rtsopts -eventlog -with-rtsopts=-N   build-depends:       QuickCheck >=2.12.6.1 && <2.14     , async >=2.2.1 && <2.3@@ -103,6 +103,7 @@     , bytestring     , conduit >=1.3.1 && <1.4     , conduit-extra >=1.3.0 && <1.4+    , connection >=0.2.0 && <0.4     , containers >=0.5.0 && <0.7     , deepseq >=1.4.4.0 && <1.5     , net-mqtt@@ -112,7 +113,6 @@     , 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@@ -134,6 +134,7 @@     , bytestring >=0.10.8 && <0.11     , conduit >=1.3.1 && <1.4     , conduit-extra >=1.3.0 && <1.4+    , connection >=0.2.0 && <0.4     , containers >=0.5.0 && <0.7     , deepseq >=1.4.4.0 && <1.5     , net-mqtt@@ -145,5 +146,4 @@     , tasty-quickcheck     , text >=1.2.3 && <1.3     , websockets >=0.12.5.3 && <0.13-    , wuss >=1.1.12 && <1.2   default-language: Haskell2010

@@ -33,9 +33,10 @@   runMQTTConduit, MQTTConduit, isConnectedSTM   ) where -import           Control.Concurrent         (threadDelay)-import           Control.Concurrent.Async   (Async, async, cancel, cancelWith,-                                             link, race_, wait, withAsync)+import           Control.Concurrent         (myThreadId, threadDelay)+import           Control.Concurrent.Async   (Async, async, asyncThreadId,+                                             cancel, cancelWith, link, race_,+                                             wait, waitAnyCancel, withAsync) import           Control.Concurrent.STM     (STM, TChan, TVar, atomically,                                              modifyTVar', newTChan, newTChanIO,                                              newTVarIO, readTChan, readTVar,@@ -54,18 +55,26 @@ 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.Conduit.Network.TLS   (runTLSClient, tlsClientConfig,+                                             tlsClientTLSSettings) import           Data.Map.Strict            (Map) import qualified Data.Map.Strict            as Map import           Data.Maybe                 (fromMaybe) import           Data.Text                  (Text) import qualified Data.Text.Encoding         as TE import           Data.Word                  (Word16)+import           GHC.Conc                   (labelThread)+import           Network.Connection         (ConnectionParams (..),+                                             TLSSettings (..), connectTo,+                                             connectionClose,+                                             connectionGetChunk, connectionPut,+                                             initConnectionContext) import           Network.URI                (URI (..), unEscapeString, uriPort,                                              uriRegName, uriUserInfo) import qualified Network.WebSockets         as WS+import           Network.WebSockets.Stream  (makeStream)+import           System.IO.Error            (catchIOError, isEOFError) import           System.Timeout             (timeout)-import qualified Wuss                       as WSS  import           Network.MQTT.Topic         (Filter, Topic) import           Network.MQTT.Types         as T@@ -93,7 +102,6 @@   _ch         :: TChan MQTTPkt   , _pktID    :: TVar Word16   , _cb       :: MessageCallback-  , _ts       :: TVar [Async ()]   , _acks     :: TVar (Map (DispatchType,Word16) (TChan MQTTPkt))   , _st       :: TVar ConnState   , _ct       :: TVar (Async ())@@ -115,6 +123,7 @@   , _username       :: Maybe String -- ^ Optional username (parsed from the URI)   , _password       :: Maybe String -- ^ Optional password (parsed from the URI)   , _connectTimeout :: Int -- ^ Connection timeout (microseconds)+  , _tlsSettings    :: TLSSettings -- ^ TLS Settings for secure connections   }  -- | A default 'MQTTConfig'.  A '_connID' /may/ be required depending on@@ -128,7 +137,8 @@                         _cleanSession=True, _lwt=Nothing,                         _msgCB=NoCallback,                         _protocol=Protocol311, _connProps=mempty,-                        _connectTimeout=180000000}+                        _connectTimeout=180000000,+                        _tlsSettings=TLSSettingsSimple False False False}  -- | Connect to an MQTT server by URI. --@@ -153,7 +163,7 @@       (Just a) = uriAuthority uri       (u,p) = up (uriUserInfo a) -  v <- timeout _connectTimeout $+  v <- namedTimeout "MQTT connect" _connectTimeout $     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}@@ -184,7 +194,8 @@  -- | Set up and run a client connected via TLS. runClientTLS :: MQTTConfig -> IO MQTTClient-runClientTLS cfg@MQTTConfig{..} = tcpCompat (runTLSClient (tlsClientConfig _port (BCS.pack _hostname))) cfg+runClientTLS cfg@MQTTConfig{..} = tcpCompat (runTLSClient tlsConf) cfg+  where tlsConf = (tlsClientConfig _port (BCS.pack _hostname)) {tlsClientTLSSettings=_tlsSettings}  -- Compatibility mechanisms for TCP Conduit bits. tcpCompat :: ((AppData -> IO ()) -> IO ()) -> MQTTConfig -> IO MQTTClient@@ -199,8 +210,9 @@     adapt mk f = mk (f . adaptor)     adaptor s = (wsSource s, wsSink s) +    cf :: Bool -> String -> Int -> String -> WS.ConnectionOptions -> WS.Headers -> WS.ClientApp () -> IO ()     cf False = WS.runClientWith-    cf True  = \h p -> WSS.runSecureClientWith h ((fromInteger . toInteger) p)+    cf True  = runWSS      wsSource :: WS.Connection -> ConduitT () BCS.ByteString IO ()     wsSource ws = loop@@ -213,13 +225,48 @@       where         loop = await >>= maybe (pure ()) (\bs -> liftIO (WS.sendBinaryData ws bs) >> loop) +    runWSS :: String -> Int -> String -> WS.ConnectionOptions -> WS.Headers -> WS.ClientApp () -> IO ()+    runWSS host port path options hdrs' app = do+      let connectionParams = ConnectionParams+            { connectionHostname = host+            , connectionPort =  toEnum port+            , connectionUseSecure = Just _tlsSettings+            , connectionUseSocks = Nothing+            } +      context <- initConnectionContext+      E.bracket (connectTo context connectionParams) connectionClose+        (\conn -> do+            stream <- makeStream (reader conn) (writer conn)+            WS.runClientWithStream stream host path options hdrs' app)++        where+          reader conn =+            catchIOError (Just <$> connectionGetChunk conn)+            (\e -> if isEOFError e then pure Nothing else E.throwIO e)++          writer conn maybeBytes =+            case maybeBytes of+              Nothing    -> pure ()+              Just bytes -> connectionPut conn (BC.toStrict bytes)+ pingPeriod :: Int pingPeriod = 30000000 -- 30 seconds  mqttFail :: String -> a mqttFail = E.throw . MQTTException +-- A couple async utilities to get our stuff named.++namedAsync :: String -> IO a -> IO (Async a)+namedAsync s a = async a >>= \p -> labelThread (asyncThreadId p) s >> pure p++namedWithAsync :: String -> IO a -> (Async a -> IO b) -> IO b+namedWithAsync n a t = withAsync a (\p -> labelThread (asyncThreadId p) n >> t p)++namedTimeout :: String -> Int -> IO a -> IO (Maybe a)+namedTimeout n to a = timeout to (myThreadId >>= \tid -> labelThread tid n >> a)+ -- | MQTTConduit provides a source and sink for data as used by 'runMQTTConduit'. type MQTTConduit = (ConduitT () BCS.ByteString IO (), ConduitT BCS.ByteString Void IO ()) @@ -234,7 +281,6 @@ runMQTTConduit mkconn MQTTConfig{..} = do   _ch <- newTChanIO   _pktID <- newTVarIO 1-  _ts <- newTVarIO []   _acks <- newTVarIO mempty   _st <- newTVarIO Starting   _ct <- newTVarIO undefined@@ -244,7 +290,7 @@   let _cb = _msgCB       cli = MQTTClient{..} -  t <- async $ clientThread cli+  t <- namedAsync "MQTT clientThread" $ clientThread cli   s <- atomically (waitForLaunch cli t)    when (s == Disconnected) $ wait t@@ -256,8 +302,7 @@   where     clientThread cli = E.finally connectAndRun markDisco       where-        connectAndRun = mkconn $ \ad ->-          E.bracket (start cli ad) cancelAll (run ad)+        connectAndRun = mkconn $ \ad -> (start cli ad) >>= (run ad)         markDisco = atomically $ do           st <- readTVar (_st cli)           guard $ st == Starting || st == Connected@@ -276,20 +321,20 @@       pure c      run (src,sink) c@MQTTClient{..} = do-      o <- async $ whenConnected >> processOut       pch <- newTChanIO-      p <- async $ whenConnected >> doPing-      to <- async (watchdog pch)-      link to--      atomically $ modifyTVar' _ts (\l -> o:p:to:l)+      o <- namedAsync "MQTT out" $ onceConnected >> processOut+      p <- namedAsync "MQTT ping" $ onceConnected >> doPing+      w <- namedAsync "MQTT watchdog" $ watchdog pch+      s <- namedAsync "MQTT in" $ doSrc pch -      runConduit $ src-        .| conduitParser (parsePacket _protocol)-        .| C.mapM_ (\(_,x) -> liftIO (dispatch c pch x))+      void $ waitAnyCancel [o, p, w, s]        where-        whenConnected = atomically $ do+        doSrc pch = runConduit $ src+                    .| conduitParser (parsePacket _protocol)+                    .| C.mapM_ (\(_,x) -> liftIO (dispatch c pch x))++        onceConnected = atomically $ do           s <- readTVar _st           if s /= Connected then retry else pure () @@ -301,7 +346,7 @@         doPing = forever $ threadDelay pingPeriod >> sendPacketIO c PingPkt          watchdog ch = do-          r <- timeout (pingPeriod * 3) w+          r <- namedTimeout "MQTT ping timeout" (pingPeriod * 3) w           case r of             Nothing -> killConn c Timeout             Just _  -> watchdog ch@@ -313,8 +358,6 @@       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. -- An exception is thrown if the client didn't terminate expectedly. waitForClient :: MQTTClient -> IO ()@@ -347,7 +390,7 @@ dispatch c@MQTTClient{..} pch pkt =   case pkt of     (ConnACKPkt p)                            -> connACKd p-    (PublishPkt p)                            -> void $ async (pubMachine p) >>= link+    (PublishPkt p)                            -> void $ namedAsync "MQTT pub machine" (pubMachine p) >>= link     (SubACKPkt (SubscribeResponse i _ _))     -> delegate DSubACK i     (UnsubACKPkt (UnsubscribeResponse i _ _)) -> delegate DUnsubACK i     (PubACKPkt (PubACK i _ _))                -> delegate DPubACK i@@ -417,7 +460,7 @@                     pure ()                    manageQoS2' ch = do-                    v <- timeout 10000000 (sendREC ch)+                    v <- namedTimeout "QoS2 sendREC" 10000000 (sendREC ch)                     case v of                       Nothing -> killConn c Timeout                       _ -> notify >> sendPacketIO c (PubCOMPPkt (PubCOMP _pubPktID 0 mempty))@@ -527,7 +570,7 @@     where       types = [DPubACK, DPubREC, DPubCOMP]       publishAndWait _ pid QoS0 = sendPacketIO c (pkt False pid)-      publishAndWait ch pid _   = withAsync (pub False pid) (\p -> satisfyQoS p ch pid)+      publishAndWait ch pid _   = namedWithAsync "MQTT satisfyQoS" (pub False pid) (\p -> satisfyQoS p ch pid)        pub dup pid = do         sendPacketIO c (pkt dup pid)