mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-09-01 18:08:36 +00:00
Evgeny PoberezkinandAlexander Bondarenko
91cc48aabe
* agent: do not mark subscriptions on expired sessions as active, do mark delayed subscriptions as active on the same session, SUBOK response in the next SMP protocol version
* client: prevent sub actions from zombie sessions (#1122)
* client: prevent sub actions from zombie sessions
* error handling
* add AERR to pass background errors to client
* switch to activeClientSession
* put closeClient under activeClientSession
* rename
* remove AERR, do not skip processing
* move check and state update to one transaction
* catch extra UPs
* fix
* check queue is still pending before making it active
---------
Co-authored-by: Evgeny Poberezkin <evgeny@poberezkin.com>
* do not forward agent error
* revert not expiring sending subs
* fixes
* track subscription responses better
* add pending connection
* Revert "revert not expiring sending subs"
This reverts commit 4310a69391.
* do not expire sending commands
* rename
* fix race
* function
---------
Co-authored-by: Alexander Bondarenko <486682+dpwiz@users.noreply.github.com>
43 lines
1.5 KiB
Haskell
43 lines
1.5 KiB
Haskell
{-# LANGUAGE LambdaCase #-}
|
|
{-# LANGUAGE NamedFieldPuns #-}
|
|
{-# LANGUAGE ScopedTypeVariables #-}
|
|
|
|
module Simplex.Messaging.Session where
|
|
|
|
import Control.Concurrent.STM
|
|
import Control.Monad
|
|
import Data.Composition ((.:.))
|
|
import Data.Functor (($>))
|
|
import Simplex.Messaging.TMap (TMap)
|
|
import qualified Simplex.Messaging.TMap as TM
|
|
import Simplex.Messaging.Util (($>>=))
|
|
|
|
data SessionVar a = SessionVar
|
|
{ sessionVar :: TMVar a,
|
|
sessionVarId :: Int
|
|
}
|
|
|
|
getSessVar :: forall k a. Ord k => TVar Int -> k -> TMap k (SessionVar a) -> STM (Either (SessionVar a) (SessionVar a))
|
|
getSessVar sessSeq sessKey vs = maybe (Left <$> newSessionVar) (pure . Right) =<< TM.lookup sessKey vs
|
|
where
|
|
newSessionVar :: STM (SessionVar a)
|
|
newSessionVar = do
|
|
sessionVar <- newEmptyTMVar
|
|
sessionVarId <- stateTVar sessSeq $ \next -> (next, next + 1)
|
|
let v = SessionVar {sessionVar, sessionVarId}
|
|
TM.insert sessKey v vs
|
|
pure v
|
|
|
|
removeSessVar :: Ord k => SessionVar a -> k -> TMap k (SessionVar a) -> STM ()
|
|
removeSessVar = void .:. removeSessVar'
|
|
{-# INLINE removeSessVar #-}
|
|
|
|
removeSessVar' :: Ord k => SessionVar a -> k -> TMap k (SessionVar a) -> STM Bool
|
|
removeSessVar' v sessKey vs =
|
|
TM.lookup sessKey vs >>= \case
|
|
Just v' | sessionVarId v == sessionVarId v' -> TM.delete sessKey vs $> True
|
|
_ -> pure False
|
|
|
|
tryReadSessVar :: Ord k => k -> TMap k (SessionVar a) -> STM (Maybe a)
|
|
tryReadSessVar sessKey vs = TM.lookup sessKey vs $>>= (tryReadTMVar . sessionVar)
|