107c96f · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Move Streaming type from Core to Streams

107c96fHarendra Kumar9 years ago

src/Asyncly.hs+9−10

compare with master

unifiedsplit

@@ -54,7 +54,6 @@

54module Asyncly

55 ( MonadAsync

56 , Streaming

57− , EndOfStream

59 -- * Product Style Composition

58 -- * Product Style Composition

60 , StreamT

59 , StreamT

@@ -64,14 +63,6 @@ module Asyncly

64 , ZipStream

63 , ZipStream

65 , ZipAsync

64 , ZipAsync

67− -- * Sum Style Composition

68− -- $monoidal

69− , (<=>)

70− , (<|)

71−

72− -- * Custom Composition

73− , async

74−

75 -- * Stream Type Adapters

66 -- * Stream Type Adapters

76 , serially

67 , serially

77 , interleaving

68 , interleaving

@@ -90,6 +81,15 @@ module Asyncly

90 , runZipStream

81 , runZipStream

91 , runZipAsync

82 , runZipAsync

84+ -- * Transformation

85+ , EndOfStream

86+ , async

87+

88+ -- * Sum Style Composition

89+ -- $monoidal

90+ , (<=>)

91+ , (<|)

92+

93 -- * Fold Utilities

94 , foldWith

95 , foldMapWith

@@ -105,7 +105,6 @@ module Asyncly

105 )

106where

107

108−import Asyncly.Core

109import Asyncly.Streams

108import Asyncly.Streams

110import Data.Semigroup (Semigroup(..))

109import Data.Semigroup (Semigroup(..))

111import Control.Applicative (Alternative(..))

110import Control.Applicative (Alternative(..))

src/Asyncly/Core.hs+2−123

@@ -18,19 +18,12 @@

18module Asyncly.Core

19 ( Stream (..)

20 , MonadAsync

21− , Streaming (..)

23 -- * Construction

22 -- * Construction

24− , streamBuild

25 , yielding

23 , yielding

27− -- * Elimination

28− , streamFold

29− , runStreaming

30−

31 -- * Composition

25 -- * Composition

32 , interleave

26 , interleave

33− , (<=>)

35 -- * Asynchronous Operations

28 -- * Asynchronous Operations

36 , CtxType (..)

29 , CtxType (..)

@@ -40,13 +33,8 @@ module Asyncly.Core

40 , parallel

33 , parallel

41 , parAlt

34 , parAlt

42 , parLeft

35 , parLeft

43− , (<|)

36+ , pushOneToCtx

44− , async

37+ , pullFromCtx

45−

46− -- * Fold Utilities

47− , foldWith

48− , foldMapWith

49− , forEachWith

50 )

38 )

51where

39where

@@ -591,112 +579,3 @@ instance MonadState s m => MonadState s (Stream m) where

591 get = lift get

579 get = lift get

592 put x = lift (put x)

580 put x = lift (put x)

593 state k = lift (state k)

581 state k = lift (state k)

594−

595−------------------------------------------------------------------------------

596−-- Types that can behave as a Stream

597−------------------------------------------------------------------------------

598−

599−-- | Types that can represent a stream of elements.

600−class Streaming t where

601− toStream :: t m a -> Stream m a

602− fromStream :: Stream m a -> t m a

603−

604−------------------------------------------------------------------------------

605−-- Some basic type agnostic (working on the Streaming type class) APIs

606−------------------------------------------------------------------------------

607−

608−-- XXX The async API is useful for exploring each stream arbitrarily when

609−-- zipping or merging two streams. We can use a newtype wrapper with a monad

610−-- instance that composes like regular streaming libraries to facilitate linear

611−-- composition. We will also need a yield API for that.

612−

613−-- | Run a computation asynchronously, triggers the computation and returns

614−-- another computation that when executed produces the output generated by the

615−-- original computation. Note that the returned action must be executed exactly

616−-- once and drained completely. If not executed or not drained fully we may

617−-- have a thread blocked forever and if executed more than once a

618−-- 'EndOfStream' exception will be raised.

619−

620−async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)

621−async m = do

622− ctx <- pushOneToCtx (CtxType Disjunction LIFO) (toStream m)

623− return $ fromStream $ pullFromCtx ctx

624−

625−infixr 5 <=>

626−

627−-- | Sequential interleaved composition, in contrast to '<>' this operator

