net-mqtt 0.8.3.0 → 0.8.5.0
raw patch · 5 files changed
+82/−64 lines, 5 filesdep +crypton-connectiondep −connectiondep ~network-conduit-tlsPVP ok
version bump matches the API change (PVP)
Dependencies added: crypton-connection
Dependencies removed: connection
Dependency ranges changed: network-conduit-tls
API changes (from Hackage documentation)
+ Network.MQTT.Client: stopClient :: MQTTClient -> IO ()
Files
- Changelog.md +4/−0
- app/mqtt-watch/Main.hs +7/−6
- net-mqtt.cabal +11/−12
- src/Network/MQTT/Client.hs +10/−1
- src/Network/MQTT/Types.hs +50/−45
Changelog.md view
@@ -1,5 +1,9 @@ # Changelog for net-mqtt +## 0.8.5.0++Switch to crypton-connection+ ## 0.8.3.0 Ping time is configurable.
app/mqtt-watch/Main.hs view
@@ -77,7 +77,7 @@ where go ch uref = do uri <- R.readIORef uref- when optVerbose $ putStrLn ("Connecting to " <> show uri)+ verbose ("Connecting to " <> show uri) mc <- connectURI mqttConfig{_msgCB=SimpleCallback (showme ch), _protocol=Protocol50, _cleanSession=optSessionTime == 0, _connProps=[PropReceiveMaximum 65535,@@ -89,13 +89,14 @@ (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- subres <- subscribe mc [(t, subOptions{_subQoS=optQoS}) | t <- optTopics] mempty- when optVerbose $ print subres+ verbose (if sp == ExistingSession then "<resuming session>" else "<new session>")+ verbose ("Properties: " <> show props)+ when (sp == NewSession || optSubResume) $ subscribe mc [(t, subOptions{_subQoS=optQoS}) | t <- optTopics] mempty >>= verboseShow print =<< waitForClient mc++ verbose = when optVerbose . putStrLn+ verboseShow = verbose . show showme ch _ t m props = atomically $ writeTChan ch $ Msg t m props
net-mqtt.cabal view
@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: net-mqtt-version: 0.8.3.0+version: 0.8.5.0 synopsis: An MQTT Protocol Implementation. description: Please see the README on GitHub at <https://github.com/dustin/mqtt-hs#readme> category: Network@@ -35,7 +35,6 @@ Paths_net_mqtt hs-source-dirs: src- ghc-options: -Wall build-depends: QuickCheck >=2.12.6.1 && <2.15 , async >=2.2.1 && <2.3@@ -46,10 +45,10 @@ , bytestring >=0.10.8 && <0.12 , 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+ , crypton-connection >=0.3.0 , deepseq >=1.4.3.0 && <1.5- , network-conduit-tls >=1.3.2 && <1.5+ , network-conduit-tls ==1.4.* , network-uri >=2.6.1 && <2.7 , stm >=2.4.0 && <2.6 , text >=1.2.3 && <2.1.0@@ -62,7 +61,7 @@ Paths_net_mqtt hs-source-dirs: app/example- ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N+ ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: QuickCheck >=2.12.6.1 && <2.15 , async >=2.2.1 && <2.3@@ -73,11 +72,11 @@ , bytestring >=0.10.8 && <0.12 , 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+ , crypton-connection >=0.3.0 , deepseq >=1.4.3.0 && <1.5 , net-mqtt- , network-conduit-tls >=1.3.2 && <1.5+ , network-conduit-tls ==1.4.* , network-uri >=2.6.1 && <2.7 , stm >=2.4.0 && <2.6 , text >=1.2.3 && <2.1.0@@ -90,7 +89,7 @@ Paths_net_mqtt hs-source-dirs: app/mqtt-watch- ghc-options: -Wall -threaded -rtsopts -eventlog -with-rtsopts=-N+ ghc-options: -threaded -rtsopts -eventlog -with-rtsopts=-N build-depends: QuickCheck >=2.12.6.1 && <2.15 , async >=2.2.1 && <2.3@@ -101,11 +100,11 @@ , bytestring , 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+ , crypton-connection >=0.3.0 , deepseq >=1.4.3.0 && <1.5 , net-mqtt- , network-conduit-tls >=1.3.2 && <1.5+ , network-conduit-tls ==1.4.* , network-uri >=2.6.1 && <2.7 , optparse-applicative , stm >=2.4.0 && <2.6@@ -135,11 +134,11 @@ , checkers , 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+ , crypton-connection >=0.3.0 , deepseq >=1.4.3.0 && <1.5 , net-mqtt- , network-conduit-tls >=1.3.2 && <1.5+ , network-conduit-tls ==1.4.* , network-uri >=2.6.1 && <2.7 , stm >=2.4.0 && <2.6 , tasty
src/Network/MQTT/Client.hs view
@@ -27,7 +27,7 @@ -- * Running and waiting for the client. waitForClient, connectURI, isConnected,- disconnect, normalDisconnect,+ disconnect, normalDisconnect, stopClient, -- * General client interactions. subscribe, unsubscribe, publish, publishq, pubAliased, svrProps, connACK, MQTTException(..),@@ -372,6 +372,15 @@ waitForClient c@MQTTClient{..} = flip E.finally (stopCallbackThread c) $ do void . traverse wait =<< readTVarIO _ct maybe (pure ()) E.throwIO =<< atomically (stateX c Stopped)++-- | Stops the client and closes the connection without sending a DISCONNECT+-- message to the broker. This will cause the last-will message to be delivered+-- by the broker if it has been defined.+stopClient :: MQTTClient -> IO ()+stopClient MQTTClient{..} = do+ void . traverse cancel =<< readTVarIO _ct+ void . traverse cancel =<< readTVarIO _cbHandle+ atomically (writeTVar _st Stopped) stateX :: MQTTClient -> ConnState -> STM (Maybe E.SomeException) stateX MQTTClient{..} want = f <$> readTVar _st
src/Network/MQTT/Types.hs view
@@ -27,7 +27,7 @@ parseSubOptions, ByteSize(..) ) where -import Control.Applicative (asum, liftA2, (<|>))+import Control.Applicative (liftA2, (<|>)) import Control.Monad (replicateM, when) import Data.Attoparsec.Binary (anyWord16be, anyWord32be) import qualified Data.Attoparsec.ByteString as AS@@ -35,6 +35,7 @@ import Data.Binary.Put (putWord32be, runPut) import Data.Bits (Bits (..), shiftL, testBit, (.&.), (.|.)) import qualified Data.ByteString.Lazy as BL+import Data.Foldable (asum) import Data.Functor (($>)) import Data.Maybe (fromMaybe, isJust) import Data.Word (Word16, Word32, Word8)@@ -239,45 +240,50 @@ toByteString _ (PropSharedSubscriptionAvailable x) = peW8 0x2a x +oneOf :: [(Word8, p)] -> A.Parser p+oneOf = asum . fmap (\(w, p) -> A.word8 w $> p)++oneOfp :: [(Word8, A.Parser p)] -> A.Parser p+oneOfp = asum . fmap (\(b,p) -> A.word8 b *> p)+ parseProperty :: A.Parser Property-parseProperty = (A.word8 0x01 >> PropPayloadFormatIndicator <$> A.anyWord8)- <|> (A.word8 0x02 >> PropMessageExpiryInterval <$> aWord32)- <|> (A.word8 0x03 >> PropContentType <$> aString)- <|> (A.word8 0x08 >> PropResponseTopic <$> aString)- <|> (A.word8 0x09 >> PropCorrelationData <$> aString)- <|> (A.word8 0x0b >> PropSubscriptionIdentifier <$> decodeVarInt)- <|> (A.word8 0x11 >> PropSessionExpiryInterval <$> aWord32)- <|> (A.word8 0x12 >> PropAssignedClientIdentifier <$> aString)- <|> (A.word8 0x13 >> PropServerKeepAlive <$> aWord16)- <|> (A.word8 0x15 >> PropAuthenticationMethod <$> aString)- <|> (A.word8 0x16 >> PropAuthenticationData <$> aString)- <|> (A.word8 0x17 >> PropRequestProblemInformation <$> A.anyWord8)- <|> (A.word8 0x18 >> PropWillDelayInterval <$> aWord32)- <|> (A.word8 0x19 >> PropRequestResponseInformation <$> A.anyWord8)- <|> (A.word8 0x1a >> PropResponseInformation <$> aString)- <|> (A.word8 0x1c >> PropServerReference <$> aString)- <|> (A.word8 0x1f >> PropReasonString <$> aString)- <|> (A.word8 0x21 >> PropReceiveMaximum <$> aWord16)- <|> (A.word8 0x22 >> PropTopicAliasMaximum <$> aWord16)- <|> (A.word8 0x23 >> PropTopicAlias <$> aWord16)- <|> (A.word8 0x24 >> PropMaximumQoS <$> A.anyWord8)- <|> (A.word8 0x25 >> PropRetainAvailable <$> A.anyWord8)- <|> (A.word8 0x26 >> PropUserProperty <$> aString <*> aString)- <|> (A.word8 0x27 >> PropMaximumPacketSize <$> aWord32)- <|> (A.word8 0x28 >> PropWildcardSubscriptionAvailable <$> A.anyWord8)- <|> (A.word8 0x29 >> PropSubscriptionIdentifierAvailable <$> A.anyWord8)- <|> (A.word8 0x2a >> PropSharedSubscriptionAvailable <$> A.anyWord8)+parseProperty = oneOfp [ (0x01, PropPayloadFormatIndicator <$> A.anyWord8),+ (0x02, PropMessageExpiryInterval <$> aWord32),+ (0x03, PropContentType <$> aString),+ (0x08, PropResponseTopic <$> aString),+ (0x09, PropCorrelationData <$> aString),+ (0x0b, PropSubscriptionIdentifier <$> decodeVarInt),+ (0x11, PropSessionExpiryInterval <$> aWord32),+ (0x12, PropAssignedClientIdentifier <$> aString),+ (0x13, PropServerKeepAlive <$> aWord16),+ (0x15, PropAuthenticationMethod <$> aString),+ (0x16, PropAuthenticationData <$> aString),+ (0x17, PropRequestProblemInformation <$> A.anyWord8),+ (0x18, PropWillDelayInterval <$> aWord32),+ (0x19, PropRequestResponseInformation <$> A.anyWord8),+ (0x1a, PropResponseInformation <$> aString),+ (0x1c, PropServerReference <$> aString),+ (0x1f, PropReasonString <$> aString),+ (0x21, PropReceiveMaximum <$> aWord16),+ (0x22, PropTopicAliasMaximum <$> aWord16),+ (0x23, PropTopicAlias <$> aWord16),+ (0x24, PropMaximumQoS <$> A.anyWord8),+ (0x25, PropRetainAvailable <$> A.anyWord8),+ (0x26, PropUserProperty <$> aString <*> aString),+ (0x27, PropMaximumPacketSize <$> aWord32),+ (0x28, PropWildcardSubscriptionAvailable <$> A.anyWord8),+ (0x29, PropSubscriptionIdentifierAvailable <$> A.anyWord8),+ (0x2a, PropSharedSubscriptionAvailable <$> A.anyWord8)+ ] bsProps :: ProtocolLevel -> [Property] -> BL.ByteString-bsProps Protocol311 _ = mempty-bsProps p l = let b = foldMap (toByteString p) l in- (BL.pack . encodeLength . fromIntegral . BL.length) b <> b+bsProps Protocol311 = const mempty+bsProps p = withLength . foldMap (toByteString p) + parseProperties :: ProtocolLevel -> A.Parser [Property] parseProperties Protocol311 = pure mempty-parseProperties Protocol50 = do- len <- decodeVarInt- either fail pure . AS.parseOnly (A.many' parseProperty) =<< A.take len+parseProperties Protocol50 = either fail pure . AS.parseOnly (A.many' parseProperty) =<< A.take =<< decodeVarInt -- | MQTT Protocol Levels data ProtocolLevel = Protocol311 -- ^ MQTT 3.1.1@@ -402,9 +408,7 @@ aWord32 = anyWord32be aString :: A.Parser BL.ByteString-aString = do- n <- aWord16- BL.fromStrict <$> A.take (fromIntegral n)+aString = fmap BL.fromStrict . A.take . fromIntegral =<< aWord16 -- | Parse a CONNect packet. This is useful when examining the -- beginning of the stream as it allows you to determine the protocol@@ -524,7 +528,7 @@ let pbytes = BL.unpack $ bsProps prot props in [0x20] <> encodeVarInt (2 + length pbytes)- <>[ boolBit (sp /= NewSession), toByte rc] <> pbytes+ <> [boolBit (sp /= NewSession), toByte rc] <> pbytes parseConnectACK :: A.Parser MQTTPkt parseConnectACK = do@@ -821,13 +825,14 @@ 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)+ unsubACK = oneOf [(0x00, UnsubSuccess),+ (0x11, UnsubNoSubscriptionExisted),+ (0x80, UnsubUnspecifiedError),+ (0x83, UnsubImplementationSpecificError),+ (0x87, UnsubNotAuthorized),+ (0x8F, UnsubTopicFilterInvalid),+ (0x91, UnsubPacketIdentifierInUse)+ ] data AuthRequest = AuthRequest Word8 [Property] deriving (Eq, Show)