@@ -19,12 +19,16 @@ module Asyncly.Core
1919 ( Stream (..)
2020 , MonadAsync
2121 , Streaming (..)
22− , EndOfStream (..)
2322
24− -- * Running the Monad
23+ -- * Construction
24+ , streamBuild
25+ , yielding
26+
27+ -- * Elimination
28+ , streamFold
2529 , runStreaming
2630
27− -- * Interleaving
31+ -- * Composition
2832 , interleave
2933 , (<=>)
3034
@@ -32,7 +36,7 @@ module Asyncly.Core
3236 , CtxType (..)
3337 , SchedPolicy (..)
3438 , Dimension (..)
35− , yielding
39+ , EndOfStream (..)
3640 , parallel
3741 , parAlt
3842 , parLeft
@@ -119,6 +123,8 @@ data Context m a =
119123-- The stream type
120124------------------------------------------------------------------------------
121125
126+-- TBD use a functor instead of the bare type a?
127+
122128-- | Represents a monadic stream of values of type 'a' constructed using
123129-- actions in monad 'm'. Streams can be composed sequentially or in parallel in
124130-- product style compositions (monadic bind multiplies streams in a ListT
@@ -596,7 +602,7 @@ class Streaming t where
596602 fromStream :: Stream m a -> t m a
597603
598604------------------------------------------------------------------------------
599−-- Some basic stream type agnostic APIs
605+-- Some basic type agnostic (working on the Streaming type class) APIs
600606------------------------------------------------------------------------------
601607
602608-- XXX The async API is useful for exploring each stream arbitrarily when
@@ -609,7 +615,7 @@ class Streaming t where
609615-- original computation. Note that the returned action must be executed exactly
610616-- once and drained completely. If not executed or not drained fully we may
611617-- have a thread blocked forever and if executed more than once a
612−-- 'ContextUsedAfterEOF' exception will be raised.
618+-- 'EndOfStream' exception will be raised.
613619
614620async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
615621async m = do
@@ -637,22 +643,40 @@ m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
637643m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
638644
639645------------------------------------------------------------------------------
640−-- Running the monad
646+-- Constructing a stream
641647------------------------------------------------------------------------------
642648
643−-- | Run a composed streaming computation, discard the results.
644−runStreaming :: (Monad m, Streaming t) => t m a -> m ()
645−runStreaming m = go (toStream m)
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
646657
647− where
648−
649− go m1 = (runStream m1) Nothing stop yield
658+------------------------------------------------------------------------------
659+-- Destroying a stream
660+------------------------------------------------------------------------------
650661
651− stop = return ()
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
652670
653− {-# INLINE yield #-}
654− yield _ Nothing = stop
655− yield _ (Just x) = go x
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
656680
657681------------------------------------------------------------------------------
658682-- Utilities