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
46 , 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
105 , zipAsyncWith -- deprecated
114import Control.Monad (ap)
115import Control.Monad.Base (MonadBase (..), liftBaseDefault)
116import Control.Monad.Catch (MonadThrow, throwM)
117-- import Control.Monad.Error.Class (MonadError(..))
118import Control.Monad.IO.Class (MonadIO(..))
119import Control.Monad.Reader.Class (MonadReader(..))
120import Control.Monad.State.Class (MonadState(..))
121import Control.Monad.Trans.Class (MonadTrans (lift))
122import Data.Semigroup (Semigroup(..))
123import Prelude hiding (zipWith)
124import Streamly.Core ( MonadParallel
126 , SVar, SVarStyle(..)
127 , SVarTag(..), SVarSched(..))
128import qualified Streamly.Core as S
130------------------------------------------------------------------------------
131-- Types that can behave as a Stream
132------------------------------------------------------------------------------
134-- | Class of types that can represent a stream of elements of some type 'a' in
136class IsStream t where
137 toStream :: t m a -> S.Stream m a
138 fromStream :: S.Stream m a -> t m a
140-- | Same as 'IsStream'.
141{-# Deprecated Streaming "Please use IsStream instead." #-}
142type Streaming = IsStream
144------------------------------------------------------------------------------
145-- Constructing a stream
146------------------------------------------------------------------------------
148-- | Represesnts an empty stream just like @[]@ represents an empty list.
149nil :: IsStream t => t m a
150nil = fromStream S.nil
154-- | Constructs a stream by adding a pure value at the head of an existing
155-- stream, just like ':' constructs lists. For example:
158-- > let stream = 1 \`cons` 2 \`cons` 3 \`cons` nil
159-- > (toList . streamly) stream
162cons :: IsStream t => a -> t m a -> t m a
163cons a r = fromStream $ S.cons a (toStream r)
167-- | Operator equivalent of 'cons' so that you can construct a stream of pure
168-- values more succinctly like this:
171-- > let stream = 1 .: 2 .: 3 .: nil
172-- > (toList . streamly) stream
176-- '.:' constructs a stream just like ':' constructs a list.
178-- Also note that another equivalent way of building streams from pure values
182-- > let stream = pure 1 <> pure 2 <> pure 3
183-- > (toList . streamly) stream
187-- In the first method we construct a stream by adding one element at a time.
188-- In the second method we first construct singleton streams using 'pure' and
189-- then compose all those streams together using the 'Semigroup' style
190-- composition of streams. The former method is a bit more efficient than the
193(.:) :: IsStream t => a -> t m a -> t m a
196-- | Build a stream from its church encoding. The function passed maps
197-- directly to the underlying representation of the stream type. The second
198-- parameter to the function is the "yield" function yielding a value and the
199-- remaining stream if any otherwise 'Nothing'. The third parameter is to
200-- represent an "empty" stream.
201streamBuild :: IsStream t
202 => (forall r. Maybe (SVar m a)
203 -> (a -> t m a -> m r)
208streamBuild k = fromStream $ S.Stream $ \svr stp sng yld ->
209 let yield a r = yld a (toStream r)
210 in k svr yield sng stp
212-- | Build a singleton stream from a callback function.
213fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
214fromCallback k = fromStream $ S.Stream $ \_ _ sng _ -> k sng
216-- | Read an SVar to get a stream.
217fromSVar :: (MonadParallel m, IsStream t) => SVar m a -> t m a
218fromSVar sv = fromStream $ S.fromStreamVar sv
220------------------------------------------------------------------------------
221-- Destroying a stream
222------------------------------------------------------------------------------
224-- | Fold a stream using its church encoding. The second argument is the "step"
225-- function consuming an element and the remaining stream, if any. The third
226-- argument is for consuming an "empty" stream that yields nothing.
230 -> (a -> t m a -> m r)
235streamFold svr step single blank m =
236 let yield a x = step a (fromStream x)
237 in (S.runStream (toStream m)) svr blank single yield
239-- | Run a streaming composition, discard the results. By default it interprets
240-- the stream as 'StreamT', to run other types of streams use the type adapting
241-- combinators for example @runStream . 'parallely'@.
242runStream :: Monad m => StreamT m a -> m ()
243runStream m = go (toStream m)
249 in (S.runStream m1) Nothing stop single yield
251-- | Same as 'runStream'
252{-# Deprecated runStreaming "Please use runStream instead." #-}
253runStreaming :: (Monad m, IsStream t) => t m a -> m ()
254runStreaming = runStream . adapt
256-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
257-- be read back from the SVar using 'fromSVar'.
258toSVar :: (IsStream t, MonadParallel m) => SVar m a -> t m a -> m ()
259toSVar sv m = S.toStreamVar sv (toStream m)
261------------------------------------------------------------------------------
263------------------------------------------------------------------------------
265-- XXX Get rid of this?
266-- | Make a stream asynchronous, triggers the computation and returns a stream
267-- in the underlying monad representing the output generated by the original
268-- computation. The returned action is exhaustible and must be drained once. If
269-- not drained fully we may have a thread blocked forever and once exhausted it
270-- will always return 'empty'.
272async :: (IsStream t, MonadParallel m) => t m a -> m (t m a)
274 sv <- S.newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)
277------------------------------------------------------------------------------
278-- CPP macros for common instances
279------------------------------------------------------------------------------
281-- XXX use template haskell instead and include Monoid and IsStream instances
284#define MONADPARALLEL , MonadParallel m
286#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
287instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
288 pure = STREAM . singleton; \
291#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
292instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\
293 liftBase = liftBaseDefault }; \
295instance (MonadIO m CONSTRAINT) => MonadIO (STREAM m) where { \
296 liftIO = lift . liftIO }; \
298instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \
299 throwM = lift . throwM }; \
302instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \
303 throwError = lift . throwError; \
305 fromStream $ S.withCatchError (toStream m) (\e -> toStream $ h e) }; \
308instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \
310 local f m = fromStream $ S.withLocal f (toStream m) }; \
312instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
314 put x = lift (put x); \
315 state k = lift (state k) }
317------------------------------------------------------------------------------
319------------------------------------------------------------------------------
321-- | The 'Semigroup' instance of 'StreamT' appends two streams sequentially,
322-- yielding all elements from the first stream, and then all elements from the
326-- main = ('toList' . 'streamly' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
332-- The 'Monad' instance runs the /monadic continuation/ for each
333-- element of the stream, streamly.
336-- main = 'runStream' . 'streamly' $ do
337-- x <- return 1 \<\> return 2
345-- 'StreamT' nests streams streamly in a depth first manner.
348-- main = 'runStream' . 'streamly' $ do
349-- x <- return 1 \<\> return 2
350-- y <- return 3 \<\> return 4
351-- liftIO $ print (x, y)
360-- This behavior of 'StreamT' is exactly like a list transformer. We call the
361-- monadic code being run for each element of the stream a monadic
362-- continuation. In imperative paradigm we can think of this composition as
363-- nested @for@ loops and the monadic continuation is the body of the loop. The
364-- loop iterates for all elements of the stream.
366-- The 'streamly' combinator can be omitted as the default stream type is
368-- Note that serial composition can be used to combine an infinite number of
369-- streams as it explores only one stream at a time.
371newtype StreamT m a = StreamT {getStreamT :: S.Stream m a}
372 deriving (Semigroup, Monoid, Functor, MonadTrans)
374instance IsStream StreamT where
375 toStream = getStreamT
378------------------------------------------------------------------------------
380------------------------------------------------------------------------------
382-- | Polymorphic version of the 'Semigroup' operation '<>' of 'StreamT'.
383-- Appends two streams sequentially, yielding all elements from the first
384-- stream, and then all elements from the second stream.
386splice :: IsStream t => t m a -> t m a -> t m a
387splice m1 m2 = fromStream $ S.splice (toStream m1) (toStream m2)
389------------------------------------------------------------------------------
391------------------------------------------------------------------------------
393instance Monad m => Monad (StreamT m) where
395 (StreamT (S.Stream m)) >>= f = StreamT $ S.Stream $ \_ stp sng yld ->
396 let run x = (S.runStream x) Nothing stp sng yld
397 single a = run $ toStream (f a)
398 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
399 in m Nothing stp single yield
401------------------------------------------------------------------------------
403------------------------------------------------------------------------------
405MONAD_APPLICATIVE_INSTANCE(StreamT,)
406MONAD_COMMON_INSTANCES(StreamT,)
408------------------------------------------------------------------------------
410------------------------------------------------------------------------------
412-- | The 'Semigroup' instance of 'CostreamT' interleaves two streams,
413-- yielding one element from each stream alternately.
416-- main = ('toList' . 'interleaving' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
422-- Similarly, the 'Monad' instance fairly interleaves the iterations of the
423-- inner and the outer loop, nesting loops in a breadth first manner.
427-- main = 'runStream' . 'interleaving' $ do
428-- x <- return 1 \<\> return 2
429-- y <- return 3 \<\> return 4
430-- liftIO $ print (x, y)
439-- Note that interleaving composition can only combine a finite number of
440-- streams as it needs to retain state for each unfinished stream.
442newtype CostreamT m a = CostreamT {getCostreamT :: S.Stream m a}
443 deriving (Functor, MonadTrans)
445{-# DEPRECATED InterleavedT "Please use 'CostreamT' instead." #-}
446type InterleavedT = CostreamT
448instance IsStream CostreamT where
449 toStream = getCostreamT
450 fromStream = CostreamT
452------------------------------------------------------------------------------
454------------------------------------------------------------------------------
456-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CostreamT'.
457-- Interleaves two streams, yielding one element from each stream alternately.
458{-# INLINE cosplice #-}
459cosplice :: IsStream t => t m a -> t m a -> t m a
460cosplice m1 m2 = fromStream $ S.cosplice (toStream m1) (toStream m2)
462instance Semigroup (CostreamT m a) where
467-- | Same as 'cosplice'.
468{-# DEPRECATED (<=>) "Please use 'cosplice' instead." #-}
470(<=>) :: IsStream t => t m a -> t m a -> t m a
473------------------------------------------------------------------------------
475------------------------------------------------------------------------------
477instance Monoid (CostreamT m a) where
481------------------------------------------------------------------------------
483------------------------------------------------------------------------------
485instance Monad m => Monad (CostreamT m) where
487 (CostreamT (S.Stream m)) >>= f = CostreamT $ S.Stream $ \_ stp sng yld ->
488 let run x = (S.runStream x) Nothing stp sng yld
489 single a = run $ toStream (f a)
490 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
491 in m Nothing stp single yield
493------------------------------------------------------------------------------
495------------------------------------------------------------------------------
497MONAD_APPLICATIVE_INSTANCE(CostreamT,)
498MONAD_COMMON_INSTANCES(CostreamT,)
500------------------------------------------------------------------------------
502------------------------------------------------------------------------------
504-- | Adaptive parallel, in a left to right 'Semigroup' composition it tries to
505-- yield elements from the left stream as long as it can, but it can run the
506-- right stream in parallel if it needs to based on demand. The right stream
507-- can be run if the left stream blocks on IO or cannot produce elements fast
508-- enough for the consumer.
511-- main = ('toList' . 'coparallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
517-- Similarly, the monad instance of 'ParallelT' /may/ run each iteration
518-- concurrently based on demand. More concurrent iterations are started only
519-- if the previous iterations are not able to produce enough output for the
524-- import Control.Concurrent
526-- main = 'runStream' . 'coparallely' $ do
527-- n <- return 3 \<\> return 2 \<\> return 1
529-- threadDelay (n * 1000000)
530-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
533-- ThreadId 40: Delay 1
534-- ThreadId 39: Delay 2
535-- ThreadId 38: Delay 3
538-- All iterations may run in the same thread if they do not block.
540-- Note that this composition can be used to combine infinite number of streams
541-- as it explores only a bounded number of streams at a time.
543newtype CoparallelT m a = CoparallelT {getCoparallelT :: S.Stream m a}
544 deriving (Functor, MonadTrans)
546{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}
547type AsyncT = CoparallelT
549instance IsStream CoparallelT where
550 toStream = getCoparallelT
551 fromStream = CoparallelT
553------------------------------------------------------------------------------
555------------------------------------------------------------------------------
557-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CoparallelT', but
558-- polymorphic. Merges two streams possibly concurrently, preferring the
559-- elements from the left one when available.
560{-# INLINE coparallel #-}
561coparallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
562coparallel m1 m2 = fromStream $ S.coparallel (toStream m1) (toStream m2)
564instance MonadParallel m => Semigroup (CoparallelT m a) where
567-- | Same as 'coparallel'.
568{-# DEPRECATED (<|) "Please use 'coparallel' instead." #-}
570(<|) :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
573------------------------------------------------------------------------------
575------------------------------------------------------------------------------
577instance MonadParallel m => Monoid (CoparallelT m a) where
581------------------------------------------------------------------------------
583------------------------------------------------------------------------------
585{-# INLINE parbind #-}
587 :: (forall c. S.Stream m c -> S.Stream m c -> S.Stream m c)
589 -> (a -> S.Stream m b)
591parbind par m f = go m
594 S.Stream $ \ctx stp sng yld ->
595 let run x = (S.runStream x) ctx stp sng yld
597 yield a r = run $ f a `par` go r
598 in g Nothing stp single yield
600instance MonadParallel m => Monad (CoparallelT m) where
602 (CoparallelT m) >>= f = CoparallelT $ parbind par m (getCoparallelT . f)
603 where par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)
605------------------------------------------------------------------------------
607------------------------------------------------------------------------------
609MONAD_APPLICATIVE_INSTANCE(CoparallelT,MONADPARALLEL)
610MONAD_COMMON_INSTANCES(CoparallelT, MONADPARALLEL)
612------------------------------------------------------------------------------
614------------------------------------------------------------------------------
616-- | Round robin concurrent composition.
618-- The Semigroup instance of 'ParallelT' concurrently /merges/ streams in a
619-- round robin fashion, yielding elements from both streams alternately.
622-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
628-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations fairly
629-- concurrently using a round robin scheduling.
633-- import Control.Concurrent
635-- main = 'runStream' . 'parallely' $ do
636-- n <- return 3 \<\> return 2 \<\> return 1
638-- threadDelay (n * 1000000)
639-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
642-- ThreadId 40: Delay 1
643-- ThreadId 39: Delay 2
644-- ThreadId 38: Delay 3
647-- Unlike 'CoparallelT' all iterations are guaranteed to run fairly
648-- concurrently, unconditionally.
650-- Note that round robin composition can only combine a finite number of
651-- streams as it needs to retain state for each unfinished stream.
653newtype ParallelT m a = ParallelT {getParallelT :: S.Stream m a}
654 deriving (Functor, MonadTrans)
656instance IsStream ParallelT where
657 toStream = getParallelT
658 fromStream = ParallelT
660------------------------------------------------------------------------------
662------------------------------------------------------------------------------
664-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'.
665-- Merges two streams concurrently choosing elements from both fairly.
666{-# INLINE parallel #-}
667parallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
668parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
670instance MonadParallel m => Semigroup (ParallelT m a) where
673------------------------------------------------------------------------------
675------------------------------------------------------------------------------
677instance MonadParallel m => Monoid (ParallelT m a) where
681------------------------------------------------------------------------------
683------------------------------------------------------------------------------
685instance MonadParallel m => Monad (ParallelT m) where
687 (ParallelT m) >>= f = ParallelT $ parbind par m (getParallelT . f)
688 where par = S.joinStreamVar2 (SVarStyle Conjunction FIFO)
690------------------------------------------------------------------------------
692------------------------------------------------------------------------------
694MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
695MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
697------------------------------------------------------------------------------
698-- Serially Zipping Streams
699------------------------------------------------------------------------------
701-- | Zip two streams serially using a pure zipping function.
702zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
703zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
705 go mx my = S.Stream $ \_ stp sng yld -> do
707 let single2 b = sng (f a b)
708 yield2 b rb = yld (f a b) (go ra rb)
709 in (S.runStream my) Nothing stp single2 yield2
710 let single1 a = merge a S.nil
711 yield1 a ra = merge a ra
712 (S.runStream mx) Nothing stp single1 yield1
714-- | The applicative instance of 'ZipStreamM' zips a number of streams serially
715-- i.e. it produces one element from each stream serially and then zips all
719-- main = (toList . 'zipStreamly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
720-- where s1 = fromFoldable [1, 2]
721-- s2 = fromFoldable [3, 4]
722-- s3 = fromFoldable [5, 6]
728-- The 'Semigroup' instance of this type works the same way as that of
731newtype ZipStreamM m a = ZipStreamM {getZipStreamM :: S.Stream m a}
732 deriving (Functor, Semigroup, Monoid)
734instance Monad m => Applicative (ZipStreamM m) where
735 pure = ZipStreamM . S.repeat
738instance IsStream ZipStreamM where
739 toStream = getZipStreamM
740 fromStream = ZipStreamM
742------------------------------------------------------------------------------
743-- Parallely Zipping Streams
744------------------------------------------------------------------------------
746-- | Zip two streams concurrently (i.e. both the elements being zipped are
747-- generated concurrently) using a pure zipping function.
748zipParallelWith :: (IsStream t, MonadParallel m)
749 => (a -> b -> c) -> t m a -> t m b -> t m c
750zipParallelWith f m1 m2 = fromStream $ S.Stream $ \_ stp sng yld -> do
753 (S.runStream (toStream (zipWith f ma mb))) Nothing stp sng yld
755{-# DEPRECATED zipAsyncWith "Please use zipParallelWith instead." #-}
756zipAsyncWith :: (IsStream t, MonadParallel m)
757 => (a -> b -> c) -> t m a -> t m b -> t m c
758zipAsyncWith = zipParallelWith
760-- | Like 'ZipStreamM' but zips in parallel, it generates all the elements to
761-- be zipped concurrently.
764-- main = (toList . 'zipParallely' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
765-- where s1 = fromFoldable [1, 2]
766-- s2 = fromFoldable [3, 4]
767-- s3 = fromFoldable [5, 6]
773-- The 'Semigroup' instance of this type works the same way as that of
776newtype ZipParallelM m a = ZipParallelM {getZipParallelM :: S.Stream m a}
777 deriving (Functor, Semigroup, Monoid)
779{-# DEPRECATED ZipAsync "Please use ZipParallelM instead." #-}
780type ZipAsync = ZipParallelM
782instance MonadParallel m => Applicative (ZipParallelM m) where
783 pure = ZipParallelM . S.repeat
784 (<*>) = zipParallelWith id
786instance IsStream ZipParallelM where
787 toStream = getZipParallelM
788 fromStream = ZipParallelM
790-------------------------------------------------------------------------------
791-- Type adapting combinators
792-------------------------------------------------------------------------------
794-- | Adapt any specific stream type to any other specific stream type.
795adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
796adapt = fromStream . toStream
798-- | Fix the type of a polymorphic stream as 'StreamT'.
799streamly :: IsStream t => StreamT m a -> t m a
802-- | Fix the type of a polymorphic stream as 'CostreamT'.
803costreamly :: IsStream t => CostreamT m a -> t m a
806-- | Same as 'costreamly'.
807{-# DEPRECATED interleaving "Please use costreamly instead." #-}
808interleaving :: IsStream t => CostreamT m a -> t m a
809interleaving = costreamly
811-- | Fix the type of a polymorphic stream as 'CoparallelT'.
812coparallely :: IsStream t => CoparallelT m a -> t m a
815-- | Same as 'coparallely'.
816{-# DEPRECATED asyncly "Please use coparallely instead." #-}
817asyncly :: IsStream t => CoparallelT m a -> t m a
820-- | Fix the type of a polymorphic stream as 'ParallelT'.
821parallely :: IsStream t => ParallelT m a -> t m a
824-- | Fix the type of a polymorphic stream as 'ZipStreamM'.
825zipStreamly :: IsStream t => ZipStreamM m a -> t m a
828-- | Same as 'zipStreamly'.
829{-# DEPRECATED zipping "Please use zipStreamly instead." #-}
830zipping :: IsStream t => ZipStreamM m a -> t m a
833-- | Fix the type of a polymorphic stream as 'ZipParallelM'.
834zipParallely :: IsStream t => ZipParallelM m a -> t m a
837-- | Same as 'zipParallely'.
838{-# DEPRECATED zippingAsync "Please use zipParallely instead." #-}
839zippingAsync :: IsStream t => ZipParallelM m a -> t m a
840zippingAsync = zipParallely
842-------------------------------------------------------------------------------
843-- Running Streams, convenience functions specialized to types
844-------------------------------------------------------------------------------
846-- | Same as @runStream@.
847{-# DEPRECATED runStreamT "Please use runStream instead." #-}
848runStreamT :: Monad m => StreamT m a -> m ()
849runStreamT = runStream
851-- | Same as @runStream . costreamly@.
852{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}
853runInterleavedT :: Monad m => InterleavedT m a -> m ()
854runInterleavedT = runStream . costreamly
856-- | Same as @runStream . coparallely@.
857{-# DEPRECATED runAsyncT "Please use 'runStream . coparallely' instead." #-}
858runAsyncT :: Monad m => CoparallelT m a -> m ()
859runAsyncT = runStream . coparallely
861-- | Same as @runStream . parallely@.
862{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}
863runParallelT :: Monad m => ParallelT m a -> m ()
864runParallelT = runStream . parallely
866-- | Same as @runStream . zipping@.
867{-# DEPRECATED runZipStream "Please use 'runStream . zipStreamly instead." #-}
868runZipStream :: Monad m => ZipStreamM m a -> m ()
869runZipStream = runStream . zipStreamly
871-- | Same as @runStream . zippingAsync@.
872{-# DEPRECATED runZipAsync "Please use 'runStream . zipParallely instead." #-}
873runZipAsync :: Monad m => ZipParallelM m a -> m ()
874runZipAsync = runStream . zipParallely
876------------------------------------------------------------------------------
878------------------------------------------------------------------------------
880-- | A serial IO stream of elements of type @a@. See 'StreamT' documentation
882type Stream a = StreamT IO a
884-- | An interleaving serial IO stream of elements of type @a@. See 'CostreamT'
885-- documentation for more details.
886type Costream a = CostreamT IO a
888-- | A demand driven left biased parallelly composing IO stream of elements of
889-- type @a@. See 'CoparallelT' documentation for more details.
890type Coparallel a = CoparallelT IO a
892-- | A round robin parallelly composing IO stream of elements of type @a@.
893-- See 'ParallelT' documentation for more details.
894type Parallel a = ParallelT IO a
896-- | An IO stream whose applicative instance zips streams serially.
897type ZipStream a = ZipStreamM IO a
899-- | An IO stream whose applicative instance zips streams parallely.
900type ZipParallel a = ZipParallelM IO a
902-- | Lift a value from IO monad to any of the stream types. This works for all
903-- stream types including those that are not monads e.g. 'ZipStream' and
905fromIO :: (IsStream t, MonadIO m) => IO a -> t m a
906fromIO = streamly . liftIO
908fromM :: (IsStream t, Monad m) => m a -> t m a
909fromM = streamly . lift
911------------------------------------------------------------------------------
913------------------------------------------------------------------------------
915-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'
916-- container of streams using the specified stream sum operation.
918-- @foldWith 'parallel' $ map return [1..3]@
919{-# INLINABLE foldWith #-}
920foldWith :: (IsStream t, Foldable f)
921 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
922foldWith f = foldr f nil
924-- | A variant of 'foldMap' that allows you to map a monadic streaming action
925-- on a 'Foldable' container and then fold it using the specified stream sum
928-- @foldMapWith 'parallel' return [1..3]@
929{-# INLINABLE foldMapWith #-}
930foldMapWith :: (IsStream t, Foldable f)
931 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
932foldMapWith f g = foldr (f . g) nil
934-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
935-- monadic streaming function is the last argument.
936{-# INLINABLE forEachWith #-}
937forEachWith :: (IsStream t, Foldable f)
938 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
939forEachWith 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