1{-# LANGUAGE FlexibleContexts #-}
2{-# LANGUAGE FlexibleInstances #-}
3{-# LANGUAGE GeneralizedNewtypeDeriving#-}
4{-# LANGUAGE MultiParamTypeClasses #-}
5{-# LANGUAGE RankNTypes #-}
6{-# LANGUAGE StandaloneDeriving #-}
7{-# LANGUAGE UndecidableInstances #-} -- XXX
10-- Module : Streamly.Streams
11-- Copyright : (c) 2017 Harendra Kumar
14-- Maintainer : harendra.kumar@gmail.com
15-- Stability : experimental
19module Streamly.Streams
21 -- * Product Style Composition
77 -- * Sum Style Composition
89import Control.Applicative (Alternative (..), liftA2)
90import Control.Monad (MonadPlus(..), ap)
91import Control.Monad.Base (MonadBase (..))
92import Control.Monad.Catch (MonadThrow)
93import Control.Monad.Error.Class (MonadError(..))
94import Control.Monad.IO.Class (MonadIO(..))
95import Control.Monad.Reader.Class (MonadReader(..))
96import Control.Monad.State.Class (MonadState(..))
97import Control.Monad.Trans.Class (MonadTrans)
98import Data.Semigroup (Semigroup(..))
99import Prelude hiding (drop, take, zipWith)
104-- Different styles of streams represented by different types dictate the
105-- manner in which a stream is processed in monadic or applicative composition.
106-- Each type differs in the processing order of the elements or processing
107-- parallelism. The types have the same underlying representation and
108-- therefore can be converted back and forth easily to switch the composition
109-- style at any point.
111-- 'StreamT' and 'InterleavedT' process the streams serially but differ in the
112-- order in which the elements of the streams are processed. 'StreamT' goes
113-- depth first that is it combine one element of the first stream to every
114-- element of the second stream before it comes back to process the second
115-- element of the first stream. 'InterleavedT' takes turns when processing
116-- elements from each stream and therefore all iterations finish at the same
117-- time rather than one iteration at a time as in the case of 'StreamT'.
118-- Depending on the cache behavior or IO behavior 'InterleavedT' might be
119-- slightly more efficient than 'StreamT' or vice-versa.
121-- 'AsyncT' and 'ParallelT' can process the individual elements of the streams
122-- in parallel. 'AsyncT' is the parallel version of 'StreamT', it follows the
123-- order of 'StreamT', processing multiple elements of a stream concurrently.
124-- 'ParallelT' follows the order of 'InterleavedT', processing and interleaving
125-- both streams concurrently. Also, note that 'fmap' on these streams would
126-- also work in a concurrent fashion i.e. multiple elements in a stream or
127-- multiple streams can be mapped concurrently.
129-- Zipping types have a different style of composition altogether, they zip the
130-- streams rather than multiplying them. These types provide only Applicative
131-- composition and no Monad instances. 'ZipStream' zips serially i.e. it
132-- produces one element from each stream in a serial manner and then zips the
133-- two items. 'ZipAsync' on the other hand zips in parallel, it produces one
134-- element from each stream concurrently and then zips the two.
136-- It may help develop a better understanding if we think in terms of stream
137-- generation (production) side parallelism versus stream processing
138-- (consumption) side parallelism. The sum style operators (e.g. '<|' or '<|>')
139-- generate streams in parallel and then combine the results into a unified
140-- stream. On the other hand the product style composition (e.g. 'AsyncT' or
141-- 'ParallelT' or 'ZipAsync') dictate how the elements of a stream are
142-- processed when it is consumed.
144------------------------------------------------------------------------------
145-- Types that can behave as a Stream
146------------------------------------------------------------------------------
148-- | Class of types that can represent a stream of elements of some type 'a' in
150class Streaming t where
151 toStream :: t m a -> Stream m a
152 fromStream :: Stream m a -> t m a
154------------------------------------------------------------------------------
155-- Constructing a stream
156------------------------------------------------------------------------------
158cons :: (Streaming t) => a -> t m a -> t m a
159cons a r = fromStream $ scons a (Just (toStream r))
161nil :: Streaming t => t m a
162nil = fromStream $ snil
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.
169streamBuild :: Streaming t
170 => (forall r. Maybe (SVar m a)
171 -> (a -> Maybe (t m a) -> m r)
175streamBuild k = fromStream $ Stream $ \sv stp yld ->
176 let yield a Nothing = yld a Nothing
177 yield a (Just r) = yld a (Just (toStream r))
180-- | Build a singleton stream from a callback function.
181fromCallback :: (Streaming t) => (forall r. (a -> m r) -> m r) -> t m a
182fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)
184-- | Read an SVar to get a stream.
185fromSVar :: (MonadAsync m, Streaming t) => SVar m a -> t m a
186fromSVar sv = fromStream $ fromStreamVar sv
188------------------------------------------------------------------------------
189-- Destroying a stream
190------------------------------------------------------------------------------
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.
195streamFold :: Streaming t
196 => Maybe (SVar m a) -> (a -> Maybe (t m a) -> m r) -> m r -> t m a -> m r
197streamFold 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
202-- | Run a streaming composition, discard the results.
203runStreaming :: (Monad m, Streaming t) => t m a -> m ()
204runStreaming m = go (toStream m)
208 yield _ Nothing = stop
209 yield _ (Just x) = go x
210 in (runStream m1) Nothing stop yield
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'.
214toSVar :: (Streaming t, MonadAsync m) => SVar m a -> t m a -> m ()
215toSVar sv m = toStreamVar sv (toStream m)
217------------------------------------------------------------------------------
219------------------------------------------------------------------------------
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'.
228async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
230 sv <- newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)
233------------------------------------------------------------------------------
234-- StreamT iterates serially in the Monad and Applicative compositions
235------------------------------------------------------------------------------
237-- | Note that 'StreamT' style composition works well operationally for
238-- infinite streams. However if the second stream is infinite we will never
239-- come back to the second iteration of the first stream.
241newtype StreamT m a = StreamT {getStreamT :: Stream m a}
242 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
244deriving instance MonadAsync m => Alternative (StreamT m)
245deriving instance MonadAsync m => MonadPlus (StreamT m)
246deriving instance (MonadBase b m, Monad m) => MonadBase b (StreamT m)
247deriving instance MonadError e m => MonadError e (StreamT m)
248deriving instance MonadReader r m => MonadReader r (StreamT m)
249deriving instance MonadState s m => MonadState s (StreamT m)
251instance Streaming StreamT where
252 toStream = getStreamT
255-- XXX The Functor/Applicative/Num instances for all the types are exactly the
256-- same, how can we reduce this boilerplate (use TH)? We cannot derive them
257-- from a single base type because they depend on the Monad instance which is
258-- different for each type.
260------------------------------------------------------------------------------
262------------------------------------------------------------------------------
264-- | Execute a monadic action sequentially for each element in the 'AsyncT'
265-- stream, i.e. an iteration finishes completely before the next one starts.
266instance Monad m => Monad (StreamT m) where
268 (StreamT (Stream m)) >>= f = StreamT $ Stream $ \_ stp yld ->
269 let run x = (runStream x) Nothing stp yld
270 yield a Nothing = run $ getStreamT (f a)
271 yield a (Just r) = run $ getStreamT (f a)
272 <> getStreamT (StreamT r >>= f)
273 in m Nothing stp yield
275------------------------------------------------------------------------------
277------------------------------------------------------------------------------
279instance Monad m => Applicative (StreamT m) where
280 pure a = StreamT $ scons a Nothing
283------------------------------------------------------------------------------
285------------------------------------------------------------------------------
287instance Monad m => Functor (StreamT m) where
288 fmap f (StreamT (Stream m)) = StreamT $ Stream $ \_ stp yld ->
289 let yield a Nothing = yld (f a) Nothing
290 yield a (Just r) = yld (f a)
291 (Just (getStreamT (fmap f (StreamT r))))
292 in m Nothing stp yield
294------------------------------------------------------------------------------
296------------------------------------------------------------------------------
298instance (Monad m, Num a) => Num (StreamT m a) where
299 fromInteger n = pure (fromInteger n)
309instance (Monad m, Fractional a) => Fractional (StreamT m a) where
310 fromRational n = pure (fromRational n)
316instance (Monad m, Floating a) => Floating (StreamT m a) where
336 logBase = liftA2 logBase
338------------------------------------------------------------------------------
339-- InterleavedT interleaves the iterations in the Monad and Applicative
341------------------------------------------------------------------------------
343-- | Note that 'InterleavedT' style composition cannot work well operationally
344-- on infinite streams. It will keep accumulating the elements of the stream
345-- resulting in continuous increase in space consumption.
347newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
348 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
350deriving instance MonadAsync m => Alternative (InterleavedT m)
351deriving instance MonadAsync m => MonadPlus (InterleavedT m)
352deriving instance (MonadBase b m, Monad m) => MonadBase b (InterleavedT m)
353deriving instance MonadError e m => MonadError e (InterleavedT m)
354deriving instance MonadReader r m => MonadReader r (InterleavedT m)
355deriving instance MonadState s m => MonadState s (InterleavedT m)
357instance Streaming InterleavedT where
358 toStream = getInterleavedT
359 fromStream = InterleavedT
361-- | Execute a monadic action for each element in the stream, in a fairly
362-- interleaved manner i.e. iterations yield alternately.
363instance Monad m => Monad (InterleavedT m) where
365 (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->
366 let run x = (runStream x) Nothing stp yld
367 yield a Nothing = run $ getInterleavedT (f a)
368 yield a (Just r) = run $ getInterleavedT (f a)
370 getInterleavedT (InterleavedT r >>= f)
371 in m Nothing stp yield
373------------------------------------------------------------------------------
375------------------------------------------------------------------------------
377instance Monad m => Applicative (InterleavedT m) where
378 pure a = InterleavedT $ scons a Nothing
381------------------------------------------------------------------------------
383------------------------------------------------------------------------------
385instance Monad m => Functor (InterleavedT m) where
386 fmap f (InterleavedT (Stream m)) = InterleavedT $ Stream $ \_ stp yld ->
387 let yield a Nothing = yld (f a) Nothing
389 yld (f a) (Just (getInterleavedT (fmap f (InterleavedT r))))
390 in m Nothing stp yield
392------------------------------------------------------------------------------
394------------------------------------------------------------------------------
396instance (Monad m, Num a) => Num (InterleavedT m a) where
397 fromInteger n = pure (fromInteger n)
407instance (Monad m, Fractional a) => Fractional (InterleavedT m a) where
408 fromRational n = pure (fromRational n)
414instance (Monad m, Floating a) => Floating (InterleavedT m a) where
434 logBase = liftA2 logBase
436------------------------------------------------------------------------------
437-- AsyncT runs the iterations asynchronously in the Monad and Applicative
438-- compositions. More iterations are started in parallel on demand. It can work
439-- with infinite streams.
440------------------------------------------------------------------------------
442-- | Just like 'StreamT', 'AsyncT' style composition works well operationally
443-- for infinite streams. However, if the second stream is infinite we will
444-- never be able to finish the first stream.
446-- Note that 'fmap' on 'AsyncT' would map on multiple elements concurrently.
448newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
449 deriving (Semigroup, Monoid, MonadTrans)
451deriving instance MonadAsync m => Alternative (AsyncT m)
452deriving instance MonadAsync m => MonadPlus (AsyncT m)
453deriving instance MonadAsync m => MonadIO (AsyncT m)
454deriving instance MonadAsync m => MonadThrow (AsyncT m)
455deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m)
456deriving instance (MonadError e m, MonadAsync m) => MonadError e (AsyncT m)
457deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)
458deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)
460instance Streaming AsyncT where
464{-# INLINE parbind #-}
466 :: (forall c. Stream m c -> Stream m c -> Stream m c)
470parbind par m f = go m
473 Stream $ \ctx stp yld ->
474 let run x = (runStream x) ctx stp yld
475 yield a Nothing = run $ f a
476 yield a (Just r) = run $ f a `par` (go r)
477 in g Nothing stp yield
479-- | Execute a monadic action for each element in the stream, running
480-- iterations in parallel, but giving preference to iterations started
482instance MonadAsync m => Monad (AsyncT m) where
484 (AsyncT m) >>= f = AsyncT $ parbind par m g
485 where g x = getAsyncT (f x)
486 par = joinStreamVar2 (SVarStyle Conjunction LIFO)
488------------------------------------------------------------------------------
490------------------------------------------------------------------------------
492instance MonadAsync m => Applicative (AsyncT m) where
493 pure a = AsyncT $ scons a Nothing
496------------------------------------------------------------------------------
498------------------------------------------------------------------------------
500instance Monad m => Functor (AsyncT m) where
501 fmap f (AsyncT (Stream m)) = AsyncT $ Stream $ \_ stp yld ->
502 let yield a Nothing = yld (f a) Nothing
503 yield a (Just r) = yld (f a) (Just (getAsyncT (fmap f (AsyncT r))))
504 in m Nothing stp yield
506------------------------------------------------------------------------------
508------------------------------------------------------------------------------
510instance (MonadAsync m, Num a) => Num (AsyncT m a) where
511 fromInteger n = pure (fromInteger n)
521instance (MonadAsync m, Fractional a) => Fractional (AsyncT m a) where
522 fromRational n = pure (fromRational n)
528instance (MonadAsync m, Floating a) => Floating (AsyncT m a) where
548 logBase = liftA2 logBase
550------------------------------------------------------------------------------
551-- ParallelT runs all iterations in parallel in the Monad and Applicative
552-- compositions. Iterations are scheduled in a round-robin fashion.
553------------------------------------------------------------------------------
555-- | Note that, just like 'InterleavedT', 'ParallelT' style composition cannot
556-- work well operationally on infinite streams. It will keep accumulating the
557-- elements of the stream resulting in continuous increase in space
560-- Note that 'fmap' on 'ParallelT' would map on both streams concurrently.
562newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
563 deriving (Semigroup, Monoid, MonadTrans)
565deriving instance MonadAsync m => Alternative (ParallelT m)
566deriving instance MonadAsync m => MonadPlus (ParallelT m)
567deriving instance MonadAsync m => MonadIO (ParallelT m)
568deriving instance MonadAsync m => MonadThrow (ParallelT m)
569deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)
570deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)
571deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)
572deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)
574instance Streaming ParallelT where
575 toStream = getParallelT
576 fromStream = ParallelT
578-- | Execute a monadic action for each element in the stream, running
579-- iterations in a fairly parallel manner, i.e. all iterations are equally
581instance MonadAsync m => Monad (ParallelT m) where
583 (ParallelT m) >>= f = ParallelT $ parbind par m g
584 where g x = getParallelT (f x)
585 par = joinStreamVar2 (SVarStyle Conjunction FIFO)
587------------------------------------------------------------------------------
589------------------------------------------------------------------------------
591instance MonadAsync m => Applicative (ParallelT m) where
592 pure a = ParallelT $ scons a Nothing
595------------------------------------------------------------------------------
597------------------------------------------------------------------------------
599instance Monad m => Functor (ParallelT m) where
600 fmap f (ParallelT (Stream m)) = ParallelT $ Stream $ \_ stp yld ->
601 let yield a Nothing = yld (f a) Nothing
602 yield a (Just r) = yld (f a)
603 (Just (getParallelT (fmap f (ParallelT r))))
604 in m Nothing stp yield
606------------------------------------------------------------------------------
608------------------------------------------------------------------------------
610instance (MonadAsync m, Num a) => Num (ParallelT m a) where
611 fromInteger n = pure (fromInteger n)
621instance (MonadAsync m, Fractional a) => Fractional (ParallelT m a) where
622 fromRational n = pure (fromRational n)
628instance (MonadAsync m, Floating a) => Floating (ParallelT m a) where
648 logBase = liftA2 logBase
650------------------------------------------------------------------------------
651-- Serially Zipping Streams
652------------------------------------------------------------------------------
654-- | Zip two streams serially using a pure zipping function.
655zipWith :: Streaming t => (a -> b -> c) -> t m a -> t m b -> t m c
656zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
658 go mx my = Stream $ \_ stp yld -> do
660 let yield2 b Nothing = yld (f a b) Nothing
661 yield2 b (Just rb) = yld (f a b) (Just (go ra rb))
662 in (runStream my) Nothing stp yield2
663 let yield1 a Nothing = merge a snil
664 yield1 a (Just ra) = merge a ra
665 (runStream mx) Nothing stp yield1
667-- | 'ZipStream' zips serially i.e. it produces one element from each stream in
668-- a serial manner and then zips the two elements.
669newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}
670 deriving (Semigroup, Monoid)
672deriving instance MonadAsync m => Alternative (ZipStream m)
674instance Monad m => Functor (ZipStream m) where
675 fmap f (ZipStream (Stream m)) = ZipStream $ Stream $ \_ stp yld ->
676 let yield a Nothing = yld (f a) Nothing
677 yield a (Just r) = yld (f a)
678 (Just (getZipStream (fmap f (ZipStream r))))
679 in m Nothing stp yield
681instance Monad m => Applicative (ZipStream m) where
682 pure a = ZipStream $ scons a Nothing
685instance Streaming ZipStream where
686 toStream = getZipStream
687 fromStream = ZipStream
689instance (Monad m, Num a) => Num (ZipStream m a) where
690 fromInteger n = pure (fromInteger n)
700instance (Monad m, Fractional a) => Fractional (ZipStream m a) where
701 fromRational n = pure (fromRational n)
707instance (Monad m, Floating a) => Floating (ZipStream m a) where
727 logBase = liftA2 logBase
729------------------------------------------------------------------------------
730-- Parallely Zipping Streams
731------------------------------------------------------------------------------
733-- | Zip two streams asyncly (i.e. both the elements being zipped are generated
734-- concurrently) using a pure zipping function.
735zipAsyncWith :: (Streaming t, MonadAsync m)
736 => (a -> b -> c) -> t m a -> t m b -> t m c
737zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
740 (runStream (toStream (zipWith f ma mb))) Nothing stp yld
742-- | 'ZipAsync' zips in parallel, it produces one element from each stream
743-- concurrently and then zips the two.
744newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}
745 deriving (Semigroup, Monoid)
747deriving instance MonadAsync m => Alternative (ZipAsync m)
749instance Monad m => Functor (ZipAsync m) where
750 fmap f (ZipAsync (Stream m)) = ZipAsync $ Stream $ \_ stp yld ->
751 let yield a Nothing = yld (f a) Nothing
752 yield a (Just r) = yld (f a)
753 (Just (getZipAsync (fmap f (ZipAsync r))))
754 in m Nothing stp yield
756instance MonadAsync m => Applicative (ZipAsync m) where
757 pure a = ZipAsync $ scons a Nothing
758 (<*>) = zipAsyncWith id
760instance Streaming ZipAsync where
761 toStream = getZipAsync
762 fromStream = ZipAsync
764instance (MonadAsync m, Num a) => Num (ZipAsync m a) where
765 fromInteger n = pure (fromInteger n)
775instance (MonadAsync m, Fractional a) => Fractional (ZipAsync m a) where
776 fromRational n = pure (fromRational n)
782instance (MonadAsync m, Floating a) => Floating (ZipAsync m a) where
802 logBase = liftA2 logBase
804-------------------------------------------------------------------------------
805-- Type adapting combinators
806-------------------------------------------------------------------------------
808-- | Adapt one streaming type to another.
809adapt :: (Streaming t1, Streaming t2) => t1 m a -> t2 m a
810adapt = fromStream . toStream
812-- | Interpret an ambiguously typed stream as 'StreamT'.
813serially :: StreamT m a -> StreamT m a
816-- | Interpret an ambiguously typed stream as 'InterleavedT'.
817interleaving :: InterleavedT m a -> InterleavedT m a
820-- | Interpret an ambiguously typed stream as 'AsyncT'.
821asyncly :: AsyncT m a -> AsyncT m a
824-- | Interpret an ambiguously typed stream as 'ParallelT'.
825parallely :: ParallelT m a -> ParallelT m a
828-- | Interpret an ambiguously typed stream as 'ZipStream'.
829zipping :: ZipStream m a -> ZipStream m a
832-- | Interpret an ambiguously typed stream as 'ZipAsync'.
833zippingAsync :: ZipAsync m a -> ZipAsync m a
836-------------------------------------------------------------------------------
837-- Running Streams, convenience functions specialized to types
838-------------------------------------------------------------------------------
840-- | Run a 'StreamT' computation. Same as @runStreaming . serially@.
841runStreamT :: Monad m => StreamT m a -> m ()
842runStreamT = runStreaming
844-- | Run an 'InterleavedT' computation. Same as @runStreaming . interleaving@.
845runInterleavedT :: Monad m => InterleavedT m a -> m ()
846runInterleavedT = runStreaming
848-- | Run an 'AsyncT' computation. Same as @runStreaming . asyncly@.
849runAsyncT :: Monad m => AsyncT m a -> m ()
850runAsyncT = runStreaming
852-- | Run a 'ParallelT' computation. Same as @runStreaming . parallely@.
853runParallelT :: Monad m => ParallelT m a -> m ()
854runParallelT = runStreaming
856-- | Run a 'ZipStream' computation. Same as @runStreaming . zipping@.
857runZipStream :: Monad m => ZipStream m a -> m ()
858runZipStream = runStreaming
860-- | Run a 'ZipAsync' computation. Same as @runStreaming . zippingAsync@.
861runZipAsync :: Monad m => ZipAsync m a -> m ()
862runZipAsync = runStreaming
864------------------------------------------------------------------------------
865-- Sum Style Composition
866------------------------------------------------------------------------------
870-- | Sequential interleaved composition, in contrast to '<>' this operator
871-- fairly interleaves the two streams instead of appending them; yielding one
872-- element from each stream alternately. Unlike '<>' it cannot be used to fold
873-- an infinite container of streams.
875(<=>) :: Streaming t => t m a -> t m a -> t m a
876m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
878-- | Parallel interleaved composition, in contrast to '<|>' this operator
879-- "merges" streams in a left biased manner rather than fairly interleaving
880-- them. It keeps yielding from the stream on the left as long as it can. If
881-- the left stream blocks or cannot keep up with the pace of the consumer it
882-- can yield from the stream on the right in parallel. Unlike '<|>' it can be
883-- used to fold infinite containers of streams.
885(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
886m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
888------------------------------------------------------------------------------
890------------------------------------------------------------------------------
893-- These utilities are designed to pass the first argument as one of the sum
894-- style composition operators (i.e. '<>', '<=>', '<|', '<|>') to conveniently
895-- fold a container using any style of stream composition.
897-- | Like the 'Prelude' 'fold' but allows you to specify a binary sum style
898-- stream composition operator to fold a container of streams.
900-- @foldWith (<>) $ map return [1..3]@
901{-# INLINABLE foldWith #-}
902foldWith :: (Streaming t, Foldable f)
903 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
904foldWith f = foldr f nil
906-- | Like 'foldMap' but allows you to specify a binary sum style composition
907-- operator to fold a container of streams. Maps a monadic streaming action on
908-- the container before folding it.
910-- @foldMapWith (<>) return [1..3]@
911{-# INLINABLE foldMapWith #-}
912foldMapWith :: (Streaming t, Foldable f)
913 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
914foldMapWith f g = foldr (f . g) nil
916-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
917-- monadic streaming function is the last argument.
918{-# INLINABLE forEachWith #-}
919forEachWith :: (Streaming t, Foldable f)
920 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
921forEachWith f xs g = foldr (f . g) nil xs
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list