@@ -0,0 +1,862 @@
2+-- Module : Streamly.Internal.Data.Stream.Time
3+-- Copyright : (c) 2020 Composewell Technologies
4+-- License : BSD-3-Clause
5+-- Maintainer : streamly@composewell.com
6+-- Stability : experimental
10+module Streamly.Internal.Data.Stream.Time
20+ , classifySessionsByGeneric
23+ , classifyKeepAliveSessions
27+ , sampleIntervalStart
34+import Control.Concurrent (threadDelay)
35+import Control.Exception (assert)
36+import Control.Monad.IO.Class (MonadIO(..))
37+import Data.Heap (Entry(..))
38+import Data.Kind (Type)
40+import Data.Maybe (isNothing)
41+import Data.Proxy (Proxy(..))
42+import Streamly.Data.Fold (Fold)
43+import Streamly.Internal.Data.Fold.Type (Fold (..))
44+import Streamly.Internal.Data.IsMap (IsMap(..))
45+import Streamly.Internal.Data.Stream.Type (Stream)
46+import Streamly.Internal.Data.Time.Units
53+import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)
55+import qualified Data.Heap as H
56+import qualified Streamly.Data.Fold as Fold
57+import qualified Streamly.Data.Stream as Stream
58+import qualified Streamly.Data.Unfold as Unfold
59+import qualified Streamly.Internal.Data.Fold as Fold (Step(..))
60+import qualified Streamly.Internal.Data.IsMap as IsMap
61+import qualified Streamly.Internal.Data.Stream as Stream
68+import Streamly.Internal.Data.Stream.Concurrent
72+-- Imports for example snippets in this module.
75+-- >>> import Control.Concurrent (threadDelay)
76+-- >>> import qualified Streamly.Data.Array.Unboxed as Array
77+-- >>> import qualified Streamly.Data.Fold as Fold
78+-- >>> import qualified Streamly.Data.Parser as Parser
79+-- >>> import qualified Streamly.Data.Stream as Stream
80+-- >>> import qualified Streamly.Internal.Data.Stream as Stream (timestamped)
81+-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Concur
82+-- >>> import qualified Streamly.Internal.Data.Stream.Time as Concur
83+-- >>> import Prelude hiding (concatMap, concat)
86+-- threadDelay (n * 1000000) -- sleep for n seconds
87+-- putStrLn (show n ++ " sec") -- print "n sec"
91+--------------------------------------------------------------------------------
93+--------------------------------------------------------------------------------
95+-- | Intersperse a monadic action into the input stream after every @n@
98+-- >>> Stream.fold Fold.drain $ Concur.interjectSuffix 1.05 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
102+{-# INLINE interjectSuffix #-}
103+interjectSuffix :: MonadAsync m => Double -> m a -> Stream m a -> Stream m a
104+interjectSuffix n f xs = parallelFst [xs, repeatM timed]
108+ timed = liftIO (threadDelay (round $ n * 1000000)) >> f
109+ repeatM = Stream.sequence . Stream.repeat
111+-- XXX Notes from D.takeByTime (which was removed)
112+-- XXX using getTime in the loop can be pretty expensive especially for
113+-- computations where iterations are lightweight. We have the following
116+-- 1) Run a timeout thread updating a flag asynchronously and check that
117+-- flag here, that way we can have a cheap termination check.
119+-- 2) Use COARSE clock to get time with lower resolution but more efficiently.
121+-- 3) Use rdtscp/rdtsc to get time directly from the processor, compute the
122+-- termination value of rdtsc in the beginning and then in each iteration just
123+-- get rdtsc and check if we should terminate.
126+-- | @takeInterval duration@ yields stream elements upto specified time
127+-- @duration@ in seconds. The duration starts when the stream is evaluated for
128+-- the first time, before the first element is yielded. The time duration is
129+-- checked before generating each element, if the duration has expired the
132+-- The total time taken in executing the stream is guaranteed to be /at least/
133+-- @duration@, however, because the duration is checked before generating an
134+-- element, the upper bound is indeterminate and depends on the time taken in
135+-- generating and processing the last element.
137+-- No element is yielded if the duration is zero. At least one element is
138+-- yielded if the duration is non-zero.
142+{-# INLINE takeInterval #-}
143+takeInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
146+ . Stream.takeWhile isNothing
147+ . interjectSuffix d (return Nothing) . fmap Just
149+-- | @dropInterval duration@ drops stream elements until specified @duration@ in
150+-- seconds has passed. The duration begins when the stream is evaluated for the
151+-- first time. The time duration is checked /after/ generating a stream element,
152+-- the element is yielded if the duration has expired otherwise it is dropped.
154+-- The time elapsed before starting to generate the first element is /at most/
155+-- @duration@, however, because the duration expiry is checked after the
156+-- element is generated, the lower bound is indeterminate and depends on the
157+-- time taken in generating an element.
159+-- All elements are yielded if the duration is zero.
163+{-# INLINE dropInterval #-}
164+dropInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
167+ . Stream.dropWhile isNothing
168+ . interjectSuffix d (return Nothing) . fmap Just
170+-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
171+-- XXX add this example after fixing the serial stream rate control
173+-- | Group the input stream into windows of @n@ second each and then fold each
174+-- group using the provided fold function.
176+-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
177+-- [...,...,...,...,...]
180+{-# INLINE intervalsOf #-}
181+intervalsOf :: MonadAsync m => Double -> Fold m a b -> Stream m a -> Stream m b
184+ (Fold.takeEndBy isNothing (Fold.catMaybes f))
185+ (interjectSuffix n (return Nothing) (fmap Just xs))
187+-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
188+-- parallel threads and resetting it whenever a span is emitted.
190+-- | Like 'chunksOf' but if the chunk is not completed within the specified
191+-- time interval then emit whatever we have collected till now. The chunk
192+-- timeout is reset whenever a chunk is emitted. The granularity of the clock
195+-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
196+-- >>> f = Stream.fold (Fold.drainBy print) $ Stream.chunksOfTimeout 5 1 Fold.toList s
199+{-# INLINE chunksOfTimeout #-}
200+chunksOfTimeout :: MonadAsync m
201+ => Int -> Double -> Fold m a b -> Stream m a -> Stream m b
202+chunksOfTimeout n timeout f =
204+ . classifySessionsBy
205+ 0.1 False (const (return False)) timeout (Fold.take n f)
206+ . Stream.timestamped
210+------------------------------------------------------------------------------
211+-- Windowed classification
212+------------------------------------------------------------------------------
214+-- TODO: To mark the position in space or time we can have Indexed or
215+-- TimeStamped types. That can make it easy to deal with the position indices
218+------------------------------------------------------------------------------
219+-- Keyed Sliding Windows
220+------------------------------------------------------------------------------
223+{-# INLINABLE classifySlidingChunks #-}
224+classifySlidingChunks
225+ :: (IsStream t, MonadAsync m, Ord k)
226+ => Int -- ^ window size
227+ -> Int -- ^ window slide
228+ -> Fold m a b -- ^ Fold to be applied to window events
229+ -> t m (k, a, Bool) -- ^ window key, data, close event
231+classifySlidingChunks wsize wslide (Fold step initial extract) str
234+-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
235+-- slide the send window when an ack is received and we slide the receive
236+-- window when a sequence is complete. Sliding is stateful in case of TCP,
237+-- sliding releases the send buffer or makes data available to the user from
238+-- the receive buffer.
239+{-# INLINABLE classifySlidingSessions #-}
240+classifySlidingSessions
241+ :: (IsStream t, MonadAsync m, Ord k)
242+ => Double -- ^ timer tick in seconds
243+ -> Double -- ^ time window size
244+ -> Double -- ^ window slide
245+ -> Fold m a b -- ^ Fold to be applied to window events
246+ -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
248+classifySlidingSessions tick interval slide (Fold step initial extract) str
252+------------------------------------------------------------------------------
253+-- Sliding Window Buffers
254+------------------------------------------------------------------------------
256+-- These buffered versions could be faster than concurrent incremental folds of
257+-- all overlapping windows as in many cases we may not need all the values to
258+-- compute the fold, we can just compute the result using the old value and new
259+-- value. However, we may need the buffer once in a while, for example for
260+-- string search we usually compute the hash incrementally but when the hash
261+-- matches the hash of the pattern we need to compare the whole string.
263+-- XXX we should be able to implement sequence based splitting combinators
264+-- using this combinator.
267+-- | Buffer n elements of the input in a ring buffer. When t new elements are
268+-- collected, slide the window to remove the same number of oldest elements,
269+-- insert the new elements, and apply an incremental fold on the sliding
270+-- window, supplying the outgoing elements, the new ring buffer as arguments.
272+ :: (IsStream t, Monad m, Ord a, Unboxed a)
273+ => Int -- window size
274+ -> Int -- window slide
275+ -> Fold m (Ring a, Array a) b
278+slidingChunkBuffer = undefined
280+-- Buffer n seconds worth of stream elements of the input in a radix tree.
281+-- Every t seconds, remove the items that are older than n seconds, and apply
282+-- an incremental fold on the sliding window, supplying the outgoing elements,
283+-- and the new radix tree buffer as arguments.
285+ :: (IsStream t, Monad m, Ord a, Unboxed a)
286+ => Int -- window size
288+ -> Fold m (RTree a, Array a) b
291+slidingSessionBuffer = undefined
294+------------------------------------------------------------------------------
295+-- Keyed Session Windows
296+------------------------------------------------------------------------------
299+-- | Keyed variable size space windows. Close the window if we do not receive a
300+-- window event in the next "spaceout" elements.
301+{-# INLINABLE classifyChunksBy #-}
303+ :: (IsStream t, MonadAsync m, Ord k)
304+ => Int -- ^ window spaceout (spread)
305+ -> Bool -- ^ reset the spaceout when a chunk window element is received
306+ -> Fold m a b -- ^ Fold to be applied to chunk window elements
307+ -> t m (k, a, Bool) -- ^ chunk key, data, last element
309+classifyChunksBy spanout reset (Fold step initial extract) str = undefined
311+-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
312+-- received within the chunk size window. The chunk gets closed only if no
313+-- element is received within the chunk window.
315+{-# INLINABLE classifyKeepAliveChunks #-}
316+classifyKeepAliveChunks
317+ :: (IsStream t, MonadAsync m, Ord k)
318+ => Int -- ^ window spaceout (spread)
319+ -> Fold m a b -- ^ Fold to be applied to chunk window elements
320+ -> t m (k, a, Bool) -- ^ chunk key, data, last element
322+classifyKeepAliveChunks spanout = classifyChunksBy spanout True
325+data SessionState t m f s b = SessionState
326+ { sessionCurTime :: !AbsTime -- ^ time since last event
327+ , sessionEventTime :: !AbsTime -- ^ time as per last event
328+ -- We can use the Map size instead of maintaining a count, but if we have
329+ -- to switch to HashMap then it can be useful.
330+ , sessionCount :: !Int -- ^ total number of sessions in progress
331+ , sessionTimerHeap :: H.Heap (H.Entry AbsTime (Key f)) -- ^ heap for timeouts
332+ , sessionKeyValueMap :: f s -- ^ Stored sessions for keys
333+ , sessionOutputStream :: t (m :: Type -> Type) (Key f, b) -- ^ Completed sessions
336+data SessionEntry s = LiveSession !AbsTime !s | ZombieSession
338+-- delete from map and output the fold accumulator
339+ejectEntry :: (Monad m, IsMap f) =>
343+ -> Stream m (Key f, b)
347+ -> m (heap, f entry, Stream m (Key f, b), Int)
348+ejectEntry extract hp mp out cnt acc key = do
350+ let out1 = (key, sess) `Stream.cons` out
351+ let mp1 = IsMap.mapDelete key mp
352+ return (hp, mp1, out1, cnt - 1)
354+{-# NOINLINE flush #-}
355+flush :: (Monad m, IsMap f) =>
357+ -> SessionState Stream m f (SessionEntry s) b
358+ -> m (SessionState Stream m f (SessionEntry s) b)
359+flush extract session@SessionState{..} = do
360+ (hp', mp', out, count) <-
363+ , sessionKeyValueMap
368+ { sessionCount = count
369+ , sessionTimerHeap = hp'
370+ , sessionKeyValueMap = mp'
371+ , sessionOutputStream = out
376+ ejectAll (hp, mp, out, !cnt) = do
377+ let hres = H.uncons hp
379+ Just (Entry _ key, hp1) -> do
380+ r <- case IsMap.mapLookup key mp of
381+ Nothing -> return (hp1, mp, out, cnt)
382+ Just ZombieSession ->
383+ return (hp1, IsMap.mapDelete key mp, out, cnt)
384+ Just (LiveSession _ acc) ->
385+ ejectEntry extract hp1 mp out cnt acc key
388+ assert (IsMap.mapNull mp) (return ())
389+ return (hp, mp, out, cnt)
391+{-# NOINLINE ejectOne #-}
392+ejectOne :: (IsMap f, Monad m) =>
395+ -> ( H.Heap (Entry AbsTime (Key f))
396+ , f (SessionEntry acc)
397+ , Stream m (Key f, b)
400+ -> m ( H.Heap (Entry AbsTime (Key f))
401+ , f (SessionEntry acc)
402+ , Stream m (Key f, b), Int
404+ejectOne reset extract = go
408+ go (hp, mp, out, !cnt) = do
409+ let hres = H.uncons hp
411+ Just (Entry expiry key, hp1) ->
412+ case IsMap.mapLookup key mp of
413+ Nothing -> go (hp1, mp, out, cnt)
414+ Just ZombieSession ->
415+ go (hp1, IsMap.mapDelete key mp, out, cnt)
416+ Just (LiveSession expiry1 acc) -> do
417+ if not reset || expiry1 <= expiry
418+ then ejectEntry extract hp1 mp out cnt acc key
420+ -- reset the session timeout and continue
421+ let hp2 = H.insert (Entry expiry1 key) hp1
422+ in go (hp2, mp, out, cnt)
424+ assert (IsMap.mapNull mp) (return ())
425+ return (hp, mp, out, cnt)
427+{-# NOINLINE ejectExpired #-}
428+ejectExpired :: (IsMap f, Monad m) =>
432+ -> SessionState Stream m f (SessionEntry acc) b
434+ -> m (SessionState Stream m f (SessionEntry acc) b)
435+ejectExpired reset ejectPred extract session@SessionState{..} curTime = do
436+ (hp', mp', out, count) <-
438+ sessionTimerHeap sessionKeyValueMap Stream.nil sessionCount
440+ { sessionCurTime = curTime
441+ , sessionCount = count
442+ , sessionTimerHeap = hp'
443+ , sessionKeyValueMap = mp'
444+ , sessionOutputStream = out
449+ ejectLoop hp mp out !cnt = do
450+ let hres = H.uncons hp
452+ Just (Entry expiry key, hp1) -> do
454+ if curTime >= expiry
455+ then return (True, False)
461+ case IsMap.mapLookup key mp of
462+ Nothing -> ejectLoop hp1 mp out cnt
463+ Just ZombieSession ->
464+ ejectLoop hp1 (IsMap.mapDelete key mp) out cnt
465+ Just (LiveSession expiry1 acc) -> do
466+ if expiry1 <= curTime || not reset || force
468+ (hp2,mp1,out1,cnt1) <-
469+ ejectEntry extract hp1 mp out cnt acc key
470+ ejectLoop hp2 mp1 out1 cnt1
472+ -- reset the session timeout and continue
473+ let hp2 = H.insert (Entry expiry1 key) hp1
474+ in ejectLoop hp2 mp out cnt
475+ else return (hp, mp, out, cnt)
477+ assert (IsMap.mapNull mp) (return ())
478+ return (hp, mp, out, cnt)
480+-- XXX Use mutable IORef in accumulator
481+{-# INLINE classifySessionsByGeneric #-}
482+classifySessionsByGeneric
483+ :: forall m f a b. (MonadAsync m, IsMap f)
484+ => Proxy (f :: (Type -> Type))
485+ -> Double -- ^ timer tick in seconds
486+ -> Bool -- ^ reset the timer when an event is received
487+ -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
488+ -> Double -- ^ session timeout in seconds
489+ -> Fold m a b -- ^ Fold to be applied to session data
490+ -> Stream m (AbsTime, (Key f, a)) -- ^ timestamp, (session key, session
492+ -> Stream m (Key f, b) -- ^ session key, fold result
493+classifySessionsByGeneric _ tick reset ejectPred tmout
494+ (Fold step initial extract) input =
495+ Stream.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
496+ $ Stream.scanlMAfter' sstep (return szero) (flush extract)
497+ $ interjectSuffix tick (return Nothing)
502+ timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
503+ tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
504+ szero = SessionState
505+ { sessionCurTime = toAbsTime (0 :: MilliSecond64)
506+ , sessionEventTime = toAbsTime (0 :: MilliSecond64)
508+ , sessionTimerHeap = H.empty
509+ , sessionKeyValueMap = IsMap.mapEmpty :: f s
510+ , sessionOutputStream = Stream.nil
513+ -- We can eject sessions based on the current session count to limit
514+ -- memory consumption. There are two possible strategies:
516+ -- 1) Eject old sessions or sessions beyond a certain/lower timeout
517+ -- threshold even before timeout, effectively reduce the timeout.
518+ -- 2) Drop creation of new sessions but keep accepting new events for the
521+ -- We use the first strategy as of now.
523+ -- Got a new stream input element
524+ sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
525+ -- XXX instead of a heap we could use a timer wheel.
527+ -- XXX if the key is an Int, we can also use an IntMap for slightly
528+ -- better performance.
532+ -- Values for each key are collected in a map using the supplied fold.
533+ -- When we insert a key in the Map we insert an entry into the heap as
534+ -- well with the session expiry as the sort key. The Map entry
535+ -- consists of the fold result, and the expiry time of the session. If
536+ -- "reset" is True the expiry time is readjusted whenever a new event
537+ -- is processed. If the fold terminates and a new session is started
538+ -- for the same key the expiry time is set to the first timestamp of
541+ -- The heap must have at most one entry for any given key. The heap is
542+ -- processed periodically to remove the expired entries. We pick up an
543+ -- expired entry from the top of the heap and if the session has
544+ -- expired based on the expiry time in the Map entry then we remove the
545+ -- session from the Map and yield its fold output. Otherwise, we
546+ -- reinsert the entry into the heap based on the current expiry in the
549+ -- If an entry is removed from the Map and not removed from the heap
550+ -- and in the meantime it is inserted again in the Map (using the same
551+ -- key) then how do we avoid inserting multiple entries in the heap?
552+ -- For this reason we maintain the invariant that the Map entry is
553+ -- removed only when the heap entry is removed. Even if the fold has
554+ -- finished we still keep a dummy Map entry (ZombieSession) until the
555+ -- heap entry is removed. That way if we have a Map entry we do not
556+ -- insert a heap entry because we know it is already there.
557+ -- XXX The ZombieSession mechanism does not work as expected as we
558+ -- ignore ZombieSession when inserting a new entry. Anyway, we can
559+ -- remove this mechanism as at most only two heap entries may be
560+ -- created and they will be ultimately cleaned up.
562+ -- Heap processing needs the map and map processing needs the heap,
563+ -- therefore we cannot separate the two for modularity unless we have a
564+ -- way to achieve mutual recursion.
566+ let curTime = max sessionEventTime timestamp
567+ mOld = IsMap.mapLookup key sessionKeyValueMap
569+ -- deleting a key from the heap is expensive, so we never
570+ -- delete a key from heap, we just purge it from the Map and it
571+ -- gets purged from the heap on timeout. We just need an extra
572+ -- lookup in the Map when the key is purged from the heap, that
573+ -- should not be expensive.
575+ let (mp, cnt) = case mOld of
576+ Just (LiveSession _ _) ->
578+ key ZombieSession sessionKeyValueMap
581+ _ -> (sessionKeyValueMap, sessionCount)
583+ { sessionCurTime = curTime
584+ , sessionEventTime = curTime
585+ , sessionCount = cnt
586+ , sessionKeyValueMap = mp
587+ , sessionOutputStream = Stream.fromPure (key, fb)
590+ let expiry = addToAbsTime timestamp timeoutMs
591+ (hp1, mp1, out1, cnt1) <- do
592+ let vars = (sessionTimerHeap, sessionKeyValueMap,
593+ Stream.nil, sessionCount)
595+ -- inserting new entry
597+ -- Eject a session from heap and map if needed
598+ eject <- ejectPred sessionCount
599+ (hp, mp, out, cnt) <-
601+ then ejectOne reset extract vars
604+ -- Insert the new session in heap
605+ let hp' = H.insert (Entry expiry key) hp
606+ in return (hp', mp, out, cnt + 1)
607+ -- updating old entry
608+ Just _ -> return vars
610+ let acc = LiveSession expiry fs1
611+ mp2 = IsMap.mapInsert key acc mp1
612+ return $ SessionState
613+ { sessionCurTime = curTime
614+ , sessionEventTime = curTime
615+ , sessionCount = cnt1
616+ , sessionTimerHeap = hp1
617+ , sessionKeyValueMap = mp2
618+ , sessionOutputStream = out1
622+ Just (LiveSession _ acc) -> return $ Fold.Partial acc
626+ error $ "classifySessionsBy: "
627+ ++ "The supplied fold must consume at least one input"
628+ Fold.Partial fs -> do
629+ res <- step fs value
631+ Fold.Done fb -> done fb
632+ Fold.Partial fs1 -> partial fs1
634+ -- Got a timer tick event
635+ sstep sessionState@SessionState{..} Nothing =
636+ let curTime = addToAbsTime sessionCurTime tickMs
637+ in ejectExpired reset ejectPred extract sessionState curTime
639+-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
640+-- classifies an input event @stream@ consisting of @(timestamp, (key,
641+-- value))@ into sessions based on the @key@, folding all the values
642+-- corresponding to the same key into a session using the supplied @fold@.
644+-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
645+-- session key and the folded value is emitted in the output stream. The
646+-- timeout is measured from the first event in the session. If the @keepalive@
647+-- option is set to 'True' the timeout is reset to 0 whenever an event is
650+-- The @timestamp@ in the input stream is an absolute time from some epoch,
651+-- characterizing the time when the input event was generated. The notion of
652+-- current time is maintained by a monotonic event time clock using the
653+-- timestamps seen in the input stream. The latest timestamp seen till now is
654+-- used as the base for the current time. When no new events are seen, a timer
655+-- is started with a clock resolution of @tick@ seconds. This timer is used to
656+-- detect session timeouts in the absence of new events.
658+-- To ensure an upper bound on the memory used the number of sessions can be
659+-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
660+-- oldest session is ejected before inserting a new session.
662+-- When the stream ends any buffered sessions are ejected immediately.
664+-- If a session key is received even after a session has finished, another
665+-- session is created for that key.
668+-- Stream.fold (Fold.drainBy print)
669+-- $ Concur.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
670+-- $ Stream.timestamped
671+-- $ Stream.delay 0.1
672+-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
679+{-# INLINE classifySessionsBy #-}
681+ :: (MonadAsync m, Ord k)
682+ => Double -- ^ timer tick in seconds
683+ -> Bool -- ^ reset the timer when an event is received
684+ -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
685+ -> Double -- ^ session timeout in seconds
686+ -> Fold m a b -- ^ Fold to be applied to session data
687+ -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
688+ -> Stream m (k, b) -- ^ session key, fold result
689+classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k))
691+-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
692+-- option set to 'True'.
695+-- classifyKeepAliveSessions = classifySessionsBy 1 True
700+{-# INLINE classifyKeepAliveSessions #-}
701+classifyKeepAliveSessions ::
702+ (MonadAsync m, Ord k)
703+ => (Int -> m Bool) -- ^ predicate to eject sessions on session count
704+ -> Double -- ^ session inactive timeout
705+ -> Fold m a b -- ^ Fold to be applied to session payload data
706+ -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
708+classifyKeepAliveSessions = classifySessionsBy 1 True
710+------------------------------------------------------------------------------
711+-- Keyed tumbling windows
712+------------------------------------------------------------------------------
714+-- Tumbling windows is a special case of sliding windows where the window slide
715+-- is the same as the window size. Or it can be a special case of session
716+-- windows where the reset flag is set to False.
718+-- XXX instead of using the early termination flag in the stream, we can use an
719+-- early terminating fold instead.
722+-- | Split the stream into fixed size chunks of specified size. Within each
723+-- such chunk fold the elements in buckets identified by the keys. A particular
724+-- bucket fold can be terminated early if a closing flag is encountered in an
725+-- element for that key.
728+{-# INLINABLE classifyChunksOf #-}
730+ :: (IsStream t, MonadAsync m, Ord k)
731+ => Int -- ^ window size
732+ -> Fold m a b -- ^ Fold to be applied to window events
733+ -> t m (k, a, Bool) -- ^ window key, data, close event
735+classifyChunksOf wsize = classifyChunksBy wsize False
738+-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
739+-- option set to 'False'.
742+-- classifySessionsOf = classifySessionsBy 1 False
747+{-# INLINE classifySessionsOf #-}
748+classifySessionsOf ::
749+ (MonadAsync m, Ord k)
750+ => (Int -> m Bool) -- ^ predicate to eject sessions on session count
751+ -> Double -- ^ time window size
752+ -> Fold m a b -- ^ Fold to be applied to session data
753+ -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
755+classifySessionsOf = classifySessionsBy 1 False
757+------------------------------------------------------------------------------
759+------------------------------------------------------------------------------
761+-- | Continuously evaluate the input stream and sample the last event in time
762+-- window of @n@ seconds.
764+-- This is also known as @throttle@ in some libraries.
767+-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
772+{-# INLINE sampleIntervalEnd #-}
773+sampleIntervalEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
774+sampleIntervalEnd n = Stream.catMaybes . intervalsOf n Fold.last
776+-- | Like 'sampleInterval' but samples at the beginning of the time window.
779+-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
784+{-# INLINE sampleIntervalStart #-}
785+sampleIntervalStart :: MonadAsync m => Double -> Stream m a -> Stream m a
786+sampleIntervalStart n = Stream.catMaybes . intervalsOf n Fold.one
788+data BurstState t x =
792+ | BurstDoneNext !x !t !x
794+{-# INLINE sampleBurst #-}
795+sampleBurst :: MonadAsync m => Bool -> Double -> Stream m a -> Stream m a
796+sampleBurst sampleAtEnd gap xs =
797+ -- XXX Ideally we should schedule a timer event exactly after gap time,
798+ -- but the tick stream should work well as long as the timer
799+ -- granularity is small enough compared to the gap.
800+ Stream.mapMaybe extract
801+ $ Stream.scan (Fold.foldl' step BurstNone)
802+ $ Stream.timeIndexed
803+ $ interjectSuffix 0.01 (return Nothing) (fmap Just xs)
807+ gap1 = toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
810+ step BurstNone (t1, Just x1) = BurstWait t1 x1
811+ step BurstNone _ = BurstNone
813+ step (BurstDone _) (t1, Just x1) = BurstWait t1 x1
814+ step (BurstDone _) _ = BurstNone
816+ step old@(BurstWait t0 x0) (t1, Nothing)
817+ | t1 - t0 >= gap1 = BurstDone x0
819+ -- This can happen due to scheduling delays, if we received back to
820+ -- back events spaced by more than the timeout without an
821+ -- intervening timeout event then we emit the old event instead of
822+ -- replacing it by the new.
823+ step (BurstWait t0 x0) (t1, Just x1)
824+ | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
825+ | sampleAtEnd = BurstWait t1 x1
826+ | otherwise = BurstWait t1 x0
828+ step (BurstDoneNext _ t0 x0) (t1, Nothing)
829+ | t1 - t0 >= gap1 = BurstDone x0
830+ | otherwise = BurstWait t0 x0
831+ step (BurstDoneNext _ t0 x0) (t1, Just x1)
832+ | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
833+ | sampleAtEnd = BurstWait t1 x1
834+ | otherwise = BurstWait t1 x0
836+ {-# INLINE extract #-}
837+ extract (BurstDoneNext x _ _) = Just x
838+ extract (BurstDone x) = Just x
841+-- | Sample one event at the end of each burst of events. A burst is a group
842+-- of events close together in time, it ends when an event is spaced by more
843+-- than the specified time interval (in seconds) from the previous event.
845+-- This is known as @debounce@ in some libraries.
847+-- The clock granularity is 10 ms.
851+{-# INLINE sampleBurstEnd #-}
852+sampleBurstEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
853+sampleBurstEnd = sampleBurst True
855+-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
856+-- instead of at the end of it.
860+{-# INLINE sampleBurstStart #-}
861+sampleBurstStart :: MonadAsync m => Double -> Stream m a -> Stream m a
862+sampleBurstStart = sampleBurst False