2d6caf0 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Move `newCallbackStream` to StreamD.Generate.

2d6caf0Pranay Sashank6 years ago

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

compare with master

unifiedsplit

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

64 , fromProducer

65 , fromSVar

66 , toStreamD

67+

68+ , newCallbackStream

67 )

69 )

68where

70where

70−import Control.Concurrent (threadDelay)

72+import Control.Concurrent (myThreadId, threadDelay)

71−import Control.Monad (forever)

73+import Control.Monad (void, forever)

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

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

73import Data.Int (Int64)

75import Data.Int (Int64)

74import Streamly.Internal.Data.Time.Units

76import Streamly.Internal.Data.Time.Units

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

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

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

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

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

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

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

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

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

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

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

412 {-# INLINE_LATE step #-}

415 {-# INLINE_LATE step #-}

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

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

414 step _ [] = return Stop

417 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

89 -- * Concurrent Application

90 , mkParallel

91 , mkParallelD

92− , newCallbackStream

93 )

92 )

94where

93where

96−import Control.Concurrent (myThreadId)

97import Control.Monad (void)

95import Control.Monad (void)

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

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

99import Data.Functor.Identity (Identity(..))

97import Data.Functor.Identity (Identity(..))

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

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

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

904mkParallel = fromStreamD . mkParallelD . toStreamD

902mkParallel = fromStreamD . mkParallelD . toStreamD

905

903

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−

930------------------------------------------------------------------------------

904------------------------------------------------------------------------------

931-- Time related

905-- Time related

932------------------------------------------------------------------------------

906------------------------------------------------------------------------------