mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-12 02:22:45 +00:00
agent: background mode for agent NSE (#924)
This commit is contained in:
@@ -174,14 +174,17 @@ import UnliftIO.STM
|
||||
-- import GHC.Conc (unsafeIOToSTM)
|
||||
|
||||
-- | Creates an SMP agent client instance
|
||||
getSMPAgentClient :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> InitialAgentServers -> SQLiteStore -> m AgentClient
|
||||
getSMPAgentClient cfg initServers store =
|
||||
getSMPAgentClient :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient
|
||||
getSMPAgentClient cfg initServers store backgroundMode =
|
||||
liftIO (newSMPAgentEnv cfg store) >>= runReaderT runAgent
|
||||
where
|
||||
runAgent = do
|
||||
c <- getAgentClient initServers
|
||||
void $ raceAny_ [subscriber c, runNtfSupervisor c, cleanupManager c] `forkFinally` const (disconnectAgentClient c)
|
||||
void $ runAgentThreads c `forkFinally` const (disconnectAgentClient c)
|
||||
pure c
|
||||
runAgentThreads c
|
||||
| backgroundMode = subscriber c
|
||||
| otherwise = raceAny_ [subscriber c, runNtfSupervisor c, cleanupManager c]
|
||||
|
||||
disconnectAgentClient :: MonadUnliftIO m => AgentClient -> m ()
|
||||
disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns, xftpAgent = xa}} = do
|
||||
@@ -428,8 +431,9 @@ rcDiscoverCtrl' pairings = do
|
||||
liftError RCP $ discoverRCCtrl subs pairings
|
||||
|
||||
-- | Activate operations
|
||||
foregroundAgent :: MonadUnliftIO m => AgentClient -> m ()
|
||||
foregroundAgent c = withAgentEnv c $ foregroundAgent' c
|
||||
-- stopPending = True is used in iOS NSE
|
||||
foregroundAgent :: MonadUnliftIO m => AgentClient -> Bool -> m ()
|
||||
foregroundAgent c stopPending = withAgentEnv c $ foregroundAgent' c stopPending
|
||||
|
||||
-- | Suspend operations with max delay to deliver pending messages
|
||||
suspendAgent :: MonadUnliftIO m => AgentClient -> Int -> m ()
|
||||
@@ -1800,8 +1804,11 @@ sendNtfConnCommands c cmd = do
|
||||
setNtfServers' :: AgentMonad' m => AgentClient -> [NtfServer] -> m ()
|
||||
setNtfServers' c = atomically . writeTVar (ntfServers c)
|
||||
|
||||
foregroundAgent' :: AgentMonad' m => AgentClient -> m ()
|
||||
foregroundAgent' c = do
|
||||
-- stopPending = True is used in iOS NSE
|
||||
foregroundAgent' :: AgentMonad' m => AgentClient -> Bool -> m ()
|
||||
foregroundAgent' c stopPending = do
|
||||
-- to prevent race conditions with message deliveries and async commands in the app
|
||||
when stopPending $ stopPendingActions c
|
||||
atomically $ writeTVar (agentState c) ASForeground
|
||||
mapM_ activate $ reverse agentOperations
|
||||
where
|
||||
|
||||
@@ -29,6 +29,7 @@ module Simplex.Messaging.Agent.Client
|
||||
closeAgentClient,
|
||||
closeProtocolServerClients,
|
||||
closeXFTPServerClient,
|
||||
stopPendingActions,
|
||||
runSMPServerTest,
|
||||
runXFTPServerTest,
|
||||
getXFTPWorkPath,
|
||||
@@ -604,16 +605,25 @@ closeAgentClient c = liftIO $ do
|
||||
closeProtocolServerClients c xftpClients
|
||||
cancelActions . actions $ reconnections c
|
||||
cancelActions . actions $ asyncClients c
|
||||
cancelActions $ smpQueueMsgDeliveries c
|
||||
cancelActions $ asyncCmdProcesses c
|
||||
stopPendingActions c
|
||||
atomically . RQ.clear $ activeSubs c
|
||||
atomically . RQ.clear $ pendingSubs c
|
||||
clear subscrConns
|
||||
clear getMsgLocks
|
||||
where
|
||||
clear :: Monoid m => (AgentClient -> TVar m) -> IO ()
|
||||
clear sel = atomically $ writeTVar (sel c) mempty
|
||||
|
||||
stopPendingActions :: MonadIO m => AgentClient -> m ()
|
||||
stopPendingActions c = liftIO $ do
|
||||
-- messages
|
||||
cancelActions $ smpQueueMsgDeliveries c
|
||||
clear pendingMsgsQueued
|
||||
clear smpQueueMsgQueues
|
||||
-- commands
|
||||
cancelActions $ asyncCmdProcesses c
|
||||
clear connCmdsQueued
|
||||
clear asyncCmdQueues
|
||||
clear getMsgLocks
|
||||
where
|
||||
clear :: Monoid m => (AgentClient -> TVar m) -> IO ()
|
||||
clear sel = atomically $ writeTVar (sel c) mempty
|
||||
|
||||
@@ -956,7 +956,7 @@ testRatchetSyncSuspendForeground t = do
|
||||
|
||||
suspendAgent bob2 0
|
||||
threadDelay 100000
|
||||
foregroundAgent bob2
|
||||
foregroundAgent bob2 False
|
||||
|
||||
withSmpServerStoreMsgLogOn t testPort $ \_ -> do
|
||||
runRight_ $ do
|
||||
@@ -1119,7 +1119,7 @@ testSuspendingAgent =
|
||||
5 <- sendMessage a bId SMP.noMsgFlags "hello 2"
|
||||
get a ##> ("", bId, SENT 5)
|
||||
Nothing <- 100000 `timeout` get b
|
||||
foregroundAgent b
|
||||
foregroundAgent b False
|
||||
get b =##> \case ("", c, Msg "hello 2") -> c == aId; _ -> False
|
||||
|
||||
testSuspendingAgentCompleteSending :: ATransport -> IO ()
|
||||
@@ -2065,7 +2065,7 @@ testTwoUsers = withAgentClients2 $ \a b -> do
|
||||
getSMPAgentClient' :: AgentConfig -> InitialAgentServers -> FilePath -> IO AgentClient
|
||||
getSMPAgentClient' cfg' initServers dbPath = do
|
||||
Right st <- liftIO $ createAgentStore dbPath "" False MCError
|
||||
getSMPAgentClient cfg' initServers st
|
||||
getSMPAgentClient cfg' initServers st False
|
||||
|
||||
testServerMultipleIdentities :: HasCallStack => IO ()
|
||||
testServerMultipleIdentities =
|
||||
|
||||
Reference in New Issue
Block a user