Skip to content

Commit 7c25b3b

Browse files
epoberezkinepoberezkin
authored
smp protocol: send DELD when subscribed queue is deleted (#1312)
* smp protocol: send DELD when subscribed queue is deleted * fix, test * refactor * send DELD event only if the client supports it (version 10); send END otherwise * fix test * notify on notifier rotation * increase test delays
1 parent a70bd02 commit 7c25b3b

File tree

12 files changed

+111
-66
lines changed

12 files changed

+111
-66
lines changed

‎src/Simplex/Messaging/Agent.hs‎

Lines changed: 5 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -166,7 +166,7 @@ import Simplex.Messaging.Agent.Store
166166
import Simplex.Messaging.Agent.Store.SQLite
167167
import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
168168
import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
169-
import Simplex.Messaging.Client (ProtocolClient (..), SMPClientError, ServerTransmission (..), ServerTransmissionBatch, temporaryClientError, unexpectedResponse)
169+
import Simplex.Messaging.Client (SMPClientError, ServerTransmission (..), ServerTransmissionBatch, temporaryClientError, unexpectedResponse)
170170
import qualified Simplex.Messaging.Crypto as C
171171
import Simplex.Messaging.Crypto.File (CryptoFile, CryptoFileArgs)
172172
import Simplex.Messaging.Crypto.Ratchet (PQEncryption, PQSupport (..), pattern PQEncOff, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
@@ -181,7 +181,7 @@ import Simplex.Messaging.Protocol (BrokerMsg, Cmd (..), ErrorType (AUTH), MsgBod
181181
import qualified Simplex.Messaging.Protocol as SMP
182182
import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
183183
import qualified Simplex.Messaging.TMap as TM
184-
import Simplex.Messaging.Transport (SMPVersion, THandleParams (sessionId))
184+
import Simplex.Messaging.Transport (SMPVersion)
185185
import Simplex.Messaging.Util
186186
import Simplex.Messaging.Version
187187
import Simplex.RemoteControl.Client
@@ -2450,17 +2450,14 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
24502450
handleNotifyAck :: AM ACKd -> AM ACKd
24512451
handleNotifyAck m = m `catchAgentError` \e -> notify (ERR e) >> ack
24522452
SMP.END ->
2453-
atomically (TM.lookup tSess (smpClients c) $>>= (tryReadTMVar . sessionVar) >>= processEND)
2453+
atomically (ifM (activeClientSession c tSess sessId) (removeSubscription c connId $> True) (pure False))
24542454
>>= notifyEnd
24552455
where
2456-
processEND = \case
2457-
Just (Right clnt)
2458-
| sessId == sessionId (thParams $ connectedClient clnt) ->
2459-
removeSubscription c connId $> True
2460-
_ -> pure False
24612456
notifyEnd removed
24622457
| removed = notify END >> logServer "<--" c srv rId "END"
24632458
| otherwise = logServer "<--" c srv rId "END from disconnected client - ignored"
2459+
-- Possibly, we need to add some flag to connection that it was deleted
2460+
SMP.DELD -> atomically (removeSubscription c connId) >> notify DELD
24642461
SMP.ERR e -> notify $ ERR $ SMP (B.unpack $ strEncode srv) e
24652462
r -> unexpected r
24662463
where

‎src/Simplex/Messaging/Agent/Protocol.hs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -344,6 +344,7 @@ data AEvent (e :: AEntity) where
344344
INFO :: PQSupport -> ConnInfo -> AEvent AEConn
345345
CON :: PQEncryption -> AEvent AEConn -- notification that connection is established
346346
END :: AEvent AEConn
347+
DELD :: AEvent AEConn
347348
CONNECT :: AProtocolType -> TransportHost -> AEvent AENone
348349
DISCONNECT :: AProtocolType -> TransportHost -> AEvent AENone
349350
DOWN :: SMPServer -> [ConnId] -> AEvent AENone
@@ -413,6 +414,7 @@ data AEventTag (e :: AEntity) where
413414
INFO_ :: AEventTag AEConn
414415
CON_ :: AEventTag AEConn
415416
END_ :: AEventTag AEConn
417+
DELD_ :: AEventTag AEConn
416418
CONNECT_ :: AEventTag AENone
417419
DISCONNECT_ :: AEventTag AENone
418420
DOWN_ :: AEventTag AENone
@@ -466,6 +468,7 @@ aEventTag = \case
466468
INFO {} -> INFO_
467469
CON _ -> CON_
468470
END -> END_
471+
DELD -> DELD_
469472
CONNECT {} -> CONNECT_
470473
DISCONNECT {} -> DISCONNECT_
471474
DOWN {} -> DOWN_

‎src/Simplex/Messaging/Notifications/Protocol.hs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -443,6 +443,8 @@ data NtfSubStatus
443443
NSInactive
444444
| -- | END received
445445
NSEnd
446+
| -- | DELD received (connection was deleted)
447+
NSDeleted
446448
| -- | SMP AUTH error
447449
NSAuth
448450
| -- | SMP error other than AUTH
@@ -456,6 +458,7 @@ ntfShouldSubscribe = \case
456458
NSActive -> True
457459
NSInactive -> True
458460
NSEnd -> False
461+
NSDeleted -> False
459462
NSAuth -> False
460463
NSErr _ -> False
461464

@@ -466,6 +469,7 @@ instance Encoding NtfSubStatus where
466469
NSActive -> "ACTIVE"
467470
NSInactive -> "INACTIVE"
468471
NSEnd -> "END"
472+
NSDeleted -> "DELETED"
469473
NSAuth -> "AUTH"
470474
NSErr err -> "ERR " <> err
471475
smpP =
@@ -475,6 +479,7 @@ instance Encoding NtfSubStatus where
475479
"ACTIVE" -> pure NSActive
476480
"INACTIVE" -> pure NSInactive
477481
"END" -> pure NSEnd
482+
"DELETED" -> pure NSDeleted
478483
"AUTH" -> pure NSAuth
479484
"ERR" -> NSErr <$> (A.space *> A.takeByteString)
480485
_ -> fail "bad NtfSubStatus"

‎src/Simplex/Messaging/Notifications/Server.hs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -226,6 +226,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
226226
Right SMP.END ->
227227
whenM (atomically $ activeClientSession' ca sessionId srv) $
228228
updateSubStatus smpQueue NSEnd
229+
Right SMP.DELD -> updateSubStatus smpQueue NSDeleted
229230
Right (SMP.ERR e) -> logError $ "SMP server error: " <> tshow e
230231
Right _ -> logError "SMP server unexpected response"
231232
Left e -> logError $ "SMP client error: " <> tshow e

‎src/Simplex/Messaging/Protocol.hs‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -484,6 +484,7 @@ data BrokerMsg where
484484
RRES :: EncFwdResponse -> BrokerMsg -- relay to proxy
485485
PRES :: EncResponse -> BrokerMsg -- proxy to client
486486
END :: BrokerMsg
487+
DELD :: BrokerMsg
487488
INFO :: QueueInfo -> BrokerMsg
488489
OK :: BrokerMsg
489490
ERR :: ErrorType -> BrokerMsg
@@ -705,6 +706,7 @@ data BrokerMsgTag
705706
| RRES_
706707
| PRES_
707708
| END_
709+
| DELD_
708710
| INFO_
709711
| OK_
710712
| ERR_
@@ -778,6 +780,7 @@ instance Encoding BrokerMsgTag where
778780
RRES_ -> "RRES"
779781
PRES_ -> "PRES"
780782
END_ -> "END"
783+
DELD_ -> "DELD"
781784
INFO_ -> "INFO"
782785
OK_ -> "OK"
783786
ERR_ -> "ERR"
@@ -794,6 +797,7 @@ instance ProtocolMsgTag BrokerMsgTag where
794797
"RRES" -> Just RRES_
795798
"PRES" -> Just PRES_
796799
"END" -> Just END_
800+
"DELD" -> Just DELD_
797801
"INFO" -> Just INFO_
798802
"OK" -> Just OK_
799803
"ERR" -> Just ERR_
@@ -1423,6 +1427,9 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
14231427
RRES (EncFwdResponse encBlock) -> e (RRES_, ' ', Tail encBlock)
14241428
PRES (EncResponse encBlock) -> e (PRES_, ' ', Tail encBlock)
14251429
END -> e END_
1430+
DELD
1431+
| v >= deletedEventSMPVersion -> e DELD_
1432+
| otherwise -> e END_
14261433
INFO info -> e (INFO_, ' ', info)
14271434
OK -> e OK_
14281435
ERR err -> e (ERR_, ' ', err)
@@ -1448,6 +1455,7 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
14481455
RRES_ -> RRES <$> (EncFwdResponse . unTail <$> _smpP)
14491456
PRES_ -> PRES <$> (EncResponse . unTail <$> _smpP)
14501457
END_ -> pure END
1458+
DELD_ -> pure DELD
14511459
INFO_ -> INFO <$> _smpP
14521460
OK_ -> pure OK
14531461
ERR_ -> ERR <$> _smpP

‎src/Simplex/Messaging/Server.hs‎

Lines changed: 47 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -136,9 +136,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
136136
expired <- restoreServerMessages
137137
restoreServerStats expired
138138
raceAny_
139-
( serverThread s "server subscribedQ" subscribedQ subscribers pendingENDs subscriptions cancelSub
140-
: serverThread s "server ntfSubscribedQ" ntfSubscribedQ Env.notifiers pendingNtfENDs ntfSubscriptions (\_ -> pure ())
141-
: sendPendingENDsThread s
139+
( serverThread s "server subscribedQ" True subscribedQ subscribers pendingENDs subscriptions cancelSub
140+
: serverThread s "server deletedQ" False deletedQ subscribers pendingDELDs subscriptions cancelSub
141+
: serverThread s "server ntfSubscribedQ" True ntfSubscribedQ Env.notifiers pendingNtfENDs ntfSubscriptions (\_ -> pure ())
142+
: serverThread s "server ntfDeletedQ" False ntfDeletedQ Env.notifiers pendingNtfDELDs ntfSubscriptions (\_ -> pure ())
143+
: sendPendingEvtsThread s
142144
: receiveFromProxyAgent pa
143145
: map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg
144146
)
@@ -163,22 +165,23 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
163165
forall s.
164166
Server ->
165167
String ->
166-
(Server -> TQueue (QueueId, ClientId, Subscribed)) ->
168+
Subscribed ->
169+
(Server -> TQueue (QueueId, ClientId)) ->
167170
(Server -> TMap QueueId (TVar Client)) ->
168171
(Server -> TVar (IM.IntMap (NonEmpty RecipientId))) ->
169172
(Client -> TMap QueueId s) ->
170173
(s -> IO ()) ->
171174
M ()
172-
serverThread s label subQ subs ends clientSubs unsub = do
175+
serverThread s label subscribed subQ subs pendingEvts clientSubs unsub = do
173176
labelMyThread label
174177
cls <- asks clients
175178
liftIO . forever $
176179
(atomically (readTQueue $ subQ s) >>= atomically . updateSubscribers cls)
177180
$>>= endPreviousSubscriptions
178181
>>= mapM_ unsub
179182
where
180-
updateSubscribers :: TVar (IM.IntMap (Maybe Client)) -> (QueueId, ClientId, Bool) -> STM (Maybe (QueueId, Client))
181-
updateSubscribers cls (qId, clntId, subscribed) =
183+
updateSubscribers :: TVar (IM.IntMap (Maybe Client)) -> (QueueId, ClientId) -> STM (Maybe (QueueId, Client))
184+
updateSubscribers cls (qId, clntId) =
182185
-- Client lookup by ID is in the same STM transaction.
183186
-- In case client disconnects during the transaction,
184187
-- it will be re-evaluated, and the client won't be stored as subscribed.
@@ -201,37 +204,41 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
201204
| otherwise = (\yes -> if yes then Just (qId, c') else Nothing) <$> readTVar (connected c')
202205
endPreviousSubscriptions :: (QueueId, Client) -> IO (Maybe s)
203206
endPreviousSubscriptions (qId, c) = do
204-
atomically $ modifyTVar' (ends s) $ IM.alter (Just . maybe [qId] (qId <|)) (clientId c)
207+
atomically $ modifyTVar' (pendingEvts s) $ IM.alter (Just . maybe [qId] (qId <|)) (clientId c)
205208
atomically $ TM.lookupDelete qId (clientSubs c)
206209

207-
sendPendingENDsThread :: Server -> M ()
208-
sendPendingENDsThread s = do
210+
sendPendingEvtsThread :: Server -> M ()
211+
sendPendingEvtsThread s = do
209212
endInt <- asks $ pendingENDInterval . config
210213
cls <- asks clients
211214
forever $ do
212215
threadDelay endInt
213-
sendPending cls $ pendingENDs s
214-
sendPending cls $ pendingNtfENDs s
216+
sendPending cls END $ pendingENDs s
217+
sendPending cls DELD $ pendingDELDs s
218+
sendPending cls END $ pendingNtfENDs s
219+
sendPending cls DELD $ pendingNtfDELDs s
215220
where
216-
sendPending cls ref = do
221+
sendPending cls evt ref = do
217222
ends <- atomically $ swapTVar ref IM.empty
218223
unless (null ends) $ forM_ (IM.assocs ends) $ \(cId, qIds) ->
219-
mapM_ (queueENDs qIds) . join . IM.lookup cId =<< readTVarIO cls
220-
queueENDs qIds c@Client {connected, sndQ = q} =
224+
mapM_ (queueEvts qIds evt) . join . IM.lookup cId =<< readTVarIO cls
225+
queueEvts qIds evt c@Client {connected, sndQ = q} =
221226
whenM (readTVarIO connected) $ do
222227
sent <- atomically $ ifM (isFullTBQueue q) (pure False) (writeTBQueue q ts $> True)
223228
if sent
224229
then updateEndStats
225230
else -- if queue is full it can block
226-
forkClient c ("sendPendingENDsThread.queueENDs") $
231+
forkClient c ("sendPendingEvtsThread.queueEvts") $
227232
atomically (writeTBQueue q ts) >> updateEndStats
228233
where
229-
ts = L.map (CorrId "",,END) qIds
230-
updateEndStats = do
231-
stats <- asks serverStats
232-
let len = L.length qIds
233-
liftIO $ atomicModifyIORef'_ (qSubEnd stats) (+ len)
234-
liftIO $ atomicModifyIORef'_ (qSubEndB stats) (+ (len `div` 255 + 1)) -- up to 255 ENDs in the batch
234+
ts = L.map (CorrId "",,evt) qIds
235+
updateEndStats = case evt of
236+
END -> do
237+
stats <- asks serverStats
238+
let len = L.length qIds
239+
liftIO $ atomicModifyIORef'_ (qSubEnd stats) (+ len)
240+
liftIO $ atomicModifyIORef'_ (qSubEndB stats) (+ (len `div` 255 + 1)) -- up to 255 ENDs in the batch
241+
_ -> pure ()
235242

236243
receiveFromProxyAgent :: ProxyAgent -> M ()
237244
receiveFromProxyAgent ProxyAgent {smpAgent = SMPClientAgent {agentQ}} =
@@ -892,7 +899,7 @@ forkClient Client {endThreads, endThreadSeq} label action = do
892899
mkWeakThreadId t >>= atomically . modifyTVar' endThreads . IM.insert tId
893900

894901
client :: THandleParams SMPVersion 'TServer -> Client -> Server -> M ()
895-
client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId, procThreads} Server {subscribedQ, ntfSubscribedQ, subscribers, notifiers} = do
902+
client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId, procThreads} Server {subscribedQ, deletedQ, ntfSubscribedQ, ntfDeletedQ, subscribers, notifiers} = do
896903
labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands"
897904
forever $
898905
atomically (readTBQueue rcvQ)
@@ -985,11 +992,9 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
985992
processCommand (qr_, (corrId, entId, cmd)) = case cmd of
986993
Cmd SProxiedClient command -> processProxiedCmd (corrId, entId, command)
987994
Cmd SSender command -> Just <$> case command of
988-
SKEY sKey -> (corrId,entId,) <$> case qr_ of
989-
Just qr@QueueRec {sndSecure}
990-
| sndSecure -> secureQueue_ "SKEY" qr sKey
991-
| otherwise -> pure $ ERR AUTH
992-
Nothing -> pure $ ERR INTERNAL
995+
SKEY sKey ->
996+
withQueue $ \QueueRec {sndSecure, recipientId} ->
997+
(corrId,entId,) <$> if sndSecure then secureQueue_ "SKEY" recipientId sKey else pure $ ERR AUTH
993998
SEND flags msgBody -> withQueue $ \qr -> sendMessage qr flags msgBody
994999
PING -> pure (corrId, NoEntity, PONG)
9951000
RFWD encBlock -> (corrId, NoEntity,) <$> processForwardedCommand encBlock
@@ -1009,9 +1014,9 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
10091014
SUB -> withQueue (`subscribeQueue` entId)
10101015
GET -> withQueue getMessage
10111016
ACK msgId -> withQueue (`acknowledgeMsg` msgId)
1012-
KEY sKey -> (corrId,entId,) <$> case qr_ of
1013-
Just qr -> secureQueue_ "KEY" qr sKey
1014-
Nothing -> pure $ ERR INTERNAL
1017+
KEY sKey ->
1018+
withQueue $ \QueueRec {recipientId} ->
1019+
(corrId,entId,) <$> secureQueue_ "KEY" recipientId sKey
10151020
NKEY nKey dhKey -> addQueueNotifier_ st nKey dhKey
10161021
NDEL -> deleteQueueNotifier_ st
10171022
OFF -> suspendQueue_ st
@@ -1063,10 +1068,9 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
10631068
n <- asks $ queueIdBytes . config
10641069
liftM2 (,) (randomId n) (randomId n)
10651070

1066-
secureQueue_ :: T.Text -> QueueRec -> SndPublicAuthKey -> M BrokerMsg
1067-
secureQueue_ name qr@QueueRec {recipientId = rId} sKey = time name $ do
1071+
secureQueue_ :: T.Text -> RecipientId -> SndPublicAuthKey -> M BrokerMsg
1072+
secureQueue_ name rId sKey = time name $ do
10681073
withLog $ \s -> logSecureQueue s rId sKey
1069-
updateQueueDate qr
10701074
st <- asks queueStore
10711075
stats <- asks serverStats
10721076
incStat $ qSecured stats
@@ -1086,20 +1090,22 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
10861090
liftIO (addQueueNotifier st entId ntfCreds) >>= \case
10871091
Left DUPLICATE_ -> addNotifierRetry (n - 1) rcvPublicDhKey rcvNtfDhSecret
10881092
Left e -> pure $ ERR e
1089-
Right _ -> do
1093+
Right nId_ -> do
10901094
withLog $ \s -> logAddNotifier s entId ntfCreds
10911095
incStat . ntfCreated =<< asks serverStats
1096+
forM_ nId_ $ \nId -> atomically $ writeTQueue ntfDeletedQ (nId, clientId)
10921097
pure $ NID notifierId rcvPublicDhKey
10931098

10941099
deleteQueueNotifier_ :: QueueStore -> M (Transmission BrokerMsg)
10951100
deleteQueueNotifier_ st = do
10961101
withLog (`logDeleteNotifier` entId)
10971102
liftIO (deleteQueueNotifier st entId) >>= \case
1098-
Right () -> do
1103+
Right (Just nId) -> do
10991104
-- Possibly, the same should be done if the queue is suspended, but currently we do not use it
1100-
atomically $ writeTQueue ntfSubscribedQ (entId, clientId, False)
1105+
atomically $ writeTQueue ntfDeletedQ (nId, clientId)
11011106
incStat . ntfDeleted =<< asks serverStats
11021107
pure ok
1108+
Right Nothing -> pure ok
11031109
Left e -> pure $ err e
11041110

11051111
suspendQueue_ :: QueueStore -> M (Transmission BrokerMsg)
@@ -1124,7 +1130,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
11241130
where
11251131
newSub :: M Sub
11261132
newSub = time "SUB newSub" . atomically $ do
1127-
writeTQueue subscribedQ (rId, clientId, True)
1133+
writeTQueue subscribedQ (rId, clientId)
11281134
sub <- newSubscription NoSub
11291135
TM.insert rId sub subscriptions
11301136
pure sub
@@ -1199,7 +1205,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
11991205
pure ok
12001206
where
12011207
newSub = do
1202-
writeTQueue ntfSubscribedQ (entId, clientId, True)
1208+
writeTQueue ntfSubscribedQ (entId, clientId)
12031209
TM.insert entId () ntfSubscriptions
12041210

12051211
acknowledgeMsg :: QueueRec -> MsgId -> M (Transmission BrokerMsg)
@@ -1480,9 +1486,9 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
14801486
liftIO (deleteQueue st entId $>>= \q -> delMsgQueue ms entId $> Right q) >>= \case
14811487
Right q -> do
14821488
-- Possibly, the same should be done if the queue is suspended, but currently we do not use it
1483-
atomically $ writeTQueue subscribedQ (entId, clientId, False)
1489+
atomically $ writeTQueue deletedQ (entId, clientId)
14841490
forM_ (notifierId <$> notifier q) $ \nId ->
1485-
atomically $ writeTQueue ntfSubscribedQ (nId, clientId, False)
1491+
atomically $ writeTQueue ntfDeletedQ (nId, clientId)
14861492
updateDeletedStats q
14871493
pure ok
14881494
Left e -> pure $ err e

‎src/Simplex/Messaging/Server/Env/STM.hs‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -136,12 +136,16 @@ data Env = Env
136136
type Subscribed = Bool
137137

138138
data Server = Server
139-
{ subscribedQ :: TQueue (RecipientId, ClientId, Subscribed),
139+
{ subscribedQ :: TQueue (RecipientId, ClientId),
140+
deletedQ :: TQueue (RecipientId, ClientId),
140141
subscribers :: TMap RecipientId (TVar Client),
141-
ntfSubscribedQ :: TQueue (NotifierId, ClientId, Subscribed),
142+
ntfSubscribedQ :: TQueue (NotifierId, ClientId),
143+
ntfDeletedQ :: TQueue (NotifierId, ClientId),
142144
notifiers :: TMap NotifierId (TVar Client),
143145
pendingENDs :: TVar (IntMap (NonEmpty RecipientId)),
146+
pendingDELDs :: TVar (IntMap (NonEmpty RecipientId)),
144147
pendingNtfENDs :: TVar (IntMap (NonEmpty NotifierId)),
148+
pendingNtfDELDs :: TVar (IntMap (NonEmpty NotifierId)),
145149
savingLock :: Lock
146150
}
147151

@@ -181,13 +185,17 @@ data Sub = Sub
181185
newServer :: IO Server
182186
newServer = do
183187
subscribedQ <- newTQueueIO
188+
deletedQ <- newTQueueIO
184189
subscribers <- TM.emptyIO
185190
ntfSubscribedQ <- newTQueueIO
191+
ntfDeletedQ <- newTQueueIO
186192
notifiers <- TM.emptyIO
187193
pendingENDs <- newTVarIO IM.empty
194+
pendingDELDs <- newTVarIO IM.empty
188195
pendingNtfENDs <- newTVarIO IM.empty
196+
pendingNtfDELDs <- newTVarIO IM.empty
189197
savingLock <- atomically createLock
190-
return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers, pendingENDs, pendingNtfENDs, savingLock}
198+
return Server {subscribedQ, deletedQ, subscribers, ntfSubscribedQ, ntfDeletedQ, notifiers, pendingENDs, pendingDELDs, pendingNtfENDs, pendingNtfDELDs, savingLock}
191199

192200
newClient :: ClientId -> Natural -> VersionSMP -> ByteString -> SystemTime -> IO Client
193201
newClient clientId qSize thVersion sessionId createdAt = do

0 commit comments

Comments
 (0)