@@ -30,6 +30,21 @@ module Asyncly.Streams
3030 , SVar
3131 , newEmptySVar
3232
33+ -- * Construction
34+ , streamBuild
35+ , fromCallback
36+ , fromSVar
37+
38+ -- * Elimination
39+ , cons
40+ , nil
41+ , streamFold
42+ , runStreaming
43+ , toSVar
44+
45+ -- * Transformation
46+ , async
47+
3348 -- * Stream Styles
3449 , StreamT
3550 , InterleavedT
@@ -47,18 +62,6 @@ module Asyncly.Streams
4762 , zippingAsync
4863 , adapt
4964
50− -- * Construction
51− , streamBuild
52− , fromCallback
53− , fromSVar
54−
55− -- * Elimination
56− , cons
57− , nil
58− , streamFold
59− , runStreaming
60− , toSVar
61−
6265 -- * Running Streams
6366 , runStreamT
6467 , runInterleavedT
@@ -67,9 +70,6 @@ module Asyncly.Streams
6770 , runZipStream
6871 , runZipAsync
6972
70− -- * Transformation
71− , async
72−
7373 -- * Zipping
7474 , zipWith
7575 , zipAsyncWith
@@ -151,6 +151,85 @@ class Streaming t where
151151 toStream :: t m a -> Stream m a
152152 fromStream :: Stream m a -> t m a
153153
154+------------------------------------------------------------------------------
155+-- Constructing a stream
156+------------------------------------------------------------------------------
157+
158+cons :: (Streaming t) => a -> t m a -> t m a
159+cons a r = fromStream $ scons a (Just (toStream r))
160+
161+nil :: Streaming t => t m a
162+nil = fromStream $ snil
163+
164+-- | Build a stream from its church encoding. The function passed maps
165+-- directly to the underlying representation of the stream type. The second
166+-- parameter to the function is the "yield" function yielding a value and the
167+-- remaining stream if any otherwise 'Nothing'. The third parameter is to
168+-- represent an "empty" stream.
169+streamBuild :: Streaming t
170+ => (forall r. Maybe (SVar m a)
171+ -> (a -> Maybe (t m a) -> m r)
172+ -> m r
173+ -> m r)
174+ -> t m a
175+streamBuild k = fromStream $ Stream $ \sv stp yld ->
176+ let yield a Nothing = yld a Nothing
177+ yield a (Just r) = yld a (Just (toStream r))
178+ in k sv yield stp
179+
180+-- | Build a singleton stream from a callback function.
181+fromCallback :: (Streaming t) => (forall r. (a -> m r) -> m r) -> t m a
182+fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)
183+
184+-- | Read an SVar to get a stream.
185+fromSVar :: (MonadAsync m, Streaming t) => SVar m a -> t m a
186+fromSVar sv = fromStream $ fromStreamVar sv
187+
188+------------------------------------------------------------------------------
189+-- Destroying a stream
190+------------------------------------------------------------------------------
191+
192+-- | Fold a stream using its church encoding. The second argument is the "step"
193+-- function consuming an element and the remaining stream, if any. The third
194+-- argument is for consuming an "empty" stream that yields nothing.
195+streamFold :: Streaming t
196+ => Maybe (SVar m a) -> (a -> Maybe (t m a) -> m r) -> m r -> t m a -> m r
197+streamFold sv step blank m =
198+ let yield a Nothing = step a Nothing
199+ yield a (Just x) = step a (Just (fromStream x))
200+ in (runStream (toStream m)) sv blank yield
201+
202+-- | Run a streaming composition, discard the results.
203+runStreaming :: (Monad m, Streaming t) => t m a -> m ()
204+runStreaming m = go (toStream m)
205+ where
206+ go m1 =
207+ let stop = return ()
208+ yield _ Nothing = stop
209+ yield _ (Just x) = go x
210+ in (runStream m1) Nothing stop yield
211+
212+-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
213+-- be read back from the SVar using 'fromSVar'.
214+toSVar :: (Streaming t, MonadAsync m) => SVar m a -> t m a -> m ()
215+toSVar sv m = toStreamVar sv (toStream m)
216+
217+------------------------------------------------------------------------------
218+-- Transformation
219+------------------------------------------------------------------------------
220+
221+-- XXX Get rid of this?
222+-- | Make a stream asynchronous, triggers the computation and returns a stream
223+-- in the underlying monad representing the output generated by the original
224+-- computation. The returned action is exhaustible and must be drained once. If
225+-- not drained fully we may have a thread blocked forever and once exhausted it
226+-- will always return 'empty'.
227+
228+async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
229+async m = do
230+ sv <- newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)
231+ return $ fromSVar sv
232+
154233------------------------------------------------------------------------------
155234-- StreamT iterates serially in the Monad and Applicative compositions
156235------------------------------------------------------------------------------
@@ -739,69 +818,6 @@ zipping x = x
739818zippingAsync :: ZipAsync m a -> ZipAsync m a
740819zippingAsync x = x
741820
742−------------------------------------------------------------------------------
743−-- Constructing a stream
744−------------------------------------------------------------------------------
745−
746−cons :: (Streaming t) => a -> t m a -> t m a
747−cons a r = fromStream $ scons a (Just (toStream r))
748−
749−nil :: Streaming t => t m a
750−nil = fromStream $ snil
751−
752−-- | Build a stream from its church encoding. The function passed maps
753−-- directly to the underlying representation of the stream type. The second
754−-- parameter to the function is the "yield" function yielding a value and the
755−-- remaining stream if any otherwise 'Nothing'. The third parameter is to
756−-- represent an "empty" stream.
757−streamBuild :: Streaming t
758− => (forall r. Maybe (SVar m a)
759− -> (a -> Maybe (t m a) -> m r)
760− -> m r
761− -> m r)
762− -> t m a
763−streamBuild k = fromStream $ Stream $ \sv stp yld ->
764− let yield a Nothing = yld a Nothing
765− yield a (Just r) = yld a (Just (toStream r))
766− in k sv yield stp
767−
768−-- | Build a singleton stream from a callback function.
769−fromCallback :: (Streaming t) => (forall r. (a -> m r) -> m r) -> t m a
770−fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)
771−
772−-- | Read an SVar to get a stream.
773−fromSVar :: (MonadAsync m, Streaming t) => SVar m a -> t m a
774−fromSVar sv = fromStream $ fromStreamVar sv
775−
776−------------------------------------------------------------------------------
777−-- Destroying a stream
778−------------------------------------------------------------------------------
779−
780−-- | Fold a stream using its church encoding. The second argument is the "step"
781−-- function consuming an element and the remaining stream, if any. The third
782−-- argument is for consuming an "empty" stream that yields nothing.
783−streamFold :: Streaming t
784− => Maybe (SVar m a) -> (a -> Maybe (t m a) -> m r) -> m r -> t m a -> m r
785−streamFold sv step blank m =
786− let yield a Nothing = step a Nothing
787− yield a (Just x) = step a (Just (fromStream x))
788− in (runStream (toStream m)) sv blank yield
789−
790−-- | Run a streaming composition, discard the results.
791−runStreaming :: (Monad m, Streaming t) => t m a -> m ()
792−runStreaming m = go (toStream m)
793− where
794− go m1 =
795− let stop = return ()
796− yield _ Nothing = stop
797− yield _ (Just x) = go x
798− in (runStream m1) Nothing stop yield
799−
800−-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
801−-- be read back from the SVar using 'fromSVar'.
802−toSVar :: (Streaming t, MonadAsync m) => SVar m a -> t m a -> m ()
803−toSVar sv m = toStreamVar sv (toStream m)
804−
805821-------------------------------------------------------------------------------
806822-- Running Streams, convenience functions specialized to types
807823-------------------------------------------------------------------------------
@@ -830,22 +846,6 @@ runZipStream = runStreaming
830846runZipAsync :: Monad m => ZipAsync m a -> m ()
831847runZipAsync = runStreaming
832848
833−------------------------------------------------------------------------------
834−-- Transformation
835−------------------------------------------------------------------------------
836−
837−-- XXX Get rid of this?
838−-- | Make a stream asynchronous, triggers the computation and returns a stream
839−-- in the underlying monad representing the output generated by the original
840−-- computation. The returned action is exhaustible and must be drained once. If
841−-- not drained fully we may have a thread blocked forever and once exhausted it
842−-- will always return 'empty'.
843−
844−async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
845−async m = do
846− sv <- newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)
847− return $ fromSVar sv
848−
849849------------------------------------------------------------------------------
850850-- Sum Style Composition
851851------------------------------------------------------------------------------