628−-- fairly interleaves the two streams instead of appending them; yielding one

629−-- element from each stream alternately. Unlike '<>' it cannot be used to fold

630−-- an infinite container of streams.

631−{-# INLINE (<=>) #-}

632−(<=>) :: Streaming t => t m a -> t m a -> t m a

633−m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)

634−

635−-- | Parallel interleaved composition, in contrast to '<|>' this operator

636−-- "merges" streams in a left biased manner rather than fairly interleaving

637−-- them. It keeps yielding from the stream on the left as long as it can. If

638−-- the left stream blocks or cannot keep up with the pace of the consumer it

639−-- can yield from the stream on the right in parallel. Unlike '<|>' it can be

640−-- used to fold infinite containers of streams.

641−{-# INLINE (<|) #-}

642−(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a

643−m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)

644−

645−------------------------------------------------------------------------------

646−-- Constructing a stream

647−------------------------------------------------------------------------------

648−

649−-- XXX Need to accept a context as well

650−-- | Build a stream from its church encoding.

651−streamBuild :: Streaming t

652− => (forall r. (a -> Maybe (t m a) -> m r) -> m r -> m r) -> t m a

653−streamBuild f = fromStream $ Stream $ \_ stp yld ->

654− let yield a Nothing = yld a Nothing

655− yield a (Just r) = yld a (Just (toStream r))

656− in f yield stp

657−

658−------------------------------------------------------------------------------

659−-- Destroying a stream

660−------------------------------------------------------------------------------

661−

662−-- | Fold a stream using its church encoding. The first argument is the

663−-- "step" function consuming one element and the rest of the stream. The second

664−-- argument is the "done" function that is called when the stream is over.

665−streamFold :: Streaming t => (a -> Maybe (t m a) -> m r) -> m r -> t m a -> m r

666−streamFold step done m =

667− let yield a Nothing = step a Nothing

668− yield a (Just x) = step a (Just (fromStream x))

669− in (runStream (toStream m)) Nothing done yield

670−

671−-- | Run a streaming composition, discard the results.

672−runStreaming :: (Monad m, Streaming t) => t m a -> m ()

673−runStreaming m = go (toStream m)

674− where

675− go m1 =

676− let stop = return ()

677− yield _ Nothing = stop

678− yield _ (Just x) = go x

679− in (runStream m1) Nothing stop yield

680−

681−------------------------------------------------------------------------------

682−-- Utilities

683−------------------------------------------------------------------------------

684−

685−-- | Fold a 'Foldable' container using the given function.

686−{-# INLINABLE foldWith #-}

687−foldWith :: (Monoid b, Foldable t) => (a -> b -> b) -> t a -> b

688−foldWith f = foldr f mempty

689−

690−-- | Fold a 'Foldable' container using a function that is a composition of the

691−-- two arguments.

692−{-# INLINABLE foldMapWith #-}

693−foldMapWith :: (Monoid b, Foldable t) =>

694− (b1 -> b -> b) -> (a -> b1) -> t a -> b

695−foldMapWith f g = foldr (f . g) mempty

696−

697−-- | Fold a 'Foldable' container using a function that is a composition of the

698−-- first and the third argument.

699−{-# INLINABLE forEachWith #-}

700−forEachWith :: (Monoid b, Foldable t) =>

701− (b1 -> b -> b) -> t a -> (a -> b1) -> b

702−forEachWith f xs g = foldr (f . g) mempty xs

src/Asyncly/Prelude.hs+1−26

@@ -49,6 +49,7 @@ import Data.Semigroup (Semigroup(..))

49import Prelude hiding (drop, take, zipWith, foldr,

50 foldl)

51import Asyncly.Core

52+import Asyncly.Streams

53------------------------------------------------------------------------------

54------------------------------------------------------------------------------

54-- Elimination

55-- Elimination

@@ -123,10 +124,6 @@ unfoldr step = fromStream . go

123 Nothing -> stp

124 Nothing -> stp

124 Just (a, b) -> yld a (Just (go b))

125 Just (a, b) -> yld a (Just (go b))

125

126

126−-- | Convert a callback into a stream.

127−fromCallback :: (forall r. (a -> m r) -> m r) -> Stream m a

128−fromCallback k = Stream $ \_ _ yld -> k (\a -> yld a Nothing)

129−

130------------------------------------------------------------------------------

127------------------------------------------------------------------------------

131-- Special folds

128-- Special folds

132------------------------------------------------------------------------------

129------------------------------------------------------------------------------

@@ -177,19 +174,6 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

177 (runStream mx) Nothing stp yield1

174 (runStream mx) Nothing stp yield1

178 g a b = toStream $ f a b

175 g a b = toStream $ f a b

179

176

180−-- | Zip two streams serially using a pure zipping function.

181−zipWith :: Streaming t => (a -> b -> c) -> t m a -> t m b -> t m c

182−zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

183− where

184− go mx my = Stream $ \_ stp yld -> do

185− let merge a ra =

186− let yield2 b Nothing = yld (f a b) Nothing

187− yield2 b (Just rb) = yld (f a b) (Just (go ra rb))

188− in (runStream my) Nothing stp yield2

189− let yield1 a Nothing = merge a mempty

190− yield1 a (Just ra) = merge a ra

191− (runStream mx) Nothing stp yield1

192−

193------------------------------------------------------------------------------

177------------------------------------------------------------------------------

194-- Parallely Zipping Streams

178-- Parallely Zipping Streams

195------------------------------------------------------------------------------

179------------------------------------------------------------------------------

@@ -202,12 +186,3 @@ zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp yld -> do

202 ma <- async m1

186 ma <- async m1

203 mb <- async m2

187 mb <- async m2

204 (runStream (toStream (zipWithM f ma mb))) Nothing stp yld

188 (runStream (toStream (zipWithM f ma mb))) Nothing stp yld

205−

206−-- | Zip two streams asyncly (i.e. both the elements being zipped are generated

207−-- concurrently) using a pure zipping function.

208−zipAsyncWith :: (Streaming t, MonadAsync m)

209− => (a -> b -> c) -> t m a -> t m b -> t m c

210−zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do

211− ma <- async m1

212− mb <- async m2

213− (runStream (toStream (zipWith f ma mb))) Nothing stp yld

src/Asyncly/Streams.hs+166−3

@@ -20,14 +20,18 @@ module Asyncly.Streams

20 (

21 -- * Product Style Composition

22 -- $product

23+ Streaming (..)

24+ , MonadAsync

25+

26+ -- * Stream Styles

23− StreamT

27+ , StreamT

24 , InterleavedT

28 , InterleavedT

25 , AsyncT

29 , AsyncT

26 , ParallelT

30 , ParallelT

27 , ZipStream

31 , ZipStream

28 , ZipAsync

32 , ZipAsync

30− -- * Stream Type Adapters

34+ -- * Type Adapters

31 , serially

35 , serially

32 , interleaving

36 , interleaving

33 , asyncly

37 , asyncly

@@ -36,6 +40,14 @@ module Asyncly.Streams

36 , zippingAsync

40 , zippingAsync

37 , adapt

41 , adapt

43+ -- * Construction

44+ , streamBuild

45+ , fromCallback

46+

47+ -- * Elimination

48+ , streamFold

49+ , runStreaming

50+

39 -- * Running Streams

51 -- * Running Streams

40 , runStreamT

52 , runStreamT

41 , runInterleavedT

53 , runInterleavedT

@@ -43,6 +55,23 @@ module Asyncly.Streams

43 , runParallelT

55 , runParallelT

44 , runZipStream

56 , runZipStream

45 , runZipAsync

57 , runZipAsync

58+

59+ -- * Transformation

60+ , EndOfStream (..)

61+ , async

62+

63+ -- * Zipping

64+ , zipWith

65+ , zipAsyncWith

66+

67+ -- * Sum Style Composition

68+ , (<=>)

69+ , (<|)

70+

71+ -- * Fold Utilities

72+ , foldWith

73+ , foldMapWith

74+ , forEachWith

46 )

75 )

47where

76where

@@ -58,7 +87,6 @@ import Control.Monad.Trans.Class (MonadTrans)

58import Data.Semigroup (Semigroup(..))

87import Data.Semigroup (Semigroup(..))

59import Prelude hiding (drop, take, zipWith)

88import Prelude hiding (drop, take, zipWith)

60import Asyncly.Core

89import Asyncly.Core

61−import Asyncly.Prelude

63-- $product

91-- $product

64--

92--

@@ -102,6 +130,16 @@ import Asyncly.Prelude

102-- 'ParallelT' or 'ZipAsync') dictate how the elements of a stream are

130-- 'ParallelT' or 'ZipAsync') dictate how the elements of a stream are

103-- processed when it is consumed.

131-- processed when it is consumed.

104

132

133+------------------------------------------------------------------------------

134+-- Types that can behave as a Stream

135+------------------------------------------------------------------------------

136+

137+-- | Class of types that can represent a stream of elements of some type 'a' in

138+-- some monad 'm'.

139+class Streaming t where

140+ toStream :: t m a -> Stream m a

141+ fromStream :: Stream m a -> t m a

142+

105------------------------------------------------------------------------------

143------------------------------------------------------------------------------

106-- StreamT iterates serially in the Monad and Applicative compositions

144-- StreamT iterates serially in the Monad and Applicative compositions

107------------------------------------------------------------------------------

145------------------------------------------------------------------------------

@@ -508,6 +546,19 @@ instance (MonadAsync m, Floating a) => Floating (ParallelT m a) where

508-- Serially Zipping Streams

546-- Serially Zipping Streams

509------------------------------------------------------------------------------

547------------------------------------------------------------------------------

510

548

549+-- | Zip two streams serially using a pure zipping function.

550+zipWith :: Streaming t => (a -> b -> c) -> t m a -> t m b -> t m c

551+zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

552+ where

553+ go mx my = Stream $ \_ stp yld -> do

554+ let merge a ra =

555+ let yield2 b Nothing = yld (f a b) Nothing

556+ yield2 b (Just rb) = yld (f a b) (Just (go ra rb))

557+ in (runStream my) Nothing stp yield2

558+ let yield1 a Nothing = merge a mempty

559+ yield1 a (Just ra) = merge a ra

560+ (runStream mx) Nothing stp yield1

561+

511-- | 'ZipStream' zips serially i.e. it produces one element from each stream in

562-- | 'ZipStream' zips serially i.e. it produces one element from each stream in

512-- a serial manner and then zips the two elements.

563-- a serial manner and then zips the two elements.

513newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}

564newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}

@@ -574,6 +625,15 @@ instance (Monad m, Floating a) => Floating (ZipStream m a) where

574-- Parallely Zipping Streams

625-- Parallely Zipping Streams

575------------------------------------------------------------------------------

626------------------------------------------------------------------------------

576

627

628+-- | Zip two streams asyncly (i.e. both the elements being zipped are generated

629+-- concurrently) using a pure zipping function.

630+zipAsyncWith :: (Streaming t, MonadAsync m)

631+ => (a -> b -> c) -> t m a -> t m b -> t m c

632+zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do

633+ ma <- async m1

634+ mb <- async m2

635+ (runStream (toStream (zipWith f ma mb))) Nothing stp yld

636+

577-- | 'ZipAsync' zips in parallel, it produces one element from each stream

637-- | 'ZipAsync' zips in parallel, it produces one element from each stream

578-- concurrently and then zips the two.

638-- concurrently and then zips the two.

579newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}

639newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}

@@ -668,6 +728,47 @@ zipping x = x

668zippingAsync :: ZipAsync m a -> ZipAsync m a

728zippingAsync :: ZipAsync m a -> ZipAsync m a

669zippingAsync x = x

729zippingAsync x = x

670

730

731+------------------------------------------------------------------------------

732+-- Constructing a stream

