@@ -200,13 +200,13 @@ module Streamly.Internal.Data.Stream.IsStream.Nesting
200200 -- -- *** Splitting By Streams
201201 -- -- | Splitting a stream using another stream as separator.
202202
203− -- ** Windowed Classification
203+ -- ** Keyed Window Classification
204204
205− -- | Split the stream into windows or chunks in space or time. Each window
205+ -- | Split the stream into chunks or windows in space or time. Each window
206206 -- can be associated with a key, all events associated with a particular
207207 -- key in the window can be folded to a single result. The stream is split
208− -- into windows of specified size, the window can be terminated early if
209− -- the closing flag is specified in the input stream.
208+ -- into windows of specified size, the window termination can be
209+ -- dynamically controlled by the fold.
210210 --
211211 -- The term "chunk" is used for a space window and the term "session" is
212212 -- used for a time window.
@@ -1902,32 +1902,41 @@ data SessionState t m k a b = SessionState
19021902 }
19031903
19041904-- XXX Perhaps we should use an "Event a" type to represent timestamped data.
1905−-- | @classifySessionsBy tick timeout idle pred f stream@ groups timestamped
1906−-- events in an input event stream into sessions based on a session key. Each
1907−-- element in the input stream is an event consisting of a triple @(session key,
1908−-- sesssion data, timestamp)@. @session key@ is a key that uniquely identifies
1909−-- the session. All the events belonging to a session are folded using the fold
1910−-- @f@ until the fold terminates or a timeout has occurred. The session key and
1911−-- the result of the fold are emitted in the output stream when the session is
1912−-- purged.
1913−--
1914−-- When @idle@ is 'False', @timeout@ is the maximum lifetime of a session in
1915−-- seconds, measured from the @timestamp@ of the first event in that session.
1916−-- When @idle@ is 'True' then the timeout is an idle timeout, it is reset after
1917−-- every event received in the session.
1918−--
1919−-- @timestamp@ in an event characterizes the time when the input event was
1920−-- generated, this is an absolute time measured from some @Epoch@. The notion
1921−-- of current time is maintained by a monotonic event time clock using the
1905+
1906+-- | @classifySessionsBy tick timeout keepalive predicate fold stream@ classifies
1907+-- an input event @stream@ consisting of @(key, value, timestamp)@ into
1908+-- sessions based on the @key@, folding all the values corresponding to the
1909+-- same key into a session using the supplied @fold@.
1910+--
1911+-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
1912+-- session key and the folded value is emitted in the output stream. The
1913+-- timeout is measured from the first event in the session. If the @keepalive@
1914+-- option is set to 'True' the timeout is reset to 0 whenever an event is
1915+-- received.
1916+--
1917+-- The @timestamp@ in the input stream is an absolute time from some epoch,
1918+-- characterizing the time when the input event was generated. The notion of
1919+-- current time is maintained by a monotonic event time clock using the
19221920-- timestamps seen in the input stream. The latest timestamp seen till now is
19231921-- used as the base for the current time. When no new events are seen, a timer
1924−-- is started with a tick duration specified by @tick@. This timer is used to
1922+-- is started with a clock resolution of @tick@ seconds. This timer is used to
19251923-- detect session timeouts in the absence of new events.
19261924--
1927−-- The predicate @pred@ is invoked with the current session count, if it
1928−-- returns 'True' a session is ejected from the session cache before inserting
1929−-- a new session. This could be useful to alert or eject sessions when the
1930−-- number of sessions becomes too high.
1925+-- To ensure an upper bound on the memory used the number of sessions can be
1926+-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1927+-- oldest session is ejected before inserting a new session.
1928+--
1929+-- >>> :{
1930+-- Stream.mapM_ print
1931+-- $ Stream.classifySessionsBy 1 3 False (const (return False)) (Fold.takeLE 3 Fold.toList)
1932+-- $ Stream.map (\\(ts,(k,a)) -> (k, a, ts))
1933+-- $ Stream.timestamped
1934+-- $ Stream.delay 1
1935+-- $ (,) \<$> Stream.fromList [1,2,3] \<*> Stream.fromList [\'a',\'b',\'c']
1936+-- :}
1937+-- (1,"abc")
1938+-- (2,"abc")
1939+-- (3,"abc")
19311940--
19321941-- /Pre-release/
19331942--
@@ -2152,12 +2161,8 @@ classifySessionsBy tick tmout reset ejectPred
21522161 liftIO $ threadDelay (round $ tick * 1000000)
21532162 return Nothing
21542163
2155−-- | Like 'classifySessionsOf' but the session is kept alive if an event is
2156−-- received within the session window. The session times out and gets closed
2157−-- only if no event is received within the specified session window size.
2158−--
2159−-- If the ejection predicate returns 'True', the session that was idle for
2160−-- the longest time is ejected before inserting a new session.
2164+-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
2165+-- option set to 'True'.
21612166--
21622167-- @
21632168-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred
@@ -2204,38 +2209,13 @@ classifyChunksOf
22042209classifyChunksOf wsize = classifyChunksBy wsize False
22052210-}
22062211
2207−-- | Split the stream into fixed size time windows of specified interval in
2208−-- seconds. Within each such window, fold the elements in sessions identified
2209−-- by the session keys. The fold result is emitted in the output stream if the
2210−-- fold returns a 'Left' result or if the time window ends.
2211−--
2212−-- Session @timestamp@ in the input stream is an absolute time from some epoch,
2213−-- characterizing the time when the input element was generated. To detect
2214−-- session window end, a monotonic event time clock is maintained synced with
2215−-- the timestamps with a clock resolution of 1 second.
2216−--
2217−-- If the ejection predicate returns 'True', the session with the longest
2218−-- lifetime is ejected before inserting a new session.
2212+-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
2213+-- option set to 'False'.
22192214--
22202215-- @
22212216-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred
22222217-- @
22232218--
2224−-- @
2225−-- > :{
2226−-- Stream.mapM_ print
2227−-- $ Stream.classifySessionsOf 3 (const (return False)) (fmap Right Fold.toList)
2228−-- $ Stream.map (\\(ts,(k,a)) -> (k, a, ts))
2229−-- $ Stream.timestamped
2230−-- $ Stream.delay 1
2231−-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList [1,2,3]
2232−-- :}
2233−-- (1,Right [1,2,3])
2234−-- (2,Right [1,2,3])
2235−-- (3,Right [1,2,3])
2236−--
2237−-- @
2238−--
22392219-- /Pre-release/
22402220--
22412221{-# INLINABLE classifySessionsOf #-}