net-mqtt 0.8.1.0 → 0.8.2.0

raw patch · 3 files changed

+83/−17 lines, 3 filesdep ~attoparsecPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: attoparsec

API changes (from Hackage documentation)

Files

• Changelog.md +11/−0
• net-mqtt.cabal +6/−6
• src/Network/MQTT/Client.hs +66/−11

@@ -1,5 +1,16 @@ # Changelog for net-mqtt +## 0.8.2.0++Added `OrderedCallback` and `OrderedLowLevelCallback` to guarantee+processing the messages in order.++`SimpleCallback` and `LowLevelCallback` both run callbacks in isolated+threads such that when a message takes longer to process than another,+a second message might finish before the first.  Most of the time,+this probably doesn't matter this model is simpler and keeps latency+lower.  Now applications can decide what's appropriate for them.+ ## 0.8.1.0  Added `toFilter` to convert `Topic`s to `Filter`s.

@@ -1,11 +1,11 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.34.4.+-- This file has been generated from package.yaml by hpack version 0.34.6. -- -- see: https://github.com/sol/hpack  name:           net-mqtt-version:        0.8.1.0+version:        0.8.2.0 synopsis:       An MQTT Protocol Implementation. description:    Please see the README on GitHub at <https://github.com/dustin/mqtt-hs#readme> category:       Network@@ -39,7 +39,7 @@   build-depends:       QuickCheck >=2.12.6.1 && <2.15     , async >=2.2.1 && <2.3-    , attoparsec >=0.13.2 && <0.15+    , attoparsec >=0.13.2 && <0.14     , attoparsec-binary >=0.2 && <1.0     , base >=4.7 && <5     , binary >=0.8.5 && <0.9@@ -66,7 +66,7 @@   build-depends:       QuickCheck >=2.12.6.1 && <2.15     , async >=2.2.1 && <2.3-    , attoparsec >=0.13.2 && <0.15+    , attoparsec >=0.13.2 && <0.14     , attoparsec-binary >=0.2 && <1.0     , base >=4.7 && <5     , binary >=0.8.5 && <0.9@@ -94,7 +94,7 @@   build-depends:       QuickCheck >=2.12.6.1 && <2.15     , async >=2.2.1 && <2.3-    , attoparsec >=0.13.2 && <0.15+    , attoparsec >=0.13.2 && <0.14     , attoparsec-binary >=0.2 && <1.0     , base >=4.7 && <5     , binary >=0.8.5 && <0.9@@ -125,7 +125,7 @@       HUnit     , QuickCheck >=2.12.6.1 && <2.15     , async >=2.2.1 && <2.3-    , attoparsec >=0.13.2 && <0.15+    , attoparsec >=0.13.2 && <0.14     , attoparsec-binary >=0.2 && <1.0     , base >=4.7 && <5     , binary >=0.8.5 && <0.9

