7013f3a · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Fix delayPost, add more interspersing combinators

7013f3aHarendra Kumar6 years ago

src/Streamly/Internal/Data/Stream/StreamD.hs+30

compare with master

unifiedsplit

@@ -269,8 +269,10 @@ module Streamly.Internal.Data.Stream.StreamD

269

270 -- * Inserting

271 , intersperseM

272+ , intersperseM_

272 , intersperse

273 , intersperse

273 , intersperseSuffix

274 , intersperseSuffix

275+ , intersperseSuffix_

274 , intersperseSuffixBySpan

276 , intersperseSuffixBySpan

275 , insertBy

277 , insertBy

276

278

@@ -4023,6 +4025,20 @@ intersperseM m (Stream step state) = Stream step' (FirstYield state)

4023

4025

4024 step' _ (YieldAndCarry x st) = return $ Yield x (InterspersingYield st)

4026 step' _ (YieldAndCarry x st) = return $ Yield x (InterspersingYield st)

4025

4027

4028+{-# INLINE_NORMAL intersperseM_ #-}

4029+intersperseM_ :: Monad m => m b -> Stream m a -> Stream m a

4030+intersperseM_ m (Stream step1 state1) = Stream step (Left (pure (), state1))

4031+ where

4032+ {-# INLINE_LATE step #-}

4033+ step gst (Left (eff, st)) = do

4034+ r <- step1 gst st

4035+ case r of

4036+ Yield x s -> eff >> return (Yield x (Right s))

4037+ Skip s -> return $ Skip (Left (eff, s))

4038+ Stop -> return Stop

4039+

4040+ step _ (Right st) = return $ Skip $ Left (void m, st)

4041+

4026data SuffixState s a

4042data SuffixState s a

4027 = SuffixElem s

4043 = SuffixElem s

4028 | SuffixSuffix s

4044 | SuffixSuffix s

@@ -4045,6 +4061,20 @@ intersperseSuffix action (Stream step state) = Stream step' (SuffixElem state)

4045

4061

4046 step' _ (SuffixYield x next) = return $ Yield x next

4062 step' _ (SuffixYield x next) = return $ Yield x next

4047

4063

4064+{-# INLINE_NORMAL intersperseSuffix_ #-}

4065+intersperseSuffix_ :: Monad m => m b -> Stream m a -> Stream m a

4066+intersperseSuffix_ m (Stream step1 state1) = Stream step (Left state1)

4067+ where

4068+ {-# INLINE_LATE step #-}

4069+ step gst (Left st) = do

4070+ r <- step1 gst st

4071+ case r of

4072+ Yield x s -> return $ Yield x (Right s)

4073+ Skip s -> return $ Skip $ Left s

4074+ Stop -> m >> return Stop

4075+

4076+ step _ (Right st) = m >> return (Skip (Left st))

4077+

4048data SuffixSpanState s a

4078data SuffixSpanState s a

4049 = SuffixSpanElem s Int

4079 = SuffixSpanElem s Int

4050 | SuffixSpanSuffix s

4080 | SuffixSpanSuffix s

src/Streamly/Internal/Prelude.hs+94−15

@@ -191,6 +191,7 @@ module Streamly.Internal.Prelude

191 -- ** Special Maps

192 , mapM_

193 , trace

194+ , trace_

194 , tap

195 , tap

195 , tapOffsetEvery

196 , tapOffsetEvery

196 , tapAsync

197 , tapAsync

@@ -252,8 +253,10 @@ module Streamly.Internal.Prelude

252

253

253 , insertBy

254 , insertBy

254 , intersperseM

255 , intersperseM

256+ , intersperseM_

255 , intersperse

257 , intersperse

256 , intersperseSuffix

258 , intersperseSuffix

259+ , interspersePrefix_

257 , intersperseSuffixBySpan

260 , intersperseSuffixBySpan

258 -- , intersperseBySpan

261 -- , intersperseBySpan

259 -- , intersperseByIndices -- using an index function/stream

262 -- , intersperseByIndices -- using an index function/stream

@@ -263,6 +266,7 @@ module Streamly.Internal.Prelude

263 -- , intersperseByEvent

266 -- , intersperseByEvent

264 , interjectSuffix

267 , interjectSuffix

265 , delayPost

268 , delayPost

269+ , delayPre

266

270

267 -- ** Indexing

271 -- ** Indexing

268 , indexed

272 , indexed

@@ -2458,13 +2462,21 @@ reverse' s = fromStreamD $ D.reverse' $ toStreamD s

2458

2462

2459-- intersperseM = intersperseBySpan 1

2463-- intersperseM = intersperseBySpan 1

2460

2464

2461−-- | Generate a stream by inserting the result of a monadic action between

2465+-- | Insert the result of an effectful action between consecutive elements of

2462−-- consecutive elements of the given stream. Note that the monadic action is

2466+-- the stream.

2463−-- performed after the stream action before which its result is inserted.

2464--

2467--

2465-- @

2468-- @

2469+-- >>> S.toList $ S.trace putChar $ S.intersperseM (putChar '.' >> return ',') $ S.fromList "hello"

2470+-- > h.,e.,l.,l.,o"h,e,l,l,o"

2471+-- @

2472+--

2473+-- After yielding the first element, for every next element the effectful

2474+-- action is evaluated /after/ generating the element, however, its result is

2475+-- yielded /before/ the element.

2476+--

2477+-- @

2466−-- > S.toList $ S.intersperseM (return ',') $ S.fromList "hello"

2478+-- > S.toList $ S.intersperseM (putChar '.' >> return ',') $ S.trace putChar $ S.fromList "hello"

2467−-- "h,e,l,l,o"

2479+-- he.l.l.o."h,e,l,l,o"

2468-- @

2480-- @

2469--

2481--

2470-- @since 0.5.0

2482-- @since 0.5.0

@@ -2472,8 +2484,27 @@ reverse' s = fromStreamD $ D.reverse' $ toStreamD s

2472intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

2484intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

2473intersperseM m = fromStreamS . S.intersperseM m . toStreamS

2485intersperseM m = fromStreamS . S.intersperseM m . toStreamS

2474

2486

2475−-- | Generate a stream by inserting a given element between consecutive

2487+-- | Insert a side effect between consecutive elements of the stream.

2488+--

2489+-- @

2490+-- >>> S.drain $ S.trace putChar $ S.intersperseM_ (putChar '.') $ S.fromList "hello"

2491+-- > h.e.l.l.o

2492+-- @

2493+--

2494+-- After yielding the first element, for every next element the effect

2495+-- is evaluated after generating the element but before yielding it.

2496+--

2497+-- @

2498+-- >>> S.drain $ S.intersperseM_ (putChar '.') $ S.trace putChar $ S.fromList "hello"

2499+-- > he.l.l.o.

2500+-- @

2501+--

2502+-- /Internal/

2503+{-# INLINE intersperseM_ #-}

2504+intersperseM_ :: (IsStream t, MonadAsync m) => m b -> t m a -> t m a

2505+intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD

2506+

2476−-- elements of the given stream.

2507+-- | Insert a pure value between successive elements of a stream.

2477--

2508--

2478-- @

2509-- @

2479-- > S.toList $ S.intersperse ',' $ S.fromList "hello"

2510-- > S.toList $ S.intersperse ',' $ S.fromList "hello"

@@ -2485,16 +2516,19 @@ intersperseM m = fromStreamS . S.intersperseM m . toStreamS

2485intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a

2516intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a

2486intersperse a = fromStreamS . S.intersperse a . toStreamS

2517intersperse a = fromStreamS . S.intersperse a . toStreamS

2487

2518

2488−-- | Insert a monadic action after each element in the stream.

2519+-- | Insert an effectful action after each element of a stream.

2520+--

2521+-- @

2522+-- >>> S.toList $ S.trace putChar $ S.intersperseSuffix (putChar '.' >> return ',') $ S.fromList "hello"

2523+-- > h.,e.,l.,l.,o.,"h,e,l,l,o,"

2524+-- @

2489--

2525--

2490-- /Internal/

2526-- /Internal/

2491{-# INLINE intersperseSuffix #-}

2527{-# INLINE intersperseSuffix #-}

2492intersperseSuffix :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

2528intersperseSuffix :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

2493intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD

2529intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD

2494

2530

2495−-- | Perform a side effect after each element of a stream. The output of the

2531+-- | Insert a side effect after each element of a stream.

2496−-- effectful action is discarded, therefore, the input stream remains

2497−-- unchanged.

2498--

2532--

2499-- @

2533-- @

2500-- > S.mapM_ putChar $ S.intersperseSuffix_ (threadDelay 1000000) $ S.fromList "hello"

2534-- > S.mapM_ putChar $ S.intersperseSuffix_ (threadDelay 1000000) $ S.fromList "hello"

@@ -2504,9 +2538,28 @@ intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD

2504--

2538--

2505{-# INLINE intersperseSuffix_ #-}

2539{-# INLINE intersperseSuffix_ #-}

2506intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a

2540intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a

2541+intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD

2542+

2543+-- | Insert a side effect before each element of a stream.

2544+--

2545+-- @

2546+-- >>> S.toList $ S.trace putChar $ S.interspersePrefix_ (putChar '.' >> return ',') $ S.fromList "hello"

2547+-- > .h.e.l.l.o"hello"

2548+-- @

2549+--

2550+-- See also: 'trace_'

2551+--

2552+-- /Concurrent/

2553+--

2554+-- /Internal/

2555+--

2556+{-# INLINE interspersePrefix_ #-}

2557+interspersePrefix_ :: (IsStream t, MonadAsync m) => m b -> t m a -> t m a

2507−intersperseSuffix_ m = Serial.mapM (\x -> void m >> return x)

2558+interspersePrefix_ m = mapM (\x -> void m >> return x)

2508

2559

2560+-- Note: delay must be serial.

2561+--

2509−-- | Introduces a delay of specified seconds after each element of a stream.

2562+-- | Introduce a delay of specified seconds after each element of a stream.

2510--

2563--

2511-- /Internal/

2564-- /Internal/

2512--

2565--

@@ -2514,8 +2567,18 @@ intersperseSuffix_ m = Serial.mapM (\x -> void m >> return x)

2514delayPost :: (IsStream t, MonadIO m) => Double -> t m a -> t m a

2567delayPost :: (IsStream t, MonadIO m) => Double -> t m a -> t m a

2515delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000

2568delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000

2516

2569

2570+-- Note: delay must be serial, that's why 'trace_' is used.

2571+--

2572+-- | Introduce a delay of specified seconds before each element of a stream.

2573+--

2574+-- /Internal/

2575+--

2576+{-# INLINE delayPre #-}

2577+delayPre :: (IsStream t, MonadIO m) => Double -> t m a -> t m a

2578+delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000

2579+

2517−-- | Like 'intersperseSuffix' but intersperses a monadic action into the input

2580+-- | Like 'intersperseSuffix' but intersperses an effectful action into the

2518−-- stream after every @n@ elements and after the last element.

2581+-- input stream after every @n@ elements and after the last element.

2519--

2582--

2520-- @

2583-- @

2521-- > S.toList $ S.intersperseSuffixBySpan 2 (return ',') $ S.fromList "hello"

2584-- > S.toList $ S.intersperseSuffixBySpan 2 (return ',') $ S.fromList "hello"

@@ -4094,6 +4157,22 @@ tapRate n f xs = D.fromStreamD $ D.tapRate n f $ D.toStreamD xs

4094trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a

4157trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a

4095trace f = mapM (\x -> void (f x) >> return x)

4158trace f = mapM (\x -> void (f x) >> return x)

4096

4159

4160+-- | Perform a side effect before yielding each element of the stream and

4161+-- discard the results.

4162+--

4163+-- @

4164+-- > S.drain $ S.trace_ (print "got here") (S.enumerateFromTo 1 2)

4165+-- "got here"

4166+-- "got here"

4167+-- @

4168+--

4169+-- See also: 'trace', 'interspersePrefix_'

4170+--

4171+-- /Internal/

4172+{-# INLINE trace_ #-}

4173+trace_ :: (IsStream t, Monad m) => m b -> t m a -> t m a

4174+trace_ eff = Serial.mapM (\x -> eff >> return x)

4175+

4097------------------------------------------------------------------------------

4176------------------------------------------------------------------------------

4098-- Windowed classification

4177-- Windowed classification

4099------------------------------------------------------------------------------

4178------------------------------------------------------------------------------