733+------------------------------------------------------------------------------

734+

735+-- XXX Need to accept a context as well

736+-- | Build a stream from its church encoding. The function passed maps

737+-- directly to the underlying representation of the stream type.

738+streamBuild :: Streaming t

739+ => (forall r. (a -> Maybe (t m a) -> m r) -> m r -> m r) -> t m a

740+streamBuild k = fromStream $ Stream $ \_ stp yld ->

741+ let yield a Nothing = yld a Nothing

742+ yield a (Just r) = yld a (Just (toStream r))

743+ in k yield stp

744+

745+-- | Convert a callback into a stream.

746+fromCallback :: (Streaming t) => (forall r. (a -> m r) -> m r) -> t m a

747+fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)

748+

749+------------------------------------------------------------------------------

750+-- Destroying a stream

751+------------------------------------------------------------------------------

752+

753+-- | Fold a stream using its church encoding. The first argument is the

754+-- "step" function consuming one element and the rest of the stream. The second

755+-- argument is the "done" function that is called when the stream is over.

756+streamFold :: Streaming t => (a -> Maybe (t m a) -> m r) -> m r -> t m a -> m r

757+streamFold step done m =

758+ let yield a Nothing = step a Nothing

759+ yield a (Just x) = step a (Just (fromStream x))

