f7681dc · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Reorder the arguments in classifySessionsBy

f7681dcHarendra Kumar6 years ago

benchmark/Streamly/Benchmark/Prelude/Serial/Transformation2.hs+1−2

compare with master

unifiedsplit

@@ -201,8 +201,7 @@ o_n_heap_buffering value =

201classifySessionsOf :: (S.MonadAsync m) => SerialT m Int -> m ()

202classifySessionsOf =

203 S.drain

204− . Internal.classifySessionsOf

204+ . Internal.classifySessionsOf (const (return False)) 3 FL.drain

205− 3 (const (return False)) (fmap Right FL.drain)

206 . S.map (\(ts,(k,a)) -> (k, a, ts))

205 . S.map (\(ts,(k,a)) -> (k, a, ts))

207 . Internal.timestamped

206 . Internal.timestamped

208 . S.concatMap (\x -> S.map (x,) (S.enumerateFromTo 1 (10 :: Int)))

207 . S.concatMap (\x -> S.map (x,) (S.enumerateFromTo 1 (10 :: Int)))

src/Streamly/Internal/Data/Stream/IsStream/Nesting.hs+16−18

@@ -1903,10 +1903,10 @@ data SessionState t m k a b = SessionState

1903

1904-- XXX Perhaps we should use an "Event a" type to represent timestamped data.

1905

1906−-- | @classifySessionsBy tick timeout keepalive predicate fold stream@ classifies

1906+-- | @classifySessionsBy tick keepalive predicate timeout fold stream@

1907−-- an input event @stream@ consisting of @(key, value, timestamp)@ into

1907+-- classifies an input event @stream@ consisting of @(key, value, timestamp)@

1908−-- sessions based on the @key@, folding all the values corresponding to the

1908+-- into sessions based on the @key@, folding all the values corresponding to

1909−-- same key into a session using the supplied @fold@.

1909+-- the same key into a session using the supplied @fold@.

1910--

1911-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the

1912-- session key and the folded value is emitted in the output stream. The

@@ -1928,7 +1928,7 @@ data SessionState t m k a b = SessionState

1928--

1929-- >>> :{

1930-- Stream.mapM_ print

1931−-- $ Stream.classifySessionsBy 1 3 False (const (return False)) (Fold.takeLE 3 Fold.toList)

1931+-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.takeLE 3 Fold.toList)

1932-- $ Stream.map (\\(ts,(k,a)) -> (k, a, ts))

1933-- $ Stream.timestamped

1934-- $ Stream.delay 1

@@ -1944,13 +1944,13 @@ data SessionState t m k a b = SessionState

1944classifySessionsBy

1945 :: (IsStream t, MonadAsync m, Ord k)

1946 => Double -- ^ timer tick in seconds

1947− -> Double -- ^ session timeout in seconds

1948− -> Bool -- ^ reset the timeout when an event is received

1947+ -> Bool -- ^ reset the timer when an event is received

1949 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count

1948 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count

1949+ -> Double -- ^ session timeout in seconds

1950 -> Fold m a b -- ^ Fold to be applied to session events

1951 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp

1952 -> t m (k, b) -- ^ session key, fold result

1953−classifySessionsBy tick tmout reset ejectPred

1953+classifySessionsBy tick reset ejectPred tmout

1954 (Fold step initial extract) str =

1955 concatMap sessionOutputStream $

1956 scanlMAfter' sstep (return szero) flush stream

@@ -2165,7 +2165,7 @@ classifySessionsBy tick tmout reset ejectPred

2165-- option set to 'True'.

2166--

2167-- @

2168−-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred

2168+-- classifyKeepAliveSessions = classifySessionsBy 1 True

2169-- @

2170--

2171-- /Pre-release/

@@ -2173,13 +2173,12 @@ classifySessionsBy tick tmout reset ejectPred

2173{-# INLINABLE classifyKeepAliveSessions #-}

2174classifyKeepAliveSessions ::

2175 (IsStream t, MonadAsync m, Ord k)

2176− => Double -- ^ session inactive timeout

2177− -> (Int -> m Bool) -- ^ predicate to eject sessions on session count

2176+ => (Int -> m Bool) -- ^ predicate to eject sessions on session count

2177+ -> Double -- ^ session inactive timeout

2178 -> Fold m a b -- ^ Fold to be applied to session payload data

2179 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp

2180 -> t m (k, b)

2181−classifyKeepAliveSessions tmout =

2181+classifyKeepAliveSessions = classifySessionsBy 1 True

2182− classifySessionsBy 1 tmout True

2183

2182

2184------------------------------------------------------------------------------

2183------------------------------------------------------------------------------

2185-- Keyed tumbling windows

2184-- Keyed tumbling windows

@@ -2213,7 +2212,7 @@ classifyChunksOf wsize = classifyChunksBy wsize False

2213-- option set to 'False'.

2212-- option set to 'False'.

2214--

2213--

2215-- @

2214-- @

2216−-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred

2215+-- classifySessionsOf = classifySessionsBy 1 False

2217-- @

2216-- @

2218--

2217--

2219-- /Pre-release/

2218-- /Pre-release/

@@ -2221,13 +2220,12 @@ classifyChunksOf wsize = classifyChunksBy wsize False

2221{-# INLINABLE classifySessionsOf #-}

2220{-# INLINABLE classifySessionsOf #-}

2222classifySessionsOf ::

2221classifySessionsOf ::

2223 (IsStream t, MonadAsync m, Ord k)

2222 (IsStream t, MonadAsync m, Ord k)

2224− => Double -- ^ time window size

2225− -> (Int -> m Bool) -- ^ predicate to eject sessions on session count

2223+ => (Int -> m Bool) -- ^ predicate to eject sessions on session count

2224+ -> Double -- ^ time window size

2226 -> Fold m a b -- ^ Fold to be applied to session events

2225 -> Fold m a b -- ^ Fold to be applied to session events

2227 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp

2226 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp

2228 -> t m (k, b)

2227 -> t m (k, b)

2229−classifySessionsOf interval =

2230− classifySessionsBy 1 interval False

2228+classifySessionsOf = classifySessionsBy 1 False

2231

2229

2232------------------------------------------------------------------------------

2230------------------------------------------------------------------------------

2233-- Nested Split

2231-- Nested Split