diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -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 diff --git a/app/mqtt-watch/Main.hs b/app/mqtt-watch/Main.hs --- a/app/mqtt-watch/Main.hs +++ b/app/mqtt-watch/Main.hs @@ -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 "" else "") 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 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: 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 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 @@ -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