net-mqtt 0.7.0.0 → 0.7.0.1
raw patch · 4 files changed
+78/−34 lines, 4 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- Changelog.md +20/−0
- app/mqtt-watch/Main.hs +29/−19
- net-mqtt.cabal +2/−2
- src/Network/MQTT/Client.hs +27/−13
Changelog.md view
@@ -1,5 +1,25 @@ # Changelog for net-mqtt +## 0.7.0.1++Fixed an error where there'd be an ugly crash in a situation where+connections were failing regularly and we detected the failure before+a connection thread spun up. I was using `undefined` for the default+thread value because it was intended to be immediately set, but did+find a way to get there in a failure storm. It's a `Maybe` now.++fail on unexpected packets. I had a `print` in there from very early+on. Proper sequences are covered, but if a broker sends the client an+unexpected packet, it'd be good to not just ignore it.++When publishing, "no matching subscribers" should not be considered a+failure. It's also not bubbled up to the caller, but it is returned+as an error from the broker to basically say the publish was+successful, but nobody cares about the thing you published.++mqtt-watch will automatically reestablish sessions by default without+reissuing subscriptions (including auto-generated client IDs).+ ## 0.7.0.0 `ConnACKFlags` now has a `SessionReuse` type which makes it very clear
app/mqtt-watch/Main.hs view
@@ -4,22 +4,24 @@ module Main where -import Control.Concurrent (threadDelay)-import Control.Concurrent.Async (async, link)-import Control.Concurrent.STM (TChan, atomically, newTChanIO, readTChan, writeTChan)-import Control.Exception (Handler (..), IOException, catches)-import Control.Monad (forever, when)-import qualified Data.ByteString.Lazy as BL-import Data.Maybe (fromJust)-import qualified Data.Text.IO as TIO-import Data.Word (Word32)+import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (async, link)+import Control.Concurrent.STM (TChan, atomically, newTChanIO, readTChan, writeTChan)+import Control.Exception (Handler (..), IOException, catches)+import Control.Monad (foldM_, forever, when)+import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Lazy.Char8 as BCS+import qualified Data.IORef as R+import Data.Maybe (fromJust)+import qualified Data.Text.IO as TIO+import Data.Word (Word32) import Network.MQTT.Client-import Network.MQTT.Types (ConnACKFlags (..), SessionReuse (..))+import Network.MQTT.Types (ConnACKFlags (..), SessionReuse (..)) import Network.URI-import Options.Applicative (Parser, argument, auto, execParser, fullDesc, help, helper, info, long,- maybeReader, metavar, option, progDesc, short, showDefault, some, str,- switch, value, (<**>))-import System.IO (stdout)+import Options.Applicative (Parser, argument, auto, execParser, fullDesc, help, helper, info, long,+ maybeReader, metavar, option, progDesc, short, showDefault, some, str,+ switch, value, (<**>))+import System.IO (stdout) data Msg = Msg Topic BL.ByteString [Property] @@ -37,7 +39,7 @@ options = Options <$> option (maybeReader parseURI) (long "mqtt-uri" <> short 'u' <> showDefault <> value (fromJust $ parseURI "mqtt://localhost/") <> help "mqtt broker URI") <*> switch (short 'p' <> help "hide properties")- <*> option auto (long "session-timeout" <> showDefault <> value 0 <> help "mqtt session timeout (0 == clean)")+ <*> option auto (long "session-timeout" <> showDefault <> value 300 <> help "mqtt session timeout (0 == clean)") <*> switch (short 'v' <> long "verbose" <> help "enable debug logging") <*> option (toEnum <$> auto) (long "qos" <> short 'q' <> showDefault <> value QoS0 <> help "QoS level (0-2)") <*> switch (long "always-subscribe" <> help "subscribe even when resuming a connection")@@ -55,12 +57,15 @@ run Options{..} = do ch <- newTChanIO async (printer ch (not optHideProps)) >>= link+ uref <- R.newIORef optUri - forever $ catches (go ch) [Handler (\(ex :: MQTTException) -> handler (show ex)),- Handler (\(ex :: IOException) -> handler (show ex))]+ forever $ catches (go ch uref) [Handler (\(ex :: MQTTException) -> handler (show ex)),+ Handler (\(ex :: IOException) -> handler (show ex))] where- go ch = do+ go ch uref = do+ uri <- R.readIORef uref+ when optVerbose $ putStrLn ("Connecting to " <> show uri) mc <- connectURI mqttConfig{_msgCB=SimpleCallback (showme ch), _protocol=Protocol50, _cleanSession=optSessionTime == 0, _connProps=[PropReceiveMaximum 65535,@@ -68,9 +73,10 @@ PropSessionExpiryInterval optSessionTime, PropRequestResponseInformation 1, PropRequestProblemInformation 1]}- optUri+ uri (ConnACKFlags sp _ props) <- connACK mc+ when (optSessionTime > 0) $ updateURI uref props when optVerbose $ putStrLn (if sp == ExistingSession then "<resuming session>" else "<new session>") when optVerbose $ putStrLn ("Properties: " <> show props) when (sp == NewSession || optSubResume) $ do@@ -82,6 +88,10 @@ showme ch _ t m props = atomically $ writeTChan ch $ Msg t m props handler e = putStrLn ("ERROR: " <> e) >> threadDelay 1000000++ updateURI uref = foldM_ up ()+ where up _ (PropAssignedClientIdentifier i) = R.modifyIORef uref (\u -> u{uriFragment='#':BCS.unpack i})+ up a _ = pure a main :: IO () main = run =<< execParser opts
net-mqtt.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 4fb6103336dacf66ccbff3745a76866a8f449a508a712b440117af7be4011cac+-- hash: 20018fa092913650fb64e9a4118baa239d6ba287d900b6089922dd5396e552c6 name: net-mqtt-version: 0.7.0.0+version: 0.7.0.1 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
@@ -97,7 +97,7 @@ , _acks :: TVar (Map (DispatchType,Word16) (TChan MQTTPkt)) , _inflight :: TVar (Map Word16 PublishRequest) , _st :: TVar ConnState- , _ct :: TVar (Async ())+ , _ct :: TVar (Maybe (Async ())) , _outA :: TVar (Map Topic Word16) , _inA :: TVar (Map Word16 Topic) , _connACKFlags :: TVar ConnACKFlags@@ -273,7 +273,7 @@ _acks <- newTVarIO mempty _inflight <- newTVarIO mempty _st <- newTVarIO Starting- _ct <- newTVarIO undefined+ _ct <- newTVarIO Nothing _outA <- newTVarIO mempty _inA <- newTVarIO mempty _connACKFlags <- newTVarIO (ConnACKFlags NewSession ConnUnspecifiedError mempty)@@ -340,7 +340,7 @@ when timedOut $ killConn c Timeout waitForLaunch MQTTClient{..} t = do- writeTVar _ct t+ writeTVar _ct (Just t) c <- readTVar _st if c == Starting then retry else pure c @@ -348,7 +348,7 @@ -- An exception is thrown if the client didn't terminate expectedly. waitForClient :: MQTTClient -> IO () waitForClient c@MQTTClient{..} = do- wait =<< readTVarIO _ct+ void . traverse wait =<< readTVarIO _ct e <- atomically $ stateX c Stopped case e of Nothing -> pure ()@@ -385,8 +385,16 @@ (PubCOMPPkt (PubCOMP i _ _)) -> delegate DPubCOMP i (DisconnectPkt req) -> disco req PongPkt -> atomically . writeTChan pch $ True- x -> print x + -- Not implemented+ (AuthPkt p) -> mqttFail ("unexpected incoming auth: " <> show p)++ -- Things clients shouldn't see+ PingPkt -> mqttFail "unexpected incoming ping packet"+ (ConnPkt _ _) -> mqttFail "unexpected incoming connect"+ (SubscribePkt _) -> mqttFail "unexpected incoming subscribe"+ (UnsubscribePkt _) -> mqttFail "unexpected incoming unsubscribe"+ where connACKd connr@(ConnACKFlags _ val _) = case val of ConnAccepted -> atomically $ do writeTVar _connACKFlags connr@@ -394,10 +402,10 @@ _ -> do t <- readTVarIO _ct atomically $ writeTVar _st (ConnErr connr)- cancelWith t (MQTTException $ show connr)+ maybeCancelWith (MQTTException $ show connr) t pub p@PublishRequest{_pubQoS=QoS0} = atomically (resolve p) >>= notify Nothing- pub p@PublishRequest{_pubQoS=QoS1, _pubPktID} = do+ pub p@PublishRequest{_pubQoS=QoS1, _pubPktID} = notify (Just (PubACKPkt (PubACK _pubPktID 0 mempty))) =<< atomically (resolve p) pub p@PublishRequest{_pubQoS=QoS2} = atomically $ do p'@PublishRequest{..} <- resolve p@@ -456,12 +464,14 @@ disco req = do- t <- readTVarIO _ct atomically $ writeTVar _st (DiscoErr req)- cancelWith t (Discod req)+ maybeCancelWith (Discod req) =<< readTVarIO _ct +maybeCancelWith :: E.Exception e => e -> Maybe (Async ()) -> IO ()+maybeCancelWith e = void . traverse (`cancelWith` e)+ killConn :: E.Exception e => MQTTClient -> e -> IO ()-killConn MQTTClient{..} e = readTVarIO _ct >>= \t -> cancelWith t e+killConn MQTTClient{..} e = readTVarIO _ct >>= maybeCancelWith e checkConnected :: MQTTClient -> STM () checkConnected mc = maybe (pure ()) E.throw =<< stateX mc Connected@@ -576,16 +586,20 @@ | q == QoS0 = pure () | q == QoS1 = void $ do (PubACKPkt (PubACK _ st pprops)) <- atomically $ checkConnected c >> readTChan ch- when (st /= 0) $ mqttFail ("qos 1 publish error: " <> show st <> " " <> show pprops)+ unless (isOK st) $ mqttFail ("qos 1 publish error: " <> show st <> " " <> show pprops) | q == QoS2 = waitRec | otherwise = error "invalid QoS" where+ isOK 0 = True -- success+ isOK 16 = True -- It worked, but nobody cares (no matching subscribers)+ isOK _ = False+ waitRec = do rpkt <- atomically $ checkConnected c >> readTChan ch case rpkt of PubRECPkt (PubREC _ st recprops) -> do- when (st /= 0) $ mqttFail ("qos 2 REC publish error: " <> show st <> " " <> show recprops)+ unless (isOK st) $ mqttFail ("qos 2 REC publish error: " <> show st <> " " <> show recprops) sendPacketIO c (PubRELPkt $ PubREL pid 0 mempty) PubCOMPPkt (PubCOMP _ st' compprops) -> when (st' /= 0) $ mqttFail ("qos 2 COMP publish error: " <> show st' <> " " <> show compprops)@@ -597,7 +611,7 @@ where getDisconnected = do sendPacketIO c (DisconnectPkt $ DisconnectRequest reason props)- wait =<< readTVarIO _ct+ void . traverse wait =<< readTVarIO _ct atomically $ writeTVar _st Stopped orDieTrying = threadDelay 10000000 >> killConn c Timeout