diff --git a/cabal.project b/cabal.project index 27f82d85bd..6614d6df52 100644 --- a/cabal.project +++ b/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 diff --git a/src/Simplex/Chat.hs b/src/Simplex/Chat.hs index ec17614db3..6aeaff2b66 100644 --- a/src/Simplex/Chat.hs +++ b/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 diff --git a/src/Simplex/Chat/Library/Commands.hs b/src/Simplex/Chat/Library/Commands.hs index 4342a8821e..d8c1ae6723 100644 --- a/src/Simplex/Chat/Library/Commands.hs +++ b/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 ())