@@ -244,6 +244,9 @@ module Streamly.Fold
244 , groupsBy
244 , groupsBy
245 , groupsRollingBy
245 , groupsRollingBy
248+ , sessionsByLifeTime
247 -- * Distributing
250 -- * Distributing
249 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy
252 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy
@@ -312,6 +315,7 @@ import Control.Concurrent (threadDelay)
312import Control.Monad (void)
315import Control.Monad (void)
313import Control.Monad.IO.Class (MonadIO(..))
316import Control.Monad.IO.Class (MonadIO(..))
314import Data.Functor.Identity (Identity)
317import Data.Functor.Identity (Identity)
318+import Data.Heap (Entry(..))
315import Data.Map.Strict (Map)
319import Data.Map.Strict (Map)
316import Data.Maybe (fromJust, isJust, isNothing)
320import Data.Maybe (fromJust, isJust, isNothing)
@@ -324,6 +328,7 @@ import Prelude
324 scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip,
328 scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip,
325 span, splitAt, break)
329 span, splitAt, break)
331+import qualified Data.Heap as H
327import qualified Data.Map.Strict as Map
332import qualified Data.Map.Strict as Map
329import Streamly (MonadAsync, parallel)
334import Streamly (MonadAsync, parallel)
@@ -331,6 +336,9 @@ import Streamly.Fold.Types (Fold(..))
331import Streamly.Mem.Array (Array)
336import Streamly.Mem.Array (Array)
332import Streamly.Streams.Serial (SerialT)
337import Streamly.Streams.Serial (SerialT)
333import Streamly.Streams.StreamK (IsStream())
338import Streamly.Streams.StreamK (IsStream())
339+import Streamly.Time.Clock (Clock(..), getTime)
340+import Streamly.Time.Units
341+ (AbsTime, MilliSecond64(..), diffAbsTime, fromRelTime)
335import Streamly.Strict
343import Streamly.Strict
@@ -2097,3 +2105,106 @@ lconcatMap s f1 f2 = undefined
2097lchunksOf :: Int -> Fold m a b -> Fold m b c -> Fold m a c
2105lchunksOf :: Int -> Fold m a b -> Fold m b c -> Fold m a c
2098lchunksOf n f1 f2 = undefined
2106lchunksOf n f1 f2 = undefined
2109+------------------------------------------------------------------------------
2111+------------------------------------------------------------------------------
2113+-- | @sessionsByLifeTime lifetime tick k f stream@ groups together all input
2114+-- stream elements that belong to the same session. @lifetime@ is the maximum
2115+-- lifetime of a session in seconds. All elements belonging to a session are
2116+-- purged after this duration. The session duration is measured using the
2117+-- timestamp of the first message seen for that session. The timestamp is
2118+-- compared with the current system time, we assume that the system time is in
2119+-- sync with the timestamps in the input stream. @tick@ is the timer
2120+-- granularity specified in seconds to check and purge the sessions that have
2123+-- The function @k@ returns a 3-tuple @(session key, timestamp, session close)@
2124+-- from an input element in the stream. @session key@ is a key that uniquely
2125+-- identifies the session for the given element, @timestamp@ characterizes the
2126+-- time when the input element was generated, this must be a time measured from
2127+-- the Unix @Epoch@. @session close@ is a boolean indicating whether this
2128+-- element marks the closing of the session. When a message with
2129+-- @session close@ set to @True@ is seen the session is purged immediately.
2131+-- All the messages belonging to a session are collected using the fold @f@.
2132+-- The fold result is emitted when the session is purged either via the session
2133+-- close event or via the session liftime timeout.
2136+{-# INLINABLE sessionsByLifeTime #-}
2138+ :: (IsStream t, MonadAsync m, Ord k)
2139+ => Double -- lifetime of a session
2140+ -> Double -- timer tick in secoonds
2141+ -> (a -> (k, AbsTime, Bool)) -- (key, timestamp, session close)
2145+sessionsByLifeTime lifetime tick getKey (Fold step initial extract) str =
2146+ S.concatMap (\(Tuple3' _ _ s) -> s) $ S.scanlM' sstep szero stream
2150+ -- XXX what if the timestamps and the system time are too far off, we may
2151+ -- accumulate too many entries, have a protection in that case? If the
2152+ -- incoming logs seem too new or too old than the current system time then
2154+ -- XXX we have to use a heap in pinned memory to scale it to a large size
2155+ szero = Tuple3' H.empty Map.empty S.nil
2157+ -- Got a new input element
2158+ sstep (Tuple3' hp mp _) (Right a) =
2159+ -- deleting a key from the heap is expensive, so we never delete a
2160+ -- key, we just purge it from the Map and it gets purged from the
2161+ -- heap on timeout. We just need an extra lookup in the Map when
2162+ -- the key is purged from the heap, that should not be expensive.
2163+ let (key, ts, closed) = getKey a
2164+ accumulate v = maybe initial return v >>= \r -> step r a
2167+ let (v, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
2170+ return $ Tuple3' hp mp' (S.yield res)
2172+ let v = Map.lookup key mp
2173+ acc <- accumulate v
2174+ let mp' = Map.insert key acc mp
2175+ let hp' = case v of
2176+ Nothing -> H.insert (Entry ts key) hp
2178+ return $ Tuple3' hp' mp' S.nil
2180+ -- Got timer tick event
2181+ -- XXX can we yield the entries without accumulating them?
2182+ sstep (Tuple3' heap mmp _) (Left t) = do
2183+ (hp', mp', out) <- go heap mmp S.nil
2184+ return $ Tuple3' hp' mp' out
2189+ let r = H.uncons hp
2191+ Just (Entry ts k, hp') -> do
2192+ let duration = diffAbsTime ts t
2193+ lt = round (lifetime * 1000) :: MilliSecond64
2194+ if fromRelTime duration >= lt
2196+ let (v, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) k mp
2198+ Nothing -> go hp' mp out
2201+ go hp' mp' (res `S.cons` out)
2202+ else return (hp, mp, out)
2203+ Nothing -> return (hp, mp, out)
2205+ -- merge timer events in the stream
2206+ stream = S.map (\x -> Right x) str `parallel` S.repeatM timer
2208+ liftIO $ threadDelay (round $ tick * 1000000)
2209+ now <- liftIO $ getTime Realtime