diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,9 @@ # Changelog for net-mqtt +## 0.8.3.0 + +Ping time is configurable. + ## 0.8.2.4 Minor build changes and code cleanup. Includes LTS update. diff --git a/app/example/Main.hs b/app/example/Main.hs --- a/app/example/Main.hs +++ b/app/example/Main.hs @@ -11,7 +11,7 @@ main :: IO () main = do - let (Just uri) = parseURI "mqtt://test.mosquitto.org" + uri <- maybe (fail "Failed to parse URL") pure $ parseURI "mqtt://test.mosquitto.org" mc <- connectURI mqttConfig{_lwt=Just $ (mkLWT "tmp/haskquit" "bye for now" False){ _willProps=[PropUserProperty "lwt" "prop"]}, diff --git a/net-mqtt.cabal b/net-mqtt.cabal --- a/net-mqtt.cabal +++ b/net-mqtt.cabal @@ -1,11 +1,11 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.35.0. +-- This file has been generated from package.yaml by hpack version 0.35.2. -- -- see: https://github.com/sol/hpack name: net-mqtt -version: 0.8.2.5 +version: 0.8.3.0 synopsis: An MQTT Protocol Implementation. description: Please see the README on GitHub at category: Network @@ -44,12 +44,12 @@ , base >=4.7 && <5 , binary >=0.8.5 && <0.9 , bytestring >=0.10.8 && <0.12 - , conduit >=1.3.1 && <1.4 - , conduit-extra >=1.3.0 && <1.4 + , conduit >=1.3.1 && <1.5 + , conduit-extra >=1.3.0 && <1.5 , connection >=0.2.0 && <0.4 , containers >=0.5.0 && <0.7 , deepseq >=1.4.3.0 && <1.5 - , network-conduit-tls >=1.3.2 && <1.4 + , network-conduit-tls >=1.3.2 && <1.5 , network-uri >=2.6.1 && <2.7 , stm >=2.4.0 && <2.6 , text >=1.2.3 && <2.1.0 @@ -71,13 +71,13 @@ , base >=4.7 && <5 , binary >=0.8.5 && <0.9 , bytestring >=0.10.8 && <0.12 - , conduit >=1.3.1 && <1.4 - , conduit-extra >=1.3.0 && <1.4 + , conduit >=1.3.1 && <1.5 + , conduit-extra >=1.3.0 && <1.5 , connection >=0.2.0 && <0.4 , containers >=0.5.0 && <0.7 , deepseq >=1.4.3.0 && <1.5 , net-mqtt - , network-conduit-tls >=1.3.2 && <1.4 + , network-conduit-tls >=1.3.2 && <1.5 , network-uri >=2.6.1 && <2.7 , stm >=2.4.0 && <2.6 , text >=1.2.3 && <2.1.0 @@ -99,13 +99,13 @@ , base >=4.7 && <5 , binary >=0.8.5 && <0.9 , bytestring - , conduit >=1.3.1 && <1.4 - , conduit-extra >=1.3.0 && <1.4 + , conduit >=1.3.1 && <1.5 + , conduit-extra >=1.3.0 && <1.5 , connection >=0.2.0 && <0.4 , containers >=0.5.0 && <0.7 , deepseq >=1.4.3.0 && <1.5 , net-mqtt - , network-conduit-tls >=1.3.2 && <1.4 + , network-conduit-tls >=1.3.2 && <1.5 , network-uri >=2.6.1 && <2.7 , optparse-applicative , stm >=2.4.0 && <2.6 @@ -133,13 +133,13 @@ , binary >=0.8.5 && <0.9 , bytestring >=0.10.8 && <0.12 , checkers - , conduit >=1.3.1 && <1.4 - , conduit-extra >=1.3.0 && <1.4 + , conduit >=1.3.1 && <1.5 + , conduit-extra >=1.3.0 && <1.5 , connection >=0.2.0 && <0.4 , containers >=0.5.0 && <0.7 , deepseq >=1.4.3.0 && <1.5 , net-mqtt - , network-conduit-tls >=1.3.2 && <1.4 + , network-conduit-tls >=1.3.2 && <1.5 , network-uri >=2.6.1 && <2.7 , stm >=2.4.0 && <2.6 , tasty diff --git a/src/Network/MQTT/Arbitrary.hs b/src/Network/MQTT/Arbitrary.hs --- a/src/Network/MQTT/Arbitrary.hs +++ b/src/Network/MQTT/Arbitrary.hs @@ -29,7 +29,7 @@ import Data.Maybe (mapMaybe) import Data.Text (Text) import qualified Data.Text as Text -import Network.MQTT.Topic (Filter, Topic, mkFilter, mkTopic, unTopic, unFilter) +import Network.MQTT.Topic (Filter, Topic, mkFilter, mkTopic, unFilter, unTopic) import Network.MQTT.Types as MT import Test.QuickCheck as QC 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 @@ -15,6 +15,7 @@ are supported over plain TCP, TLS, WebSockets and Secure WebSockets. -} +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} @@ -65,12 +66,13 @@ import GHC.Conc (labelThread) import Network.Connection (ConnectionParams (..), TLSSettings (..), connectTo, connectionClose, connectionGetChunk, connectionPut, initConnectionContext) -import Network.URI (URI (..), unEscapeString, uriPort, uriRegName, uriUserInfo) +import Network.URI (URI (..), nullURIAuth, 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 Data.Set as Set import Network.MQTT.Topic (Filter, Topic, mkTopic, unFilter, unTopic) import Network.MQTT.Types as T @@ -137,6 +139,8 @@ , _password :: Maybe String -- ^ Optional password (parsed from the URI) , _connectTimeout :: Int -- ^ Connection timeout (microseconds) , _tlsSettings :: TLSSettings -- ^ TLS Settings for secure connections + , _pingPeriod :: Int -- ^ Time in seconds between pings + , _pingPatience :: Int -- ^ Time in seconds for which there must be no incoming packets before the broker is considered dead. Should be more than the ping period plus the maximum expected ping round trip time. } -- | A default 'MQTTConfig'. A '_connID' /may/ be required depending on @@ -151,7 +155,9 @@ _msgCB=NoCallback, _protocol=Protocol311, _connProps=mempty, _connectTimeout=180000000, - _tlsSettings=TLSSettingsSimple False False False} + _tlsSettings=TLSSettingsSimple False False False, + _pingPeriod=30000000, + _pingPatience=90000000} -- | Connect to an MQTT server by URI. -- @@ -173,7 +179,7 @@ "wss:" -> runWS uri True us -> mqttFail $ "invalid URI scheme: " <> us - (Just a) = uriAuthority uri + a = fromMaybe nullURIAuth $ uriAuthority uri (u,p) = up (uriUserInfo a) v <- namedTimeout "MQTT connect" _connectTimeout $ @@ -181,9 +187,7 @@ _hostname=uriRegName a, _port=port (uriPort a) (uriScheme uri), Network.MQTT.Client._username=u, Network.MQTT.Client._password=p} - case v of - Nothing -> mqttFail $ "connection to " <> show uri <> " timed out" - Just x -> pure x + maybe (mqttFail $ "connection to " <> show uri <> " timed out") pure v where port "" "mqtt:" = 1883 @@ -261,9 +265,6 @@ writer conn = maybe (pure ()) (connectionPut conn . BC.toStrict) -pingPeriod :: Int -pingPeriod = 30000000 -- 30 seconds - mqttFail :: String -> a mqttFail = E.throw . MQTTException @@ -353,10 +354,10 @@ .| C.map (BL.toStrict . toByteString _protocol) .| sink - doPing = forever $ threadDelay pingPeriod >> sendPacketIO c PingPkt + doPing = forever $ threadDelay _pingPeriod >> sendPacketIO c PingPkt watchdog ch = forever $ do - toch <- registerDelay (pingPeriod * 3) + toch <- registerDelay _pingPatience timedOut <- atomically $ ((check =<< readTVar toch) >> pure True) `orElse` (readTChan ch >> pure False) when timedOut $ killConn c Timeout @@ -468,16 +469,10 @@ resolveTopic Nothing = pure (blToTopic _pubTopic) resolveTopic (Just x) = do when (_pubTopic /= "") $ modifyTVar' _inA (Map.insert x (blToTopic _pubTopic)) - m <- readTVar _inA - case Map.lookup x m of - Nothing -> mqttFail ("failed to lookup topic alias " <> show x) - Just t -> pure t + maybe (mqttFail ("failed to lookup topic alias " <> show x)) pure . Map.lookup x =<< readTVar _inA - delegate dt pid = atomically $ do - m <- readTVar _acks - case Map.lookup (dt, pid) m of - Nothing -> nak dt - Just ch -> writeTChan ch pkt + delegate dt pid = atomically $ + maybe (nak dt) (`writeTChan` pkt) . Map.lookup (dt, pid) =<< readTVar _acks where nak DPubREC = sendPacket c (PubRELPkt (PubREL pid 0x92 mempty)) @@ -571,7 +566,7 @@ releasePktIDs :: MQTTClient -> [(DispatchType,Word16)] -> STM () releasePktIDs c@MQTTClient{..} ks = checkConnected c >> modifyTVar' _acks deleteMany - where deleteMany m = foldr Map.delete m ks + where deleteMany m = Map.withoutKeys m (Set.fromList ks) sendAndWait :: MQTTClient -> DispatchType -> (Word16 -> MQTTPkt) -> IO MQTTPkt sendAndWait c@MQTTClient{..} dt f = do @@ -592,11 +587,12 @@ subscribe :: MQTTClient -> [(Filter, SubOptions)] -> [Property] -> IO ([Either SubErr QoS], [Property]) subscribe c ls props = do runCallbackThread c - r <- sendAndWait c DSubACK (\pid -> SubscribePkt $ SubscribeRequest pid ls' props) - let (SubACKPkt (SubscribeResponse _ rs aprops)) = r - pure (rs, aprops) - where ls' = map (first filterToBL) ls + sendAndWait c DSubACK (\pid -> SubscribePkt $ SubscribeRequest pid ls' props) >>= \case + SubACKPkt (SubscribeResponse _ rs aprops) -> pure (rs, aprops) + pkt -> mqttFail $ "unexpected response to subscribe: " <> show pkt + where ls' = map (first filterToBL) ls + -- | Unsubscribe from a list of topic filters. -- -- In MQTT 3.1.1, there is no body to an unsubscribe response, so it @@ -656,9 +652,7 @@ isOK 16 = True -- It worked, but nobody cares (no matching subscribers) isOK _ = False - waitRec = do - rpkt <- atomically $ checkConnected c >> readTChan ch - case rpkt of + waitRec = atomically (checkConnected c >> readTChan ch) >>= \case PubRECPkt (PubREC _ st recprops) -> do unless (isOK st) $ mqttFail ("qos 2 REC publish error: " <> show st <> " " <> show recprops) sendPacketIO c (PubRELPkt $ PubREL pid 0 mempty) @@ -747,8 +741,8 @@ -- -- This registration will remain in place until unregisterCorrelated is called to remove it. registerCorrelated :: MQTTClient -> BL.ByteString -> MessageCallback -> STM () -registerCorrelated MQTTClient{_corr} bs cb = modifyTVar' _corr (Map.insert bs cb) +registerCorrelated MQTTClient{_corr} bs = modifyTVar' _corr . Map.insert bs -- | Unregister a callback handler for the given correlated data identifier. unregisterCorrelated :: MQTTClient -> BL.ByteString -> STM () -unregisterCorrelated MQTTClient{_corr} bs = modifyTVar' _corr (Map.delete bs) +unregisterCorrelated MQTTClient{_corr} = modifyTVar' _corr . Map.delete 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 @@ -81,4 +81,4 @@ -- | Convert a 'Topic' to a 'Filter' as all 'Topic's are valid 'Filter's toFilter :: Topic -> Filter -toFilter (Topic t) = Filter t +toFilter = Filter . unTopic diff --git a/src/Network/MQTT/Types.hs b/src/Network/MQTT/Types.hs --- a/src/Network/MQTT/Types.hs +++ b/src/Network/MQTT/Types.hs @@ -27,7 +27,7 @@ parseSubOptions, ByteSize(..) ) where -import Control.Applicative (liftA2, (<|>)) +import Control.Applicative (asum, liftA2, (<|>)) import Control.Monad (replicateM, when) import Data.Attoparsec.Binary (anyWord16be, anyWord32be) import qualified Data.Attoparsec.ByteString as AS @@ -386,14 +386,14 @@ toByteString p (AuthPkt x) = toByteString p x parsePacket :: ProtocolLevel -> A.Parser MQTTPkt -parsePacket p = parseConnect <|> parseConnectACK - <|> parsePublish p <|> parsePubACK - <|> parsePubREC <|> parsePubREL <|> parsePubCOMP - <|> parseSubscribe p <|> parseSubACK p - <|> parseUnsubscribe p <|> parseUnsubACK p - <|> PingPkt <$ A.string "\192\NUL" <|> PongPkt <$ A.string "\208\NUL" - <|> parseDisconnect p - <|> parseAuth +parsePacket p = asum [parseConnect, parseConnectACK, + parsePublish p, parsePubACK, + parsePubREC, parsePubREL, parsePubCOMP, + parseSubscribe p, parseSubACK p, + parseUnsubscribe p, parseUnsubACK p, + PingPkt <$ A.string "\192\NUL", PongPkt <$ A.string "\208\NUL", + parseDisconnect p, + parseAuth] aWord16 :: A.Parser Word16 aWord16 = anyWord16be @@ -404,8 +404,7 @@ aString :: A.Parser BL.ByteString aString = do n <- aWord16 - s <- A.take (fromIntegral n) - pure $ BL.fromStrict s + BL.fromStrict <$> A.take (fromIntegral n) -- | Parse a CONNect packet. This is useful when examining the -- beginning of the stream as it allows you to determine the protocol