concurrent processing

cabal.project

@@ -21,7 +21,7 @@ constraints: zip +disable-bzip2 +disable-zstd
 source-repository-package
     type: git
     location: https://github.com/simplex-chat/simplexmq.git
-    tag: 61ee188ee0839c34de16bc17934f04ebc7fd4873
+    tag: 906da42de095b59965859d9397d2fd01f6c128c1
 
 source-repository-package
     type: git

src/Simplex/Chat.hs

@@ -20,6 +20,7 @@ import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
+import Control.Monad.Reader (runReaderT)
 import Data.Bifunctor (bimap, second)
 import Data.List (partition, sortOn)
 import Data.List.NonEmpty (NonEmpty (..))
@@ -153,10 +154,14 @@ newChatController
     agentXFTP <- randomServerCfgs "agent XFTP servers" SPXFTP opDomains rndSrvs
     let randomAgentServers = RandomAgentServers {smpServers = agentSMP, xftpServers = agentXFTP}
     servers <- withTransaction chatStore $ \db -> agentServers db config randomPresetServers randomAgentServers
-    runExceptT (getSMPAgentClient aCfg {tbqSize} servers agentStore backgroundMode)
-      >>= mapM (mkChatController config randomPresetServers randomAgentServers)
+    ccVar <- newEmptyTMVarIO
+    let processEvent t = do
+          cc <- atomically $ readTMVar ccVar
+          runReaderT (processAgentEvent t) cc
+    runExceptT (getSMPAgentClient aCfg {tbqSize} servers agentStore processEvent)
+      >>= mapM (mkChatController config randomPresetServers randomAgentServers ccVar)
     where
-      mkChatController config randomPresetServers randomAgentServers smpAgent = do
+      mkChatController config randomPresetServers randomAgentServers ccVar smpAgent = do
         currentUser <- newTVarIO user
         currentRemoteHost <- newTVarIO Nothing
         agentAsync <- newTVarIO Nothing
@@ -193,52 +198,54 @@ newChatController
         tempDirectory <- newTVarIO optTempDirectory
         assetsDirectory <- newTVarIO Nothing
         contactMergeEnabled <- newTVarIO True
-        pure
-          ChatController
-            { firstTime = dbNew chatStore,
-              currentUser,
-              randomPresetServers,
-              randomAgentServers,
-              currentRemoteHost,
-              smpAgent,
-              agentAsync,
-              chatStore,
-              chatStoreChanged,
-              random,
-              eventSeq,
-              inputQ,
-              outputQ,
-              subscriptionMode,
-              chatLock,
-              entityLocks,
-              sndFiles,
-              rcvFiles,
-              currentCalls,
-              localDeviceName,
-              multicastSubscribers,
-              remoteSessionSeq,
-              remoteHostSessions,
-              remoteHostsFolder,
-              remoteCtrlSession,
-              config,
-              filesFolder,
-              deliveryTaskWorkers,
-              deliveryJobWorkers,
-              relayRequestWorkers,
-              relayGroupLinkChecksAsync,
-              chatRelayTests,
-              expireCIThreads,
-              expireCIFlags,
-              cleanupManagerAsync,
-              timedItemThreads,
-              chatActivated,
-              showLiveItems,
-              encryptLocalFiles,
-              tempDirectory,
-              assetsDirectory,
-              logFilePath = logFile,
-              contactMergeEnabled
-            }
+        let cc = ChatController
+              { firstTime = dbNew chatStore,
+                currentUser,
+                randomPresetServers,
+                randomAgentServers,
+                currentRemoteHost,
+                smpAgent,
+                agentAsync,
+                chatStore,
+                chatStoreChanged,
+                random,
+                eventSeq,
+                inputQ,
+                outputQ,
+                subscriptionMode,
+                chatLock,
+                entityLocks,
+                sndFiles,
+                rcvFiles,
+                currentCalls,
+                localDeviceName,
+                multicastSubscribers,
+                remoteSessionSeq,
+                remoteHostSessions,
+                remoteHostsFolder,
+                remoteCtrlSession,
+                config,
+                filesFolder,
+                deliveryTaskWorkers,
+                deliveryJobWorkers,
+                relayRequestWorkers,
+                relayGroupLinkChecksAsync,
+                chatRelayTests,
+                expireCIThreads,
+                expireCIFlags,
+                cleanupManagerAsync,
+                timedItemThreads,
+                chatActivated,
+                showLiveItems,
+                encryptLocalFiles,
+                tempDirectory,
+                assetsDirectory,
+                logFilePath = logFile,
+                contactMergeEnabled
+              }
+        atomically $ putTMVar ccVar cc
+        startSMPAgentClient smpAgent backgroundMode
+        pure cc
       presetServers' :: PresetServers
       presetServers' = presetServers {operators = operators', netCfg = netCfg'}
         where

src/Simplex/Chat/Library/Commands.hs

@@ -185,7 +185,7 @@ startChatController mainApp enableSndFiles = do
         withFastStore' setConnectionsSyncTs
         toView $ CEvtConnectionsDiff (AgentUserId <$> userDiff) (AgentConnId <$> connDiff)
     start s users = do
-      a1 <- async agentSubscriber
+      a1 <- async $ forever (liftIO $ threadDelay maxBound)
       a2 <-
         if mainApp
           then Just <$> async (subscribeUsers False users)
@@ -4709,22 +4709,15 @@ setAllExpireCIFlags b = do
     keys <- M.keys <$> readTVar expireFlags
     forM_ keys $ \k -> TM.insert k b expireFlags
 
-agentSubscriber :: CM' ()
-agentSubscriber = do
-  q <- asks $ subQ . smpAgent
-  forever (atomically (readTBQueue q) >>= process)
-    `catchOwn` \e -> do
-      eToView' $ chatErrorAgent $ CRITICAL True $ "Message reception stopped: " <> show e
-      E.throwIO e
+processAgentEvent :: ATransmission -> CM' ()
+processAgentEvent (corrId, entId, AEvt e msg) = run $ case e of
+  SAENone -> processAgentMessageNoConn msg
+  SAEConn -> processAgentMessage corrId entId msg
+  SAERcvFile -> processAgentMsgRcvFile corrId entId msg
+  SAESndFile -> processAgentMsgSndFile corrId entId msg
   where
-    process :: (ACorrId, AEntityId, AEvt) -> CM' ()
-    process (corrId, entId, AEvt e msg) = run $ case e of
-      SAENone -> processAgentMessageNoConn msg
-      SAEConn -> processAgentMessage corrId entId msg
-      SAERcvFile -> processAgentMsgRcvFile corrId entId msg
-      SAESndFile -> processAgentMsgSndFile corrId entId msg
-      where
-        run action = action `catchAllOwnErrors'` eToView'
+    run action = action `catchAllOwnErrors'` eToView'
+
 
 type AgentSubResult = Map ConnId (Either AgentErrorType ())