diff --git a/src/Simplex/Chat.hs b/src/Simplex/Chat.hs index 61a9b3f4e6..abc7f5dc18 100644 --- a/src/Simplex/Chat.hs +++ b/src/Simplex/Chat.hs @@ -164,7 +164,8 @@ newChatController mkChatController config randomPresetServers randomAgentServers ccVar smpAgent = do currentUser <- newTVarIO user currentRemoteHost <- newTVarIO Nothing - agentAsync <- newTVarIO Nothing + chatRunning <- newTVarIO False + subscribeAsync <- newTVarIO Nothing random <- liftIO C.newRandom eventSeq <- newTVarIO 0 inputQ <- newTBQueueIO tbqSize @@ -206,7 +207,8 @@ newChatController randomAgentServers, currentRemoteHost, smpAgent, - agentAsync, + chatRunning, + subscribeAsync, chatStore, chatStoreChanged, random, diff --git a/src/Simplex/Chat/Controller.hs b/src/Simplex/Chat/Controller.hs index 27b3fedae6..76736f7f44 100644 --- a/src/Simplex/Chat/Controller.hs +++ b/src/Simplex/Chat/Controller.hs @@ -231,7 +231,8 @@ data ChatController = ChatController currentRemoteHost :: TVar (Maybe RemoteHostId), firstTime :: Bool, smpAgent :: AgentClient, - agentAsync :: TVar (Maybe (Async (), Maybe (Async ()))), + chatRunning :: TVar Bool, + subscribeAsync :: TVar (Maybe (Async ())), chatStore :: DBStore, chatStoreChanged :: TVar Bool, -- if True, chat should be fully restarted random :: TVar ChaChaDRG, diff --git a/src/Simplex/Chat/Core.hs b/src/Simplex/Chat/Core.hs index bd6cac2110..3c98628b60 100644 --- a/src/Simplex/Chat/Core.hs +++ b/src/Simplex/Chat/Core.hs @@ -76,11 +76,10 @@ runSimplexChat :: ChatConfig -> ChatOpts -> User -> ChatController -> (User -> C runSimplexChat ChatConfig {testView} ChatOpts {coreOptions = CoreChatOpts {chatRelay, maintenance}} u cc@ChatController {config = ChatConfig {chatHooks}} chat | maintenance = wait =<< async (chat u cc) | otherwise = do - a1 <- runReaderT (startChatController True True) cc + runReaderT (startChatController True True) cc when (chatRelay && not testView) $ askCreateRelayAddress cc u forM_ (postStartHook chatHooks) ($ cc) - a2 <- async $ chat u cc - waitEither_ a1 a2 + chat u cc sendChatCmdStr :: ChatController -> String -> IO (Either ChatError ChatResponse) sendChatCmdStr cc s = runReaderT (execChatCommand Nothing (encodeUtf8 $ T.pack s) 0) cc diff --git a/src/Simplex/Chat/Library/Commands.hs b/src/Simplex/Chat/Library/Commands.hs index d8c1ae6723..529f1caf93 100644 --- a/src/Simplex/Chat/Library/Commands.hs +++ b/src/Simplex/Chat/Library/Commands.hs @@ -165,7 +165,7 @@ videoFilePrefix :: String videoFilePrefix = "video_" -- enableSndFiles has no effect when mainApp is True -startChatController :: Bool -> Bool -> CM' (Async ()) +startChatController :: Bool -> Bool -> CM' () startChatController mainApp enableSndFiles = do asks smpAgent >>= liftIO . resumeAgentClient unless mainApp $ chatWriteVar' subscriptionMode SMOnlyCreate @@ -174,8 +174,7 @@ startChatController mainApp enableSndFiles = do Left e -> liftIO $ putStrLn $ "Error synchronizing connections: " <> show e Right _ -> pure () restoreCalls - s <- asks agentAsync - readTVarIO s >>= maybe (start s users) (pure . fst) + unlessM (chatReadVar' chatRunning) $ start users where syncConnections' users = whenM (withFastStore' shouldSyncConnections) $ do @@ -184,13 +183,13 @@ startChatController mainApp enableSndFiles = do (userDiff, connDiff) <- withAgent (\a -> syncConnections a aUserIds connIds) withFastStore' setConnectionsSyncTs toView $ CEvtConnectionsDiff (AgentUserId <$> userDiff) (AgentConnId <$> connDiff) - start s users = do - a1 <- async $ forever (liftIO $ threadDelay maxBound) - a2 <- + start users = do + a <- if mainApp then Just <$> async (subscribeUsers False users) else pure Nothing - atomically . writeTVar s $ Just (a1, a2) + chatWriteVar' chatRunning True + chatWriteVar' subscribeAsync a if mainApp then do startXFTP xftpStartWorkers @@ -201,7 +200,6 @@ startChatController mainApp enableSndFiles = do void $ forkIO $ mapM_ startExpireCIs users startRelayChecks users else when enableSndFiles $ startXFTP xftpStartSndWorkers - pure a1 startXFTP startWorkers = do tmp <- readTVarIO =<< asks tempDirectory runExceptT (withAgent $ \a -> startWorkers a tmp) >>= \case @@ -279,17 +277,18 @@ restoreCalls = do atomically $ writeTVar calls callsMap stopChatController :: ChatController -> IO () -stopChatController ChatController {smpAgent, agentAsync = s, sndFiles, rcvFiles, expireCIFlags, remoteHostSessions, remoteCtrlSession} = do +stopChatController ChatController {smpAgent, chatRunning, subscribeAsync, sndFiles, rcvFiles, expireCIFlags, remoteHostSessions, remoteCtrlSession} = do readTVarIO remoteHostSessions >>= mapM_ (cancelRemoteHost False . snd) atomically (stateTVar remoteCtrlSession (,Nothing)) >>= mapM_ (cancelRemoteCtrl False . snd) disconnectAgentClient smpAgent - readTVarIO s >>= mapM_ (\(a1, a2) -> forkIO $ uninterruptibleCancel a1 >> mapM_ uninterruptibleCancel a2) + readTVarIO subscribeAsync >>= mapM_ (void . forkIO . uninterruptibleCancel) closeFiles sndFiles closeFiles rcvFiles atomically $ do keys <- M.keys <$> readTVar expireCIFlags forM_ keys $ \k -> TM.insert k False expireCIFlags - writeTVar s Nothing + writeTVar chatRunning False + writeTVar subscribeAsync Nothing where closeFiles :: TVar (Map Int64 Handle) -> IO () closeFiles files = do @@ -481,10 +480,9 @@ processChatCommand vr nm = \case withChatLock "deleteUser" $ deleteChatUser user' delSMPQueues DeleteUser uName delSMPQueues viewPwd_ -> withUserName uName $ \userId -> APIDeleteUser userId delSMPQueues viewPwd_ StartChat {mainApp, enableSndFiles} -> withUser' $ \_ -> - asks agentAsync >>= readTVarIO >>= \case - Just _ -> pure CRChatRunning - _ -> checkStoreNotChanged . lift $ startChatController mainApp enableSndFiles $> CRChatStarted - CheckChatRunning -> maybe CRChatStopped (const CRChatRunning) <$> chatReadVar agentAsync + ifM (chatReadVar chatRunning) (pure CRChatRunning) $ + checkStoreNotChanged . lift $ startChatController mainApp enableSndFiles $> CRChatStarted + CheckChatRunning -> ifM (chatReadVar chatRunning) (pure CRChatRunning) (pure CRChatStopped) APIStopChat -> do ask >>= liftIO . stopChatController pure CRChatStopped @@ -3465,7 +3463,7 @@ processChatCommand vr nm = \case CTGroup -> withFastStore' $ \db -> getMessageMentions db user chatId msg _ -> pure [] checkChatStopped :: CM ChatResponse -> CM ChatResponse - checkChatStopped a = asks agentAsync >>= readTVarIO >>= maybe a (const $ throwChatError CEChatNotStopped) + checkChatStopped a = ifM (chatReadVar chatRunning) (throwChatError CEChatNotStopped) a setStoreChanged :: CM () setStoreChanged = asks chatStoreChanged >>= atomically . (`writeTVar` True) #if !defined(dbPostgres) diff --git a/src/Simplex/Chat/Library/Internal.hs b/src/Simplex/Chat/Library/Internal.hs index eb0fd564e3..8e0f184258 100644 --- a/src/Simplex/Chat/Library/Internal.hs +++ b/src/Simplex/Chat/Library/Internal.hs @@ -2804,16 +2804,16 @@ checkSameUser :: UserId -> User -> CM () checkSameUser userId User {userId = activeUserId} = when (userId /= activeUserId) $ throwChatError (CEDifferentActiveUser userId activeUserId) chatStarted :: CM' Bool -chatStarted = fmap isJust . readTVarIO =<< asks agentAsync +chatStarted = readTVarIO =<< asks chatRunning waitChatStartedAndActivated :: CM' () waitChatStartedAndActivated = do - agentStarted <- asks agentAsync + chatRunning' <- asks chatRunning chatActivated <- asks chatActivated atomically $ do - started <- readTVar agentStarted + started <- readTVar chatRunning' activated <- readTVar chatActivated - unless (isJust started && activated) retry + unless (started && activated) retry chatVersionRange :: CM VersionRangeChat chatVersionRange = lift chatVersionRange' diff --git a/tests/ChatClient.hs b/tests/ChatClient.hs index ede3c1f2a2..1741d97abf 100644 --- a/tests/ChatClient.hs +++ b/tests/ChatClient.hs @@ -62,6 +62,7 @@ import Simplex.Messaging.Server.Env.STM (ServerConfig (..), ServerStoreCfg (..), import Simplex.Messaging.Server.MsgStore.STM (STMMsgStore) import Simplex.Messaging.Transport import Simplex.Messaging.Transport.Server (ServerCredentials (..), mkTransportServerConfig) +import Simplex.Messaging.Util (unlessM) import Simplex.Messaging.Version import Simplex.Messaging.Version.Internal import System.Directory (createDirectoryIfMissing, removeDirectoryRecursive) @@ -318,7 +319,7 @@ startTestChat_ TestParams {printOutput} db cfg opts@ChatOpts {coreOptions = Core Right cc <- newChatController db (Just user) cfg opts False void $ execChatCommand' (SetTempFolder "tests/tmp/tmp") 0 `runReaderT` cc chatAsync <- async $ runSimplexChat cfg opts user cc $ \_u cc' -> runChatTerminal ct cc' opts - unless maintenance $ atomically $ readTVar (agentAsync cc) >>= \a -> when (isNothing a) retry + unless maintenance $ atomically $ unlessM (readTVar $ chatRunning cc) retry termQ <- newTQueueIO termAsync <- async $ readTerminalOutput t termQ pure TestCC {chatController = cc, virtualTerminal = t, chatAsync, termAsync, termQ, printOutput}