@@ -244,6 +244,9 @@ module Streamly.Fold
244244 , groupsBy
245245 , groupsRollingBy
246246
247+ -- ** Sessions
248+ , sessionsByLifeTime
249+
247250 -- * Distributing
248251 -- |
249252 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy
@@ -312,6 +315,7 @@ import Control.Concurrent (threadDelay)
312315import Control.Monad (void)
313316import Control.Monad.IO.Class (MonadIO(..))
314317import Data.Functor.Identity (Identity)
318+import Data.Heap (Entry(..))
315319import Data.Map.Strict (Map)
316320import Data.Maybe (fromJust, isJust, isNothing)
317321
@@ -324,6 +328,7 @@ import Prelude
324328 scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip,
325329 span, splitAt, break)
326330
331+import qualified Data.Heap as H
327332import qualified Data.Map.Strict as Map
328333
329334import Streamly (MonadAsync, parallel)
@@ -331,6 +336,9 @@ import Streamly.Fold.Types (Fold(..))
331336import Streamly.Mem.Array (Array)
332337import Streamly.Streams.Serial (SerialT)
333338import Streamly.Streams.StreamK (IsStream())
339+import Streamly.Time.Clock (Clock(..), getTime)
340+import Streamly.Time.Units
341+ (AbsTime, MilliSecond64(..), diffAbsTime, fromRelTime)
334342
335343import Streamly.Strict
336344
@@ -2097,3 +2105,106 @@ lconcatMap s f1 f2 = undefined
20972105lchunksOf :: Int -> Fold m a b -> Fold m b c -> Fold m a c
20982106lchunksOf n f1 f2 = undefined
20992107-}
2108+
2109+------------------------------------------------------------------------------
2110+-- Session Windows
2111+------------------------------------------------------------------------------
2112+
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
2121+-- timed out.
2122+--
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.
2130+--
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.
2134+--
2135+-- @since 0.7.0
2136+{-# INLINABLE sessionsByLifeTime #-}
2137+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)
2142+ -> Fold m a b
2143+ -> t m a
2144+ -> t m b
2145+sessionsByLifeTime lifetime tick getKey (Fold step initial extract) str =
2146+ S.concatMap (\(Tuple3' _ _ s) -> s) $ S.scanlM' sstep szero stream
2147+
2148+ where
2149+
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
2153+ -- send an alert.
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
2156+
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
2165+ in if closed
2166+ then do
2167+ let (v, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
2168+ r <- accumulate v
2169+ res <- extract r
2170+ return $ Tuple3' hp mp' (S.yield res)
2171+ else do
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
2177+ Just _ -> hp
2178+ return $ Tuple3' hp' mp' S.nil
2179+
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
2185+
2186+ where
2187+
2188+ go hp mp out = do
2189+ let r = H.uncons hp
2190+ case r of
2191+ Just (Entry ts k, hp') -> do
2192+ let duration = diffAbsTime ts t
2193+ lt = round (lifetime * 1000) :: MilliSecond64
2194+ if fromRelTime duration >= lt
2195+ then do
2196+ let (v, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) k mp
2197+ case v of
2198+ Nothing -> go hp' mp out
2199+ Just x -> do
2200+ res <- extract x
2201+ go hp' mp' (res `S.cons` out)
2202+ else return (hp, mp, out)
2203+ Nothing -> return (hp, mp, out)
2204+
2205+ -- merge timer events in the stream
2206+ stream = S.map (\x -> Right x) str `parallel` S.repeatM timer
2207+ timer = do
2208+ liftIO $ threadDelay (round $ tick * 1000000)
2209+ now <- liftIO $ getTime Realtime
2210+ return $ Left now