diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -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). 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: 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 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 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 @@ -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)