net-mqtt 0.2.3.1 → 0.2.4.0
raw patch · 4 files changed
+50/−19 lines, 4 filesPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
API changes (from Hackage documentation)
+ Network.MQTT.Topic: type Filter = Text
- Network.MQTT.Client: subscribe :: MQTTClient -> [(Topic, QoS)] -> IO [Maybe QoS]
+ Network.MQTT.Client: subscribe :: MQTTClient -> [(Filter, QoS)] -> IO [Maybe QoS]
- Network.MQTT.Client: unsubscribe :: MQTTClient -> [Topic] -> IO ()
+ Network.MQTT.Client: unsubscribe :: MQTTClient -> [Filter] -> IO ()
- Network.MQTT.Topic: match :: Text -> Topic -> Bool
+ Network.MQTT.Topic: match :: Filter -> Topic -> Bool
Files
- Changelog.md +11/−0
- net-mqtt.cabal +2/−2
- src/Network/MQTT/Client.hs +30/−13
- src/Network/MQTT/Topic.hs +7/−4
Changelog.md view
@@ -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.
net-mqtt.cabal view
@@ -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 <https://github.com/dustin/mqtt-hs#readme> category: Network
src/Network/MQTT/Client.hs view
@@ -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))
src/Network/MQTT/Topic.hs view
@@ -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 -- <http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718107 MQTT 3.1.1 specification>.-match :: Text -> Topic -> Bool+match :: Filter -> Topic -> Bool match pat top = cmp (splitOn "/" pat) (splitOn "/" top) where