@@ -200,13 +200,13 @@ module Streamly.Internal.Data.Stream.IsStream.Nesting
200 -- -- *** Splitting By Streams
200 -- -- *** Splitting By Streams
201 -- -- | Splitting a stream using another stream as separator.
201 -- -- | Splitting a stream using another stream as separator.
203− -- ** Windowed Classification
203+ -- ** Keyed Window Classification
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
206 -- can be associated with a key, all events associated with a particular
206 -- can be associated with a key, all events associated with a particular
207 -- key in the window can be folded to a single result. The stream is split
207 -- 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
208+ -- into windows of specified size, the window termination can be
209− -- the closing flag is specified in the input stream.
209+ -- dynamically controlled by the fold.
211 -- The term "chunk" is used for a space window and the term "session" is
211 -- The term "chunk" is used for a space window and the term "session" is
212 -- used for a time window.
212 -- used for a time window.
@@ -1902,32 +1902,41 @@ data SessionState t m k a b = SessionState
1904-- XXX Perhaps we should use an "Event a" type to represent timestamped data.
1904-- XXX Perhaps we should use an "Event a" type to represent timestamped data.
1905−-- | @classifySessionsBy tick timeout idle pred f stream@ groups timestamped
1906+-- | @classifySessionsBy tick timeout keepalive predicate fold stream@ classifies
1906−-- events in an input event stream into sessions based on a session key. Each
1907+-- an input event @stream@ consisting of @(key, value, timestamp)@ into
1907−-- element in the input stream is an event consisting of a triple @(session key,
1908+-- sessions based on the @key@, folding all the values corresponding to the
1908−-- sesssion data, timestamp)@. @session key@ is a key that uniquely identifies
1909+-- same key into a session using the supplied @fold@.
1909−-- the session. All the events belonging to a session are folded using the fold
1910+--
1910−-- @f@ until the fold terminates or a timeout has occurred. The session key and
1911+-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
1911−-- the result of the fold are emitted in the output stream when the session is
1912+-- session key and the folded value is emitted in the output stream. The
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.
1913+-- timeout is measured from the first event in the session. If the @keepalive@
1916−-- When @idle@ is 'True' then the timeout is an idle timeout, it is reset after
1914+-- option is set to 'True' the timeout is reset to 0 whenever an event is
1917−-- every event received in the session.
1915+-- received.
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
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
1921−-- of current time is maintained by a monotonic event time clock using the
1919+-- current time is maintained by a monotonic event time clock using the
1922-- timestamps seen in the input stream. The latest timestamp seen till now is
1920-- timestamps seen in the input stream. The latest timestamp seen till now is
1923-- used as the base for the current time. When no new events are seen, a timer
1921-- 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
1925-- detect session timeouts in the absence of new events.
1923-- detect session timeouts in the absence of new events.
1927−-- The predicate @pred@ is invoked with the current session count, if it
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
1928−-- returns 'True' a session is ejected from the session cache before inserting
1927+-- oldest session is ejected before inserting a new session.
1929−-- a new session. This could be useful to alert or eject sessions when the
1928+--
1930−-- number of sessions becomes too high.
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
1935+-- $ (,) \<$> Stream.fromList [1,2,3] \<*> Stream.fromList [\'a',\'b',\'c']
1932-- /Pre-release/
1941-- /Pre-release/
@@ -2152,12 +2161,8 @@ classifySessionsBy tick tmout reset ejectPred
2152 liftIO $ threadDelay (round $ tick * 1000000)
2161 liftIO $ threadDelay (round $ tick * 1000000)
2153 return Nothing
2162 return Nothing
2155−-- | Like 'classifySessionsOf' but the session is kept alive if an event is
2164+-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
2156−-- received within the session window. The session times out and gets closed
2165+-- option set to 'True'.
2157−-- only if no event is received within the specified session window size.
2159−-- If the ejection predicate returns 'True', the session that was idle for
2160−-- the longest time is ejected before inserting a new session.
2163-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred
2168-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred
@@ -2204,38 +2209,13 @@ classifyChunksOf
2204classifyChunksOf wsize = classifyChunksBy wsize False
2209classifyChunksOf wsize = classifyChunksBy wsize False
2207−-- | Split the stream into fixed size time windows of specified interval in
2212+-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
2208−-- seconds. Within each such window, fold the elements in sessions identified
2213+-- option set to 'False'.
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.
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.
2217−-- If the ejection predicate returns 'True', the session with the longest
2218−-- lifetime is ejected before inserting a new session.
2221-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred
2216-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred
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
2231−-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList [1,2,3]
2233−-- (1,Right [1,2,3])
2234−-- (2,Right [1,2,3])
2235−-- (3,Right [1,2,3])
2239-- /Pre-release/
2219-- /Pre-release/
2241{-# INLINABLE classifySessionsOf #-}
2221{-# INLINABLE classifySessionsOf #-}