dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-08-28 13:44:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

separate starting agent

Browse Source
This commit is contained in:
Evgeny @ SimpleX Chat
2026-05-31 15:11:13 +00:00
parent b9b1de7ef6
commit 906da42de0
2 changed files with 23 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Simplex/Messaging/Agent.hs
+21 -17
View File
@@ -40,6 +40,7 @@ module Simplex.Messaging.Agent
vrValue,
getSMPAgentClient,
getSMPAgentClient_,
startSMPAgentClient,
disconnectAgentClient,
disposeAgentClient,
resumeAgentClient,
@@ -256,42 +257,45 @@ import UnliftIO.STM
type AE a = ExceptT AgentErrorType IO a
-- | Creates an SMP agent client instance
getSMPAgentClient :: AgentConfig -> InitialAgentServers -> DBStore -> Bool -> (ATransmission -> IO ()) -> AE AgentClient
getSMPAgentClient :: AgentConfig -> InitialAgentServers -> DBStore -> (ATransmission -> IO ()) -> AE AgentClient
getSMPAgentClient = getSMPAgentClient_ 1
{-# INLINE getSMPAgentClient #-}
getSMPAgentClient_ :: Int -> AgentConfig -> InitialAgentServers -> DBStore -> Bool -> (ATransmission -> IO ()) -> AE AgentClient
getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp, netCfg, useServices, presetServers} store backgroundMode processEvent = do
getSMPAgentClient_ :: Int -> AgentConfig -> InitialAgentServers -> DBStore -> (ATransmission -> IO ()) -> AE AgentClient
getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp, netCfg, useServices, presetServers} store processEvent = do
-- This error should be prevented in the app
when (any id useServices && sessionMode netCfg == TSMEntity) $ throwE $ CMD PROHIBITED "newAgentClient"
liftIO $ newSMPAgentEnv cfg store >>= runReaderT runAgent
liftIO $ newSMPAgentEnv cfg store >>= runReaderT createAgent
where
runAgent = do
createAgent = do
liftIO $ checkServers "SMP" smp >> checkServers "XFTP" xftp
currentTs <- liftIO getCurrentTime
notices <- liftIO $ withTransaction store (`getClientNotices` presetServers) `catchAll_` pure []
env <- ask
let processMsg c t = subscriber c t `runReaderT` env
c@AgentClient {acThread, generalQ} <- liftIO $ newAgentClient clientId initServers currentTs notices processEvent processMsg env
void $ liftIO $ forkIO $ connWorkerLoop c generalQ
unless backgroundMode $ do
t <- runAgentThreads c `forkFinally` const (liftIO $ disconnectAgentClient c)
atomically . writeTVar acThread . Just =<< mkWeakThreadId t
pure c
liftIO $ newAgentClient clientId initServers currentTs notices processEvent processMsg env
checkServers protocol srvs =
forM_ (M.assocs srvs) $ \(userId, srvs') -> checkUserServers ("getSMPAgentClient " <> protocol <> " " <> tshow userId) srvs'
runAgentThreads c = do
startSMPAgentClient :: AgentClient -> Bool -> IO ()
startSMPAgentClient c@AgentClient {acThread, generalQ, agentEnv} backgroundMode = do
void $ forkIO $ connWorkerLoop c generalQ
unless backgroundMode $ do
t <- runAgentThreads `forkFinally` const (disconnectAgentClient c)
atomically . writeTVar acThread . Just =<< mkWeakThreadId t
where
runAgentThreads = flip runReaderT agentEnv $ do
restoreServersStats c
raceAny_
[ run c "runNtfSupervisor" $ runNtfSupervisor c,
run c "cleanupManager" $ cleanupManager c,
run c "logServersStats" $ logServersStats c
[ run "runNtfSupervisor" $ runNtfSupervisor c,
run "cleanupManager" $ cleanupManager c,
run "logServersStats" $ logServersStats c
]
`E.finally` saveServersStats c
run c'@AgentClient {acThread} name a =
run name a =
a `E.catchAny` \e -> whenM (isJust <$> readTVarIO acThread) $ do
logError $ "Agent thread " <> name <> " crashed: " <> tshow e
liftIO $ notifyEvent c' ("", "", AEvt SAEConn $ ERR $ CRITICAL True $ show e)
liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR $ CRITICAL True $ show e)
logServersStats :: AgentClient -> AM' ()
logServersStats c = do
tests/AgentTests/FunctionalAPITests.hs
+2 -1
View File
@@ -4333,8 +4333,9 @@ getSMPAgentClient' :: Int -> AgentConfig -> InitialAgentServers -> String -> IO
getSMPAgentClient' clientId cfg' initServers dbPath = do
Right st <- liftIO $ createStore dbPath
subQ <- newTBQueueIO 1024
Right client <- runExceptT $ getSMPAgentClient_ clientId cfg' initServers st False (atomically . writeTBQueue subQ)
Right client <- runExceptT $ getSMPAgentClient_ clientId cfg' initServers st (atomically . writeTBQueue subQ)
when (dbNew st) $ insertUser st
startSMPAgentClient client False
pure AgentClient {client, subQ}
#if defined(dbPostgres)