@@ -36,13 +36,15 @@   ) where  import           Control.Concurrent         (myThreadId, threadDelay)-import           Control.Concurrent.Async   (Async, async, asyncThreadId, cancelWith, link, race_, wait, waitAnyCancel)+import           Control.Concurrent.Async   (Async, async, asyncThreadId, cancel, cancelWith, link, race_, wait,+                                             waitAnyCancel)+import           Control.Concurrent.MVar    (MVar, newEmptyMVar, putMVar, takeMVar) import           Control.Concurrent.STM     (STM, TChan, TVar, atomically, check, modifyTVar', newTChan, newTChanIO,                                              newTVarIO, orElse, readTChan, readTVar, readTVarIO, registerDelay, retry,                                              writeTChan, writeTVar) import           Control.DeepSeq            (force) import qualified Control.Exception          as E-import           Control.Monad              (forever, guard, unless, void, when)+import           Control.Monad              (forever, guard, join, unless, void, when) import           Control.Monad.IO.Class     (liftIO) import           Data.Bifunctor             (first) import qualified Data.ByteString.Char8      as BCS@@ -53,6 +55,7 @@ import qualified Data.Conduit.Combinators   as C import           Data.Conduit.Network       (AppData, appSink, appSource, clientSettings, runTCPClient) import           Data.Conduit.Network.TLS   (runTLSClient, tlsClientConfig, tlsClientTLSSettings)+import           Data.Foldable              (traverse_) import           Data.Map.Strict            (Map) import qualified Data.Map.Strict            as Map import           Data.Maybe                 (fromJust, fromMaybe)@@ -84,8 +87,22 @@  -- | Callback invoked on each incoming subscribed message. data MessageCallback = NoCallback+  -- | Callbacks will be invoked asynchronously, ordering is likely preserved, but not guaranteed.+  -- In high throughput scenarios, slow callbacks may result in a high number of Haskell threads,+  -- potentially bringing down the entire application when running out of memory.+  -- Typically faster than `OrderedCallback`.   | SimpleCallback (MQTTClient -> Topic -> BL.ByteString -> [Property] -> IO ())+  -- | Callbacks are guaranteed to be invoked in the same order messages are received.+  -- In high throughput scenarios, slow callbacks may cause the underlying TCP connection to block,+  -- potentially being terminated by the broker.+  -- Typically slower than `SimpleCallback`.+  | OrderedCallback (MQTTClient -> Topic -> BL.ByteString -> [Property] -> IO ())+  -- | A LowLevelCallback receives the client and the entire publish request, providing+  -- access to all of the fields of the request.  This is slightly harder to use than+  -- SimpleCallback for common cases, but there are cases where you need all the things.   | LowLevelCallback (MQTTClient -> PublishRequest -> IO ())+  -- | A low level callback that is ordered.+  | OrderedLowLevelCallback (MQTTClient -> PublishRequest -> IO ())  -- | The MQTT client. --@@ -102,6 +119,8 @@   , _inA          :: TVar (Map Word16 Topic)   , _connACKFlags :: TVar ConnACKFlags   , _corr         :: TVar (Map BL.ByteString MessageCallback)+  , _cbM          :: MVar (IO ())+  , _cbHandle     :: TVar (Maybe (Async ()))   }  -- | Configuration for setting up an MQTT client.@@ -278,6 +297,8 @@   _inA <- newTVarIO mempty   _connACKFlags <- newTVarIO (ConnACKFlags NewSession ConnUnspecifiedError mempty)   _corr <- newTVarIO mempty+  _cbM <- newEmptyMVar+  _cbHandle <- newTVarIO Nothing   let _cb = _msgCB       cli = MQTTClient{..} @@ -299,7 +320,7 @@           guard $ st == Starting || st == Connected           writeTVar (_st cli) Disconnected -    start c@MQTTClient{..} (_,sink) = do+    start c (_,sink) = do       void . runConduit $ do         let req = connectRequest{T._connID=BC.pack _connID,                                  T._lastWill=_lwt,@@ -347,7 +368,7 @@ -- | Wait for a client to terminate its connection. -- An exception is thrown if the client didn't terminate expectedly. waitForClient :: MQTTClient -> IO ()-waitForClient c@MQTTClient{..} = do+waitForClient c@MQTTClient{..} = flip E.finally (stopCallbackThread c) $ do   void . traverse wait =<< readTVarIO _ct   e <- atomically $ stateX c Stopped   case e of@@ -425,13 +446,16 @@           atomically $ modifyTVar' _inflight (Map.delete _pubPktID)           corrs <- readTVarIO _corr           E.evaluate . force =<< case maybe _cb (\cd -> Map.findWithDefault _cb cd corrs) cdata of-                                   NoCallback         -> pure ()-                                   SimpleCallback f   -> call (f c (blToTopic _pubTopic) _pubBody _pubProps)-                                   LowLevelCallback f -> call (f c p)+                                   NoCallback                -> pure ()+                                   SimpleCallback f          -> call (f c (blToTopic _pubTopic) _pubBody _pubProps)+                                   OrderedCallback f         -> callOrd (f c (blToTopic _pubTopic) _pubBody _pubProps)+                                   LowLevelCallback f        -> call (f c p)+                                   OrderedLowLevelCallback f -> callOrd (f c p)              where               call a = link =<< namedAsync "notifier" (a >> respond)-              respond = void $ traverse (sendPacketIO c) rpkt+              callOrd a = putMVar _cbM $ a >> respond+              respond = traverse_ (sendPacketIO c) rpkt               cdata = foldr f Nothing _pubProps                 where f (PropCorrelationData x) _ = Just x                       f _ o                       = o@@ -467,6 +491,37 @@           atomically $ writeTVar _st (DiscoErr req)           maybeCancelWith (Discod req) =<< readTVarIO _ct +-- Run OrderedCallbacks in a background thread. Does nothing for other callback types.+-- We keep the async handle in a TVar and make sure only one of these threads is running.+runCallbackThread :: MQTTClient -> IO ()+runCallbackThread MQTTClient{_cb, _cbM, _cbHandle}+  | isOrdered _cb = do+      -- We always spawn a thread, but may kill it if we already have+      -- one.  The new thread won't start until we atomically confirm+      -- it's the only one.+      latch <- newTVarIO False+      handle <- namedAsync "ordered callbacks" (waitFor latch *> runOrderedCallbacks)+      join . atomically $ do+        cbThread <- readTVar _cbHandle+        case cbThread of+          Nothing -> do+            -- This is the first thread.  Flip the latch and put it in place.+            writeTVar latch True+            writeTVar _cbHandle (Just handle)+            pure (pure ())+          Just _ -> pure (cancel handle) -- otherwise, cancel the temporary thread.+  | otherwise = pure ()+    where isOrdered (OrderedCallback _)         = True+          isOrdered (OrderedLowLevelCallback _) = True+          isOrdered _                           = False+          waitFor latch = atomically (check =<< readTVar latch)+          -- Keep running callbacks from the MVar+          runOrderedCallbacks = forever . join . takeMVar $ _cbM++-- Stop the background thread for OrderedCallbacks. Does nothing for other callback types.+stopCallbackThread :: MQTTClient -> IO ()+stopCallbackThread MQTTClient{_cbHandle} = maybe (pure ()) cancel =<< readTVarIO _cbHandle+ maybeCancelWith :: E.Exception e => e -> Maybe (Async ()) -> IO () maybeCancelWith e = void . traverse (`cancelWith` e) @@ -538,11 +593,11 @@ -- | Subscribe to a list of topic filters with their respective 'QoS'es. -- The accepted 'QoS'es are returned in the same order as requested. subscribe :: MQTTClient -> [(Filter, SubOptions)] -> [Property] -> IO ([Either SubErr QoS], [Property])-subscribe c@MQTTClient{..} ls props = do+subscribe c ls props = do+  runCallbackThread c   r <- sendAndWait c DSubACK (\pid -> SubscribePkt $ SubscribeRequest pid ls' props)   let (SubACKPkt (SubscribeResponse _ rs aprops)) = r   pure (rs, aprops)-     where ls' = map (first filterToBL) ls  -- | Unsubscribe from a list of topic filters.@@ -553,7 +608,7 @@ -- 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+unsubscribe c ls props = do   (UnsubACKPkt (UnsubscribeResponse _ rsn rprop)) <- sendAndWait c DUnsubACK (\pid -> UnsubscribePkt $ UnsubscribeRequest pid (map filterToBL ls) props)   pure (rprop, rsn)