@@ -89,6 +89,7 @@ module Streamly.Internal.Data.Stream.IsStream.Reduce
8989 -- | A new window starts after the previous window is finished.
9090
9191 -- , classifyChunksOf
92+ , classifySessionsByGeneric
9293 , classifySessionsBy
9394 , classifySessionsOf
9495
@@ -153,10 +154,13 @@ import Control.Monad.Catch (MonadThrow)
153154import Control.Monad.IO.Class (MonadIO(..))
154155import Data.Heap (Entry(..))
155156import Data.Kind (Type)
157+import Data.Map (Map)
156158import Data.Maybe (isNothing)
159+import Data.Proxy (Proxy(..))
157160import Foreign.Storable (Storable)
158161import Streamly.Internal.Control.Concurrent (MonadAsync)
159162import Streamly.Internal.Data.Fold.Type (Fold (..))
163+import Streamly.Internal.Data.IsMap (IsMap(..))
160164import Streamly.Internal.Data.Refold.Type (Refold (..))
161165import Streamly.Internal.Data.Parser (Parser (..))
162166import Streamly.Internal.Data.Array.Foreign.Type (Array)
@@ -178,6 +182,7 @@ import qualified Data.Heap as H
178182import qualified Data.Map.Strict as Map
179183import qualified Streamly.Internal.Data.Array.Foreign.Type as A
180184import qualified Streamly.Internal.Data.Fold as FL
185+import qualified Streamly.Internal.Data.IsMap as IsMap
181186import qualified Streamly.Internal.Data.Parser.ParserD as PRD
182187import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
183188import qualified Streamly.Internal.Data.Stream.StreamD as D
@@ -1150,64 +1155,23 @@ data SessionState t m k a b = SessionState
11501155
11511156data SessionEntry a b = LiveSession !a !b | ZombieSession
11521157
1153−-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
1154−-- classifies an input event @stream@ consisting of @(timestamp, (key,
1155−-- value))@ into sessions based on the @key@, folding all the values
1156−-- corresponding to the same key into a session using the supplied @fold@.
1157−--
1158−-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
1159−-- session key and the folded value is emitted in the output stream. The
1160−-- timeout is measured from the first event in the session. If the @keepalive@
1161−-- option is set to 'True' the timeout is reset to 0 whenever an event is
1162−-- received.
1163−--
1164−-- The @timestamp@ in the input stream is an absolute time from some epoch,
1165−-- characterizing the time when the input event was generated. The notion of
1166−-- current time is maintained by a monotonic event time clock using the
1167−-- timestamps seen in the input stream. The latest timestamp seen till now is
1168−-- used as the base for the current time. When no new events are seen, a timer
1169−-- is started with a clock resolution of @tick@ seconds. This timer is used to
1170−-- detect session timeouts in the absence of new events.
1171−--
1172−-- To ensure an upper bound on the memory used the number of sessions can be
1173−-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1174−-- oldest session is ejected before inserting a new session.
1175−--
1176−-- When the stream ends any buffered sessions are ejected immediately.
1177−--
1178−-- If a session key is received even after a session has finished, another
1179−-- session is created for that key.
1180−--
1181−-- >>> :{
1182−-- Stream.mapM_ print
1183−-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
1184−-- $ Stream.timestamped
1185−-- $ Stream.delay 0.1
1186−-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
1187−-- :}
1188−-- (1,"abc")
1189−-- (2,"abc")
1190−-- (3,"abc")
1191−--
1192−-- /Pre-release/
1193−
1194−-- XXX Use a proxy type for the map type to use.
11951158-- XXX Use mutable IORef in accumulator
11961159--
11971160-- XXX this fuses with INLINE but it may be too much work for the compiler.
11981161-- Maybe we can NOINLINE the heap processing stuff and then INLINE it to reduce
11991162-- the code bloat.
1200−{-# INLINABLE classifySessionsBy #-}
1201−classifySessionsBy
1202− :: (IsStream t, MonadAsync m, Ord k)
1203− => Double -- ^ timer tick in seconds
1163+{-# INLINABLE classifySessionsByGeneric #-}
1164+classifySessionsByGeneric
1165+ :: (IsStream t, MonadAsync m, Ord (Key f))
1166+ => Proxy (f s)
1167+ -> Double -- ^ timer tick in seconds
12041168 -> Bool -- ^ reset the timer when an event is received
12051169 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
12061170 -> Double -- ^ session timeout in seconds
12071171 -> Fold m a b -- ^ Fold to be applied to session data
1208− -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1209− -> t m (k, b) -- ^ session key, fold result
1210−classifySessionsBy tick reset ejectPred tmout
1172+ -> t m (AbsTime, (Key f, a)) -- ^ timestamp, (session key, session data)
1173+ -> t m (Key f, b) -- ^ session key, fold result
1174+classifySessionsByGeneric _ tick reset ejectPred tmout
12111175 (Fold step initial extract) input =
12121176 Expand.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
12131177 $ scanlMAfter' sstep (return szero) flush
@@ -1223,7 +1187,7 @@ classifySessionsBy tick reset ejectPred tmout
12231187 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
12241188 , sessionCount = 0
12251189 , sessionTimerHeap = H.empty
1226− , sessionKeyValueMap = Map.empty
1190+ , sessionKeyValueMap = IsMap.mapEmpty
12271191 , sessionOutputStream = IsStream.nil
12281192 }
12291193
@@ -1281,7 +1245,7 @@ classifySessionsBy tick reset ejectPred tmout
12811245 -- way to achieve mutual recursion.
12821246 --
12831247 let curTime = max sessionEventTime timestamp
1284− mOld = Map.lookup key sessionKeyValueMap
1248+ mOld = IsMap.mapLookup key sessionKeyValueMap
12851249 let done fb = do
12861250 -- deleting a key from the heap is expensive, so we never
12871251 -- delete a key from heap, we just purge it from the Map and it
@@ -1291,7 +1255,8 @@ classifySessionsBy tick reset ejectPred tmout
12911255 --
12921256 let (mp, cnt) = case mOld of
12931257 Just (LiveSession _ _) ->
1294− ( Map.insert key ZombieSession sessionKeyValueMap
1258+ ( IsMap.mapInsert
1259+ key ZombieSession sessionKeyValueMap
12951260 , sessionCount - 1
12961261 )
12971262 _ -> (sessionKeyValueMap, sessionCount)
@@ -1324,7 +1289,7 @@ classifySessionsBy tick reset ejectPred tmout
13241289 Just _ -> return vars
13251290
13261291 let acc = LiveSession expiry fs1
1327− mp2 = Map.insert key acc mp1
1292+ mp2 = IsMap.mapInsert key acc mp1
13281293 return $ SessionState
13291294 { sessionCurTime = curTime
13301295 , sessionEventTime = curTime
@@ -1371,17 +1336,17 @@ classifySessionsBy tick reset ejectPred tmout
13711336 ejectEntry hp mp out cnt acc key = do
13721337 sess <- extract acc
13731338 let out1 = (key, sess) `cons` out
1374− let mp1 = Map.delete key mp
1339+ let mp1 = IsMap.mapDelete key mp
13751340 return (hp, mp1, out1, cnt - 1)
13761341
13771342 ejectAll (hp, mp, out, !cnt) = do
13781343 let hres = H.uncons hp
13791344 case hres of
13801345 Just (Entry _ key, hp1) -> do
1381− r <- case Map.lookup key mp of
1346+ r <- case IsMap.mapLookup key mp of
13821347 Nothing -> return (hp1, mp, out, cnt)
13831348 Just ZombieSession ->
1384− return (hp1, Map.delete key mp, out, cnt)
1349+ return (hp1, IsMap.mapDelete key mp, out, cnt)
13851350 Just (LiveSession _ acc) ->
13861351 ejectEntry hp1 mp out cnt acc key
13871352 ejectAll r
@@ -1393,10 +1358,10 @@ classifySessionsBy tick reset ejectPred tmout
13931358 let hres = H.uncons hp
13941359 case hres of
13951360 Just (Entry expiry key, hp1) ->
1396− case Map.lookup key mp of
1361+ case IsMap.mapLookup key mp of
13971362 Nothing -> ejectOne (hp1, mp, out, cnt)
13981363 Just ZombieSession ->
1399− ejectOne (hp1, Map.delete key mp, out, cnt)
1364+ ejectOne (hp1, IsMap.mapDelete key mp, out, cnt)
14001365 Just (LiveSession expiry1 acc) -> do
14011366 if not reset || expiry1 <= expiry
14021367 then ejectEntry hp1 mp out cnt acc key
@@ -1405,7 +1370,7 @@ classifySessionsBy tick reset ejectPred tmout
14051370 let hp2 = H.insert (Entry expiry1 key) hp1
14061371 in ejectOne (hp2, mp, out, cnt)
14071372 Nothing -> do
1408− assert (Map.null mp) (return ())
1373+ assert (IsMap.mapNull mp) (return ())
14091374 return (hp, mp, out, cnt)
14101375
14111376 ejectExpired session@SessionState{..} curTime = do
@@ -1434,10 +1399,10 @@ classifySessionsBy tick reset ejectPred tmout
14341399 return (r, r)
14351400 if eject
14361401 then
1437− case Map.lookup key mp of
1402+ case IsMap.mapLookup key mp of
14381403 Nothing -> ejectLoop hp1 mp out cnt
14391404 Just ZombieSession ->
1440− ejectLoop hp1 (Map.delete key mp) out cnt
1405+ ejectLoop hp1 (IsMap.mapDelete key mp) out cnt
14411406 Just (LiveSession expiry1 acc) -> do
14421407 if expiry1 <= curTime || not reset || force
14431408 then do
@@ -1450,9 +1415,61 @@ classifySessionsBy tick reset ejectPred tmout
14501415 in ejectLoop hp2 mp out cnt
14511416 else return (hp, mp, out, cnt)
14521417 Nothing -> do
1453− assert (Map.null mp) (return ())
1418+ assert (IsMap.mapNull mp) (return ())
14541419 return (hp, mp, out, cnt)
14551420
1421+-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
1422+-- classifies an input event @stream@ consisting of @(timestamp, (key,
1423+-- value))@ into sessions based on the @key@, folding all the values
1424+-- corresponding to the same key into a session using the supplied @fold@.
1425+--
1426+-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
1427+-- session key and the folded value is emitted in the output stream. The
1428+-- timeout is measured from the first event in the session. If the @keepalive@
1429+-- option is set to 'True' the timeout is reset to 0 whenever an event is
1430+-- received.
1431+--
1432+-- The @timestamp@ in the input stream is an absolute time from some epoch,
1433+-- characterizing the time when the input event was generated. The notion of
1434+-- current time is maintained by a monotonic event time clock using the
1435+-- timestamps seen in the input stream. The latest timestamp seen till now is
1436+-- used as the base for the current time. When no new events are seen, a timer
1437+-- is started with a clock resolution of @tick@ seconds. This timer is used to
1438+-- detect session timeouts in the absence of new events.
1439+--
1440+-- To ensure an upper bound on the memory used the number of sessions can be
1441+-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1442+-- oldest session is ejected before inserting a new session.
1443+--
1444+-- When the stream ends any buffered sessions are ejected immediately.
1445+--
1446+-- If a session key is received even after a session has finished, another
1447+-- session is created for that key.
1448+--
1449+-- >>> :{
1450+-- Stream.mapM_ print
1451+-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
1452+-- $ Stream.timestamped
1453+-- $ Stream.delay 0.1
1454+-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
1455+-- :}
1456+-- (1,"abc")
1457+-- (2,"abc")
1458+-- (3,"abc")
1459+--
1460+-- /Pre-release/
1461+{-# INLINE classifySessionsBy #-}
1462+classifySessionsBy
1463+ :: (IsStream t, MonadAsync m, Ord k)
1464+ => Double -- ^ timer tick in seconds
1465+ -> Bool -- ^ reset the timer when an event is received
1466+ -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1467+ -> Double -- ^ session timeout in seconds
1468+ -> Fold m a b -- ^ Fold to be applied to session data
1469+ -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1470+ -> t m (k, b) -- ^ session key, fold result
1471+classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k s))
1472+
14561473-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
14571474-- option set to 'True'.
14581475--