8766fbf · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add concatM and fromCallback

8766fbfHarendra Kumar7 years ago

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

compare with master

unifiedsplit

@@ -312,6 +312,7 @@ module Streamly.Internal.Data.Stream.StreamD

312 -- * Concurrent Application

313 , mkParallel

314 , mkParallelD

315+ , newCallbackStream

315

316

316 , lastN

317 , lastN

317 )

318 )

@@ -4061,6 +4062,30 @@ mkParallelD m = Stream step Nothing

4061mkParallel :: (K.IsStream t, MonadAsync m) => t m a -> t m a

4062mkParallel :: (K.IsStream t, MonadAsync m) => t m a -> t m a

4062mkParallel = fromStreamD . mkParallelD . toStreamD

4063mkParallel = fromStreamD . mkParallelD . toStreamD

4063

4064

4065+-- Note: we can use another API with two callbacks stop and yield if we want

4066+-- the callback to be able to indicate end of stream.

4067+--

4068+-- | Generates a callback and a stream pair. The callback returned is used to

4069+-- queue values to the stream. The stream is infinite, there is no way for the

4070+-- callback to indicate that it is done now.

4071+--

4072+-- /Internal/

4073+--

4074+{-# INLINE_NORMAL newCallbackStream #-}

4075+newCallbackStream :: (K.IsStream t, MonadAsync m) => m ((a -> m ()), t m a)

4076+newCallbackStream = do

4077+ sv <- newParallelVar StopNone defState

4078+

4079+ -- XXX Add our own thread-id to the SVar as we can not know the callback's

4080+ -- thread-id and the callback is not run in a managed worker. We need to

4081+ -- handle this better.

4082+ liftIO myThreadId >>= modifyThread sv

4083+

4084+ let callback a = liftIO $ void $ send sv (ChildYield a)

4085+ -- XXX we can return an SVar and then the consumer can unfold from the

4086+ -- SVar?

4087+ return (callback, fromStreamD (fromSVar sv))

4088+

4064-------------------------------------------------------------------------------

4089-------------------------------------------------------------------------------

4065-- Concurrent tap

4090-- Concurrent tap

4066-------------------------------------------------------------------------------

4091-------------------------------------------------------------------------------

src/Streamly/Internal/Prelude.hs+24

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

61 , K.fromFoldable

62 , fromFoldableM

63 , fromPrimVar

64+ , fromCallback

65 -- ** Time related

66 -- ** Time related

66 , currentTime

67 , currentTime

@@ -277,6 +278,7 @@ module Streamly.Internal.Prelude

277

278

278 -- ** Nested Streams

279 -- ** Nested Streams

279 , concatMapM

280 , concatMapM

281+ , concatM

280 , concatUnfold

282 , concatUnfold

281 , concatUnfoldInterleave

283 , concatUnfoldInterleave

282 , concatUnfoldRoundrobin

284 , concatUnfoldRoundrobin

@@ -905,6 +907,19 @@ fromHandle h = go

905fromPrimVar :: (IsStream t, MonadIO m, Prim a) => Var IO a -> t m a

907fromPrimVar :: (IsStream t, MonadIO m, Prim a) => Var IO a -> t m a

906fromPrimVar = fromStreamD . D.fromPrimVar

908fromPrimVar = fromStreamD . D.fromPrimVar

907

909

910+-- | Takes a callback setter function and provides it with a callback. The

911+-- callback when invoked adds a value at the tail of the stream. Returns a

912+-- stream of values generated by the callback.

913+--

914+-- /Internal/

915+--

916+{-# INLINE fromCallback #-}

917+fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> SerialT m a

918+fromCallback setCallback = concatM $ do

919+ (callback, stream) <- D.newCallbackStream

920+ setCallback callback

921+ return stream

922+

908------------------------------------------------------------------------------

923------------------------------------------------------------------------------

909-- Time related

924-- Time related

910------------------------------------------------------------------------------

925------------------------------------------------------------------------------

@@ -2612,6 +2627,15 @@ roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)

2612concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b

2627concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b

2613concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)

2628concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)

2614

2629

2630+-- | Given a stream value in the underlying monad, lift and join the underlying

2631+-- monad with the stream monad.

2632+--

2633+-- /Internal/

2634+--

2635+{-# INLINE concatM #-}

2636+concatM :: (IsStream t, Monad m) => m (t m a) -> t m a

2637+concatM generator = concatMapM (\() -> generator) (K.yield ())

2638+

2615-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike

2639-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike

2616-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and

2640-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and

2617-- therefore provide many times better performance.

2641-- therefore provide many times better performance.