760+ in (runStream (toStream m)) Nothing done yield

761+

762+-- | Run a streaming composition, discard the results.

763+runStreaming :: (Monad m, Streaming t) => t m a -> m ()

764+runStreaming m = go (toStream m)

765+ where

766+ go m1 =

767+ let stop = return ()

768+ yield _ Nothing = stop

769+ yield _ (Just x) = go x

770+ in (runStream m1) Nothing stop yield

771+

671-------------------------------------------------------------------------------

772-------------------------------------------------------------------------------

672-- Running Streams, convenience functions specialized to types

773-- Running Streams, convenience functions specialized to types

673-------------------------------------------------------------------------------

774-------------------------------------------------------------------------------

@@ -695,3 +796,65 @@ runZipStream = runStreaming

695-- | Run a 'ZipAsync' computation. Same as @runStreaming . zippingAsync@.

796-- | Run a 'ZipAsync' computation. Same as @runStreaming . zippingAsync@.

696runZipAsync :: Monad m => ZipAsync m a -> m ()

797runZipAsync :: Monad m => ZipAsync m a -> m ()

697runZipAsync = runStreaming

798runZipAsync = runStreaming

799+

800+------------------------------------------------------------------------------

801+-- Transformation

802+------------------------------------------------------------------------------

803+

804+-- | Make a stream asynchronous, triggers the computation and returns a stream

805+-- in the underlying monad representing the output generated by the original

806+-- computation. The returned action must be drained once and only once. If not

807+-- drained fully we may have a thread blocked forever and once drained it will

808+-- throw an 'EndOfStream' exception if we try to run it again

809+

810+async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)

811+async m = do

812+ ctx <- pushOneToCtx (CtxType Disjunction LIFO) (toStream m)

813+ return $ fromStream $ pullFromCtx ctx

814+

815+------------------------------------------------------------------------------

816+-- Sum Style Composition

817+------------------------------------------------------------------------------

818+

819+infixr 5 <=>

820+

821+-- | Sequential interleaved composition, in contrast to '<>' this operator

822+-- fairly interleaves the two streams instead of appending them; yielding one

823+-- element from each stream alternately. Unlike '<>' it cannot be used to fold

824+-- an infinite container of streams.

825+{-# INLINE (<=>) #-}

826+(<=>) :: Streaming t => t m a -> t m a -> t m a

827+m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)

828+

829+-- | Parallel interleaved composition, in contrast to '<|>' this operator

830+-- "merges" streams in a left biased manner rather than fairly interleaving

831+-- them. It keeps yielding from the stream on the left as long as it can. If

832+-- the left stream blocks or cannot keep up with the pace of the consumer it

833+-- can yield from the stream on the right in parallel. Unlike '<|>' it can be

834+-- used to fold infinite containers of streams.

835+{-# INLINE (<|) #-}

836+(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a

837+m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)

838+

839+------------------------------------------------------------------------------

840+-- Fold Utilities

841+------------------------------------------------------------------------------

842+

843+-- | Fold a 'Foldable' container using the given function.

844+{-# INLINABLE foldWith #-}

845+foldWith :: (Monoid b, Foldable t) => (a -> b -> b) -> t a -> b

846+foldWith f = foldr f mempty

847+

848+-- | Fold a 'Foldable' container using a function that is a composition of the

849+-- two arguments.

850+{-# INLINABLE foldMapWith #-}

851+foldMapWith :: (Monoid b, Foldable t) =>

852+ (b1 -> b -> b) -> (a -> b1) -> t a -> b

853+foldMapWith f g = foldr (f . g) mempty

854+

855+-- | Fold a 'Foldable' container using a function that is a composition of the

856+-- first and the third argument.

857+{-# INLINABLE forEachWith #-}

858+forEachWith :: (Monoid b, Foldable t) =>

859+ (b1 -> b -> b) -> t a -> (a -> b1) -> b

860+forEachWith f xs g = foldr (f . g) mempty xs