8766fbf · streamly · git.iko.soy

git.iko.soy/streamly

Add concatM and fromCallback

commit8766fbfddec8bfce47af6617c14ff45733121f3a authorHarendra Kumar <harendra@composewell.com> dateTue 25 Feb 2020, 19:03 (7 years ago) parent8293e2d

2 files changed+49

compare with master

unifiedsplit

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

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

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

312312 -- * Concurrent Application

313313 , mkParallel

314314 , mkParallelD

315+ , newCallbackStream

315316

316317 , lastN

317318 )

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

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

40624063mkParallel = fromStreamD . mkParallelD . toStreamD

40634064

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+

40644089-------------------------------------------------------------------------------

40654090-- Concurrent tap

40664091-------------------------------------------------------------------------------

src/Streamly/Internal/Prelude.hs+24

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

6161 , K.fromFoldable

6262 , fromFoldableM

6363 , fromPrimVar

64+ , fromCallback

6465

6566 -- ** Time related

6667 , currentTime

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

277278

278279 -- ** Nested Streams

279280 , concatMapM

281+ , concatM

280282 , concatUnfold

281283 , concatUnfoldInterleave

282284 , concatUnfoldRoundrobin

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

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

906908fromPrimVar = fromStreamD . D.fromPrimVar

907909

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+

908923------------------------------------------------------------------------------

909924-- Time related

910925------------------------------------------------------------------------------

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

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

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

26142629

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+

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

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

26172641-- therefore provide many times better performance.