diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -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. diff --git a/net-mqtt.cabal b/net-mqtt.cabal --- a/net-mqtt.cabal +++ b/net-mqtt.cabal @@ -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 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 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 @@ -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)