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
83 , serially -- deprecated
85 , interleaving -- deprecated
87 , asyncly -- deprecated
90 , zipping -- deprecated
92 , zippingAsync -- deprecated
96 , runStreamT -- deprecated
97 , runInterleavedT -- deprecated
98 , runAsyncT -- deprecated
99 , runParallelT -- deprecated
100 , runZipStream -- deprecated
101 , runZipAsync -- deprecated
110import Control.Monad (ap)
111import Control.Monad.Base (MonadBase (..), liftBaseDefault)
112import Control.Monad.Catch (MonadThrow, throwM)
113-- import Control.Monad.Error.Class (MonadError(..))
114import Control.Monad.IO.Class (MonadIO(..))
115import Control.Monad.Reader.Class (MonadReader(..))
116import Control.Monad.State.Class (MonadState(..))
117import Control.Monad.Trans.Class (MonadTrans (lift))
118import Data.Semigroup (Semigroup(..))
119import Streamly.Core ( MonadParallel
120 , SVar, SVarStyle(..)
121 , SVarTag(..), SVarSched(..))
122import qualified Streamly.Core as S
124------------------------------------------------------------------------------
125-- Types that can behave as a Stream
126------------------------------------------------------------------------------
128-- | Class of types that can represent a stream of elements of some type 'a' in
132class IsStream t where
133 toStream :: t m a -> S.Stream m a
134 fromStream :: S.Stream m a -> t m a
136-- | Same as 'IsStream'.
139{-# DEPRECATED Streaming "Please use IsStream instead." #-}
140type Streaming = IsStream
142------------------------------------------------------------------------------
143-- Constructing a stream
144------------------------------------------------------------------------------
154nil :: IsStream t => t m a
155nil = fromStream S.nil
159-- | Constructs a stream by adding a monadic action at the head of an existing
160-- stream. For example:
163-- > toList $ getLine \`consM` getLine \`consM` nil
170consM :: (IsStream t, Monad m) => m a -> t m a -> t m a
171consM m r = fromStream $ S.consM m (toStream r)
175-- | Operator equivalent of 'consM'.
178-- > toList $ getLine |: getLine |: nil
185(|:) :: (IsStream t, Monad m) => m a -> t m a -> t m a
190-- | Construct a stream by adding a pure value at the head of an existing
191-- stream. Same as @consM . return@. For example:
194-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
199cons :: IsStream t => a -> t m a -> t m a
200cons a r = fromStream $ S.cons a (toStream r)
204-- | Operator equivalent of 'cons'.
207-- > toList $ 1 .: 2 .: 3 .: nil
212(.:) :: IsStream t => a -> t m a -> t m a
215-- | Build a stream from its church encoding. The function passed maps
216-- directly to the underlying representation of the stream type. The second
217-- parameter to the function is the "yield" function yielding a value and the
218-- remaining stream if any otherwise 'Nothing'. The third parameter is to
219-- represent an "empty" stream.
220streamBuild :: IsStream t
221 => (forall r. Maybe (SVar m a)
222 -> (a -> t m a -> m r)
227streamBuild k = fromStream $ S.Stream $ \svr stp sng yld ->
228 let yield a r = yld a (toStream r)
229 in k svr yield sng stp
231-- | Build a singleton stream from a callback function.
232fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
233fromCallback k = fromStream $ S.Stream $ \_ _ sng _ -> k sng
235-- | Read an SVar to get a stream.
236fromSVar :: (MonadParallel m, IsStream t) => SVar m a -> t m a
237fromSVar sv = fromStream $ S.fromStreamVar sv
239------------------------------------------------------------------------------
240-- Destroying a stream
241------------------------------------------------------------------------------
243-- | Fold a stream using its church encoding. The second argument is the "step"
244-- function consuming an element and the remaining stream, if any. The third
245-- argument is for consuming an "empty" stream that yields nothing.
249 -> (a -> t m a -> m r)
254streamFold svr step single blank m =
255 let yield a x = step a (fromStream x)
256 in (S.runStream (toStream m)) svr blank single yield
258-- | Run a streaming composition, discard the results. By default it interprets
259-- the stream as 'StreamT', to run other types of streams use the type adapting
260-- combinators for example @runStream . 'parallely'@.
263runStream :: Monad m => StreamT m a -> m ()
264runStream m = go (toStream m)
270 in (S.runStream m1) Nothing stop single yield
272-- | Same as 'runStream'
275{-# DEPRECATED runStreaming "Please use runStream instead." #-}
276runStreaming :: (Monad m, IsStream t) => t m a -> m ()
277runStreaming = runStream . adapt
279-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
280-- be read back from the SVar using 'fromSVar'.
281toSVar :: (IsStream t, MonadParallel m) => SVar m a -> t m a -> m ()
282toSVar sv m = S.toStreamVar sv (toStream m)
284------------------------------------------------------------------------------
286------------------------------------------------------------------------------
288-- XXX Get rid of this?
289-- | Make a stream asynchronous, triggers the computation and returns a stream
290-- in the underlying monad representing the output generated by the original
291-- computation. The returned action is exhaustible and must be drained once. If
292-- not drained fully we may have a thread blocked forever and once exhausted it
293-- will always return 'empty'.
296async :: (IsStream t, MonadParallel m) => t m a -> m (t m a)
298 sv <- S.newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)
301------------------------------------------------------------------------------
302-- CPP macros for common instances
303------------------------------------------------------------------------------
305-- XXX use template haskell instead and include Monoid and IsStream instances
308#define MONADPARALLEL , MonadParallel m
310#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
311instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
312 pure = STREAM . S.singleton; \
315#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
316instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\
317 liftBase = liftBaseDefault }; \
319instance (MonadIO m CONSTRAINT) => MonadIO (STREAM m) where { \
320 liftIO = lift . liftIO }; \
322instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \
323 throwM = lift . throwM }; \
326instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \
327 throwError = lift . throwError; \
329 fromStream $ S.withCatchError (toStream m) (\e -> toStream $ h e) }; \
332instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \
334 local f m = fromStream $ S.withLocal f (toStream m) }; \
336instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
338 put x = lift (put x); \
339 state k = lift (state k) }
341------------------------------------------------------------------------------
343------------------------------------------------------------------------------
345-- | The 'Semigroup' instance of 'StreamT' appends two streams sequentially,
346-- yielding all elements from the first stream, and then all elements from the
350-- main = ('toList' . 'streamly' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
356-- The 'Monad' instance runs the /monadic continuation/ for each
357-- element of the stream, streamly.
360-- main = 'runStream' . 'streamly' $ do
361-- x <- return 1 \<\> return 2
369-- 'StreamT' nests streams streamly in a depth first manner.
372-- main = 'runStream' . 'streamly' $ do
373-- x <- return 1 \<\> return 2
374-- y <- return 3 \<\> return 4
375-- liftIO $ print (x, y)
384-- This behavior of 'StreamT' is exactly like a list transformer. We call the
385-- monadic code being run for each element of the stream a monadic
386-- continuation. In imperative paradigm we can think of this composition as
387-- nested @for@ loops and the monadic continuation is the body of the loop. The
388-- loop iterates for all elements of the stream.
390-- The 'streamly' combinator can be omitted as the default stream type is
392-- Note that serial composition can be used to combine an infinite number of
393-- streams as it explores only one stream at a time.
396newtype StreamT m a = StreamT {getStreamT :: S.Stream m a}
397 deriving (Semigroup, Monoid, Functor, MonadTrans)
399instance IsStream StreamT where
400 toStream = getStreamT
403------------------------------------------------------------------------------
405------------------------------------------------------------------------------
407-- | Polymorphic version of the 'Semigroup' operation '<>' of 'StreamT'.
408-- Appends two streams sequentially, yielding all elements from the first
409-- stream, and then all elements from the second stream.
413splice :: IsStream t => t m a -> t m a -> t m a
414splice m1 m2 = fromStream $ S.splice (toStream m1) (toStream m2)
416------------------------------------------------------------------------------
418------------------------------------------------------------------------------
420instance Monad m => Monad (StreamT m) where
422 (StreamT (S.Stream m)) >>= f = StreamT $ S.Stream $ \_ stp sng yld ->
423 let run x = (S.runStream x) Nothing stp sng yld
424 single a = run $ toStream (f a)
425 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
426 in m Nothing stp single yield
428------------------------------------------------------------------------------
430------------------------------------------------------------------------------
432MONAD_APPLICATIVE_INSTANCE(StreamT,)
433MONAD_COMMON_INSTANCES(StreamT,)
435------------------------------------------------------------------------------
437------------------------------------------------------------------------------
439-- | The 'Semigroup' instance of 'CostreamT' interleaves two streams,
440-- yielding one element from each stream alternately.
443-- main = ('toList' . 'costreamly' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
449-- Similarly, the 'Monad' instance fairly interleaves the iterations of the
450-- inner and the outer loop, nesting loops in a breadth first manner.
454-- main = 'runStream' . 'costreamly' $ do
455-- x <- return 1 \<\> return 2
456-- y <- return 3 \<\> return 4
457-- liftIO $ print (x, y)
466-- Note that costreamly composition can only combine a finite number of
467-- streams as it needs to retain state for each unfinished stream.
470newtype CostreamT m a = CostreamT {getCostreamT :: S.Stream m a}
471 deriving (Functor, MonadTrans)
475{-# DEPRECATED InterleavedT "Please use 'CostreamT' instead." #-}
476type InterleavedT = CostreamT
478instance IsStream CostreamT where
479 toStream = getCostreamT
480 fromStream = CostreamT
482------------------------------------------------------------------------------
484------------------------------------------------------------------------------
486-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CostreamT'.
487-- Interleaves two streams, yielding one element from each stream alternately.
490{-# INLINE cosplice #-}
491cosplice :: IsStream t => t m a -> t m a -> t m a
492cosplice m1 m2 = fromStream $ S.cosplice (toStream m1) (toStream m2)
494instance Semigroup (CostreamT m a) where
499-- | Same as 'cosplice'.
502{-# DEPRECATED (<=>) "Please use 'cosplice' instead." #-}
504(<=>) :: IsStream t => t m a -> t m a -> t m a
507------------------------------------------------------------------------------
509------------------------------------------------------------------------------
511instance Monoid (CostreamT m a) where
515------------------------------------------------------------------------------
517------------------------------------------------------------------------------
519instance Monad m => Monad (CostreamT m) where
521 (CostreamT (S.Stream m)) >>= f = CostreamT $ S.Stream $ \_ stp sng yld ->
522 let run x = (S.runStream x) Nothing stp sng yld
523 single a = run $ toStream (f a)
524 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
525 in m Nothing stp single yield
527------------------------------------------------------------------------------
529------------------------------------------------------------------------------
531MONAD_APPLICATIVE_INSTANCE(CostreamT,)
532MONAD_COMMON_INSTANCES(CostreamT,)
534------------------------------------------------------------------------------
536------------------------------------------------------------------------------
538-- | Adaptive parallel, in a left to right 'Semigroup' composition it tries to
539-- yield elements from the left stream as long as it can, but it can run the
540-- right stream in parallel if it needs to based on demand. The right stream
541-- can be run if the left stream blocks on IO or cannot produce elements fast
542-- enough for the consumer.
545-- main = ('toList' . 'coparallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
551-- Similarly, the monad instance of 'ParallelT' /may/ run each iteration
552-- concurrently based on demand. More concurrent iterations are started only
553-- if the previous iterations are not able to produce enough output for the
558-- import Control.Concurrent
560-- main = 'runStream' . 'coparallely' $ do
561-- n <- return 3 \<\> return 2 \<\> return 1
563-- threadDelay (n * 1000000)
564-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
567-- ThreadId 40: Delay 1
568-- ThreadId 39: Delay 2
569-- ThreadId 38: Delay 3
572-- All iterations may run in the same thread if they do not block.
574-- Note that this composition can be used to combine infinite number of streams
575-- as it explores only a bounded number of streams at a time.
578newtype CoparallelT m a = CoparallelT {getCoparallelT :: S.Stream m a}
579 deriving (Functor, MonadTrans)
583{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}
584type AsyncT = CoparallelT
586instance IsStream CoparallelT where
587 toStream = getCoparallelT
588 fromStream = CoparallelT
590------------------------------------------------------------------------------
592------------------------------------------------------------------------------
594-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CoparallelT'.
595-- Merges two streams possibly concurrently, preferring the
596-- elements from the left one when available.
599{-# INLINE coparallel #-}
600coparallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
601coparallel m1 m2 = fromStream $ S.coparallel (toStream m1) (toStream m2)
603instance MonadParallel m => Semigroup (CoparallelT m a) where
606-- | Same as 'coparallel'.
609{-# DEPRECATED (<|) "Please use 'coparallel' instead." #-}
611(<|) :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
614------------------------------------------------------------------------------
616------------------------------------------------------------------------------
618instance MonadParallel m => Monoid (CoparallelT m a) where
622------------------------------------------------------------------------------
624------------------------------------------------------------------------------
626{-# INLINE parbind #-}
628 :: (forall c. S.Stream m c -> S.Stream m c -> S.Stream m c)
630 -> (a -> S.Stream m b)
632parbind par m f = go m
635 S.Stream $ \ctx stp sng yld ->
636 let run x = (S.runStream x) ctx stp sng yld
638 yield a r = run $ f a `par` go r
639 in g Nothing stp single yield
641instance MonadParallel m => Monad (CoparallelT m) where
643 (CoparallelT m) >>= f = CoparallelT $ parbind par m (getCoparallelT . f)
644 where par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)
646------------------------------------------------------------------------------
648------------------------------------------------------------------------------
650MONAD_APPLICATIVE_INSTANCE(CoparallelT,MONADPARALLEL)
651MONAD_COMMON_INSTANCES(CoparallelT, MONADPARALLEL)
653------------------------------------------------------------------------------
655------------------------------------------------------------------------------
657-- | Round robin concurrent composition.
659-- The Semigroup instance of 'ParallelT' concurrently /merges/ streams in a
660-- round robin fashion, yielding elements from both streams alternately.
663-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
669-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations fairly
670-- concurrently using a round robin scheduling.
674-- import Control.Concurrent
676-- main = 'runStream' . 'parallely' $ do
677-- n <- return 3 \<\> return 2 \<\> return 1
679-- threadDelay (n * 1000000)
680-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
683-- ThreadId 40: Delay 1
684-- ThreadId 39: Delay 2
685-- ThreadId 38: Delay 3
688-- Unlike 'CoparallelT' all iterations are guaranteed to run fairly
689-- concurrently, unconditionally.
691-- Note that round robin composition can only combine a finite number of
692-- streams as it needs to retain state for each unfinished stream.
695newtype ParallelT m a = ParallelT {getParallelT :: S.Stream m a}
696 deriving (Functor, MonadTrans)
698instance IsStream ParallelT where
699 toStream = getParallelT
700 fromStream = ParallelT
702------------------------------------------------------------------------------
704------------------------------------------------------------------------------
706-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'.
707-- Merges two streams concurrently choosing elements from both fairly.
710{-# INLINE parallel #-}
711parallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
712parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
714instance MonadParallel m => Semigroup (ParallelT m a) where
717------------------------------------------------------------------------------
719------------------------------------------------------------------------------
721instance MonadParallel m => Monoid (ParallelT m a) where
725------------------------------------------------------------------------------
727------------------------------------------------------------------------------
729instance MonadParallel m => Monad (ParallelT m) where
731 (ParallelT m) >>= f = ParallelT $ parbind par m (getParallelT . f)
732 where par = S.joinStreamVar2 (SVarStyle Conjunction FIFO)
734------------------------------------------------------------------------------
736------------------------------------------------------------------------------
738MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
739MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
741------------------------------------------------------------------------------
742-- Serially Zipping Streams
743------------------------------------------------------------------------------
745-- | The applicative instance of 'ZipStreamM' zips a number of streams serially
746-- i.e. it produces one element from each stream serially and then zips all
750-- main = (toList . 'zipStreamly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
751-- where s1 = fromFoldable [1, 2]
752-- s2 = fromFoldable [3, 4]
753-- s3 = fromFoldable [5, 6]
759-- The 'Semigroup' instance of this type works the same way as that of
763newtype ZipStreamM m a = ZipStreamM {getZipStreamM :: S.Stream m a}
764 deriving (Functor, Semigroup, Monoid)
766instance IsStream ZipStreamM where
767 toStream = getZipStreamM
768 fromStream = ZipStreamM
770instance Monad m => Applicative (ZipStreamM m) where
771 pure = ZipStreamM . S.repeat
772 m1 <*> m2 = fromStream $ S.zipWith id (toStream m1) (toStream m2)
774------------------------------------------------------------------------------
775-- Parallely Zipping Streams
776------------------------------------------------------------------------------
778-- | Like 'ZipStreamM' but zips in parallel, it generates all the elements to
779-- be zipped concurrently.
782-- main = (toList . 'zipParallely' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
783-- where s1 = fromFoldable [1, 2]
784-- s2 = fromFoldable [3, 4]
785-- s3 = fromFoldable [5, 6]
791-- The 'Semigroup' instance of this type works the same way as that of
795newtype ZipParallelM m a = ZipParallelM {getZipParallelM :: S.Stream m a}
796 deriving (Functor, Semigroup, Monoid)
800{-# DEPRECATED ZipAsync "Please use ZipParallelM instead." #-}
801type ZipAsync = ZipParallelM
803instance IsStream ZipParallelM where
804 toStream = getZipParallelM
805 fromStream = ZipParallelM
807instance MonadParallel m => Applicative (ZipParallelM m) where
808 pure = ZipParallelM . S.repeat
809 m1 <*> m2 = fromStream $ S.zipParallelWith id (toStream m1) (toStream m2)
811-------------------------------------------------------------------------------
812-- Type adapting combinators
813-------------------------------------------------------------------------------
815-- | Adapt any specific stream type to any other specific stream type.
818adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
819adapt = fromStream . toStream
821-- | Fix the type of a polymorphic stream as 'StreamT'.
824streamly :: IsStream t => StreamT m a -> t m a
827-- | Same as 'streamly'.
830{-# DEPRECATED serially "Please use streamly instead." #-}
831serially :: IsStream t => StreamT m a -> t m a
834-- | Fix the type of a polymorphic stream as 'CostreamT'.
837costreamly :: IsStream t => CostreamT m a -> t m a
840-- | Same as 'costreamly'.
843{-# DEPRECATED interleaving "Please use costreamly instead." #-}
844interleaving :: IsStream t => CostreamT m a -> t m a
845interleaving = costreamly
847-- | Fix the type of a polymorphic stream as 'CoparallelT'.
850coparallely :: IsStream t => CoparallelT m a -> t m a
853-- | Same as 'coparallely'.
856{-# DEPRECATED asyncly "Please use coparallely instead." #-}
857asyncly :: IsStream t => CoparallelT m a -> t m a
860-- | Fix the type of a polymorphic stream as 'ParallelT'.
863parallely :: IsStream t => ParallelT m a -> t m a
866-- | Fix the type of a polymorphic stream as 'ZipStreamM'.
869zipStreamly :: IsStream t => ZipStreamM m a -> t m a
872-- | Same as 'zipStreamly'.
875{-# DEPRECATED zipping "Please use zipStreamly instead." #-}
876zipping :: IsStream t => ZipStreamM m a -> t m a
879-- | Fix the type of a polymorphic stream as 'ZipParallelM'.
882zipParallely :: IsStream t => ZipParallelM m a -> t m a
885-- | Same as 'zipParallely'.
888{-# DEPRECATED zippingAsync "Please use zipParallely instead." #-}
889zippingAsync :: IsStream t => ZipParallelM m a -> t m a
890zippingAsync = zipParallely
892-------------------------------------------------------------------------------
893-- Running Streams, convenience functions specialized to types
894-------------------------------------------------------------------------------
896-- | Same as @runStream@.
899{-# DEPRECATED runStreamT "Please use runStream instead." #-}
900runStreamT :: Monad m => StreamT m a -> m ()
901runStreamT = runStream
903-- | Same as @runStream . costreamly@.
906{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}
907runInterleavedT :: Monad m => InterleavedT m a -> m ()
908runInterleavedT = runStream . costreamly
910-- | Same as @runStream . coparallely@.
913{-# DEPRECATED runAsyncT "Please use 'runStream . coparallely' instead." #-}
914runAsyncT :: Monad m => CoparallelT m a -> m ()
915runAsyncT = runStream . coparallely
917-- | Same as @runStream . parallely@.
920{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}
921runParallelT :: Monad m => ParallelT m a -> m ()
922runParallelT = runStream . parallely
924-- | Same as @runStream . zipping@.
927{-# DEPRECATED runZipStream "Please use 'runStream . zipStreamly instead." #-}
928runZipStream :: Monad m => ZipStreamM m a -> m ()
929runZipStream = runStream . zipStreamly
931-- | Same as @runStream . zippingAsync@.
934{-# DEPRECATED runZipAsync "Please use 'runStream . zipParallely instead." #-}
935runZipAsync :: Monad m => ZipParallelM m a -> m ()
936runZipAsync = runStream . zipParallely
938------------------------------------------------------------------------------
940------------------------------------------------------------------------------
942-- | A serial IO stream of elements of type @a@. See 'StreamT' documentation
946type Stream a = StreamT IO a
948-- | An interleaving serial IO stream of elements of type @a@. See 'CostreamT'
949-- documentation for more details.
952type Costream a = CostreamT IO a
954-- | A demand driven left biased parallelly composing IO stream of elements of
955-- type @a@. See 'CoparallelT' documentation for more details.
958type Coparallel a = CoparallelT IO a
960-- | A round robin parallelly composing IO stream of elements of type @a@.
961-- See 'ParallelT' documentation for more details.
964type Parallel a = ParallelT IO a
966-- | An IO stream whose applicative instance zips streams serially.
969type ZipStream a = ZipStreamM IO a
971-- | An IO stream whose applicative instance zips streams parallely.
974type ZipParallel a = ZipParallelM IO a
976------------------------------------------------------------------------------
978------------------------------------------------------------------------------
980-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'
981-- container of streams using the specified stream sum operation.
983-- @foldWith 'parallel' $ map return [1..3]@
986{-# INLINABLE foldWith #-}
987foldWith :: (IsStream t, Foldable f)
988 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
989foldWith f = foldr f nil
991-- | A variant of 'foldMap' that allows you to map a monadic streaming action
992-- on a 'Foldable' container and then fold it using the specified stream sum
995-- @foldMapWith 'parallel' return [1..3]@
998{-# INLINABLE foldMapWith #-}
999foldMapWith :: (IsStream t, Foldable f)
1000 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
1001foldMapWith f g = foldr (f . g) nil
1003-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
1004-- monadic streaming function is the last argument.
1007{-# INLINABLE forEachWith #-}
1008forEachWith :: (IsStream t, Foldable f)
1009 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
1010forEachWith 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