dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-08-02 11:38:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

core: fix leak in serverDown handler (#1061)

Browse Source
This commit is contained in:
Alexander Bondarenko
2024-03-21 12:03:19 +00:00
committed by GitHub
parent 9eacae5492
commit a48c22ea36
1 changed files with 11 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Simplex/Messaging/Agent/Client.hs
+11 -11
View File
@@ -521,11 +521,11 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
connectClient v = do
cfg <- getClientConfig c smpCfg
g <- asks random
u <- askUnliftIO
liftEitherError (protocolClientError SMP $ B.unpack $ strEncode srv) (getProtocolClient g tSess cfg (Just msgQ) $ clientDisconnected u v)
env <- ask
liftEitherError (protocolClientError SMP $ B.unpack $ strEncode srv) (getProtocolClient g tSess cfg (Just msgQ) $ clientDisconnected env v)
clientDisconnected :: UnliftIO m -> SMPClientVar -> SMPClient -> IO ()
clientDisconnected u v client = do
clientDisconnected :: Env -> SMPClientVar -> SMPClient -> IO ()
clientDisconnected env v client = do
removeClientAndSubs >>= serverDown
logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
where
@@ -548,7 +548,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
unless (null conns) $ notifySub "" $ DOWN srv conns
unless (null qs) $ do
atomically $ mapM_ (releaseGetLock c) qs
unliftIO u $ resubscribeSMPSession c tSess
runReaderT (resubscribeSMPSession c tSess) env
notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
@@ -1066,19 +1066,19 @@ subscribeQueues c qs = do
atomically $ do
modifyTVar' (subscrConns c) (`S.union` S.fromList (map qConnId qs'))
RQ.batchAddQueues (pendingSubs c) qs'
u <- askUnliftIO
env <- ask
-- only "checked" queues are subscribed
(errs <>) <$> sendTSessionBatches "SUB" 90 id (subscribeQueues_ u) c qs'
(errs <>) <$> sendTSessionBatches "SUB" 90 id (subscribeQueues_ env) c qs'
where
checkQueue rq = do
prohibited <- atomically $ hasGetLock c rq
pure $ if prohibited then Left (rq, Left $ CMD PROHIBITED) else Right rq
subscribeQueues_ :: UnliftIO m -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
subscribeQueues_ u smp qs' = do
subscribeQueues_ :: Env -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
subscribeQueues_ env smp qs' = do
rs <- sendBatch subscribeSMPQueues smp qs'
mapM_ (uncurry $ processSubResult c) rs
when (any temporaryClientError . lefts . map snd $ L.toList rs) . unliftIO u $
resubscribeSMPSession c (transportSession' smp)
when (any temporaryClientError . lefts . map snd $ L.toList rs) $
runReaderT (resubscribeSMPSession c $ transportSession' smp) env
pure rs
type BatchResponses e r = (NonEmpty (RcvQueue, Either e r))