2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE MultiParamTypeClasses #-}
7{-# LANGUAGE RankNTypes #-}
8{-# LANGUAGE StandaloneDeriving #-}
9{-# LANGUAGE UndecidableInstances #-} -- XXX
12-- Module : Streamly.Streams
13-- Copyright : (c) 2017 Harendra Kumar
16-- Maintainer : harendra.kumar@gmail.com
17-- Stability : experimental
21module Streamly.Streams
24 , Streaming -- deprecated
26 , S.MonadAsync -- deprecated
48 , runStreaming -- deprecated
70 -- * Stream Transformers
73 , InterleavedT -- deprecated
75 , AsyncT -- deprecated
79 , ZipAsync -- deprecated
84 , interleaving -- deprecated
86 , asyncly -- deprecated
89 , zipping -- deprecated
91 , zippingAsync -- deprecated
95 , runStreamT -- deprecated
96 , runInterleavedT -- deprecated
97 , runAsyncT -- deprecated
98 , runParallelT -- deprecated
99 , runZipStream -- deprecated
100 , runZipAsync -- deprecated
109import Control.Monad (ap)
110import Control.Monad.Base (MonadBase (..), liftBaseDefault)
111import Control.Monad.Catch (MonadThrow, throwM)
112-- import Control.Monad.Error.Class (MonadError(..))
113import Control.Monad.IO.Class (MonadIO(..))
114import Control.Monad.Reader.Class (MonadReader(..))
115import Control.Monad.State.Class (MonadState(..))
116import Control.Monad.Trans.Class (MonadTrans (lift))
117import Data.Semigroup (Semigroup(..))
118import Streamly.Core ( MonadParallel
119 , SVar, SVarStyle(..)
120 , SVarTag(..), SVarSched(..))
121import qualified Streamly.Core as S
123------------------------------------------------------------------------------
124-- Types that can behave as a Stream
125------------------------------------------------------------------------------
127-- | Class of types that can represent a stream of elements of some type 'a' in
129class IsStream t where
130 toStream :: t m a -> S.Stream m a
131 fromStream :: S.Stream m a -> t m a
133-- | Same as 'IsStream'.
134{-# Deprecated Streaming "Please use IsStream instead." #-}
135type Streaming = IsStream
137------------------------------------------------------------------------------
138-- Constructing a stream
139------------------------------------------------------------------------------
147nil :: IsStream t => t m a
148nil = fromStream S.nil
152-- | Constructs a stream by adding a monadic action at the head of an existing
153-- stream. For example:
156-- > toList $ getLine \`consM` getLine \`consM` nil
161consM :: (IsStream t, Monad m) => m a -> t m a -> t m a
162consM m r = fromStream $ S.consM m (toStream r)
166-- | Operator equivalent of 'consM'.
169-- > toList $ getLine |: getLine |: nil
174(|:) :: (IsStream t, Monad m) => m a -> t m a -> t m a
179-- | Construct a stream by adding a pure value at the head of an existing
180-- stream. Same as @consM . return@. For example:
183-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
186cons :: IsStream t => a -> t m a -> t m a
187cons a r = fromStream $ S.cons a (toStream r)
191-- | Operator equivalent of 'cons'.
194-- > toList $ 1 .: 2 .: 3 .: nil
198(.:) :: IsStream t => a -> t m a -> t m a
201-- | Build a stream from its church encoding. The function passed maps
202-- directly to the underlying representation of the stream type. The second
203-- parameter to the function is the "yield" function yielding a value and the
204-- remaining stream if any otherwise 'Nothing'. The third parameter is to
205-- represent an "empty" stream.
206streamBuild :: IsStream t
207 => (forall r. Maybe (SVar m a)
208 -> (a -> t m a -> m r)
213streamBuild k = fromStream $ S.Stream $ \svr stp sng yld ->
214 let yield a r = yld a (toStream r)
215 in k svr yield sng stp
217-- | Build a singleton stream from a callback function.
218fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
219fromCallback k = fromStream $ S.Stream $ \_ _ sng _ -> k sng
221-- | Read an SVar to get a stream.
222fromSVar :: (MonadParallel m, IsStream t) => SVar m a -> t m a
223fromSVar sv = fromStream $ S.fromStreamVar sv
225------------------------------------------------------------------------------
226-- Destroying a stream
227------------------------------------------------------------------------------
229-- | Fold a stream using its church encoding. The second argument is the "step"
230-- function consuming an element and the remaining stream, if any. The third
231-- argument is for consuming an "empty" stream that yields nothing.
235 -> (a -> t m a -> m r)
240streamFold svr step single blank m =
241 let yield a x = step a (fromStream x)
242 in (S.runStream (toStream m)) svr blank single yield
244-- | Run a streaming composition, discard the results. By default it interprets
245-- the stream as 'StreamT', to run other types of streams use the type adapting
246-- combinators for example @runStream . 'parallely'@.
247runStream :: Monad m => StreamT m a -> m ()
248runStream m = go (toStream m)
254 in (S.runStream m1) Nothing stop single yield
256-- | Same as 'runStream'
257{-# Deprecated runStreaming "Please use runStream instead." #-}
258runStreaming :: (Monad m, IsStream t) => t m a -> m ()
259runStreaming = runStream . adapt
261-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
262-- be read back from the SVar using 'fromSVar'.
263toSVar :: (IsStream t, MonadParallel m) => SVar m a -> t m a -> m ()
264toSVar sv m = S.toStreamVar sv (toStream m)
266------------------------------------------------------------------------------
268------------------------------------------------------------------------------
270-- XXX Get rid of this?
271-- | Make a stream asynchronous, triggers the computation and returns a stream
272-- in the underlying monad representing the output generated by the original
273-- computation. The returned action is exhaustible and must be drained once. If
274-- not drained fully we may have a thread blocked forever and once exhausted it
275-- will always return 'empty'.
277async :: (IsStream t, MonadParallel m) => t m a -> m (t m a)
279 sv <- S.newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)
282------------------------------------------------------------------------------
283-- CPP macros for common instances
284------------------------------------------------------------------------------
286-- XXX use template haskell instead and include Monoid and IsStream instances
289#define MONADPARALLEL , MonadParallel m
291#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
292instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
293 pure = STREAM . S.singleton; \
296#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
297instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\
298 liftBase = liftBaseDefault }; \
300instance (MonadIO m CONSTRAINT) => MonadIO (STREAM m) where { \
301 liftIO = lift . liftIO }; \
303instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \
304 throwM = lift . throwM }; \
307instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \
308 throwError = lift . throwError; \
310 fromStream $ S.withCatchError (toStream m) (\e -> toStream $ h e) }; \
313instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \
315 local f m = fromStream $ S.withLocal f (toStream m) }; \
317instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
319 put x = lift (put x); \
320 state k = lift (state k) }
322------------------------------------------------------------------------------
324------------------------------------------------------------------------------
326-- | The 'Semigroup' instance of 'StreamT' appends two streams sequentially,
327-- yielding all elements from the first stream, and then all elements from the
331-- main = ('toList' . 'streamly' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
337-- The 'Monad' instance runs the /monadic continuation/ for each
338-- element of the stream, streamly.
341-- main = 'runStream' . 'streamly' $ do
342-- x <- return 1 \<\> return 2
350-- 'StreamT' nests streams streamly in a depth first manner.
353-- main = 'runStream' . 'streamly' $ do
354-- x <- return 1 \<\> return 2
355-- y <- return 3 \<\> return 4
356-- liftIO $ print (x, y)
365-- This behavior of 'StreamT' is exactly like a list transformer. We call the
366-- monadic code being run for each element of the stream a monadic
367-- continuation. In imperative paradigm we can think of this composition as
368-- nested @for@ loops and the monadic continuation is the body of the loop. The
369-- loop iterates for all elements of the stream.
371-- The 'streamly' combinator can be omitted as the default stream type is
373-- Note that serial composition can be used to combine an infinite number of
374-- streams as it explores only one stream at a time.
376newtype StreamT m a = StreamT {getStreamT :: S.Stream m a}
377 deriving (Semigroup, Monoid, Functor, MonadTrans)
379instance IsStream StreamT where
380 toStream = getStreamT
383------------------------------------------------------------------------------
385------------------------------------------------------------------------------
387-- | Polymorphic version of the 'Semigroup' operation '<>' of 'StreamT'.
388-- Appends two streams sequentially, yielding all elements from the first
389-- stream, and then all elements from the second stream.
391splice :: IsStream t => t m a -> t m a -> t m a
392splice m1 m2 = fromStream $ S.splice (toStream m1) (toStream m2)
394------------------------------------------------------------------------------
396------------------------------------------------------------------------------
398instance Monad m => Monad (StreamT m) where
400 (StreamT (S.Stream m)) >>= f = StreamT $ S.Stream $ \_ stp sng yld ->
401 let run x = (S.runStream x) Nothing stp sng yld
402 single a = run $ toStream (f a)
403 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
404 in m Nothing stp single yield
406------------------------------------------------------------------------------
408------------------------------------------------------------------------------
410MONAD_APPLICATIVE_INSTANCE(StreamT,)
411MONAD_COMMON_INSTANCES(StreamT,)
413------------------------------------------------------------------------------
415------------------------------------------------------------------------------
417-- | The 'Semigroup' instance of 'CostreamT' interleaves two streams,
418-- yielding one element from each stream alternately.
421-- main = ('toList' . 'costreamly' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
427-- Similarly, the 'Monad' instance fairly interleaves the iterations of the
428-- inner and the outer loop, nesting loops in a breadth first manner.
432-- main = 'runStream' . 'costreamly' $ do
433-- x <- return 1 \<\> return 2
434-- y <- return 3 \<\> return 4
435-- liftIO $ print (x, y)
444-- Note that costreamly composition can only combine a finite number of
445-- streams as it needs to retain state for each unfinished stream.
447newtype CostreamT m a = CostreamT {getCostreamT :: S.Stream m a}
448 deriving (Functor, MonadTrans)
450{-# DEPRECATED InterleavedT "Please use 'CostreamT' instead." #-}
451type InterleavedT = CostreamT
453instance IsStream CostreamT where
454 toStream = getCostreamT
455 fromStream = CostreamT
457------------------------------------------------------------------------------
459------------------------------------------------------------------------------
461-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CostreamT'.
462-- Interleaves two streams, yielding one element from each stream alternately.
463{-# INLINE cosplice #-}
464cosplice :: IsStream t => t m a -> t m a -> t m a
465cosplice m1 m2 = fromStream $ S.cosplice (toStream m1) (toStream m2)
467instance Semigroup (CostreamT m a) where
472-- | Same as 'cosplice'.
473{-# DEPRECATED (<=>) "Please use 'cosplice' instead." #-}
475(<=>) :: IsStream t => t m a -> t m a -> t m a
478------------------------------------------------------------------------------
480------------------------------------------------------------------------------
482instance Monoid (CostreamT m a) where
486------------------------------------------------------------------------------
488------------------------------------------------------------------------------
490instance Monad m => Monad (CostreamT m) where
492 (CostreamT (S.Stream m)) >>= f = CostreamT $ S.Stream $ \_ stp sng yld ->
493 let run x = (S.runStream x) Nothing stp sng yld
494 single a = run $ toStream (f a)
495 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
496 in m Nothing stp single yield
498------------------------------------------------------------------------------
500------------------------------------------------------------------------------
502MONAD_APPLICATIVE_INSTANCE(CostreamT,)
503MONAD_COMMON_INSTANCES(CostreamT,)
505------------------------------------------------------------------------------
507------------------------------------------------------------------------------
509-- | Adaptive parallel, in a left to right 'Semigroup' composition it tries to
510-- yield elements from the left stream as long as it can, but it can run the
511-- right stream in parallel if it needs to based on demand. The right stream
512-- can be run if the left stream blocks on IO or cannot produce elements fast
513-- enough for the consumer.
516-- main = ('toList' . 'coparallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
522-- Similarly, the monad instance of 'ParallelT' /may/ run each iteration
523-- concurrently based on demand. More concurrent iterations are started only
524-- if the previous iterations are not able to produce enough output for the
529-- import Control.Concurrent
531-- main = 'runStream' . 'coparallely' $ do
532-- n <- return 3 \<\> return 2 \<\> return 1
534-- threadDelay (n * 1000000)
535-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
538-- ThreadId 40: Delay 1
539-- ThreadId 39: Delay 2
540-- ThreadId 38: Delay 3
543-- All iterations may run in the same thread if they do not block.
545-- Note that this composition can be used to combine infinite number of streams
546-- as it explores only a bounded number of streams at a time.
548newtype CoparallelT m a = CoparallelT {getCoparallelT :: S.Stream m a}
549 deriving (Functor, MonadTrans)
551{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}
552type AsyncT = CoparallelT
554instance IsStream CoparallelT where
555 toStream = getCoparallelT
556 fromStream = CoparallelT
558------------------------------------------------------------------------------
560------------------------------------------------------------------------------
562-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CoparallelT'.
563-- Merges two streams possibly concurrently, preferring the
564-- elements from the left one when available.
565{-# INLINE coparallel #-}
566coparallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
567coparallel m1 m2 = fromStream $ S.coparallel (toStream m1) (toStream m2)
569instance MonadParallel m => Semigroup (CoparallelT m a) where
572-- | Same as 'coparallel'.
573{-# DEPRECATED (<|) "Please use 'coparallel' instead." #-}
575(<|) :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
578------------------------------------------------------------------------------
580------------------------------------------------------------------------------
582instance MonadParallel m => Monoid (CoparallelT m a) where
586------------------------------------------------------------------------------
588------------------------------------------------------------------------------
590{-# INLINE parbind #-}
592 :: (forall c. S.Stream m c -> S.Stream m c -> S.Stream m c)
594 -> (a -> S.Stream m b)
596parbind par m f = go m
599 S.Stream $ \ctx stp sng yld ->
600 let run x = (S.runStream x) ctx stp sng yld
602 yield a r = run $ f a `par` go r
603 in g Nothing stp single yield
605instance MonadParallel m => Monad (CoparallelT m) where
607 (CoparallelT m) >>= f = CoparallelT $ parbind par m (getCoparallelT . f)
608 where par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)
610------------------------------------------------------------------------------
612------------------------------------------------------------------------------
614MONAD_APPLICATIVE_INSTANCE(CoparallelT,MONADPARALLEL)
615MONAD_COMMON_INSTANCES(CoparallelT, MONADPARALLEL)
617------------------------------------------------------------------------------
619------------------------------------------------------------------------------
621-- | Round robin concurrent composition.
623-- The Semigroup instance of 'ParallelT' concurrently /merges/ streams in a
624-- round robin fashion, yielding elements from both streams alternately.
627-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
633-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations fairly
634-- concurrently using a round robin scheduling.
638-- import Control.Concurrent
640-- main = 'runStream' . 'parallely' $ do
641-- n <- return 3 \<\> return 2 \<\> return 1
643-- threadDelay (n * 1000000)
644-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
647-- ThreadId 40: Delay 1
648-- ThreadId 39: Delay 2
649-- ThreadId 38: Delay 3
652-- Unlike 'CoparallelT' all iterations are guaranteed to run fairly
653-- concurrently, unconditionally.
655-- Note that round robin composition can only combine a finite number of
656-- streams as it needs to retain state for each unfinished stream.
658newtype ParallelT m a = ParallelT {getParallelT :: S.Stream m a}
659 deriving (Functor, MonadTrans)
661instance IsStream ParallelT where
662 toStream = getParallelT
663 fromStream = ParallelT
665------------------------------------------------------------------------------
667------------------------------------------------------------------------------
669-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'.
670-- Merges two streams concurrently choosing elements from both fairly.
671{-# INLINE parallel #-}
672parallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
673parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
675instance MonadParallel m => Semigroup (ParallelT m a) where
678------------------------------------------------------------------------------
680------------------------------------------------------------------------------
682instance MonadParallel m => Monoid (ParallelT m a) where
686------------------------------------------------------------------------------
688------------------------------------------------------------------------------
690instance MonadParallel m => Monad (ParallelT m) where
692 (ParallelT m) >>= f = ParallelT $ parbind par m (getParallelT . f)
693 where par = S.joinStreamVar2 (SVarStyle Conjunction FIFO)
695------------------------------------------------------------------------------
697------------------------------------------------------------------------------
699MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
700MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
702------------------------------------------------------------------------------
703-- Serially Zipping Streams
704------------------------------------------------------------------------------
706-- | The applicative instance of 'ZipStreamM' zips a number of streams serially
707-- i.e. it produces one element from each stream serially and then zips all
711-- main = (toList . 'zipStreamly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
712-- where s1 = fromFoldable [1, 2]
713-- s2 = fromFoldable [3, 4]
714-- s3 = fromFoldable [5, 6]
720-- The 'Semigroup' instance of this type works the same way as that of
723newtype ZipStreamM m a = ZipStreamM {getZipStreamM :: S.Stream m a}
724 deriving (Functor, Semigroup, Monoid)
726instance IsStream ZipStreamM where
727 toStream = getZipStreamM
728 fromStream = ZipStreamM
730instance Monad m => Applicative (ZipStreamM m) where
731 pure = ZipStreamM . S.repeat
732 m1 <*> m2 = fromStream $ S.zipWith id (toStream m1) (toStream m2)
734------------------------------------------------------------------------------
735-- Parallely Zipping Streams
736------------------------------------------------------------------------------
738-- | Like 'ZipStreamM' but zips in parallel, it generates all the elements to
739-- be zipped concurrently.
742-- main = (toList . 'zipParallely' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
743-- where s1 = fromFoldable [1, 2]
744-- s2 = fromFoldable [3, 4]
745-- s3 = fromFoldable [5, 6]
751-- The 'Semigroup' instance of this type works the same way as that of
754newtype ZipParallelM m a = ZipParallelM {getZipParallelM :: S.Stream m a}
755 deriving (Functor, Semigroup, Monoid)
757{-# DEPRECATED ZipAsync "Please use ZipParallelM instead." #-}
758type ZipAsync = ZipParallelM
760instance IsStream ZipParallelM where
761 toStream = getZipParallelM
762 fromStream = ZipParallelM
764instance MonadParallel m => Applicative (ZipParallelM m) where
765 pure = ZipParallelM . S.repeat
766 m1 <*> m2 = fromStream $ S.zipParallelWith id (toStream m1) (toStream m2)
768-------------------------------------------------------------------------------
769-- Type adapting combinators
770-------------------------------------------------------------------------------
772-- | Adapt any specific stream type to any other specific stream type.
773adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
774adapt = fromStream . toStream
776-- | Fix the type of a polymorphic stream as 'StreamT'.
777streamly :: IsStream t => StreamT m a -> t m a
780-- | Fix the type of a polymorphic stream as 'CostreamT'.
781costreamly :: IsStream t => CostreamT m a -> t m a
784-- | Same as 'costreamly'.
785{-# DEPRECATED interleaving "Please use costreamly instead." #-}
786interleaving :: IsStream t => CostreamT m a -> t m a
787interleaving = costreamly
789-- | Fix the type of a polymorphic stream as 'CoparallelT'.
790coparallely :: IsStream t => CoparallelT m a -> t m a
793-- | Same as 'coparallely'.
794{-# DEPRECATED asyncly "Please use coparallely instead." #-}
795asyncly :: IsStream t => CoparallelT m a -> t m a
798-- | Fix the type of a polymorphic stream as 'ParallelT'.
799parallely :: IsStream t => ParallelT m a -> t m a
802-- | Fix the type of a polymorphic stream as 'ZipStreamM'.
803zipStreamly :: IsStream t => ZipStreamM m a -> t m a
806-- | Same as 'zipStreamly'.
807{-# DEPRECATED zipping "Please use zipStreamly instead." #-}
808zipping :: IsStream t => ZipStreamM m a -> t m a
811-- | Fix the type of a polymorphic stream as 'ZipParallelM'.
812zipParallely :: IsStream t => ZipParallelM m a -> t m a
815-- | Same as 'zipParallely'.
816{-# DEPRECATED zippingAsync "Please use zipParallely instead." #-}
817zippingAsync :: IsStream t => ZipParallelM m a -> t m a
818zippingAsync = zipParallely
820-------------------------------------------------------------------------------
821-- Running Streams, convenience functions specialized to types
822-------------------------------------------------------------------------------
824-- | Same as @runStream@.
825{-# DEPRECATED runStreamT "Please use runStream instead." #-}
826runStreamT :: Monad m => StreamT m a -> m ()
827runStreamT = runStream
829-- | Same as @runStream . costreamly@.
830{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}
831runInterleavedT :: Monad m => InterleavedT m a -> m ()
832runInterleavedT = runStream . costreamly
834-- | Same as @runStream . coparallely@.
835{-# DEPRECATED runAsyncT "Please use 'runStream . coparallely' instead." #-}
836runAsyncT :: Monad m => CoparallelT m a -> m ()
837runAsyncT = runStream . coparallely
839-- | Same as @runStream . parallely@.
840{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}
841runParallelT :: Monad m => ParallelT m a -> m ()
842runParallelT = runStream . parallely
844-- | Same as @runStream . zipping@.
845{-# DEPRECATED runZipStream "Please use 'runStream . zipStreamly instead." #-}
846runZipStream :: Monad m => ZipStreamM m a -> m ()
847runZipStream = runStream . zipStreamly
849-- | Same as @runStream . zippingAsync@.
850{-# DEPRECATED runZipAsync "Please use 'runStream . zipParallely instead." #-}
851runZipAsync :: Monad m => ZipParallelM m a -> m ()
852runZipAsync = runStream . zipParallely
854------------------------------------------------------------------------------
856------------------------------------------------------------------------------
858-- | A serial IO stream of elements of type @a@. See 'StreamT' documentation
860type Stream a = StreamT IO a
862-- | An interleaving serial IO stream of elements of type @a@. See 'CostreamT'
863-- documentation for more details.
864type Costream a = CostreamT IO a
866-- | A demand driven left biased parallelly composing IO stream of elements of
867-- type @a@. See 'CoparallelT' documentation for more details.
868type Coparallel a = CoparallelT IO a
870-- | A round robin parallelly composing IO stream of elements of type @a@.
871-- See 'ParallelT' documentation for more details.
872type Parallel a = ParallelT IO a
874-- | An IO stream whose applicative instance zips streams serially.
875type ZipStream a = ZipStreamM IO a
877-- | An IO stream whose applicative instance zips streams parallely.
878type ZipParallel a = ZipParallelM IO a
880------------------------------------------------------------------------------
882------------------------------------------------------------------------------
884-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'
885-- container of streams using the specified stream sum operation.
887-- @foldWith 'parallel' $ map return [1..3]@
888{-# INLINABLE foldWith #-}
889foldWith :: (IsStream t, Foldable f)
890 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
891foldWith f = foldr f nil
893-- | A variant of 'foldMap' that allows you to map a monadic streaming action
894-- on a 'Foldable' container and then fold it using the specified stream sum
897-- @foldMapWith 'parallel' return [1..3]@
898{-# INLINABLE foldMapWith #-}
899foldMapWith :: (IsStream t, Foldable f)
900 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
901foldMapWith f g = foldr (f . g) nil
903-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
904-- monadic streaming function is the last argument.
905{-# INLINABLE forEachWith #-}
906forEachWith :: (IsStream t, Foldable f)
907 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
908forEachWith 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