net-mqtt 0.5.0.1 → 0.5.0.2

raw patch · 7 files changed

+111/−44 lines, 7 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• Changelog.md +13/−0
• README.md +1/−1
• app/Main.hs +7/−5
• net-mqtt.cabal +2/−2
• src/Network/MQTT/Client.hs +37/−27
• src/Network/MQTT/Types.hs +47/−7
• test/Spec.hs +4/−2

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

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

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

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

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

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

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