@@ -20,14 +20,18 @@ module Asyncly.Streams
2020 (
2121 -- * Product Style Composition
2222 -- $product
23− StreamT
23+ Streaming (..)
24+ , MonadAsync
25+
26+ -- * Stream Styles
27+ , StreamT
2428 , InterleavedT
2529 , AsyncT
2630 , ParallelT
2731 , ZipStream
2832 , ZipAsync
2933
30− -- * Stream Type Adapters
34+ -- * Type Adapters
3135 , serially
3236 , interleaving
3337 , asyncly
@@ -36,6 +40,14 @@ module Asyncly.Streams
3640 , zippingAsync
3741 , adapt
3842
43+ -- * Construction
44+ , streamBuild
45+ , fromCallback
46+
47+ -- * Elimination
48+ , streamFold
49+ , runStreaming
50+
3951 -- * Running Streams
4052 , runStreamT
4153 , runInterleavedT
@@ -43,6 +55,23 @@ module Asyncly.Streams
4355 , runParallelT
4456 , runZipStream
4557 , 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
4675 )
4776where
4877
@@ -58,7 +87,6 @@ import Control.Monad.Trans.Class (MonadTrans)
5887import Data.Semigroup (Semigroup(..))
5988import Prelude hiding (drop, take, zipWith)
6089import Asyncly.Core
61−import Asyncly.Prelude
6290
6391-- $product
6492--
@@ -102,6 +130,16 @@ import Asyncly.Prelude
102130-- 'ParallelT' or 'ZipAsync') dictate how the elements of a stream are
103131-- processed when it is consumed.
104132
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+
105143------------------------------------------------------------------------------
106144-- StreamT iterates serially in the Monad and Applicative compositions
107145------------------------------------------------------------------------------
@@ -508,6 +546,19 @@ instance (MonadAsync m, Floating a) => Floating (ParallelT m a) where
508546-- Serially Zipping Streams
509547------------------------------------------------------------------------------
510548
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+
511562-- | 'ZipStream' zips serially i.e. it produces one element from each stream in
512563-- a serial manner and then zips the two elements.
513564newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}
@@ -574,6 +625,15 @@ instance (Monad m, Floating a) => Floating (ZipStream m a) where
574625-- Parallely Zipping Streams
575626------------------------------------------------------------------------------
576627
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+
577637-- | 'ZipAsync' zips in parallel, it produces one element from each stream
578638-- concurrently and then zips the two.
579639newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}
@@ -668,6 +728,47 @@ zipping x = x
668728zippingAsync :: ZipAsync m a -> ZipAsync m a
669729zippingAsync x = x
670730
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+
671772-------------------------------------------------------------------------------
672773-- Running Streams, convenience functions specialized to types
673774-------------------------------------------------------------------------------
@@ -695,3 +796,65 @@ runZipStream = runStreaming
695796-- | Run a 'ZipAsync' computation. Same as @runStreaming . zippingAsync@.
696797runZipAsync :: Monad m => ZipAsync m a -> m ()
697798runZipAsync = 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