diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,9 @@ # Changelog for net-mqtt +## 0.8.5.0 + +Switch to crypton-connection + ## 0.8.3.0 Ping time is configurable. 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 @@ -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 "" else "") - 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 "" else "") + 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 diff --git a/net-mqtt.cabal b/net-mqtt.cabal --- a/net-mqtt.cabal +++ b/net-mqtt.cabal @@ -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 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 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 @@ -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 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 @@ -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)