{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}

module Simplex.Messaging.Agent.TSessionSubs
  ( TSessionSubs (sessionSubs),
    SessSubs (..),
    emptyIO,
    clear,
    hasActiveSub,
    hasPendingSub,
    addPendingSub,
    setSessionId,
    addActiveSub,
    batchAddActiveSubs,
    batchAddPendingSubs,
    deletePendingSub,
    batchDeletePendingSubs,
    deleteSub,
    batchDeleteSubs,
    hasPendingSubs,
    getPendingSubs,
    getActiveSubs,
    setSubsPending,
    updateClientNotices,
    foldSessionSubs,
    mapSubs,
  )
where

import Control.Concurrent.STM
import Control.Monad
import Data.Int (Int64)
import Data.List (foldl')
import Data.Map.Strict (Map)
import qualified Data.Map.Strict as M
import Data.Maybe (isJust)
import qualified Data.Set as S
import Simplex.Messaging.Agent.Protocol (SMPQueue (..))
import Simplex.Messaging.Agent.Store (RcvQueueSub (..), SomeRcvQueue)
import Simplex.Messaging.Client (SMPTransportSession, TransportSessionMode (..))
import Simplex.Messaging.Protocol (RecipientId)
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
import Simplex.Messaging.Transport
import Simplex.Messaging.Util (($>>=))

data TSessionSubs = TSessionSubs
  { TSessionSubs -> TMap SMPTransportSession SessSubs
sessionSubs :: TMap SMPTransportSession SessSubs
  }

data SessSubs = SessSubs
  { SessSubs -> TVar (Maybe ByteString)
subsSessId :: TVar (Maybe SessionId),
    SessSubs -> TMap RecipientId RcvQueueSub
activeSubs :: TMap RecipientId RcvQueueSub,
    SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs :: TMap RecipientId RcvQueueSub
  }

emptyIO :: IO TSessionSubs
emptyIO :: IO TSessionSubs
emptyIO = TMap SMPTransportSession SessSubs -> TSessionSubs
TVar
  (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
-> TSessionSubs
TSessionSubs (TVar
   (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
 -> TSessionSubs)
-> IO
     (TVar
        (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs))
-> IO TSessionSubs
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IO
  (TVar
     (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs))
forall k a. IO (TMap k a)
TM.emptyIO
{-# INLINE emptyIO #-}

clear :: TSessionSubs -> STM ()
clear :: TSessionSubs -> STM ()
clear = TVar
  (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
-> STM ()
forall k a. TMap k a -> STM ()
TM.clear (TVar
   (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
 -> STM ())
-> (TSessionSubs
    -> TVar
         (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs))
-> TSessionSubs
-> STM ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TSessionSubs -> TMap SMPTransportSession SessSubs
TSessionSubs
-> TVar
     (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
sessionSubs
{-# INLINE clear #-}

lookupSubs :: SMPTransportSession -> TSessionSubs -> STM (Maybe SessSubs)
lookupSubs :: SMPTransportSession -> TSessionSubs -> STM (Maybe SessSubs)
lookupSubs SMPTransportSession
tSess = (UserId, ProtocolServer 'PSMP, Maybe ByteString)
-> TVar
     (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
-> STM (Maybe SessSubs)
forall k a. Ord k => k -> TMap k a -> STM (Maybe a)
TM.lookup SMPTransportSession
(UserId, ProtocolServer 'PSMP, Maybe ByteString)
tSess (TVar
   (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
 -> STM (Maybe SessSubs))
-> (TSessionSubs
    -> TVar
         (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs))
-> TSessionSubs
-> STM (Maybe SessSubs)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TSessionSubs -> TMap SMPTransportSession SessSubs
TSessionSubs
-> TVar
     (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
sessionSubs
{-# INLINE lookupSubs #-}

getSessSubs :: SMPTransportSession -> TSessionSubs -> STM SessSubs
getSessSubs :: SMPTransportSession -> TSessionSubs -> STM SessSubs
getSessSubs SMPTransportSession
tSess TSessionSubs
ss = SMPTransportSession -> TSessionSubs -> STM (Maybe SessSubs)
lookupSubs SMPTransportSession
tSess TSessionSubs
ss STM (Maybe SessSubs)
-> (Maybe SessSubs -> STM SessSubs) -> STM SessSubs
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= STM SessSubs
-> (SessSubs -> STM SessSubs) -> Maybe SessSubs -> STM SessSubs
forall b a. b -> (a -> b) -> Maybe a -> b
maybe STM SessSubs
new SessSubs -> STM SessSubs
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
  where
    new :: STM SessSubs
new = do
      SessSubs
s <- TVar (Maybe ByteString)
-> TMap RecipientId RcvQueueSub
-> TMap RecipientId RcvQueueSub
-> SessSubs
SessSubs (TVar (Maybe ByteString)
 -> TMap RecipientId RcvQueueSub
 -> TMap RecipientId RcvQueueSub
 -> SessSubs)
-> STM (TVar (Maybe ByteString))
-> STM
     (TMap RecipientId RcvQueueSub
      -> TMap RecipientId RcvQueueSub -> SessSubs)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe ByteString -> STM (TVar (Maybe ByteString))
forall a. a -> STM (TVar a)
newTVar Maybe ByteString
forall a. Maybe a
Nothing STM
  (TMap RecipientId RcvQueueSub
   -> TMap RecipientId RcvQueueSub -> SessSubs)
-> STM (TMap RecipientId RcvQueueSub)
-> STM (TMap RecipientId RcvQueueSub -> SessSubs)
forall a b. STM (a -> b) -> STM a -> STM b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Map RecipientId RcvQueueSub -> STM (TMap RecipientId RcvQueueSub)
forall a. a -> STM (TVar a)
newTVar Map RecipientId RcvQueueSub
forall k a. Map k a
M.empty STM (TMap RecipientId RcvQueueSub -> SessSubs)
-> STM (TMap RecipientId RcvQueueSub) -> STM SessSubs
forall a b. STM (a -> b) -> STM a -> STM b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Map RecipientId RcvQueueSub -> STM (TMap RecipientId RcvQueueSub)
forall a. a -> STM (TVar a)
newTVar Map RecipientId RcvQueueSub
forall k a. Map k a
M.empty
      SMPTransportSession
-> SessSubs -> TMap SMPTransportSession SessSubs -> STM ()
forall k a. Ord k => k -> a -> TMap k a -> STM ()
TM.insert SMPTransportSession
tSess SessSubs
s (TMap SMPTransportSession SessSubs -> STM ())
-> TMap SMPTransportSession SessSubs -> STM ()
forall a b. (a -> b) -> a -> b
$ TSessionSubs -> TMap SMPTransportSession SessSubs
sessionSubs TSessionSubs
ss
      SessSubs -> STM SessSubs
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure SessSubs
s

hasActiveSub :: SMPTransportSession -> RecipientId -> TSessionSubs -> STM Bool
hasActiveSub :: SMPTransportSession -> RecipientId -> TSessionSubs -> STM Bool
hasActiveSub = (SessSubs -> TMap RecipientId RcvQueueSub)
-> SMPTransportSession -> RecipientId -> TSessionSubs -> STM Bool
hasQueue_ SessSubs -> TMap RecipientId RcvQueueSub
activeSubs
{-# INLINE hasActiveSub #-}

hasPendingSub :: SMPTransportSession -> RecipientId -> TSessionSubs -> STM Bool
hasPendingSub :: SMPTransportSession -> RecipientId -> TSessionSubs -> STM Bool
hasPendingSub = (SessSubs -> TMap RecipientId RcvQueueSub)
-> SMPTransportSession -> RecipientId -> TSessionSubs -> STM Bool
hasQueue_ SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs
{-# INLINE hasPendingSub #-}

hasQueue_ :: (SessSubs -> TMap RecipientId RcvQueueSub) -> SMPTransportSession -> RecipientId -> TSessionSubs -> STM Bool
hasQueue_ :: (SessSubs -> TMap RecipientId RcvQueueSub)
-> SMPTransportSession -> RecipientId -> TSessionSubs -> STM Bool
hasQueue_ SessSubs -> TMap RecipientId RcvQueueSub
subs SMPTransportSession
tSess RecipientId
rId TSessionSubs
ss = Maybe RcvQueueSub -> Bool
forall a. Maybe a -> Bool
isJust (Maybe RcvQueueSub -> Bool) -> STM (Maybe RcvQueueSub) -> STM Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (SMPTransportSession -> TSessionSubs -> STM (Maybe SessSubs)
lookupSubs SMPTransportSession
tSess TSessionSubs
ss STM (Maybe SessSubs)
-> (SessSubs -> STM (Maybe RcvQueueSub)) -> STM (Maybe RcvQueueSub)
forall (m :: * -> *) (f :: * -> *) a b.
(Monad m, Monad f, Traversable f) =>
m (f a) -> (a -> m (f b)) -> m (f b)
$>>= RecipientId
-> TMap RecipientId RcvQueueSub -> STM (Maybe RcvQueueSub)
forall k a. Ord k => k -> TMap k a -> STM (Maybe a)
TM.lookup RecipientId
rId (TMap RecipientId RcvQueueSub -> STM (Maybe RcvQueueSub))
-> (SessSubs -> TMap RecipientId RcvQueueSub)
-> SessSubs
-> STM (Maybe RcvQueueSub)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SessSubs -> TMap RecipientId RcvQueueSub
subs)
{-# INLINE hasQueue_ #-}

addPendingSub :: SMPTransportSession -> RcvQueueSub -> TSessionSubs -> STM ()
addPendingSub :: SMPTransportSession -> RcvQueueSub -> TSessionSubs -> STM ()
addPendingSub SMPTransportSession
tSess RcvQueueSub
rq TSessionSubs
ss = SMPTransportSession -> TSessionSubs -> STM SessSubs
getSessSubs SMPTransportSession
tSess TSessionSubs
ss STM SessSubs -> (SessSubs -> STM ()) -> STM ()
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= RecipientId
-> RcvQueueSub -> TMap RecipientId RcvQueueSub -> STM ()
forall k a. Ord k => k -> a -> TMap k a -> STM ()
TM.insert (RcvQueueSub -> RecipientId
rcvId RcvQueueSub
rq) RcvQueueSub
rq (TMap RecipientId RcvQueueSub -> STM ())
-> (SessSubs -> TMap RecipientId RcvQueueSub) -> SessSubs -> STM ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs

setSessionId :: SMPTransportSession -> SessionId -> TSessionSubs -> STM ()
setSessionId :: SMPTransportSession -> ByteString -> TSessionSubs -> STM ()
setSessionId SMPTransportSession
tSess ByteString
sessId TSessionSubs
ss = do
  SessSubs
s <- SMPTransportSession -> TSessionSubs -> STM SessSubs
getSessSubs SMPTransportSession
tSess TSessionSubs
ss
  TVar (Maybe ByteString) -> STM (Maybe ByteString)
forall a. TVar a -> STM a
readTVar (SessSubs -> TVar (Maybe ByteString)
subsSessId SessSubs
s) STM (Maybe ByteString) -> (Maybe ByteString -> STM ()) -> STM ()
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
    Maybe ByteString
Nothing -> TVar (Maybe ByteString) -> Maybe ByteString -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar (SessSubs -> TVar (Maybe ByteString)
subsSessId SessSubs
s) (ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
sessId)
    Just ByteString
sessId' -> Bool -> STM () -> STM ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (ByteString
sessId ByteString -> ByteString -> Bool
forall a. Eq a => a -> a -> Bool
== ByteString
sessId') (STM () -> STM ()) -> STM () -> STM ()
forall a b. (a -> b) -> a -> b
$ STM (Map RecipientId RcvQueueSub) -> STM ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (STM (Map RecipientId RcvQueueSub) -> STM ())
-> STM (Map RecipientId RcvQueueSub) -> STM ()
forall a b. (a -> b) -> a -> b
$ SessSubs -> Maybe ByteString -> STM (Map RecipientId RcvQueueSub)
setSubsPending_ SessSubs
s (Maybe ByteString -> STM (Map RecipientId RcvQueueSub))
-> Maybe ByteString -> STM (Map RecipientId RcvQueueSub)
forall a b. (a -> b) -> a -> b
$ ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
sessId

addActiveSub :: SMPTransportSession -> SessionId -> RcvQueueSub -> TSessionSubs -> STM ()
addActiveSub :: SMPTransportSession
-> ByteString -> RcvQueueSub -> TSessionSubs -> STM ()
addActiveSub SMPTransportSession
tSess ByteString
sessId RcvQueueSub
rq TSessionSubs
ss = do
  SessSubs
s <- SMPTransportSession -> TSessionSubs -> STM SessSubs
getSessSubs SMPTransportSession
tSess TSessionSubs
ss
  Maybe ByteString
sessId' <- TVar (Maybe ByteString) -> STM (Maybe ByteString)
forall a. TVar a -> STM a
readTVar (TVar (Maybe ByteString) -> STM (Maybe ByteString))
-> TVar (Maybe ByteString) -> STM (Maybe ByteString)
forall a b. (a -> b) -> a -> b
$ SessSubs -> TVar (Maybe ByteString)
subsSessId SessSubs
s
  let rId :: RecipientId
rId = RcvQueueSub -> RecipientId
rcvId RcvQueueSub
rq
  if ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
sessId Maybe ByteString -> Maybe ByteString -> Bool
forall a. Eq a => a -> a -> Bool
== Maybe ByteString
sessId'
    then do
      RecipientId
-> RcvQueueSub -> TMap RecipientId RcvQueueSub -> STM ()
forall k a. Ord k => k -> a -> TMap k a -> STM ()
TM.insert RecipientId
rId RcvQueueSub
rq (TMap RecipientId RcvQueueSub -> STM ())
-> TMap RecipientId RcvQueueSub -> STM ()
forall a b. (a -> b) -> a -> b
$ SessSubs -> TMap RecipientId RcvQueueSub
activeSubs SessSubs
s
      RecipientId -> TMap RecipientId RcvQueueSub -> STM ()
forall k a. Ord k => k -> TMap k a -> STM ()
TM.delete RecipientId
rId (TMap RecipientId RcvQueueSub -> STM ())
-> TMap RecipientId RcvQueueSub -> STM ()
forall a b. (a -> b) -> a -> b
$ SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs SessSubs
s
    else RecipientId
-> RcvQueueSub -> TMap RecipientId RcvQueueSub -> STM ()
forall k a. Ord k => k -> a -> TMap k a -> STM ()
TM.insert RecipientId
rId RcvQueueSub
rq (TMap RecipientId RcvQueueSub -> STM ())
-> TMap RecipientId RcvQueueSub -> STM ()
forall a b. (a -> b) -> a -> b
$ SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs SessSubs
s

batchAddActiveSubs :: SMPTransportSession -> SessionId -> [RcvQueueSub] -> TSessionSubs -> STM ()
batchAddActiveSubs :: SMPTransportSession
-> ByteString -> [RcvQueueSub] -> TSessionSubs -> STM ()
batchAddActiveSubs SMPTransportSession
tSess ByteString
sessId [RcvQueueSub]
rqs TSessionSubs
ss = do
  SessSubs
s <- SMPTransportSession -> TSessionSubs -> STM SessSubs
getSessSubs SMPTransportSession
tSess TSessionSubs
ss
  Maybe ByteString
sessId' <- TVar (Maybe ByteString) -> STM (Maybe ByteString)
forall a. TVar a -> STM a
readTVar (TVar (Maybe ByteString) -> STM (Maybe ByteString))
-> TVar (Maybe ByteString) -> STM (Maybe ByteString)
forall a b. (a -> b) -> a -> b
$ SessSubs -> TVar (Maybe ByteString)
subsSessId SessSubs
s
  let qs :: Map RecipientId RcvQueueSub
qs = [(RecipientId, RcvQueueSub)] -> Map RecipientId RcvQueueSub
forall k a. Ord k => [(k, a)] -> Map k a
M.fromList ([(RecipientId, RcvQueueSub)] -> Map RecipientId RcvQueueSub)
-> [(RecipientId, RcvQueueSub)] -> Map RecipientId RcvQueueSub
forall a b. (a -> b) -> a -> b
$ (RcvQueueSub -> (RecipientId, RcvQueueSub))
-> [RcvQueueSub] -> [(RecipientId, RcvQueueSub)]
forall a b. (a -> b) -> [a] -> [b]
map (\RcvQueueSub
rq -> (RcvQueueSub -> RecipientId
rcvId RcvQueueSub
rq, RcvQueueSub
rq)) [RcvQueueSub]
rqs
  if ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
sessId Maybe ByteString -> Maybe ByteString -> Bool
forall a. Eq a => a -> a -> Bool
== Maybe ByteString
sessId'
    then do
      Map RecipientId RcvQueueSub
-> TMap RecipientId RcvQueueSub -> STM ()
forall k a. Ord k => Map k a -> TMap k a -> STM ()
TM.union Map RecipientId RcvQueueSub
qs (TMap RecipientId RcvQueueSub -> STM ())
-> TMap RecipientId RcvQueueSub -> STM ()
forall a b. (a -> b) -> a -> b
$ SessSubs -> TMap RecipientId RcvQueueSub
activeSubs SessSubs
s
      TMap RecipientId RcvQueueSub
-> (Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
-> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs SessSubs
s) (Map RecipientId RcvQueueSub
-> Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub
forall k a b. Ord k => Map k a -> Map k b -> Map k a
`M.difference` Map RecipientId RcvQueueSub
qs)
    else Map RecipientId RcvQueueSub
-> TMap RecipientId RcvQueueSub -> STM ()
forall k a. Ord k => Map k a -> TMap k a -> STM ()
TM.union Map RecipientId RcvQueueSub
qs (TMap RecipientId RcvQueueSub -> STM ())
-> TMap RecipientId RcvQueueSub -> STM ()
forall a b. (a -> b) -> a -> b
$ SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs SessSubs
s

batchAddPendingSubs :: SMPTransportSession -> [RcvQueueSub] -> TSessionSubs -> STM ()
batchAddPendingSubs :: SMPTransportSession -> [RcvQueueSub] -> TSessionSubs -> STM ()
batchAddPendingSubs SMPTransportSession
tSess [RcvQueueSub]
rqs TSessionSubs
ss = do
  SessSubs
s <- SMPTransportSession -> TSessionSubs -> STM SessSubs
getSessSubs SMPTransportSession
tSess TSessionSubs
ss
  TMap RecipientId RcvQueueSub
-> (Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
-> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs SessSubs
s) ((Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
 -> STM ())
-> (Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
-> STM ()
forall a b. (a -> b) -> a -> b
$ Map RecipientId RcvQueueSub
-> Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub
forall k a. Ord k => Map k a -> Map k a -> Map k a
M.union (Map RecipientId RcvQueueSub
 -> Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
-> Map RecipientId RcvQueueSub
-> Map RecipientId RcvQueueSub
-> Map RecipientId RcvQueueSub
forall a b. (a -> b) -> a -> b
$ [(RecipientId, RcvQueueSub)] -> Map RecipientId RcvQueueSub
forall k a. Ord k => [(k, a)] -> Map k a
M.fromList ([(RecipientId, RcvQueueSub)] -> Map RecipientId RcvQueueSub)
-> [(RecipientId, RcvQueueSub)] -> Map RecipientId RcvQueueSub
forall a b. (a -> b) -> a -> b
$ (RcvQueueSub -> (RecipientId, RcvQueueSub))
-> [RcvQueueSub] -> [(RecipientId, RcvQueueSub)]
forall a b. (a -> b) -> [a] -> [b]
map (\RcvQueueSub
rq -> (RcvQueueSub -> RecipientId
rcvId RcvQueueSub
rq, RcvQueueSub
rq)) [RcvQueueSub]
rqs

deletePendingSub :: SMPTransportSession -> RecipientId -> TSessionSubs -> STM ()
deletePendingSub :: SMPTransportSession -> RecipientId -> TSessionSubs -> STM ()
deletePendingSub SMPTransportSession
tSess RecipientId
rId = SMPTransportSession -> TSessionSubs -> STM (Maybe SessSubs)
lookupSubs SMPTransportSession
tSess (TSessionSubs -> STM (Maybe SessSubs))
-> (Maybe SessSubs -> STM ()) -> TSessionSubs -> STM ()
forall (m :: * -> *) a b c.
Monad m =>
(a -> m b) -> (b -> m c) -> a -> m c
>=> (SessSubs -> STM ()) -> Maybe SessSubs -> STM ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (RecipientId -> TMap RecipientId RcvQueueSub -> STM ()
forall k a. Ord k => k -> TMap k a -> STM ()
TM.delete RecipientId
rId (TMap RecipientId RcvQueueSub -> STM ())
-> (SessSubs -> TMap RecipientId RcvQueueSub) -> SessSubs -> STM ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs)

batchDeletePendingSubs :: SMPTransportSession -> S.Set RecipientId -> TSessionSubs -> STM ()
batchDeletePendingSubs :: SMPTransportSession -> Set RecipientId -> TSessionSubs -> STM ()
batchDeletePendingSubs SMPTransportSession
tSess Set RecipientId
rIds = SMPTransportSession -> TSessionSubs -> STM (Maybe SessSubs)
lookupSubs SMPTransportSession
tSess (TSessionSubs -> STM (Maybe SessSubs))
-> (Maybe SessSubs -> STM ()) -> TSessionSubs -> STM ()
forall (m :: * -> *) a b c.
Monad m =>
(a -> m b) -> (b -> m c) -> a -> m c
>=> (SessSubs -> STM ()) -> Maybe SessSubs -> STM ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (TMap RecipientId RcvQueueSub -> STM ()
forall {a}. TVar (Map RecipientId a) -> STM ()
delete (TMap RecipientId RcvQueueSub -> STM ())
-> (SessSubs -> TMap RecipientId RcvQueueSub) -> SessSubs -> STM ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs)
  where
    delete :: TVar (Map RecipientId a) -> STM ()
delete = (TVar (Map RecipientId a)
-> (Map RecipientId a -> Map RecipientId a) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
`modifyTVar'` (Map RecipientId a -> Set RecipientId -> Map RecipientId a
forall k a. Ord k => Map k a -> Set k -> Map k a
`M.withoutKeys` Set RecipientId
rIds))

deleteSub :: SMPTransportSession -> RecipientId -> TSessionSubs -> STM ()
deleteSub :: SMPTransportSession -> RecipientId -> TSessionSubs -> STM ()
deleteSub SMPTransportSession
tSess RecipientId
rId = SMPTransportSession -> TSessionSubs -> STM (Maybe SessSubs)
lookupSubs SMPTransportSession
tSess (TSessionSubs -> STM (Maybe SessSubs))
-> (Maybe SessSubs -> STM ()) -> TSessionSubs -> STM ()
forall (m :: * -> *) a b c.
Monad m =>
(a -> m b) -> (b -> m c) -> a -> m c
>=> (SessSubs -> STM ()) -> Maybe SessSubs -> STM ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (\SessSubs
s -> RecipientId -> TMap RecipientId RcvQueueSub -> STM ()
forall k a. Ord k => k -> TMap k a -> STM ()
TM.delete RecipientId
rId (SessSubs -> TMap RecipientId RcvQueueSub
activeSubs SessSubs
s) STM () -> STM () -> STM ()
forall a b. STM a -> STM b -> STM b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> RecipientId -> TMap RecipientId RcvQueueSub -> STM ()
forall k a. Ord k => k -> TMap k a -> STM ()
TM.delete RecipientId
rId (SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs SessSubs
s))

batchDeleteSubs :: SomeRcvQueue q => SMPTransportSession -> [q] -> TSessionSubs -> STM ()
batchDeleteSubs :: forall q.
SomeRcvQueue q =>
SMPTransportSession -> [q] -> TSessionSubs -> STM ()
batchDeleteSubs SMPTransportSession
tSess [q]
rqs = SMPTransportSession -> TSessionSubs -> STM (Maybe SessSubs)
lookupSubs SMPTransportSession
tSess (TSessionSubs -> STM (Maybe SessSubs))
-> (Maybe SessSubs -> STM ()) -> TSessionSubs -> STM ()
forall (m :: * -> *) a b c.
Monad m =>
(a -> m b) -> (b -> m c) -> a -> m c
>=> (SessSubs -> STM ()) -> Maybe SessSubs -> STM ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (\SessSubs
s -> TMap RecipientId RcvQueueSub -> STM ()
forall {a}. TVar (Map RecipientId a) -> STM ()
delete (SessSubs -> TMap RecipientId RcvQueueSub
activeSubs SessSubs
s) STM () -> STM () -> STM ()
forall a b. STM a -> STM b -> STM b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> TMap RecipientId RcvQueueSub -> STM ()
forall {a}. TVar (Map RecipientId a) -> STM ()
delete (SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs SessSubs
s))
  where
    rIds :: Set RecipientId
rIds = [RecipientId] -> Set RecipientId
forall a. Ord a => [a] -> Set a
S.fromList ([RecipientId] -> Set RecipientId)
-> [RecipientId] -> Set RecipientId
forall a b. (a -> b) -> a -> b
$ (q -> RecipientId) -> [q] -> [RecipientId]
forall a b. (a -> b) -> [a] -> [b]
map q -> RecipientId
forall q. SMPQueue q => q -> RecipientId
queueId [q]
rqs
    delete :: TVar (Map RecipientId a) -> STM ()
delete = (TVar (Map RecipientId a)
-> (Map RecipientId a -> Map RecipientId a) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
`modifyTVar'` (Map RecipientId a -> Set RecipientId -> Map RecipientId a
forall k a. Ord k => Map k a -> Set k -> Map k a
`M.withoutKeys` Set RecipientId
rIds))

hasPendingSubs :: SMPTransportSession -> TSessionSubs -> STM Bool
hasPendingSubs :: SMPTransportSession -> TSessionSubs -> STM Bool
hasPendingSubs SMPTransportSession
tSess = SMPTransportSession -> TSessionSubs -> STM (Maybe SessSubs)
lookupSubs SMPTransportSession
tSess (TSessionSubs -> STM (Maybe SessSubs))
-> (Maybe SessSubs -> STM Bool) -> TSessionSubs -> STM Bool
forall (m :: * -> *) a b c.
Monad m =>
(a -> m b) -> (b -> m c) -> a -> m c
>=> STM Bool -> (SessSubs -> STM Bool) -> Maybe SessSubs -> STM Bool
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (Bool -> STM Bool
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False) ((Map RecipientId RcvQueueSub -> Bool)
-> STM (Map RecipientId RcvQueueSub) -> STM Bool
forall a b. (a -> b) -> STM a -> STM b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (Bool -> Bool
not (Bool -> Bool)
-> (Map RecipientId RcvQueueSub -> Bool)
-> Map RecipientId RcvQueueSub
-> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Map RecipientId RcvQueueSub -> Bool
forall a. Map RecipientId a -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null) (STM (Map RecipientId RcvQueueSub) -> STM Bool)
-> (SessSubs -> STM (Map RecipientId RcvQueueSub))
-> SessSubs
-> STM Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TMap RecipientId RcvQueueSub -> STM (Map RecipientId RcvQueueSub)
forall a. TVar a -> STM a
readTVar (TMap RecipientId RcvQueueSub -> STM (Map RecipientId RcvQueueSub))
-> (SessSubs -> TMap RecipientId RcvQueueSub)
-> SessSubs
-> STM (Map RecipientId RcvQueueSub)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs)

getPendingSubs :: SMPTransportSession -> TSessionSubs -> STM (Map RecipientId RcvQueueSub)
getPendingSubs :: SMPTransportSession
-> TSessionSubs -> STM (Map RecipientId RcvQueueSub)
getPendingSubs = (SessSubs -> TMap RecipientId RcvQueueSub)
-> SMPTransportSession
-> TSessionSubs
-> STM (Map RecipientId RcvQueueSub)
getSubs_ SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs
{-# INLINE getPendingSubs #-}

getActiveSubs :: SMPTransportSession -> TSessionSubs -> STM (Map RecipientId RcvQueueSub)
getActiveSubs :: SMPTransportSession
-> TSessionSubs -> STM (Map RecipientId RcvQueueSub)
getActiveSubs = (SessSubs -> TMap RecipientId RcvQueueSub)
-> SMPTransportSession
-> TSessionSubs
-> STM (Map RecipientId RcvQueueSub)
getSubs_ SessSubs -> TMap RecipientId RcvQueueSub
activeSubs
{-# INLINE getActiveSubs #-}

getSubs_ :: (SessSubs -> TMap RecipientId RcvQueueSub) -> SMPTransportSession -> TSessionSubs -> STM (Map RecipientId RcvQueueSub)
getSubs_ :: (SessSubs -> TMap RecipientId RcvQueueSub)
-> SMPTransportSession
-> TSessionSubs
-> STM (Map RecipientId RcvQueueSub)
getSubs_ SessSubs -> TMap RecipientId RcvQueueSub
subs SMPTransportSession
tSess = SMPTransportSession -> TSessionSubs -> STM (Maybe SessSubs)
lookupSubs SMPTransportSession
tSess (TSessionSubs -> STM (Maybe SessSubs))
-> (Maybe SessSubs -> STM (Map RecipientId RcvQueueSub))
-> TSessionSubs
-> STM (Map RecipientId RcvQueueSub)
forall (m :: * -> *) a b c.
Monad m =>
(a -> m b) -> (b -> m c) -> a -> m c
>=> STM (Map RecipientId RcvQueueSub)
-> (SessSubs -> STM (Map RecipientId RcvQueueSub))
-> Maybe SessSubs
-> STM (Map RecipientId RcvQueueSub)
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (Map RecipientId RcvQueueSub -> STM (Map RecipientId RcvQueueSub)
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Map RecipientId RcvQueueSub
forall k a. Map k a
M.empty) (TMap RecipientId RcvQueueSub -> STM (Map RecipientId RcvQueueSub)
forall a. TVar a -> STM a
readTVar (TMap RecipientId RcvQueueSub -> STM (Map RecipientId RcvQueueSub))
-> (SessSubs -> TMap RecipientId RcvQueueSub)
-> SessSubs
-> STM (Map RecipientId RcvQueueSub)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SessSubs -> TMap RecipientId RcvQueueSub
subs)

setSubsPending :: TransportSessionMode -> SMPTransportSession -> SessionId -> TSessionSubs -> STM (Map RecipientId RcvQueueSub)
setSubsPending :: TransportSessionMode
-> SMPTransportSession
-> ByteString
-> TSessionSubs
-> STM (Map RecipientId RcvQueueSub)
setSubsPending TransportSessionMode
mode tSess :: SMPTransportSession
tSess@(UserId
uId, ProtoServer BrokerMsg
srv, Maybe ByteString
connId_) ByteString
sessId tss :: TSessionSubs
tss@(TSessionSubs TMap SMPTransportSession SessSubs
ss)
  | Bool
entitySession Bool -> Bool -> Bool
forall a. Eq a => a -> a -> Bool
== Maybe ByteString -> Bool
forall a. Maybe a -> Bool
isJust Maybe ByteString
connId_ =
      (UserId, ProtocolServer 'PSMP, Maybe ByteString)
-> TVar
     (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
-> STM (Maybe SessSubs)
forall k a. Ord k => k -> TMap k a -> STM (Maybe a)
TM.lookup SMPTransportSession
(UserId, ProtocolServer 'PSMP, Maybe ByteString)
tSess TMap SMPTransportSession SessSubs
TVar
  (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
ss STM (Maybe SessSubs)
-> (Maybe SessSubs -> STM (Map RecipientId RcvQueueSub))
-> STM (Map RecipientId RcvQueueSub)
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (SessSubs -> STM (Map RecipientId RcvQueueSub))
-> Maybe SessSubs -> STM (Map RecipientId RcvQueueSub)
forall {k} {a}.
(SessSubs -> STM (Map k a)) -> Maybe SessSubs -> STM (Map k a)
withSessSubs (SessSubs -> Maybe ByteString -> STM (Map RecipientId RcvQueueSub)
`setSubsPending_` Maybe ByteString
forall a. Maybe a
Nothing)
  | Bool
otherwise =
      (UserId, ProtocolServer 'PSMP, Maybe ByteString)
-> TVar
     (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
-> STM (Maybe SessSubs)
forall k a. Ord k => k -> TMap k a -> STM (Maybe a)
TM.lookupDelete SMPTransportSession
(UserId, ProtocolServer 'PSMP, Maybe ByteString)
tSess TMap SMPTransportSession SessSubs
TVar
  (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
ss STM (Maybe SessSubs)
-> (Maybe SessSubs -> STM (Map RecipientId RcvQueueSub))
-> STM (Map RecipientId RcvQueueSub)
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (SessSubs -> STM (Map RecipientId RcvQueueSub))
-> Maybe SessSubs -> STM (Map RecipientId RcvQueueSub)
forall {k} {a}.
(SessSubs -> STM (Map k a)) -> Maybe SessSubs -> STM (Map k a)
withSessSubs SessSubs -> STM (Map RecipientId RcvQueueSub)
setPendingChangeMode
  where
    entitySession :: Bool
entitySession = TransportSessionMode
mode TransportSessionMode -> TransportSessionMode -> Bool
forall a. Eq a => a -> a -> Bool
== TransportSessionMode
TSMEntity
    sessEntId :: a -> Maybe a
sessEntId = if Bool
entitySession then a -> Maybe a
forall a. a -> Maybe a
Just else Maybe a -> a -> Maybe a
forall a b. a -> b -> a
const Maybe a
forall a. Maybe a
Nothing
    withSessSubs :: (SessSubs -> STM (Map k a)) -> Maybe SessSubs -> STM (Map k a)
withSessSubs SessSubs -> STM (Map k a)
run = \case
      Maybe SessSubs
Nothing -> Map k a -> STM (Map k a)
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Map k a
forall k a. Map k a
M.empty
      Just SessSubs
s -> do
        Maybe ByteString
sessId' <- TVar (Maybe ByteString) -> STM (Maybe ByteString)
forall a. TVar a -> STM a
readTVar (TVar (Maybe ByteString) -> STM (Maybe ByteString))
-> TVar (Maybe ByteString) -> STM (Maybe ByteString)
forall a b. (a -> b) -> a -> b
$ SessSubs -> TVar (Maybe ByteString)
subsSessId SessSubs
s
        if ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
sessId Maybe ByteString -> Maybe ByteString -> Bool
forall a. Eq a => a -> a -> Bool
== Maybe ByteString
sessId' then SessSubs -> STM (Map k a)
run SessSubs
s else Map k a -> STM (Map k a)
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Map k a
forall k a. Map k a
M.empty
    setPendingChangeMode :: SessSubs -> STM (Map RecipientId RcvQueueSub)
setPendingChangeMode SessSubs
s = do
      Map RecipientId RcvQueueSub
subs <- Map RecipientId RcvQueueSub
-> Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub
forall k a. Ord k => Map k a -> Map k a -> Map k a
M.union (Map RecipientId RcvQueueSub
 -> Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
-> STM (Map RecipientId RcvQueueSub)
-> STM (Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TMap RecipientId RcvQueueSub -> STM (Map RecipientId RcvQueueSub)
forall a. TVar a -> STM a
readTVar (SessSubs -> TMap RecipientId RcvQueueSub
activeSubs SessSubs
s) STM (Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
-> STM (Map RecipientId RcvQueueSub)
-> STM (Map RecipientId RcvQueueSub)
forall a b. STM (a -> b) -> STM a -> STM b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> TMap RecipientId RcvQueueSub -> STM (Map RecipientId RcvQueueSub)
forall a. TVar a -> STM a
readTVar (SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs SessSubs
s)
      Bool -> STM () -> STM ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (Map RecipientId RcvQueueSub -> Bool
forall a. Map RecipientId a -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null Map RecipientId RcvQueueSub
subs) (STM () -> STM ()) -> STM () -> STM ()
forall a b. (a -> b) -> a -> b
$
        Map RecipientId RcvQueueSub -> (RcvQueueSub -> STM ()) -> STM ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Map RecipientId RcvQueueSub
subs ((RcvQueueSub -> STM ()) -> STM ())
-> (RcvQueueSub -> STM ()) -> STM ()
forall a b. (a -> b) -> a -> b
$ \RcvQueueSub
rq -> SMPTransportSession -> RcvQueueSub -> TSessionSubs -> STM ()
addPendingSub (UserId
uId, ProtoServer BrokerMsg
srv, ByteString -> Maybe ByteString
forall a. a -> Maybe a
sessEntId (RcvQueueSub -> ByteString
connId RcvQueueSub
rq)) RcvQueueSub
rq TSessionSubs
tss
      Map RecipientId RcvQueueSub -> STM (Map RecipientId RcvQueueSub)
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Map RecipientId RcvQueueSub
subs

setSubsPending_ :: SessSubs -> Maybe SessionId -> STM (Map RecipientId RcvQueueSub)
setSubsPending_ :: SessSubs -> Maybe ByteString -> STM (Map RecipientId RcvQueueSub)
setSubsPending_ SessSubs
s Maybe ByteString
sessId_ = do
  TVar (Maybe ByteString) -> Maybe ByteString -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar (SessSubs -> TVar (Maybe ByteString)
subsSessId SessSubs
s) Maybe ByteString
sessId_
  let as :: TMap RecipientId RcvQueueSub
as = SessSubs -> TMap RecipientId RcvQueueSub
activeSubs SessSubs
s
  Map RecipientId RcvQueueSub
subs <- TMap RecipientId RcvQueueSub -> STM (Map RecipientId RcvQueueSub)
forall a. TVar a -> STM a
readTVar TMap RecipientId RcvQueueSub
as
  Bool -> STM () -> STM ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (Map RecipientId RcvQueueSub -> Bool
forall a. Map RecipientId a -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null Map RecipientId RcvQueueSub
subs) (STM () -> STM ()) -> STM () -> STM ()
forall a b. (a -> b) -> a -> b
$ do
    TMap RecipientId RcvQueueSub
-> Map RecipientId RcvQueueSub -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar TMap RecipientId RcvQueueSub
as Map RecipientId RcvQueueSub
forall k a. Map k a
M.empty
    TMap RecipientId RcvQueueSub
-> (Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
-> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs SessSubs
s) ((Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
 -> STM ())
-> (Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
-> STM ()
forall a b. (a -> b) -> a -> b
$ Map RecipientId RcvQueueSub
-> Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub
forall k a. Ord k => Map k a -> Map k a -> Map k a
M.union Map RecipientId RcvQueueSub
subs
  Map RecipientId RcvQueueSub -> STM (Map RecipientId RcvQueueSub)
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Map RecipientId RcvQueueSub
subs

updateClientNotices :: SMPTransportSession -> [(RecipientId, Maybe Int64)] -> TSessionSubs -> STM ()
updateClientNotices :: SMPTransportSession
-> [(RecipientId, Maybe UserId)] -> TSessionSubs -> STM ()
updateClientNotices SMPTransportSession
tSess [(RecipientId, Maybe UserId)]
noticeIds TSessionSubs
ss = do
  SessSubs
s <- SMPTransportSession -> TSessionSubs -> STM SessSubs
getSessSubs SMPTransportSession
tSess TSessionSubs
ss
  TMap RecipientId RcvQueueSub
-> (Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
-> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs SessSubs
s) ((Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
 -> STM ())
-> (Map RecipientId RcvQueueSub -> Map RecipientId RcvQueueSub)
-> STM ()
forall a b. (a -> b) -> a -> b
$ \Map RecipientId RcvQueueSub
m -> (Map RecipientId RcvQueueSub
 -> (RecipientId, Maybe UserId) -> Map RecipientId RcvQueueSub)
-> Map RecipientId RcvQueueSub
-> [(RecipientId, Maybe UserId)]
-> Map RecipientId RcvQueueSub
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
foldl' (\Map RecipientId RcvQueueSub
m' (RecipientId
rcvId, Maybe UserId
clientNoticeId) -> (RcvQueueSub -> RcvQueueSub)
-> RecipientId
-> Map RecipientId RcvQueueSub
-> Map RecipientId RcvQueueSub
forall k a. Ord k => (a -> a) -> k -> Map k a -> Map k a
M.adjust (\RcvQueueSub
rq -> RcvQueueSub
rq {clientNoticeId}) RecipientId
rcvId Map RecipientId RcvQueueSub
m') Map RecipientId RcvQueueSub
m [(RecipientId, Maybe UserId)]
noticeIds

foldSessionSubs :: (a -> (SMPTransportSession, SessSubs) -> IO a) -> a -> TSessionSubs -> IO a
foldSessionSubs :: forall a.
(a -> (SMPTransportSession, SessSubs) -> IO a)
-> a -> TSessionSubs -> IO a
foldSessionSubs a -> (SMPTransportSession, SessSubs) -> IO a
f a
a = (a
 -> ((UserId, ProtocolServer 'PSMP, Maybe ByteString), SessSubs)
 -> IO a)
-> a
-> [((UserId, ProtocolServer 'PSMP, Maybe ByteString), SessSubs)]
-> IO a
forall (t :: * -> *) (m :: * -> *) b a.
(Foldable t, Monad m) =>
(b -> a -> m b) -> b -> t a -> m b
foldM a -> (SMPTransportSession, SessSubs) -> IO a
a
-> ((UserId, ProtocolServer 'PSMP, Maybe ByteString), SessSubs)
-> IO a
f a
a ([((UserId, ProtocolServer 'PSMP, Maybe ByteString), SessSubs)]
 -> IO a)
-> (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs
    -> [((UserId, ProtocolServer 'PSMP, Maybe ByteString), SessSubs)])
-> Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs
-> IO a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs
-> [((UserId, ProtocolServer 'PSMP, Maybe ByteString), SessSubs)]
forall k a. Map k a -> [(k, a)]
M.assocs (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs
 -> IO a)
-> (TSessionSubs
    -> IO
         (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs))
-> TSessionSubs
-> IO a
forall (m :: * -> *) b c a.
Monad m =>
(b -> m c) -> (a -> m b) -> a -> m c
<=< TVar
  (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
-> IO
     (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
forall a. TVar a -> IO a
readTVarIO (TVar
   (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
 -> IO
      (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs))
-> (TSessionSubs
    -> TVar
         (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs))
-> TSessionSubs
-> IO
     (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TSessionSubs -> TMap SMPTransportSession SessSubs
TSessionSubs
-> TVar
     (Map (UserId, ProtocolServer 'PSMP, Maybe ByteString) SessSubs)
sessionSubs

mapSubs :: (Map RecipientId RcvQueueSub -> a) -> SessSubs -> IO (a, a)
mapSubs :: forall a.
(Map RecipientId RcvQueueSub -> a) -> SessSubs -> IO (a, a)
mapSubs Map RecipientId RcvQueueSub -> a
f SessSubs
s = do
  Map RecipientId RcvQueueSub
active <- TMap RecipientId RcvQueueSub -> IO (Map RecipientId RcvQueueSub)
forall a. TVar a -> IO a
readTVarIO (TMap RecipientId RcvQueueSub -> IO (Map RecipientId RcvQueueSub))
-> TMap RecipientId RcvQueueSub -> IO (Map RecipientId RcvQueueSub)
forall a b. (a -> b) -> a -> b
$ SessSubs -> TMap RecipientId RcvQueueSub
activeSubs SessSubs
s
  Map RecipientId RcvQueueSub
pending <- TMap RecipientId RcvQueueSub -> IO (Map RecipientId RcvQueueSub)
forall a. TVar a -> IO a
readTVarIO (TMap RecipientId RcvQueueSub -> IO (Map RecipientId RcvQueueSub))
-> TMap RecipientId RcvQueueSub -> IO (Map RecipientId RcvQueueSub)
forall a b. (a -> b) -> a -> b
$ SessSubs -> TMap RecipientId RcvQueueSub
pendingSubs SessSubs
s
  (a, a) -> IO (a, a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Map RecipientId RcvQueueSub -> a
f Map RecipientId RcvQueueSub
active, Map RecipientId RcvQueueSub -> a
f Map RecipientId RcvQueueSub
pending)