diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index b8adb8315..c8deb186c 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -271,9 +271,7 @@ getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp, netC currentTs <- liftIO getCurrentTime notices <- liftIO $ withTransaction store (`getClientNotices` presetServers) `catchAll_` pure [] env <- ask - let processMsg c t = - agentOperationBracket c AORcvNetwork waitUntilActive (processSMPTransmissions c t) `runReaderT` env - `catchOwn` \e -> atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ CRITICAL True $ "subscriber error: " <> show e) + let processMsg c t = subscriber c t `runReaderT` env c@AgentClient {acThread} <- liftIO $ newAgentClient clientId initServers currentTs notices processMsg env t <- runAgentThreads c `forkFinally` const (liftIO $ disconnectAgentClient c) atomically . writeTVar acThread . Just =<< mkWeakThreadId t @@ -2985,6 +2983,14 @@ getNextSMPServer :: AgentClient -> UserId -> [SMPServer] -> AM SMPServerWithAuth getNextSMPServer c userId = getNextServer c userId storageSrvs {-# INLINE getNextSMPServer #-} +subscriber :: AgentClient -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> AM' () +subscriber c@AgentClient {subQ} t = run $ + agentOperationBracket c AORcvNetwork waitUntilActive $ + processSMPTransmissions c t + where + run a = a `catchOwn` \e -> notify $ CRITICAL True $ "subscriber error: " <> show e + notify err = atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR err) + cleanupManager :: AgentClient -> AM' () cleanupManager c@AgentClient {subQ} = do diff --git a/src/Simplex/Messaging/Notifications/Server/Env.hs b/src/Simplex/Messaging/Notifications/Server/Env.hs index 7ece78609..67659c788 100644 --- a/src/Simplex/Messaging/Notifications/Server/Env.hs +++ b/src/Simplex/Messaging/Notifications/Server/Env.hs @@ -125,10 +125,10 @@ newNtfServerEnv config@NtfServerConfig {pushQSize, smpAgentCfg, apnsConfig, dbSt store <- newNtfDbStore dbStoreConfig tlsServerCreds <- loadServerCredential ntfCredentials XV.Fingerprint fp <- loadFingerprint ntfCredentials + let dbService = if useServiceCreds then Just $ mkDbService random store else Nothing pushServer <- newNtfPushServer pushQSize apnsConfig serverStats <- newNtfServerStats =<< getCurrentTime - let dbService = if useServiceCreds then Just $ mkDbService random store else Nothing - processMsg = mkProcessMsg store pushServer serverStats + let processMsg = mkProcessMsg store pushServer serverStats subscriber <- newNtfSubscriber smpAgentCfg processMsg dbService random pure NtfEnv {config, subscriber, pushServer, store, random, tlsServerCreds, serverIdentity = C.KeyHash fp, serverStats} where