@@ -4009,6 +4009,8 @@ data SessionState t m k a b = SessionState
4009-- is started with a tick duration specified by @tick@. This timer is used to
4009-- is started with a tick duration specified by @tick@. This timer is used to
4010-- detect session timeouts in the absence of new events.
4010-- detect session timeouts in the absence of new events.
4012+-- After @maxSessions@ limit is reached we start ejecting old sessions.
4012-- /Internal/
4014-- /Internal/
@@ -4019,10 +4021,12 @@ classifySessionsBy
4019 => Double -- ^ timer tick in seconds
4021 => Double -- ^ timer tick in seconds
4020 -> Double -- ^ session timeout in seconds
4022 -> Double -- ^ session timeout in seconds
4021 -> Bool -- ^ reset the timeout when an event is received
4023 -> Bool -- ^ reset the timeout when an event is received
4024+ -> Int -- ^ maximum number of sessions to cache
4022 -> Fold m a (Either b b) -- ^ Fold to be applied to session events
4025 -> Fold m a (Either b b) -- ^ Fold to be applied to session events
4023 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
4026 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
4024 -> t m (k, b) -- ^ session key, fold result
4027 -> t m (k, b) -- ^ session key, fold result
4025−classifySessionsBy tick timeout reset (Fold step initial extract) str =
4028+classifySessionsBy tick timeout reset maxSessions
4029+ (Fold step initial extract) str =
4026 concatMap (\session -> sessionOutputStream session)
4030 concatMap (\session -> sessionOutputStream session)
4027 $ scanlM' sstep szero stream
4031 $ scanlM' sstep szero stream
@@ -4113,10 +4117,11 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =
4113 -- Got a timer tick event
4117 -- Got a timer tick event
4114 -- XXX can we yield the entries without accumulating them?
4118 -- XXX can we yield the entries without accumulating them?
4115 sstep (session@SessionState{..}) Nothing = do
4119 sstep (session@SessionState{..}) Nothing = do
4120+ (hp', mp', out, count) <-
4116− (hp', mp', out, count) <- go sessionTimerHeap sessionKeyValueMap K.nil 0
4121+ go sessionTimerHeap sessionKeyValueMap K.nil sessionCount
4117 return $ session
4122 return $ session
4118 { sessionCurTime = curTime
4123 { sessionCurTime = curTime
4119− , sessionCount = sessionCount - count
4124+ , sessionCount = count
4120 , sessionTimerHeap = hp'
4125 , sessionTimerHeap = hp'
4121 , sessionKeyValueMap = mp'
4126 , sessionKeyValueMap = mp'
4122 , sessionOutputStream = out
4127 , sessionOutputStream = out
@@ -4125,11 +4130,12 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =
4127 curTime = addToAbsTime sessionCurTime tickMs
4132 curTime = addToAbsTime sessionCurTime tickMs
4133+ expired etime cnt = curTime >= etime || cnt > maxSessions
4128 go hp mp out cnt = do
4134 go hp mp out cnt = do
4129 let hres = H.uncons hp
4135 let hres = H.uncons hp
4130 case hres of
4136 case hres of
4131 Just (Entry expiry key, hp') -> do
4137 Just (Entry expiry key, hp') -> do
4132− if curTime >= expiry
4138+ if expired expiry
4134 let (r, mp') = Map.updateLookupWithKey
4140 let (r, mp') = Map.updateLookupWithKey
4135 (\_ _ -> Nothing) key mp
4141 (\_ _ -> Nothing) key mp
@@ -4137,14 +4143,14 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =
4137 Nothing -> go hp' mp' out cnt
4143 Nothing -> go hp' mp' out cnt
4138 Just (Tuple' latestTS acc) -> do
4144 Just (Tuple' latestTS acc) -> do
4139 let expiry' = addToAbsTime latestTS timeoutMs
4145 let expiry' = addToAbsTime latestTS timeoutMs
4140− if curTime >= expiry' || not reset
4146+ if expired expiry' || not reset
4142 sess <- extract acc
4148 sess <- extract acc
4143 let val = case sess of
4149 let val = case sess of
4144 Left x -> x
4150 Left x -> x
4145 Right x -> x
4151 Right x -> x
4146 go hp' mp' ((key, val) `K.cons` out)
4152 go hp' mp' ((key, val) `K.cons` out)
4147− (cnt + 1)
4153+ (cnt - 1)
4149 -- reset the session timeout
4155 -- reset the session timeout
4150 -- XXX purge anyway if the session size
4156 -- XXX purge anyway if the session size