@@ -19,12 +19,16 @@ module Asyncly.Core
19 ( Stream (..)
19 ( Stream (..)
20 , MonadAsync
20 , MonadAsync
21 , Streaming (..)
21 , Streaming (..)
24− -- * Running the Monad
23+ -- * Construction
25 , runStreaming
29 , runStreaming
27− -- * Interleaving
31+ -- * Composition
28 , interleave
32 , interleave
@@ -32,7 +36,7 @@ module Asyncly.Core
32 , CtxType (..)
36 , CtxType (..)
33 , SchedPolicy (..)
37 , SchedPolicy (..)
34 , Dimension (..)
38 , Dimension (..)
35− , yielding
39+ , EndOfStream (..)
36 , parallel
40 , parallel
@@ -119,6 +123,8 @@ data Context m a =
119-- The stream type
123-- The stream type
120------------------------------------------------------------------------------
124------------------------------------------------------------------------------
126+-- TBD use a functor instead of the bare type a?
122-- | Represents a monadic stream of values of type 'a' constructed using
128-- | Represents a monadic stream of values of type 'a' constructed using
123-- actions in monad 'm'. Streams can be composed sequentially or in parallel in
129-- actions in monad 'm'. Streams can be composed sequentially or in parallel in
124-- product style compositions (monadic bind multiplies streams in a ListT
130-- product style compositions (monadic bind multiplies streams in a ListT
@@ -596,7 +602,7 @@ class Streaming t where
596 fromStream :: Stream m a -> t m a
602 fromStream :: Stream m a -> t m a
598------------------------------------------------------------------------------
604------------------------------------------------------------------------------
599−-- Some basic stream type agnostic APIs
605+-- Some basic type agnostic (working on the Streaming type class) APIs
600------------------------------------------------------------------------------
606------------------------------------------------------------------------------
602-- XXX The async API is useful for exploring each stream arbitrarily when
608-- XXX The async API is useful for exploring each stream arbitrarily when
@@ -609,7 +615,7 @@ class Streaming t where
609-- original computation. Note that the returned action must be executed exactly
615-- original computation. Note that the returned action must be executed exactly
610-- once and drained completely. If not executed or not drained fully we may
616-- once and drained completely. If not executed or not drained fully we may
611-- have a thread blocked forever and if executed more than once a
617-- have a thread blocked forever and if executed more than once a
612−-- 'ContextUsedAfterEOF' exception will be raised.
618+-- 'EndOfStream' exception will be raised.
614async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
620async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
615async m = do
621async m = do
@@ -637,22 +643,40 @@ m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
637m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
643m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
639------------------------------------------------------------------------------
645------------------------------------------------------------------------------
640−-- Running the monad
646+-- Constructing a stream
641------------------------------------------------------------------------------
647------------------------------------------------------------------------------
643−-- | Run a composed streaming computation, discard the results.
649+-- XXX Need to accept a context as well
644−runStreaming :: (Monad m, Streaming t) => t m a -> m ()
650+-- | Build a stream from its church encoding.
645−runStreaming m = go (toStream m)
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))
647− where
658+------------------------------------------------------------------------------
648−
659+-- Destroying a stream
649− go m1 = (runStream m1) Nothing stop yield
660+------------------------------------------------------------------------------
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
653− {-# INLINE yield #-}
671+-- | Run a streaming composition, discard the results.
672+runStreaming :: (Monad m, Streaming t) => t m a -> m ()
673+runStreaming m = go (toStream m)
676+ let stop = return ()
654− yield _ Nothing = stop
677+ yield _ Nothing = stop
655− yield _ (Just x) = go x
678+ yield _ (Just x) = go x
679+ in (runStream m1) Nothing stop yield
657------------------------------------------------------------------------------
681------------------------------------------------------------------------------
658-- Utilities
682-- Utilities