@@ -179,7 +179,6 @@ import Streamly.Internal.Data.Time.Units
179179 , toAbsTime)
180180
181181import qualified Data.Heap as H
182−import qualified Data.Map.Strict as Map
183182import qualified Streamly.Internal.Data.Array.Foreign.Type as A
184183import qualified Streamly.Internal.Data.Fold as FL
185184import qualified Streamly.Internal.Data.IsMap as IsMap
@@ -1142,18 +1141,18 @@ classifyKeepAliveChunks
11421141classifyKeepAliveChunks spanout = classifyChunksBy spanout True
11431142-}
11441143
1145−data SessionState t m k a b = SessionState
1144+data SessionState t m f s b = SessionState
11461145 { sessionCurTime :: !AbsTime -- ^ time since last event
11471146 , sessionEventTime :: !AbsTime -- ^ time as per last event
11481147 -- We can use the Map size instead of maintaining a count, but if we have
11491148 -- to switch to HashMap then it can be useful.
11501149 , sessionCount :: !Int -- ^ total number of sessions in progress
1151− , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
1152− , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
1153− , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
1150+ , sessionTimerHeap :: H.Heap (H.Entry AbsTime (Key f)) -- ^ heap for timeouts
1151+ , sessionKeyValueMap :: f s -- ^ Stored sessions for keys
1152+ , sessionOutputStream :: t (m :: Type -> Type) (Key f, b) -- ^ Completed sessions
11541153 }
11551154
1156−data SessionEntry a b = LiveSession !a !b | ZombieSession
1155+data SessionEntry s = LiveSession !AbsTime !s | ZombieSession
11571156
11581157-- delete from map and output the fold accumulator
11591158ejectEntry :: (Monad m, IsStream t, IsMap f) =>
@@ -1172,10 +1171,10 @@ ejectEntry extract hp mp out cnt acc key = do
11721171 return (hp, mp1, out1, cnt - 1)
11731172
11741173{-# NOINLINE flush #-}
1175−flush :: (Monad m, Ord k, IsStream t) =>
1174+flush :: (Monad m, IsMap f, IsStream t) =>
11761175 (s -> m b)
1177− -> SessionState t m k (SessionEntry a s) b
1178− -> m (SessionState t m k (SessionEntry a s) b)
1176+ -> SessionState t m f (SessionEntry s) b
1177+ -> m (SessionState t m f (SessionEntry s) b)
11791178flush extract session@SessionState{..} = do
11801179 (hp', mp', out, count) <-
11811180 ejectAll
@@ -1205,20 +1204,20 @@ flush extract session@SessionState{..} = do
12051204 ejectEntry extract hp1 mp out cnt acc key
12061205 ejectAll r
12071206 Nothing -> do
1208− assert (Map.null mp) (return ())
1207+ assert (IsMap.mapNull mp) (return ())
12091208 return (hp, mp, out, cnt)
12101209
12111210{-# NOINLINE ejectOne #-}
1212−ejectOne :: (IsMap f, IsStream t, Ord k, Monad m) =>
1211+ejectOne :: (IsMap f, IsStream t, Monad m) =>
12131212 Bool
12141213 -> (acc -> m b)
1215− -> ( H.Heap (Entry k (Key f))
1216− , f (SessionEntry k acc)
1214+ -> ( H.Heap (Entry AbsTime (Key f))
1215+ , f (SessionEntry acc)
12171216 , t m (Key f, b)
12181217 , Int
12191218 )
1220− -> m ( H.Heap (Entry k (Key f))
1221− , f (SessionEntry k acc)
1219+ -> m ( H.Heap (Entry AbsTime (Key f))
1220+ , f (SessionEntry acc)
12221221 , t m (Key f, b), Int
12231222 )
12241223ejectOne reset extract = go
@@ -1245,13 +1244,13 @@ ejectOne reset extract = go
12451244 return (hp, mp, out, cnt)
12461245
12471246{-# NOINLINE ejectExpired #-}
1248−ejectExpired :: (Ord k, IsStream t, Monad m) =>
1247+ejectExpired :: (IsMap f, IsStream t, Monad m) =>
12491248 Bool
12501249 -> (Int -> m Bool)
12511250 -> (acc -> m b)
1252− -> SessionState t m k (SessionEntry AbsTime acc) b
1251+ -> SessionState t m f (SessionEntry acc) b
12531252 -> AbsTime
1254− -> m (SessionState t m k (SessionEntry AbsTime acc) b)
1253+ -> m (SessionState t m f (SessionEntry acc) b)
12551254ejectExpired reset ejectPred extract session@SessionState{..} curTime = do
12561255 (hp', mp', out, count) <-
12571256 ejectLoop
@@ -1300,8 +1299,8 @@ ejectExpired reset ejectPred extract session@SessionState{..} curTime = do
13001299-- XXX Use mutable IORef in accumulator
13011300{-# INLINE classifySessionsByGeneric #-}
13021301classifySessionsByGeneric
1303− :: (IsStream t, MonadAsync m, Ord (Key f))
1304− => Proxy (f s)
1302+ :: forall t m f a b. (IsStream t, MonadAsync m, IsMap f)
1303+ => Proxy (f :: (Type -> Type))
13051304 -> Double -- ^ timer tick in seconds
13061305 -> Bool -- ^ reset the timer when an event is received
13071306 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
@@ -1325,7 +1324,7 @@ classifySessionsByGeneric _ tick reset ejectPred tmout
13251324 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
13261325 , sessionCount = 0
13271326 , sessionTimerHeap = H.empty
1328− , sessionKeyValueMap = IsMap.mapEmpty
1327+ , sessionKeyValueMap = IsMap.mapEmpty :: f s
13291328 , sessionOutputStream = IsStream.nil
13301329 }
13311330
@@ -1505,7 +1504,7 @@ classifySessionsBy
15051504 -> Fold m a b -- ^ Fold to be applied to session data
15061505 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
15071506 -> t m (k, b) -- ^ session key, fold result
1508−classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k s))
1507+classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k))
15091508
15101509-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
15111510-- option set to 'True'.