2d6caf0 · streamly · git.iko.soy

git.iko.soy/streamly

Move `newCallbackStream` to StreamD.Generate.

commit2d6caf0ea653cd4887f2012b50a9ef1e82267c3b authorPranay Sashank <pranaysashank@composewell.com> dateSun 3 Jan 2021, 01:54 (6 years ago) parent9c483a0

2 files changed+29−28

compare with master

unifiedsplit

src/Streamly/Internal/Data/Stream/StreamD/Generate.hs+29−2 src/Streamly/Internal/Data/Stream/StreamD/Transform.hs−26

src/Streamly/Internal/Data/Stream/StreamD/Generate.hs+29−2

@@ -64,11 +64,13 @@ module Streamly.Internal.Data.Stream.StreamD.Generate

6464 , fromProducer

6565 , fromSVar

6666 , toStreamD

67+

68+ , newCallbackStream

6769 )

6870where

6971

70−import Control.Concurrent (threadDelay)

71−import Control.Monad (forever)

72+import Control.Concurrent (myThreadId, threadDelay)

73+import Control.Monad (void, forever)

7274import Control.Monad.IO.Class (MonadIO(..))

7375import Data.Int (Int64)

7476import Streamly.Internal.Data.Time.Units

@@ -79,6 +81,7 @@ import Streamly.Internal.Data.Time.Units

7981import Streamly.Internal.Data.Unfold.Types (Unfold(..))

8082

8183import qualified Streamly.Internal.Data.IORef.Prim as Prim

84+import qualified Streamly.Internal.Data.Stream.StreamK as K

8285

8386import Prelude hiding (iterate, repeat, replicate, takeWhile)

8487import Streamly.Internal.Data.Stream.StreamD.Type

@@ -412,3 +415,27 @@ fromListM = Stream step

412415 {-# INLINE_LATE step #-}

413416 step _ (m:ms) = m >>= \x -> return $ Yield x ms

414417 step _ [] = return Stop

418+

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

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

421+--

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

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

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

425+--

426+-- /Internal/

427+--

428+{-# INLINE_NORMAL newCallbackStream #-}

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

430+newCallbackStream = do

431+ sv <- newParallelVar StopNone defState

432+

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

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

435+ -- handle this better.

436+ liftIO myThreadId >>= modifyThread sv

437+

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

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

440+ -- SVar?

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

src/Streamly/Internal/Data/Stream/StreamD/Transform.hs−26

@@ -89,11 +89,9 @@ module Streamly.Internal.Data.Stream.StreamD.Transform

8989 -- * Concurrent Application

9090 , mkParallel

9191 , mkParallelD

92− , newCallbackStream

9392 )

9493where

9594

96−import Control.Concurrent (myThreadId)

9795import Control.Monad (void)

9896import Control.Monad.IO.Class (MonadIO(..))

9997import Data.Functor.Identity (Identity(..))

@@ -903,30 +901,6 @@ mkParallelD m = Stream step Nothing

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

904902mkParallel = fromStreamD . mkParallelD . toStreamD

905903

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

907−-- the callback to be able to indicate end of stream.

908−--

909−-- | Generates a callback and a stream pair. The callback returned is used to

910−-- queue values to the stream. The stream is infinite, there is no way for the

911−-- callback to indicate that it is done now.

912−--

913−-- /Internal/

914−--

915−{-# INLINE_NORMAL newCallbackStream #-}

916−newCallbackStream :: (K.IsStream t, MonadAsync m) => m ((a -> m ()), t m a)

917−newCallbackStream = do

918− sv <- newParallelVar StopNone defState

919−

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

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

922− -- handle this better.

923− liftIO myThreadId >>= modifyThread sv

924−

925− let callback a = liftIO $ void $ send sv (ChildYield a)

926− -- XXX we can return an SVar and then the consumer can unfold from the

927− -- SVar?

928− return (callback, fromStreamD (fromSVar sv))

929−

930904------------------------------------------------------------------------------

931905-- Time related

932906------------------------------------------------------------------------------