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
62 , StreamT -- deprecated
64 , InterleavedT -- deprecated
66 , AsyncT -- deprecated
69 , ZipStream -- deprecated
71 , ZipAsync -- deprecated
76 , interleaving -- deprecated
78 , asyncly -- deprecated
81 , zipping -- deprecated
83 , zippingAsync -- deprecated
87 , runStreamT -- deprecated
88 , runInterleavedT -- deprecated
89 , runAsyncT -- deprecated
90 , runParallelT -- deprecated
91 , runZipStream -- deprecated
92 , runZipAsync -- deprecated
97 , zipAsyncWith -- deprecated
106import Control.Monad (ap)
107import Control.Monad.Base (MonadBase (..), liftBaseDefault)
108import Control.Monad.Catch (MonadThrow, throwM)
109-- import Control.Monad.Error.Class (MonadError(..))
110import Control.Monad.IO.Class (MonadIO(..))
111import Control.Monad.Reader.Class (MonadReader(..))
112import Control.Monad.State.Class (MonadState(..))
113import Control.Monad.Trans.Class (MonadTrans (lift))
114import Data.Semigroup (Semigroup(..))
115import Prelude hiding (zipWith)
116import Streamly.Core ( MonadParallel, Stream(Stream)
118 , SVar, SVarStyle(..)
119 , SVarTag(..), SVarSched(..))
120import qualified Streamly.Core as S
122------------------------------------------------------------------------------
123-- Types that can behave as a Stream
124------------------------------------------------------------------------------
126-- | Class of types that can represent a stream of elements of some type 'a' in
128class IsStream t where
129 toStream :: t m a -> Stream m a
130 fromStream :: Stream m a -> t m a
132-- | Same as 'IsStream'.
133{-# Deprecated Streaming "Please use IsStream instead." #-}
134type Streaming = IsStream
136------------------------------------------------------------------------------
137-- Constructing a stream
138------------------------------------------------------------------------------
140-- | Represesnts an empty stream just like @[]@ represents an empty list.
141nil :: IsStream t => t m a
142nil = fromStream S.nil
146-- | Constructs a stream by adding a pure value at the head of an existing
147-- stream, just like ':' constructs lists. For example:
150-- > let stream = 1 \`cons` 2 \`cons` 3 \`cons` nil
151-- > (toList . serially) stream
154cons :: IsStream t => a -> t m a -> t m a
155cons a r = fromStream $ S.cons a (toStream r)
159-- | Operator equivalent of 'cons' so that you can construct a stream of pure
160-- values more succinctly like this:
163-- > let stream = 1 .: 2 .: 3 .: nil
164-- > (toList . serially) stream
168-- '.:' constructs a stream just like ':' constructs a list.
170-- Also note that another equivalent way of building streams from pure values
174-- > let stream = pure 1 <> pure 2 <> pure 3
175-- > (toList . serially) stream
179-- In the first method we construct a stream by adding one element at a time.
180-- In the second method we first construct singleton streams using 'pure' and
181-- then compose all those streams together using the 'Semigroup' style
182-- composition of streams. The former method is a bit more efficient than the
185(.:) :: IsStream t => a -> t m a -> t m a
188-- | Build a stream from its church encoding. The function passed maps
189-- directly to the underlying representation of the stream type. The second
190-- parameter to the function is the "yield" function yielding a value and the
191-- remaining stream if any otherwise 'Nothing'. The third parameter is to
192-- represent an "empty" stream.
193streamBuild :: IsStream t
194 => (forall r. Maybe (SVar m a)
195 -> (a -> t m a -> m r)
200streamBuild k = fromStream $ Stream $ \svr stp sng yld ->
201 let yield a r = yld a (toStream r)
202 in k svr yield sng stp
204-- | Build a singleton stream from a callback function.
205fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
206fromCallback k = fromStream $ Stream $ \_ _ sng _ -> k sng
208-- | Read an SVar to get a stream.
209fromSVar :: (MonadParallel m, IsStream t) => SVar m a -> t m a
210fromSVar sv = fromStream $ S.fromStreamVar sv
212------------------------------------------------------------------------------
213-- Destroying a stream
214------------------------------------------------------------------------------
216-- | Fold a stream using its church encoding. The second argument is the "step"
217-- function consuming an element and the remaining stream, if any. The third
218-- argument is for consuming an "empty" stream that yields nothing.
222 -> (a -> t m a -> m r)
227streamFold svr step single blank m =
228 let yield a x = step a (fromStream x)
229 in (S.runStream (toStream m)) svr blank single yield
231-- | Run a streaming composition, discard the results. By default it interprets
232-- the stream as 'SerialT', to run other types of streams use the type adapting
233-- combinators for example @runStream . 'parallely'@.
234runStream :: Monad m => SerialT m a -> m ()
235runStream m = go (toStream m)
241 in (S.runStream m1) Nothing stop single yield
243-- | Same as 'runStream'
244{-# Deprecated runStreaming "Please use runStream instead." #-}
245runStreaming :: (Monad m, IsStream t) => t m a -> m ()
246runStreaming = runStream . adapt
248-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
249-- be read back from the SVar using 'fromSVar'.
250toSVar :: (IsStream t, MonadParallel m) => SVar m a -> t m a -> m ()
251toSVar sv m = S.toStreamVar sv (toStream m)
253------------------------------------------------------------------------------
255------------------------------------------------------------------------------
257-- XXX Get rid of this?
258-- | Make a stream asynchronous, triggers the computation and returns a stream
259-- in the underlying monad representing the output generated by the original
260-- computation. The returned action is exhaustible and must be drained once. If
261-- not drained fully we may have a thread blocked forever and once exhausted it
262-- will always return 'empty'.
264async :: (IsStream t, MonadParallel m) => t m a -> m (t m a)
266 sv <- S.newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)
269------------------------------------------------------------------------------
270-- CPP macros for common instances
271------------------------------------------------------------------------------
273-- XXX use template haskell instead and include Monoid and IsStream instances
276#define MONADPARALLEL , MonadParallel m
278#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
279instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
280 pure = STREAM . singleton; \
283#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
284instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\
285 liftBase = liftBaseDefault }; \
287instance (MonadIO m CONSTRAINT) => MonadIO (STREAM m) where { \
288 liftIO = lift . liftIO }; \
290instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \
291 throwM = lift . throwM }; \
294instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \
295 throwError = lift . throwError; \
297 fromStream $ S.withCatchError (toStream m) (\e -> toStream $ h e) }; \
300instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \
302 local f m = fromStream $ S.withLocal f (toStream m) }; \
304instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
306 put x = lift (put x); \
307 state k = lift (state k) }
309------------------------------------------------------------------------------
311------------------------------------------------------------------------------
313-- | The 'Semigroup' instance of 'SerialT' appends two streams sequentially,
314-- yielding all elements from the first stream, and then all elements from the
318-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
324-- The 'Monad' instance runs the /monadic continuation/ for each
325-- element of the stream, serially.
328-- main = 'runStream' . 'serially' $ do
329-- x <- return 1 \<\> return 2
337-- 'SerialT' nests streams serially in a depth first manner.
340-- main = 'runStream' . 'serially' $ do
341-- x <- return 1 \<\> return 2
342-- y <- return 3 \<\> return 4
343-- liftIO $ print (x, y)
352-- This behavior of 'SerialT' is exactly like a list transformer. We call the
353-- monadic code being run for each element of the stream a monadic
354-- continuation. In imperative paradigm we can think of this composition as
355-- nested @for@ loops and the monadic continuation is the body of the loop. The
356-- loop iterates for all elements of the stream.
358-- The 'serially' combinator can be omitted as the default stream type is
360-- Note that serial composition can be used to combine an infinite number of
361-- streams as it explores only one stream at a time.
363newtype SerialT m a = SerialT {getSerialT :: Stream m a}
364 deriving (Semigroup, Monoid, Functor, MonadTrans)
366instance IsStream SerialT where
367 toStream = getSerialT
371{-# Deprecated StreamT "Please use SerialT instead." #-}
372type StreamT = SerialT
374------------------------------------------------------------------------------
376------------------------------------------------------------------------------
378-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
379-- Appends two streams sequentially, yielding all elements from the first
380-- stream, and then all elements from the second stream.
382serial :: IsStream t => t m a -> t m a -> t m a
383serial m1 m2 = fromStream $ S.serial (toStream m1) (toStream m2)
385------------------------------------------------------------------------------
387------------------------------------------------------------------------------
389instance Monad m => Monad (SerialT m) where
391 (SerialT (Stream m)) >>= f = SerialT $ Stream $ \_ stp sng yld ->
392 let run x = (S.runStream x) Nothing stp sng yld
393 single a = run $ toStream (f a)
394 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
395 in m Nothing stp single yield
397------------------------------------------------------------------------------
399------------------------------------------------------------------------------
401MONAD_APPLICATIVE_INSTANCE(SerialT,)
402MONAD_COMMON_INSTANCES(SerialT,)
404------------------------------------------------------------------------------
406------------------------------------------------------------------------------
408-- | The 'Semigroup' instance of 'CoserialT' interleaves two streams,
409-- yielding one element from each stream alternately.
412-- main = ('toList' . 'interleaving' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
418-- Similarly, the 'Monad' instance fairly interleaves the iterations of the
419-- inner and the outer loop, nesting loops in a breadth first manner.
423-- main = 'runStream' . 'interleaving' $ do
424-- x <- return 1 \<\> return 2
425-- y <- return 3 \<\> return 4
426-- liftIO $ print (x, y)
435-- Note that interleaving composition can only combine a finite number of
436-- streams as it needs to retain state for each unfinished stream.
438newtype CoserialT m a = CoserialT {getCoserialT :: Stream m a}
439 deriving (Functor, MonadTrans)
441{-# DEPRECATED InterleavedT "Please use 'CoserialT' instead." #-}
442type InterleavedT = CoserialT
444instance IsStream CoserialT where
445 toStream = getCoserialT
446 fromStream = CoserialT
448------------------------------------------------------------------------------
450------------------------------------------------------------------------------
452-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CoserialT'.
453-- Interleaves two streams, yielding one element from each stream alternately.
454{-# INLINE coserial #-}
455coserial :: IsStream t => t m a -> t m a -> t m a
456coserial m1 m2 = fromStream $ S.coserial (toStream m1) (toStream m2)
458instance Semigroup (CoserialT m a) where
463-- | Same as 'coserial'.
464{-# DEPRECATED (<=>) "Please use 'coserial' instead." #-}
466(<=>) :: IsStream t => t m a -> t m a -> t m a
469------------------------------------------------------------------------------
471------------------------------------------------------------------------------
473instance Monoid (CoserialT m a) where
477------------------------------------------------------------------------------
479------------------------------------------------------------------------------
481instance Monad m => Monad (CoserialT m) where
483 (CoserialT (Stream m)) >>= f = CoserialT $ Stream $ \_ stp sng yld ->
484 let run x = (S.runStream x) Nothing stp sng yld
485 single a = run $ toStream (f a)
486 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
487 in m Nothing stp single yield
489------------------------------------------------------------------------------
491------------------------------------------------------------------------------
493MONAD_APPLICATIVE_INSTANCE(CoserialT,)
494MONAD_COMMON_INSTANCES(CoserialT,)
496------------------------------------------------------------------------------
498------------------------------------------------------------------------------
500-- | Adaptive parallel, in a left to right 'Semigroup' composition it tries to
501-- yield elements from the left stream as long as it can, but it can run the
502-- right stream in parallel if it needs to based on demand. The right stream
503-- can be run if the left stream blocks on IO or cannot produce elements fast
504-- enough for the consumer.
507-- main = ('toList' . 'coparallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
513-- Similarly, the monad instance of 'ParallelT' /may/ run each iteration
514-- concurrently based on demand. More concurrent iterations are started only
515-- if the previous iterations are not able to produce enough output for the
520-- import Control.Concurrent
522-- main = 'runStream' . 'coparallely' $ do
523-- n <- return 3 \<\> return 2 \<\> return 1
525-- threadDelay (n * 1000000)
526-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
529-- ThreadId 40: Delay 1
530-- ThreadId 39: Delay 2
531-- ThreadId 38: Delay 3
534-- All iterations may run in the same thread if they do not block.
536-- Note that this composition can be used to combine infinite number of streams
537-- as it explores only a bounded number of streams at a time.
539newtype CoparallelT m a = CoparallelT {getCoparallelT :: Stream m a}
540 deriving (Functor, MonadTrans)
542{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}
543type AsyncT = CoparallelT
545instance IsStream CoparallelT where
546 toStream = getCoparallelT
547 fromStream = CoparallelT
549------------------------------------------------------------------------------
551------------------------------------------------------------------------------
553-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CoparallelT', but
554-- polymorphic. Merges two streams possibly concurrently, preferring the
555-- elements from the left one when available.
556{-# INLINE coparallel #-}
557coparallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
558coparallel m1 m2 = fromStream $ S.coparallel (toStream m1) (toStream m2)
560instance MonadParallel m => Semigroup (CoparallelT m a) where
563-- | Same as 'coparallel'.
564{-# DEPRECATED (<|) "Please use 'coparallel' instead." #-}
566(<|) :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
569------------------------------------------------------------------------------
571------------------------------------------------------------------------------
573instance MonadParallel m => Monoid (CoparallelT m a) where
577------------------------------------------------------------------------------
579------------------------------------------------------------------------------
581{-# INLINE parbind #-}
583 :: (forall c. Stream m c -> Stream m c -> Stream m c)
587parbind par m f = go m
590 Stream $ \ctx stp sng yld ->
591 let run x = (S.runStream x) ctx stp sng yld
593 yield a r = run $ f a `par` go r
594 in g Nothing stp single yield
596instance MonadParallel m => Monad (CoparallelT m) where
598 (CoparallelT m) >>= f = CoparallelT $ parbind par m (getCoparallelT . f)
599 where par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)
601------------------------------------------------------------------------------
603------------------------------------------------------------------------------
605MONAD_APPLICATIVE_INSTANCE(CoparallelT,MONADPARALLEL)
606MONAD_COMMON_INSTANCES(CoparallelT, MONADPARALLEL)
608------------------------------------------------------------------------------
610------------------------------------------------------------------------------
612-- | Round robin concurrent composition.
614-- The Semigroup instance of 'ParallelT' concurrently /merges/ streams in a
615-- round robin fashion, yielding elements from both streams alternately.
618-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
624-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations fairly
625-- concurrently using a round robin scheduling.
629-- import Control.Concurrent
631-- main = 'runStream' . 'parallely' $ do
632-- n <- return 3 \<\> return 2 \<\> return 1
634-- threadDelay (n * 1000000)
635-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
638-- ThreadId 40: Delay 1
639-- ThreadId 39: Delay 2
640-- ThreadId 38: Delay 3
643-- Unlike 'CoparallelT' all iterations are guaranteed to run fairly
644-- concurrently, unconditionally.
646-- Note that round robin composition can only combine a finite number of
647-- streams as it needs to retain state for each unfinished stream.
649newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
650 deriving (Functor, MonadTrans)
652instance IsStream ParallelT where
653 toStream = getParallelT
654 fromStream = ParallelT
656------------------------------------------------------------------------------
658------------------------------------------------------------------------------
660-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'.
661-- Merges two streams concurrently choosing elements from both fairly.
662{-# INLINE parallel #-}
663parallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
664parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
666instance MonadParallel m => Semigroup (ParallelT m a) where
669------------------------------------------------------------------------------
671------------------------------------------------------------------------------
673instance MonadParallel m => Monoid (ParallelT m a) where
677------------------------------------------------------------------------------
679------------------------------------------------------------------------------
681instance MonadParallel m => Monad (ParallelT m) where
683 (ParallelT m) >>= f = ParallelT $ parbind par m (getParallelT . f)
684 where par = S.joinStreamVar2 (SVarStyle Conjunction FIFO)
686------------------------------------------------------------------------------
688------------------------------------------------------------------------------
690MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
691MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
693------------------------------------------------------------------------------
694-- Serially Zipping Streams
695------------------------------------------------------------------------------
697-- | Zip two streams serially using a pure zipping function.
698zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
699zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
701 go mx my = Stream $ \_ stp sng yld -> do
703 let single2 b = sng (f a b)
704 yield2 b rb = yld (f a b) (go ra rb)
705 in (S.runStream my) Nothing stp single2 yield2
706 let single1 a = merge a S.nil
707 yield1 a ra = merge a ra
708 (S.runStream mx) Nothing stp single1 yield1
710-- | The applicative instance of 'ZipSerial' zips a number of streams serially
711-- i.e. it produces one element from each stream serially and then zips all
715-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
716-- where s1 = fromFoldable [1, 2]
717-- s2 = fromFoldable [3, 4]
718-- s3 = fromFoldable [5, 6]
724-- The 'Semigroup' instance of this type works the same way as that of
727newtype ZipSerial m a = ZipSerial {getZipSerial :: Stream m a}
728 deriving (Functor, Semigroup, Monoid)
730{-# Deprecated ZipStream "Please use ZipSerial instead." #-}
731-- | Same as ZipSerial.
732type ZipStream = ZipSerial
734instance Monad m => Applicative (ZipSerial m) where
735 pure = ZipSerial . S.repeat
738instance IsStream ZipSerial where
739 toStream = getZipSerial
740 fromStream = ZipSerial
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 $ 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 'ZipSerial' 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 ZipParallel m a = ZipParallel {getZipParallel :: Stream m a}
777 deriving (Functor, Semigroup, Monoid)
779{-# DEPRECATED ZipAsync "Please use ZipParallel instead." #-}
780type ZipAsync = ZipParallel
782instance MonadParallel m => Applicative (ZipParallel m) where
783 pure = ZipParallel . S.repeat
784 (<*>) = zipParallelWith id
786instance IsStream ZipParallel where
787 toStream = getZipParallel
788 fromStream = ZipParallel
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 'SerialT'.
799serially :: IsStream t => SerialT m a -> t m a
802-- | Fix the type of a polymorphic stream as 'CoserialT'.
803coserially :: IsStream t => CoserialT m a -> t m a
806-- | Same as 'coserially'.
807{-# DEPRECATED interleaving "Please use coserially instead." #-}
808interleaving :: IsStream t => CoserialT m a -> t m a
809interleaving = coserially
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 'ZipSerial'.
825zipSerially :: IsStream t => ZipSerial m a -> t m a
828-- | Same as 'zipParallely'.
829{-# DEPRECATED zipping "Please use zipSerially instead." #-}
830zipping :: IsStream t => ZipSerial m a -> t m a
833-- | Fix the type of a polymorphic stream as 'ZipParallel'.
834zipParallely :: IsStream t => ZipParallel m a -> t m a
837-- | Same as 'zipParallely'.
838{-# DEPRECATED zippingAsync "Please use zipParallely instead." #-}
839zippingAsync :: IsStream t => ZipParallel 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 => SerialT m a -> m ()
849runStreamT = runStream
851-- | Same as @runStream . coserially@.
852{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}
853runInterleavedT :: Monad m => InterleavedT m a -> m ()
854runInterleavedT = runStream . coserially
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 . zipSerially instead." #-}
868runZipStream :: Monad m => ZipSerial m a -> m ()
869runZipStream = runStream . zipSerially
871-- | Same as @runStream . zippingAsync@.
872{-# DEPRECATED runZipAsync "Please use 'runStream . zipParallely instead." #-}
873runZipAsync :: Monad m => ZipParallel m a -> m ()
874runZipAsync = runStream . zipParallely
876------------------------------------------------------------------------------
878------------------------------------------------------------------------------
880-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'
881-- container of streams using the specified stream sum operation.
883-- @foldWith 'parallel' $ map return [1..3]@
884{-# INLINABLE foldWith #-}
885foldWith :: (IsStream t, Foldable f)
886 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
887foldWith f = foldr f nil
889-- | A variant of 'foldMap' that allows you to map a monadic streaming action
890-- on a 'Foldable' container and then fold it using the specified stream sum
893-- @foldMapWith 'parallel' return [1..3]@
894{-# INLINABLE foldMapWith #-}
895foldMapWith :: (IsStream t, Foldable f)
896 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
897foldMapWith f g = foldr (f . g) nil
899-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
900-- monadic streaming function is the last argument.
901{-# INLINABLE forEachWith #-}
902forEachWith :: (IsStream t, Foldable f)
903 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
904forEachWith 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