diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,18 @@ # Changelog for net-mqtt +## 0.5.0.2 + +With a few attempts to misuse the library, I found some places where +error messages weren't useful enough. There were still two cases +where failures turned into indefinite STM errors instead of more +informative errors. 1) when the broker declined your connection and +2) when the broker refused your connection. These are errors at +differnet layers, so were addressed differently, but should be +informative in both cases now. + +Unsubscribe was apparently broken in MQTT 5 as well. I'd never tried +to use it, and just happened to notice it wasn't quite right. + ## 0.5.0.1 Now with no known issues. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ main :: IO main = do mc <- runClient mqttConfig{_msgCB=Just msgReceived} - print =<< subscribe mc [("tmp/topic1", subOptions), ("tmp/topic2", subOptions)] + print =<< subscribe mc [("tmp/topic1", subOptions), ("tmp/topic2", subOptions)] [] print =<< waitForClient mc -- wait for the the client to disconnect where diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -21,21 +21,23 @@ PropRequestProblemInformation 1]} putStrLn "connected!" print =<< svrProps mc - print =<< subscribe mc [("oro/#", subOptions), ("tmp/#", subOptions{_subQoS=QoS2})] + print =<< subscribe mc [("oro/#", subOptions), ("tmp/#", subOptions{_subQoS=QoS2})] mempty + print =<< unsubscribe mc ["oro/#"] mempty + let pprops = [PropUserProperty "hello" "mqttv5"] p <- async $ forever $ pubAliased mc "tmp/hi/from/haskell" "hi from haskell" False QoS1 pprops >> threadDelay 10000000 putStrLn "Publishing at QoS > 0" publishq mc "tmp/q1" "this message is at QoS 1" True QoS1 [PropUserProperty "hi" "there 1", - PropMessageExpiryInterval 30, - PropContentType "text/plain"] + PropMessageExpiryInterval 30, + PropContentType "text/plain"] putStrLn "Published!" putStrLn "Publishing at QoS > 1" publishq mc "tmp/q2" "this message is at QoS 2" True QoS2 [PropUserProperty "hi" "there 2", - PropMessageExpiryInterval 30, - PropContentType "text/plain"] + PropMessageExpiryInterval 30, + PropContentType "text/plain"] putStrLn "Published!" print =<< waitForClient mc 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: 0536d82db90a39cbd6059bf3945d34aa7b1be16e28039c56aab02768d106418d +-- hash: 3c221d190f39c9b883915e0e17701983e95297762972cda53caf5f87b53856c1 name: net-mqtt -version: 0.5.0.1 +version: 0.5.0.2 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 @@ -67,7 +67,7 @@ import Network.MQTT.Topic (Filter, Topic) import Network.MQTT.Types as T -data ConnState = Starting | Connected | Disconnected | DiscoErr DisconnectRequest deriving (Eq, Show) +data ConnState = Starting | Connected | Disconnected | DiscoErr DisconnectRequest | ConnErr ConnACKFlags deriving (Eq, Show) data DispatchType = DSubACK | DUnsubACK | DPubACK | DPubREC | DPubREL | DPubCOMP deriving (Eq, Show, Ord, Enum, Bounded) @@ -176,8 +176,10 @@ t <- async $ clientThread cli s <- atomically (waitForLaunch cli t) - when (s /= Connected) $ wait t + when (s == Disconnected) $ wait t + atomically $ checkConnected cli + pure cli where @@ -187,7 +189,7 @@ E.bracket (start cli ad) cancelAll (run ad) markDisco = atomically $ do st <- readTVar (_st cli) - guard $ st == Connected + guard $ st == Starting || st == Connected writeTVar (_st cli) Disconnected start c@MQTTClient{..} ad = do @@ -199,11 +201,13 @@ T._cleanSession=_cleanSession, T._properties=_connProps} yield (BL.toStrict $ toByteString _protocol req) .| appSink ad - (ConnACKPkt (ConnACKFlags _ val props)) <- appSource ad .| sinkParser (parsePacket _protocol) - case val of - ConnAccepted -> liftIO $ atomically $ writeTVar _svrProps props - x -> fail (show x) + (ConnACKPkt connr@(ConnACKFlags _ val props)) <- appSource ad .| sinkParser (parsePacket _protocol) + liftIO $ atomically $ do + writeTVar _svrProps props + writeTVar _st $ if val == ConnAccepted then Connected else (ConnErr connr) + when (val /= ConnAccepted) $ fail (show val) + pure c run ad c@MQTTClient{..} = do @@ -256,16 +260,16 @@ dispatch :: MQTTClient -> TChan Bool -> MQTTPkt -> IO () dispatch c@MQTTClient{..} pch pkt = case pkt of - (PublishPkt p) -> pubMachine p - (SubACKPkt (SubscribeResponse i _ _)) -> delegate DSubACK i - (UnsubACKPkt (UnsubscribeResponse i _)) -> delegate DUnsubACK i - (PubACKPkt (PubACK i _ _)) -> delegate DPubACK i - (PubRECPkt (PubREC i _ _)) -> delegate DPubREC i - (PubRELPkt (PubREL i _ _)) -> delegate DPubREL i - (PubCOMPPkt (PubCOMP i _ _)) -> delegate DPubCOMP i - (DisconnectPkt req) -> disco req - PongPkt -> atomically . writeTChan pch $ True - x -> print x + (PublishPkt p) -> pubMachine p + (SubACKPkt (SubscribeResponse i _ _)) -> delegate DSubACK i + (UnsubACKPkt (UnsubscribeResponse i _ _)) -> delegate DUnsubACK i + (PubACKPkt (PubACK i _ _)) -> delegate DPubACK i + (PubRECPkt (PubREC i _ _)) -> delegate DPubREC i + (PubRELPkt (PubREL i _ _)) -> delegate DPubREL i + (PubCOMPPkt (PubCOMP i _ _)) -> delegate DPubCOMP i + (DisconnectPkt req) -> disco req + PongPkt -> atomically . writeTChan pch $ True + x -> print x where delegate dt pid = atomically $ do m <- readTVar _acks @@ -325,6 +329,7 @@ check Connected = pure () check Disconnected = fail "disconnected" check (DiscoErr req) = fail (show req) + check (ConnErr req) = fail (show req) sendPacket :: MQTTClient -> MQTTPkt -> STM () sendPacket c@MQTTClient{..} p = checkConnected c >> writeTChan _ch p @@ -370,18 +375,25 @@ -- | 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, SubOptions)] -> IO ([Either SubErr QoS], [Property]) -subscribe c@MQTTClient{..} ls = do - r <- sendAndWait c DSubACK (\pid -> SubscribePkt $ SubscribeRequest pid ls' mempty) - let (SubACKPkt (SubscribeResponse _ rs props)) = r - pure (rs, props) +subscribe :: MQTTClient -> [(Filter, SubOptions)] -> [Property] -> IO ([Either SubErr QoS], [Property]) +subscribe c@MQTTClient{..} ls props = do + r <- sendAndWait c DSubACK (\pid -> SubscribePkt $ SubscribeRequest pid ls' props) + let (SubACKPkt (SubscribeResponse _ rs aprops)) = r + pure (rs, aprops) where ls' = map (\(s, i) -> (textToBL s, i)) ls -- | Unsubscribe from a list of topic filters. -unsubscribe :: MQTTClient -> [Filter] -> [Property] -> IO () -unsubscribe c@MQTTClient{..} ls props = - void $ sendAndWait c DUnsubACK (\pid -> UnsubscribePkt $ UnsubscribeRequest pid (map textToBL ls) props) +-- +-- In MQTT 3.1.1, there is no body to an unsubscribe response, so it +-- can be ignored. If this returns, you were unsubscribed. In MQTT +-- 5, you'll get a list of unsub status values corresponding to your +-- request filters, and whatever properties the server thought you +-- should know about. +unsubscribe :: MQTTClient -> [Filter] -> [Property] -> IO ([UnsubStatus], [Property]) +unsubscribe c@MQTTClient{..} ls props = do + (UnsubACKPkt (UnsubscribeResponse _ rsn rprop)) <- sendAndWait c DUnsubACK (\pid -> UnsubscribePkt $ UnsubscribeRequest pid (map textToBL ls) props) + pure (rprop, rsn) -- | Publish a message (QoS 0). publish :: MQTTClient @@ -426,7 +438,6 @@ | q == QoS1 = void $ do (PubACKPkt (PubACK _ st pprops)) <- atomically $ readTChan ch when (st /= 0) $ fail ("qos 1 publish error: " <> show st <> " " <> show pprops) - pure () | q == QoS2 = waitRec | otherwise = error "invalid QoS" @@ -438,7 +449,6 @@ cancel p -- must not publish after rel (PubCOMPPkt (PubCOMP _ st' compprops)) <- atomically $ readTChan ch when (st' /= 0) $ fail ("qos 2 COMP publish error: " <> show st' <> " " <> show compprops) - pure () -- | Disconnect from the MQTT server. disconnect :: MQTTClient -> DiscoReason -> [Property] -> IO () 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 @@ -19,7 +19,7 @@ ProtocolLevel(..), Property(..), AuthRequest(..), SubscribeRequest(..), SubOptions(..), subOptions, SubscribeResponse(..), SubErr(..), RetainHandling(..), DisconnectRequest(..), - UnsubscribeRequest(..), UnsubscribeResponse(..), DiscoReason(..), + UnsubscribeRequest(..), UnsubscribeResponse(..), UnsubStatus(..), DiscoReason(..), parsePacket, ByteMe(toByteString), -- for testing encodeLength, parseHdrLen, parseProperty, parseProperties, bsProps, @@ -784,19 +784,59 @@ Left x -> fail x Right x -> pure x -data UnsubscribeResponse = UnsubscribeResponse Word16 [Property] deriving(Eq, Show) +data UnsubStatus = UnsubSuccess + | UnsubNoSubscriptionExisted + | UnsubUnspecifiedError + | UnsubImplementationSpecificError + | UnsubNotAuthorized + | UnsubTopicFilterInvalid + | UnsubPacketIdentifierInUse + deriving(Show, Eq, Bounded, Enum) +instance ByteMe UnsubStatus where + toByteString _ UnsubSuccess = BL.singleton 0x00 + toByteString _ UnsubNoSubscriptionExisted = BL.singleton 0x11 + toByteString _ UnsubUnspecifiedError = BL.singleton 0x80 + toByteString _ UnsubImplementationSpecificError = BL.singleton 0x83 + toByteString _ UnsubNotAuthorized = BL.singleton 0x87 + toByteString _ UnsubTopicFilterInvalid = BL.singleton 0x8F + toByteString _ UnsubPacketIdentifierInUse = BL.singleton 0x91 + +data UnsubscribeResponse = UnsubscribeResponse Word16 [Property] [UnsubStatus] deriving(Eq, Show) + instance ByteMe UnsubscribeResponse where - toByteString prot (UnsubscribeResponse pid props) = BL.singleton 0xb0 - <> withLength (encodeWord16 pid <> bsProps prot props) + toByteString Protocol311 (UnsubscribeResponse pid _ _) = BL.singleton 0xb0 + <> withLength (encodeWord16 pid) + toByteString Protocol50 (UnsubscribeResponse pid props res) = BL.singleton 0xb0 + <> withLength (encodeWord16 pid + <> bsProps Protocol50 props + <> mconcat (fmap (toByteString Protocol50) res)) + parseUnsubACK :: ProtocolLevel -> A.Parser MQTTPkt -parseUnsubACK prot = do +parseUnsubACK Protocol311 = do _ <- A.word8 0xb0 _ <- parseHdrLen pid <- aWord16 - props <- parseProperties prot - pure $ UnsubACKPkt (UnsubscribeResponse pid props) + pure $ UnsubACKPkt (UnsubscribeResponse pid mempty mempty) + +parseUnsubACK Protocol50 = do + _ <- A.word8 0xb0 + rl <- parseHdrLen + pid <- aWord16 + props <- parseProperties Protocol50 + res <- replicateM (rl - propLen Protocol50 props - 2) unsubACK + pure $ UnsubACKPkt (UnsubscribeResponse pid props res) + + where + unsubACK :: A.Parser UnsubStatus + unsubACK = (UnsubSuccess <$ A.word8 0x00) + <|> (UnsubNoSubscriptionExisted <$ A.word8 0x11) + <|> (UnsubUnspecifiedError <$ A.word8 0x80) + <|> (UnsubImplementationSpecificError <$ A.word8 0x83) + <|> (UnsubNotAuthorized <$ A.word8 0x87) + <|> (UnsubTopicFilterInvalid <$ A.word8 0x8F) + <|> (UnsubPacketIdentifierInUse <$ A.word8 0x91) data AuthRequest = AuthRequest Word8 [Property] deriving (Eq, Show) diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -130,8 +130,10 @@ | length l == 1 = [] | otherwise = [UnsubscribeRequest p sl props | sl <- shrinkList (:[]) l, not (null sl)] +instance Arbitrary UnsubStatus where arbitrary = arbitraryBoundedEnum + instance Arbitrary UnsubscribeResponse where - arbitrary = UnsubscribeResponse <$> arbitrary <*> arbitrary + arbitrary = UnsubscribeResponse <$> arbitrary <*> arbitrary <*> arbitrary instance Arbitrary MT.Property where arbitrary = oneof [ @@ -222,7 +224,7 @@ where c = map (\(k,SubOptions{..}) -> (k,subOptions{_subQoS=_subQoS})) s v311mask (SubACKPkt (SubscribeResponse p s _)) = SubACKPkt (SubscribeResponse p s mempty) v311mask (UnsubscribePkt (UnsubscribeRequest p l _)) = UnsubscribePkt (UnsubscribeRequest p l mempty) - v311mask (UnsubACKPkt (UnsubscribeResponse p _)) = UnsubACKPkt (UnsubscribeResponse p mempty) + v311mask (UnsubACKPkt (UnsubscribeResponse p _ _)) = UnsubACKPkt (UnsubscribeResponse p mempty mempty) v311mask (PublishPkt req) = PublishPkt req{_pubProps=mempty} v311mask (DisconnectPkt _) = DisconnectPkt (DisconnectRequest DiscoNormalDisconnection mempty) v311mask (PubACKPkt (PubACK x _ _)) = PubACKPkt (PubACK x 0 mempty)