@@ -3987,20 +3987,19 @@ data SessionState t m k a b = SessionState
39873987
39883988#undef Type
39893989
3990−-- | @classifySessionsBy tick timeout reset f stream@ groups timed events in an
3990+-- | @classifySessionsBy tick timeout idle f stream@ groups timed events in an
39913991-- input event stream into sessions based on a session key. Each element in the
3992−-- stream is an event consisting of a 4-tuple @(session key, sesssion data,
3993−-- isClosingEvent, timestamp)@. @session key@ is a key that uniquely
3994−-- identifies a session. All the events belonging to a session are folded using
3995−-- the fold @f@ until an event with @isClosingEvent@ flag set arrives or a
3996−-- timeout has occurred. The session key and the result of the fold are
3997−-- emitted in the output stream when the session is purged.
3998−--
3999−-- When @reset@ is 'False', @timeout@ is the maximum lifetime of a session in
3992+-- stream is an event consisting of a triple @(session key, sesssion data,
3993+-- timestamp)@. @session key@ is a key that uniquely identifies the session.
3994+-- All the events belonging to a session are folded using the fold @f@ until
3995+-- the fold returns a 'Left' result or a timeout has occurred. The session key
3996+-- and the result of the fold are emitted in the output stream when the session
3997+-- is purged.
3998+--
3999+-- When @idle@ is 'False', @timeout@ is the maximum lifetime of a session in
40004000-- seconds, measured from the @timestamp@ of the first event in that session.
4001−-- When "reset" is 'True' then the timeout is reset after every event received
4002−-- in the session, in other words timeout is measured from the timestamp of the
4003−-- last event in the session.
4001+-- When "idle" is 'True' then the timeout is an idle timeout, it is reset after
4002+-- every event received in the session.
40044003--
40054004-- @timestamp@ in an event characterizes the time when the input event was
40064005-- generated, this is an absolute time measured from some @Epoch@. The notion
@@ -4012,14 +4011,16 @@ data SessionState t m k a b = SessionState
40124011--
40134012-- /Internal/
40144013--
4014+
4015+-- XXX Perhaps we should use an "Event a" type to represent timestamped data.
40154016{-# INLINABLE classifySessionsBy #-}
40164017classifySessionsBy
40174018 :: (IsStream t, MonadAsync m, Ord k)
40184019 => Double -- ^ timer tick in seconds
40194020 -> Double -- ^ session timeout in seconds
40204021 -> Bool -- ^ reset the timeout when an event is received
4021− -> Fold m a b -- ^ Fold to be applied to session events
4022− -> t m (k, a, Bool, AbsTime) -- ^ session key, data, timestamp, close event
4022+ -> Fold m a (Either b b) -- ^ Fold to be applied to session events
4023+ -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
40234024 -> t m (k, b) -- ^ session key, fold result
40244025classifySessionsBy tick timeout reset (Fold step initial extract) str =
40254026 concatMap (\session -> sessionOutputStream session)
@@ -4048,7 +4049,7 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =
40484049 -- old ones.
40494050 --
40504051 -- Got a new stream input element
4051− sstep (session@SessionState{..}) (Just (key, a, closing, ts)) =
4052+ sstep (session@SessionState{..}) (Just (key, a, ts)) = do
40524053 -- XXX we should use a heap in pinned memory to scale it to a large
40534054 -- size
40544055 --
@@ -4071,42 +4072,43 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =
40714072 Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
40724073 new <- step old a
40734074 return $ Tuple' ts new
4074− in if closing
4075− then do
4076− let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key
4077− sessionKeyValueMap
4078− Tuple' _ acc <- accumulate r
4079− res <- extract acc
4075+ mOld = Map.lookup key sessionKeyValueMap
4076+
4077+ acc@(Tuple' _ fres) <- accumulate mOld
4078+ res <- extract fres
4079+ case res of
4080+ Left x -> do
4081+ let mp' = case mOld of
4082+ Nothing -> sessionKeyValueMap
4083+ Just _ -> Map.delete key sessionKeyValueMap
40804084 return $ session
40814085 { sessionCurTime = curTime
40824086 , sessionEventTime = curTime
40834087 , sessionCount = sessionCount - 1
40844088 , sessionKeyValueMap = mp'
4085− , sessionOutputStream = yield (key, res)
4089+ , sessionOutputStream = yield (key, x)
4090+ }
4091+ Right _ -> do
4092+ let mp' = Map.insert key acc sessionKeyValueMap
4093+ let (hp', count) =
4094+ case mOld of
4095+ Nothing ->
4096+ let expiry = addToAbsTime ts timeoutMs
4097+ in (H.insert (Entry expiry key)
4098+ sessionTimerHeap, sessionCount + 1)
4099+ Just _ -> (sessionTimerHeap, sessionCount)
4100+ -- Event time is maintained as monotonically increasing
4101+ -- time. If we have lagged behind any of the timestamps
4102+ -- seen then we increase it to match the latest time seen
4103+ -- in the timestamps. We also increase it on timer ticks.
4104+ return $ SessionState
4105+ { sessionCurTime = curTime
4106+ , sessionEventTime = curTime
4107+ , sessionCount = count
4108+ , sessionTimerHeap = hp'
4109+ , sessionKeyValueMap = mp'
4110+ , sessionOutputStream = K.nil
40864111 }
4087− else do
4088− let r = Map.lookup key sessionKeyValueMap
4089− acc <- accumulate r
4090− let mp' = Map.insert key acc sessionKeyValueMap
4091− let (hp', count) =
4092− case r of
4093− Nothing ->
4094− let expiry = addToAbsTime ts timeoutMs
4095− in (H.insert (Entry expiry key)
4096− sessionTimerHeap, sessionCount + 1)
4097− Just _ -> (sessionTimerHeap, sessionCount)
4098− -- Event time is maintained as monotonically increasing
4099− -- time. If we have lagged behind any of the timestamps
4100− -- seen then we increase it to match the latest time seen
4101− -- in the timestamps. We also increase it on timer ticks.
4102− return $ SessionState
4103− { sessionCurTime = curTime
4104− , sessionEventTime = curTime
4105− , sessionCount = count
4106− , sessionTimerHeap = hp'
4107− , sessionKeyValueMap = mp'
4108− , sessionOutputStream = K.nil
4109− }
41104112
41114113 -- Got a timer tick event
41124114 -- XXX can we yield the entries without accumulating them?
@@ -4138,7 +4140,10 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =
41384140 if curTime >= expiry' || not reset
41394141 then do
41404142 sess <- extract acc
4141− go hp' mp' ((key, sess) `K.cons` out)
4143+ let val = case sess of
4144+ Left x -> x
4145+ Right x -> x
4146+ go hp' mp' ((key, val) `K.cons` out)
41424147 (cnt + 1)
41434148 else
41444149 -- reset the session timeout
@@ -4171,8 +4176,8 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =
41714176classifyKeepAliveSessions
41724177 :: (IsStream t, MonadAsync m, Ord k)
41734178 => Double -- ^ session inactive timeout
4174− -> Fold m a b -- ^ Fold to be applied to session payload data
4175− -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
4179+ -> Fold m a (Either b b) -- ^ Fold to be applied to session payload data
4180+ -> t m (k, a, AbsTime) -- ^ session key, data, close flag, timestamp
41764181 -> t m (k, b)
41774182classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
41784183
@@ -4205,10 +4210,9 @@ classifyChunksOf wsize = classifyChunksBy wsize False
42054210-}
42064211
42074212-- | Split the stream into fixed size time windows of specified interval in
4208−-- seconds. Within each such window, fold the elements in buckets identified by
4209−-- the keys. A particular bucket fold can be terminated early if a closing flag
4210−-- is encountered in an element for that key. Once a fold is terminated the key
4211−-- and value for that bucket are emitted in the output stream.
4213+-- seconds. Within each such window, fold the elements in sessions identified
4214+-- by the session keys. The fold result is emitted in the output stream if the
4215+-- fold returns a 'Left' result or if the time window ends.
42124216--
42134217-- Session @timestamp@ in the input stream is an absolute time from some epoch,
42144218-- characterizing the time when the input element was generated. To detect
@@ -4225,8 +4229,8 @@ classifyChunksOf wsize = classifyChunksBy wsize False
42254229classifySessionsOf
42264230 :: (IsStream t, MonadAsync m, Ord k)
42274231 => Double -- ^ time window size
4228− -> Fold m a b -- ^ Fold to be applied to window events
4229− -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
4232+ -> Fold m a (Either b b) -- ^ Fold to be applied to window events
4233+ -> t m (k, a, AbsTime) -- ^ window key, data, close flag, timestamp
42304234 -> t m (k, b)
42314235classifySessionsOf interval = classifySessionsBy 1 interval False
42324236