diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,16 @@ # Changelog for net-mqtt +## 0.2.4.0 + +Introduced `Filter` type alias to distinguish from `Topic`. + +Reintroduced timeout management at the protocol layer, dropping a +connection when a pong response hasn't come in in a while (~3x longer +than the current 30s ping rate). This was mostly after noticing +mosquitto do really weird things where it seemed to just forget about +all my active connections (other clients figured that out and dropped +and reconnected). + ## 0.2.3.1 Fixed up github links. 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: ac8bc19458d5b6eef155eb8eff472b52acc60654122e07271771dc6cf7f92552 +-- hash: 823f507cc62f2e9bd1f915a491b6404b412b35e9c9848e46e1927c1533ed6b0f name: net-mqtt -version: 0.2.3.1 +version: 0.2.4.0 synopsis: An MQTT Protocol Implementation. description: Please see the README on GitHub at category: Network 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 @@ -26,7 +26,8 @@ import Control.Concurrent (threadDelay) import Control.Concurrent.Async (Async, async, cancel, cancelWith, - race_, wait, waitCatch, withAsync) + link, race_, wait, waitCatch, + withAsync) import Control.Concurrent.STM (STM, TChan, TVar, atomically, modifyTVar', newTChan, newTChanIO, newTVarIO, readTChan, readTVar, @@ -51,10 +52,11 @@ import Data.Word (Word16) import Network.URI (URI (..), unEscapeString, uriPort, uriRegName, uriUserInfo) +import System.Timeout (timeout) -import Network.MQTT.Topic (Topic) +import Network.MQTT.Topic (Filter, Topic) import Network.MQTT.Types as T data ConnState = Starting | Connected | Disconnected deriving (Eq, Show) @@ -136,6 +138,9 @@ runClientTLS :: MQTTConfig -> IO MQTTClient runClientTLS cfg@MQTTConfig{..} = runClientAppData (runTLSClient (tlsClientConfig _port (BCS.pack _hostname))) cfg +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 @@ -185,15 +190,18 @@ run ad c@MQTTClient{..} = do o <- async processOut + pch <- newTChanIO p <- async doPing + to <- async (watchdog pch) + link to atomically $ do - modifyTVar' _ts (\l -> o:p:l) + modifyTVar' _ts (\l -> o:p:to:l) writeTVar _st Connected runConduit $ appSource ad .| conduitParser parsePacket - .| C.mapM_ (\(_,x) -> liftIO (dispatch c x)) + .| C.mapM_ (\(_,x) -> liftIO (dispatch c pch x)) where processOut = runConduit $ @@ -201,8 +209,17 @@ .| C.map (BL.toStrict . toByteString) .| appSink ad - doPing = forever $ threadDelay 30000000 >> sendPacketIO c PingPkt + doPing = forever $ threadDelay pingPeriod >> sendPacketIO c PingPkt + + watchdog ch = do + r <- timeout (pingPeriod * 3) w + case r of + Nothing -> E.throwIO Timeout + Just _ -> watchdog ch + + where w = atomically . readTChan $ ch + waitForLaunch MQTTClient{..} t = do writeTVar _ct t c <- readTVar _st @@ -218,8 +235,8 @@ instance E.Exception MQTTException -dispatch :: MQTTClient -> MQTTPkt -> IO () -dispatch c@MQTTClient{..} pkt = +dispatch :: MQTTClient -> TChan Bool -> MQTTPkt -> IO () +dispatch c@MQTTClient{..} pch pkt = case pkt of (PublishPkt p) -> pubMachine p (SubACKPkt (SubscribeResponse i _)) -> delegate DSubACK i @@ -228,7 +245,7 @@ (PubRECPkt (PubREC i)) -> delegate DPubREC i (PubRELPkt (PubREL i)) -> delegate DPubREL i (PubCOMPPkt (PubCOMP i)) -> delegate DPubCOMP i - PongPkt -> pure () + PongPkt -> atomically . writeTChan pch $ True x -> print x where delegate dt pid = atomically $ do @@ -298,9 +315,9 @@ -- Wait for the response in a separate transaction. atomically $ releasePktID c (dt,pid) >> readTChan ch --- | Subscribe to a list of topics with their respective QoSes. The --- accepted QoSes are returned in the same order as requested. -subscribe :: MQTTClient -> [(Topic, QoS)] -> IO [Maybe QoS] +-- | Subscribe to a list of topic filters with their respective QoSes. +-- The accepted QoSes are returned in the same order as requested. +subscribe :: MQTTClient -> [(Filter, QoS)] -> IO [Maybe QoS] subscribe c@MQTTClient{..} ls = do r <- sendAndWait c DSubACK (\pid -> SubscribePkt $ SubscribeRequest pid ls') let (SubACKPkt (SubscribeResponse _ rs)) = r @@ -308,8 +325,8 @@ where ls' = map (\(s, i) -> (textToBL s, i)) ls --- | Unsubscribe from a list of topics. -unsubscribe :: MQTTClient -> [Topic] -> IO () +-- | Unsubscribe from a list of topic filters. +unsubscribe :: MQTTClient -> [Filter] -> IO () unsubscribe c@MQTTClient{..} ls = void $ sendAndWait c DUnsubACK (\pid -> UnsubscribePkt $ UnsubscribeRequest pid (map textToBL ls)) diff --git a/src/Network/MQTT/Topic.hs b/src/Network/MQTT/Topic.hs --- a/src/Network/MQTT/Topic.hs +++ b/src/Network/MQTT/Topic.hs @@ -1,6 +1,6 @@ {-| Module : Network.MQTT.Topic. -Description : An MQTT client. +Description : MQTT Topic types and utilities. Copyright : (c) Dustin Sallings, 2019 License : BSD3 Maintainer : dustin@spy.net @@ -12,18 +12,21 @@ {-# LANGUAGE OverloadedStrings #-} module Network.MQTT.Topic ( - Topic, match + Filter, Topic, match ) where -import Data.Text (Text, splitOn, isPrefixOf) +import Data.Text (Text, isPrefixOf, splitOn) -- | An MQTT topic. type Topic = Text +-- | An MQTT topic filter. +type Filter = Text + -- | match returns true iff the given pattern can be matched by the -- specified Topic as defined in the -- . -match :: Text -> Topic -> Bool +match :: Filter -> Topic -> Bool match pat top = cmp (splitOn "/" pat) (splitOn "/" top) where