net-mqtt 0.8.2.5 → 0.8.3.0

raw patch · 7 files changed

+55/−58 lines, 7 filesdep ~conduitdep ~conduit-extradep ~network-conduit-tlsPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: conduit, conduit-extra, network-conduit-tls

API changes (from Hackage documentation)

Files

• Changelog.md +4/−0
• app/example/Main.hs +1/−1
• net-mqtt.cabal +14/−14
• src/Network/MQTT/Arbitrary.hs +1/−1
• src/Network/MQTT/Client.hs +24/−30
• src/Network/MQTT/Topic.hs +1/−1
• src/Network/MQTT/Types.hs +10/−11

@@ -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.

@@ -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"]},

@@ -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 <https://github.com/dustin/mqtt-hs#readme> 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

@@ -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 

@@ -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

@@ -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

@@ -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