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

agent: do not account for delivery of stored messages for agent suspension (#916)

Browse Source
This commit is contained in:
Evgeny Poberezkin
2023-12-05 23:14:33 +00:00
committed by GitHub
parent e7b6b5facd
commit a860936072
2 changed files with 11 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Simplex/Messaging/Agent.hs
+10 -7
View File
@@ -1103,20 +1103,23 @@ resumeMsgDelivery c cData@ConnData {connId} sq@SndQueue {server, sndId} = do
>>= \a -> atomically (TM.insert qKey a $ smpQueueMsgDeliveries c)
unlessM msgsQueued $
withStore' c (\db -> getPendingMsgs db connId sq)
>>= queuePendingMsgs c sq
>>= queuePendingMsgs' c False sq
where
queueDelivering qKey = atomically $ TM.member qKey (smpQueueMsgDeliveries c)
msgsQueued = atomically $ isJust <$> TM.lookupInsert (server, sndId) True (pendingMsgsQueued c)
queuePendingMsgs :: AgentMonad' m => AgentClient -> SndQueue -> [InternalId] -> m ()
queuePendingMsgs c sq msgIds = atomically $ do
modifyTVar' (msgDeliveryOp c) $ \s -> s {opsInProgress = opsInProgress s + length msgIds}
queuePendingMsgs c = queuePendingMsgs' c True
queuePendingMsgs' :: AgentMonad' m => AgentClient -> Bool -> SndQueue -> [InternalId] -> m ()
queuePendingMsgs' c setDeliveryOps sq msgIds = atomically $ do
when setDeliveryOps $ modifyTVar' (msgDeliveryOp c) $ \s -> s {opsInProgress = opsInProgress s + length msgIds}
-- s <- readTVar (msgDeliveryOp c)
-- unsafeIOToSTM $ putStrLn $ "msgDeliveryOp: " <> show (opsInProgress s)
(mq, _) <- getPendingMsgQ c sq
mapM_ (writeTQueue mq) msgIds
mapM_ (writeTQueue mq . (,setDeliveryOps)) msgIds
getPendingMsgQ :: AgentClient -> SndQueue -> STM (TQueue InternalId, TMVar ())
getPendingMsgQ :: AgentClient -> SndQueue -> STM (TQueue (InternalId, Bool), TMVar ())
getPendingMsgQ c SndQueue {server, sndId} = do
let qKey = (server, sndId)
maybe (newMsgQueue qKey) pure =<< TM.lookup qKey (smpQueueMsgQueues c)
@@ -1134,9 +1137,9 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
atomically $ endAgentOperation c AOSndNetwork
atomically $ throwWhenInactive c
atomically $ throwWhenNoDelivery c sq
msgId <- atomically $ readTQueue mq
(msgId, deliveryOp) <- atomically $ readTQueue mq
atomically $ beginAgentOperation c AOSndNetwork
atomically $ endAgentOperation c AOMsgDelivery -- this operation begins in queuePendingMsgs
when deliveryOp $ atomically $ endAgentOperation c AOMsgDelivery -- this operation begins in queuePendingMsgs
let mId = unId msgId
tryAgentError (withStore c $ \db -> getPendingMsgData db connId msgId) >>= \case
Left e -> notify $ MERR mId e
src/Simplex/Messaging/Agent/Client.hs
+1 -1
View File
@@ -234,7 +234,7 @@ data AgentClient = AgentClient
pendingSubs :: TRcvQueues,
removedSubs :: TMap (UserId, SMPServer, SMP.RecipientId) SMPClientError,
pendingMsgsQueued :: TMap SndQAddr Bool,
smpQueueMsgQueues :: TMap SndQAddr (TQueue InternalId, TMVar ()),
smpQueueMsgQueues :: TMap SndQAddr (TQueue (InternalId, Bool), TMVar ()),
smpQueueMsgDeliveries :: TMap SndQAddr (Async ()),
connCmdsQueued :: TMap ConnId Bool,
asyncCmdQueues :: TMap (Maybe SMPServer) (TQueue AsyncCmdId),