@@ -202,14 +202,13 @@ module Streamly.Internal.Data.Stream.IsStream.Nesting
202202
203203 -- ** Keyed Window Classification
204204
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
207− -- key in the window can be folded to a single result. The stream is split
208− -- into windows of specified size, the window termination can be
209− -- dynamically controlled by the fold.
205+ -- | Split the stream into chunks or windows by position or time. Each
206+ -- window can be associated with a key, all events associated with a
207+ -- particular key in the window can be folded to a single result. The
208+ -- window termination can be dynamically controlled by the fold.
210209 --
211− -- The term "chunk" is used for a space window and the term "session" is
212− -- used for a time window.
210+ -- The term "chunk" is used for a window defined by position of elements
211+ -- and the term "session" is used for a time window.
213212
214213 -- *** Tumbling Windows
215214 -- | A new window starts after the previous window is finished.
@@ -1901,12 +1900,10 @@ data SessionState t m k a b = SessionState
19011900 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
19021901 }
19031902
1904−-- XXX Perhaps we should use an "Event a" type to represent timestamped data.
1905−
19061903-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
1907−-- classifies an input event @stream@ consisting of @(key, value, timestamp)@
1908−-- into sessions based on the @key@, folding all the values corresponding to
1909−-- the same key into a session using the supplied @fold@.
1904+-- classifies an input event @stream@ consisting of @(timestamp, (key,
1905+-- value))@ into sessions based on the @key@, folding all the values
1906+-- corresponding to the same key into a session using the supplied @fold@.
19101907--
19111908-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
19121909-- session key and the folded value is emitted in the output stream. The
@@ -1928,11 +1925,10 @@ data SessionState t m k a b = SessionState
19281925--
19291926-- >>> :{
19301927-- Stream.mapM_ print
1931−-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.takeLE 3 Fold.toList)
1932−-- $ Stream.map (\\(ts,(k,a)) -> (k, a, ts))
1928+-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
19331929-- $ Stream.timestamped
19341930-- $ Stream.delay 1
1935−-- $ (,) \<$> Stream.fromList [1,2,3] \<*> Stream.fromList [\'a',\'b',\'c']
1931+-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
19361932-- :}
19371933-- (1,"abc")
19381934-- (2,"abc")
@@ -1947,8 +1943,8 @@ classifySessionsBy
19471943 -> Bool -- ^ reset the timer when an event is received
19481944 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
19491945 -> Double -- ^ session timeout in seconds
1950− -> Fold m a b -- ^ Fold to be applied to session events
1951− -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
1946+ -> Fold m a b -- ^ Fold to be applied to session data
1947+ -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
19521948 -> t m (k, b) -- ^ session key, fold result
19531949classifySessionsBy tick reset ejectPred tmout
19541950 (Fold step initial extract) str =
@@ -1979,7 +1975,7 @@ classifySessionsBy tick reset ejectPred tmout
19791975 -- We use the first strategy as of now.
19801976
19811977 -- Got a new stream input element
1982− sstep session@SessionState{..} (Just (key, value, timestamp)) = do
1978+ sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
19831979 -- XXX we should use a heap in pinned memory to scale it to a large
19841980 -- size
19851981 --
@@ -2176,7 +2172,7 @@ classifyKeepAliveSessions ::
21762172 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
21772173 -> Double -- ^ session inactive timeout
21782174 -> Fold m a b -- ^ Fold to be applied to session payload data
2179− -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
2175+ -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
21802176 -> t m (k, b)
21812177classifyKeepAliveSessions = classifySessionsBy 1 True
21822178
@@ -2222,8 +2218,8 @@ classifySessionsOf ::
22222218 (IsStream t, MonadAsync m, Ord k)
22232219 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
22242220 -> Double -- ^ time window size
2225− -> Fold m a b -- ^ Fold to be applied to session events
2226− -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
2221+ -> Fold m a b -- ^ Fold to be applied to session data
2222+ -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
22272223 -> t m (k, b)
22282224classifySessionsOf = classifySessionsBy 1 False
22292225