1{-# LANGUAGE FlexibleContexts #-}
2{-# LANGUAGE FlexibleInstances #-}
3{-# LANGUAGE GeneralizedNewtypeDeriving#-}
4{-# LANGUAGE MultiParamTypeClasses #-}
5{-# LANGUAGE RankNTypes #-}
6{-# LANGUAGE StandaloneDeriving #-}
7{-# LANGUAGE UndecidableInstances #-} -- XXX
10-- Module : Streamly.Streams
11-- Copyright : (c) 2017 Harendra Kumar
14-- Maintainer : harendra.kumar@gmail.com
15-- Stability : experimental
19module Streamly.Streams
76 -- * Sum Style Composition
88import Control.Applicative (Alternative (..), liftA2)
89import Control.Monad (MonadPlus(..), ap)
90import Control.Monad.Base (MonadBase (..))
91import Control.Monad.Catch (MonadThrow)
92import Control.Monad.Error.Class (MonadError(..))
93import Control.Monad.IO.Class (MonadIO(..))
94import Control.Monad.Reader.Class (MonadReader(..))
95import Control.Monad.State.Class (MonadState(..))
96import Control.Monad.Trans.Class (MonadTrans)
97import Data.Semigroup (Semigroup(..))
98import Prelude hiding (zipWith)
101------------------------------------------------------------------------------
102-- Types that can behave as a Stream
103------------------------------------------------------------------------------
105-- | Class of types that can represent a stream of elements of some type 'a' in
107class Streaming t where
108 toStream :: t m a -> Stream m a
109 fromStream :: Stream m a -> t m a
111------------------------------------------------------------------------------
112-- Constructing a stream
113------------------------------------------------------------------------------
115-- | Represesnts an empty stream just like @[]@ represents and empty list.
116nil :: Streaming t => t m a
121-- | Constructs a stream by adding a pure value at the head of an existing
122-- stream, just like ':' constructs lists. For example:
125-- > let stream = 1 \`cons` 2 \`cons` 3 \`cons` nil
126-- > (toList . serially) stream
129cons :: (Streaming t) => a -> t m a -> t m a
130cons a r = fromStream $ scons a (Just (toStream r))
134-- | Operator equivalent of 'cons' so that you can construct a stream of pure
135-- values more succinctly like this:
138-- > let stream = 1 .: 2 .: 3 .: nil
139-- > (toList . serially) stream
143-- '.:' constructs a stream just like ':' constructs a list.
145-- Also note that another equivalent way of building streams from pure values
149-- > let stream = pure 1 <> pure 2 <> pure 3
150-- > (toList . serially) stream
154-- In the first method we construct a stream by adding one element at a time.
155-- In the second method we first construct singleton streams using 'pure' and
156-- then compose all those streams together using the 'Semigroup' style
157-- composition of streams. The former method is a bit more efficient than the
160(.:) :: (Streaming t) => a -> t m a -> t m a
163-- | Build a stream from its church encoding. The function passed maps
164-- directly to the underlying representation of the stream type. The second
165-- parameter to the function is the "yield" function yielding a value and the
166-- remaining stream if any otherwise 'Nothing'. The third parameter is to
167-- represent an "empty" stream.
168streamBuild :: Streaming t
169 => (forall r. Maybe (SVar m a)
170 -> (a -> Maybe (t m a) -> m r)
174streamBuild k = fromStream $ Stream $ \sv stp yld ->
175 let yield a Nothing = yld a Nothing
176 yield a (Just r) = yld a (Just (toStream r))
179-- | Build a singleton stream from a callback function.
180fromCallback :: (Streaming t) => (forall r. (a -> m r) -> m r) -> t m a
181fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)
183-- | Read an SVar to get a stream.
184fromSVar :: (MonadAsync m, Streaming t) => SVar m a -> t m a
185fromSVar sv = fromStream $ fromStreamVar sv
187------------------------------------------------------------------------------
188-- Destroying a stream
189------------------------------------------------------------------------------
191-- | Fold a stream using its church encoding. The second argument is the "step"
192-- function consuming an element and the remaining stream, if any. The third
193-- argument is for consuming an "empty" stream that yields nothing.
194streamFold :: Streaming t
195 => Maybe (SVar m a) -> (a -> Maybe (t m a) -> m r) -> m r -> t m a -> m r
196streamFold sv step blank m =
197 let yield a Nothing = step a Nothing
198 yield a (Just x) = step a (Just (fromStream x))
199 in (runStream (toStream m)) sv blank yield
201-- | Run a streaming composition, discard the results.
202runStreaming :: (Monad m, Streaming t) => t m a -> m ()
203runStreaming m = go (toStream m)
207 yield _ Nothing = stop
208 yield _ (Just x) = go x
209 in (runStream m1) Nothing stop yield
211-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
212-- be read back from the SVar using 'fromSVar'.
213toSVar :: (Streaming t, MonadAsync m) => SVar m a -> t m a -> m ()
214toSVar sv m = toStreamVar sv (toStream m)
216------------------------------------------------------------------------------
218------------------------------------------------------------------------------
220-- XXX Get rid of this?
221-- | Make a stream asynchronous, triggers the computation and returns a stream
222-- in the underlying monad representing the output generated by the original
223-- computation. The returned action is exhaustible and must be drained once. If
224-- not drained fully we may have a thread blocked forever and once exhausted it
225-- will always return 'empty'.
227async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
229 sv <- newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)
232------------------------------------------------------------------------------
234------------------------------------------------------------------------------
236-- | The 'Monad' instance of 'StreamT' runs the /monadic continuation/ for each
237-- element of the stream, serially.
240-- main = 'runStreamT' $ do
241-- x <- return 1 \<\> return 2
249-- 'StreamT' nests streams serially in a depth first manner.
252-- main = 'runStreamT' $ do
253-- x <- return 1 \<\> return 2
254-- y <- return 3 \<\> return 4
255-- liftIO $ print (x, y)
264-- This behavior is exactly like a list transformer. We call the monadic code
265-- being run for each element of the stream a monadic continuation. In
266-- imperative paradigm we can think of this composition as nested @for@ loops
267-- and the monadic continuation is the body of the loop. The loop iterates for
268-- all elements of the stream.
270newtype StreamT m a = StreamT {getStreamT :: Stream m a}
271 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
273deriving instance MonadAsync m => Alternative (StreamT m)
274deriving instance MonadAsync m => MonadPlus (StreamT m)
275deriving instance (MonadBase b m, Monad m) => MonadBase b (StreamT m)
276deriving instance MonadError e m => MonadError e (StreamT m)
277deriving instance MonadReader r m => MonadReader r (StreamT m)
278deriving instance MonadState s m => MonadState s (StreamT m)
280instance Streaming StreamT where
281 toStream = getStreamT
284-- XXX The Functor/Applicative/Num instances for all the types are exactly the
285-- same, how can we reduce this boilerplate (use TH)? We cannot derive them
286-- from a single base type because they depend on the Monad instance which is
287-- different for each type.
289------------------------------------------------------------------------------
291------------------------------------------------------------------------------
293instance Monad m => Monad (StreamT m) where
295 (StreamT (Stream m)) >>= f = StreamT $ Stream $ \_ stp yld ->
296 let run x = (runStream x) Nothing stp yld
297 yield a Nothing = run $ getStreamT (f a)
298 yield a (Just r) = run $ getStreamT (f a)
299 <> getStreamT (StreamT r >>= f)
300 in m Nothing stp yield
302------------------------------------------------------------------------------
304------------------------------------------------------------------------------
306instance Monad m => Applicative (StreamT m) where
307 pure a = StreamT $ scons a Nothing
310------------------------------------------------------------------------------
312------------------------------------------------------------------------------
314instance Monad m => Functor (StreamT m) where
315 fmap f (StreamT (Stream m)) = StreamT $ Stream $ \_ stp yld ->
316 let yield a Nothing = yld (f a) Nothing
317 yield a (Just r) = yld (f a)
318 (Just (getStreamT (fmap f (StreamT r))))
319 in m Nothing stp yield
321------------------------------------------------------------------------------
323------------------------------------------------------------------------------
325instance (Monad m, Num a) => Num (StreamT m a) where
326 fromInteger n = pure (fromInteger n)
336instance (Monad m, Fractional a) => Fractional (StreamT m a) where
337 fromRational n = pure (fromRational n)
343instance (Monad m, Floating a) => Floating (StreamT m a) where
363 logBase = liftA2 logBase
365------------------------------------------------------------------------------
367------------------------------------------------------------------------------
369-- | Like 'StreamT' but different in nesting behavior. It fairly interleaves
370-- the iterations of the inner and the outer loop, nesting loops in a breadth
375-- main = 'runInterleavedT' $ do
376-- x <- return 1 \<\> return 2
377-- y <- return 3 \<\> return 4
378-- liftIO $ print (x, y)
387newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
388 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
390deriving instance MonadAsync m => Alternative (InterleavedT m)
391deriving instance MonadAsync m => MonadPlus (InterleavedT m)
392deriving instance (MonadBase b m, Monad m) => MonadBase b (InterleavedT m)
393deriving instance MonadError e m => MonadError e (InterleavedT m)
394deriving instance MonadReader r m => MonadReader r (InterleavedT m)
395deriving instance MonadState s m => MonadState s (InterleavedT m)
397instance Streaming InterleavedT where
398 toStream = getInterleavedT
399 fromStream = InterleavedT
401instance Monad m => Monad (InterleavedT m) where
403 (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->
404 let run x = (runStream x) Nothing stp yld
405 yield a Nothing = run $ getInterleavedT (f a)
406 yield a (Just r) = run $ getInterleavedT (f a)
408 getInterleavedT (InterleavedT r >>= f)
409 in m Nothing stp yield
411------------------------------------------------------------------------------
413------------------------------------------------------------------------------
415instance Monad m => Applicative (InterleavedT m) where
416 pure a = InterleavedT $ scons a Nothing
419------------------------------------------------------------------------------
421------------------------------------------------------------------------------
423instance Monad m => Functor (InterleavedT m) where
424 fmap f (InterleavedT (Stream m)) = InterleavedT $ Stream $ \_ stp yld ->
425 let yield a Nothing = yld (f a) Nothing
427 yld (f a) (Just (getInterleavedT (fmap f (InterleavedT r))))
428 in m Nothing stp yield
430------------------------------------------------------------------------------
432------------------------------------------------------------------------------
434instance (Monad m, Num a) => Num (InterleavedT m a) where
435 fromInteger n = pure (fromInteger n)
445instance (Monad m, Fractional a) => Fractional (InterleavedT m a) where
446 fromRational n = pure (fromRational n)
452instance (Monad m, Floating a) => Floating (InterleavedT m a) where
472 logBase = liftA2 logBase
474------------------------------------------------------------------------------
476------------------------------------------------------------------------------
478-- | Like 'StreamT' but /may/ run each iteration concurrently using demand
479-- driven concurrency. More concurrent iterations are started only if the
480-- previous iterations are not able to produce enough output for the consumer.
484-- import Control.Concurrent
486-- main = 'runAsyncT' $ do
487-- n <- return 3 \<\> return 2 \<\> return 1
489-- threadDelay (n * 1000000)
490-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
493-- ThreadId 40: Delay 1
494-- ThreadId 39: Delay 2
495-- ThreadId 38: Delay 3
498-- All iterations may run in the same thread if they do not block.
499newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
500 deriving (Semigroup, Monoid, MonadTrans)
502deriving instance MonadAsync m => Alternative (AsyncT m)
503deriving instance MonadAsync m => MonadPlus (AsyncT m)
504deriving instance MonadAsync m => MonadIO (AsyncT m)
505deriving instance MonadAsync m => MonadThrow (AsyncT m)
506deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m)
507deriving instance (MonadError e m, MonadAsync m) => MonadError e (AsyncT m)
508deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)
509deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)
511instance Streaming AsyncT where
515{-# INLINE parbind #-}
517 :: (forall c. Stream m c -> Stream m c -> Stream m c)
521parbind par m f = go m
524 Stream $ \ctx stp yld ->
525 let run x = (runStream x) ctx stp yld
526 yield a Nothing = run $ f a
527 yield a (Just r) = run $ f a `par` go r
528 in g Nothing stp yield
530instance MonadAsync m => Monad (AsyncT m) where
532 (AsyncT m) >>= f = AsyncT $ parbind par m g
533 where g x = getAsyncT (f x)
534 par = joinStreamVar2 (SVarStyle Conjunction LIFO)
536------------------------------------------------------------------------------
538------------------------------------------------------------------------------
540instance MonadAsync m => Applicative (AsyncT m) where
541 pure a = AsyncT $ scons a Nothing
544------------------------------------------------------------------------------
546------------------------------------------------------------------------------
548instance Monad m => Functor (AsyncT m) where
549 fmap f (AsyncT (Stream m)) = AsyncT $ Stream $ \_ stp yld ->
550 let yield a Nothing = yld (f a) Nothing
551 yield a (Just r) = yld (f a) (Just (getAsyncT (fmap f (AsyncT r))))
552 in m Nothing stp yield
554------------------------------------------------------------------------------
556------------------------------------------------------------------------------
558instance (MonadAsync m, Num a) => Num (AsyncT m a) where
559 fromInteger n = pure (fromInteger n)
569instance (MonadAsync m, Fractional a) => Fractional (AsyncT m a) where
570 fromRational n = pure (fromRational n)
576instance (MonadAsync m, Floating a) => Floating (AsyncT m a) where
596 logBase = liftA2 logBase
598------------------------------------------------------------------------------
600------------------------------------------------------------------------------
602-- | Like 'StreamT' but runs /all/ iterations fairly concurrently using a round
607-- import Control.Concurrent
609-- main = 'runParallelT' $ do
610-- n <- return 3 \<\> return 2 \<\> return 1
612-- threadDelay (n * 1000000)
613-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
616-- ThreadId 40: Delay 1
617-- ThreadId 39: Delay 2
618-- ThreadId 38: Delay 3
621-- Unlike 'AsyncT' all iterations are guaranteed to run fairly concurrently,
623newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
624 deriving (Semigroup, Monoid, MonadTrans)
626deriving instance MonadAsync m => Alternative (ParallelT m)
627deriving instance MonadAsync m => MonadPlus (ParallelT m)
628deriving instance MonadAsync m => MonadIO (ParallelT m)
629deriving instance MonadAsync m => MonadThrow (ParallelT m)
630deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)
631deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)
632deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)
633deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)
635instance Streaming ParallelT where
636 toStream = getParallelT
637 fromStream = ParallelT
639instance MonadAsync m => Monad (ParallelT m) where
641 (ParallelT m) >>= f = ParallelT $ parbind par m g
642 where g x = getParallelT (f x)
643 par = joinStreamVar2 (SVarStyle Conjunction FIFO)
645------------------------------------------------------------------------------
647------------------------------------------------------------------------------
649instance MonadAsync m => Applicative (ParallelT m) where
650 pure a = ParallelT $ scons a Nothing
653------------------------------------------------------------------------------
655------------------------------------------------------------------------------
657instance Monad m => Functor (ParallelT m) where
658 fmap f (ParallelT (Stream m)) = ParallelT $ Stream $ \_ stp yld ->
659 let yield a Nothing = yld (f a) Nothing
660 yield a (Just r) = yld (f a)
661 (Just (getParallelT (fmap f (ParallelT r))))
662 in m Nothing stp yield
664------------------------------------------------------------------------------
666------------------------------------------------------------------------------
668instance (MonadAsync m, Num a) => Num (ParallelT m a) where
669 fromInteger n = pure (fromInteger n)
679instance (MonadAsync m, Fractional a) => Fractional (ParallelT m a) where
680 fromRational n = pure (fromRational n)
686instance (MonadAsync m, Floating a) => Floating (ParallelT m a) where
706 logBase = liftA2 logBase
708------------------------------------------------------------------------------
709-- Serially Zipping Streams
710------------------------------------------------------------------------------
712-- | Zip two streams serially using a pure zipping function.
713zipWith :: Streaming t => (a -> b -> c) -> t m a -> t m b -> t m c
714zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
716 go mx my = Stream $ \_ stp yld -> do
718 let yield2 b Nothing = yld (f a b) Nothing
719 yield2 b (Just rb) = yld (f a b) (Just (go ra rb))
720 in (runStream my) Nothing stp yield2
721 let yield1 a Nothing = merge a snil
722 yield1 a (Just ra) = merge a ra
723 (runStream mx) Nothing stp yield1
725-- | 'ZipStream' zips serially i.e. it produces one element from each stream
726-- serially and then zips the two elements. Note, for convenience we have used
727-- the 'zipping' combinator in the following example instead of using a type
731-- main = (toList . 'zipping' $ (,) \<$\> s1 \<*\> s2) >>= print
732-- where s1 = pure 1 <> pure 2
733-- s2 = pure 3 <> pure 4
739-- This applicative operation can be seen as the zipping equivalent of
740-- interleaving with '<=>'.
741newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}
742 deriving (Semigroup, Monoid)
744deriving instance MonadAsync m => Alternative (ZipStream m)
746instance Monad m => Functor (ZipStream m) where
747 fmap f (ZipStream (Stream m)) = ZipStream $ Stream $ \_ stp yld ->
748 let yield a Nothing = yld (f a) Nothing
749 yield a (Just r) = yld (f a)
750 (Just (getZipStream (fmap f (ZipStream r))))
751 in m Nothing stp yield
753instance Monad m => Applicative (ZipStream m) where
754 pure a = ZipStream $ scons a Nothing
757instance Streaming ZipStream where
758 toStream = getZipStream
759 fromStream = ZipStream
761instance (Monad m, Num a) => Num (ZipStream m a) where
762 fromInteger n = pure (fromInteger n)
772instance (Monad m, Fractional a) => Fractional (ZipStream m a) where
773 fromRational n = pure (fromRational n)
779instance (Monad m, Floating a) => Floating (ZipStream m a) where
799 logBase = liftA2 logBase
801------------------------------------------------------------------------------
802-- Parallely Zipping Streams
803------------------------------------------------------------------------------
805-- | Zip two streams asyncly (i.e. both the elements being zipped are generated
806-- concurrently) using a pure zipping function.
807zipAsyncWith :: (Streaming t, MonadAsync m)
808 => (a -> b -> c) -> t m a -> t m b -> t m c
809zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
812 (runStream (toStream (zipWith f ma mb))) Nothing stp yld
814-- | Like 'ZipStream' but zips in parallel, it generates both the elements to
815-- be zipped concurrently.
818-- main = (toList . 'zippingAsync' $ (,) \<$\> s1 \<*\> s2) >>= print
819-- where s1 = pure 1 <> pure 2
820-- s2 = pure 3 <> pure 4
826-- This applicative operation can be seen as the zipping equivalent of
827-- parallel composition with '<|>'.
828newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}
829 deriving (Semigroup, Monoid)
831deriving instance MonadAsync m => Alternative (ZipAsync m)
833instance Monad m => Functor (ZipAsync m) where
834 fmap f (ZipAsync (Stream m)) = ZipAsync $ Stream $ \_ stp yld ->
835 let yield a Nothing = yld (f a) Nothing
836 yield a (Just r) = yld (f a)
837 (Just (getZipAsync (fmap f (ZipAsync r))))
838 in m Nothing stp yield
840instance MonadAsync m => Applicative (ZipAsync m) where
841 pure a = ZipAsync $ scons a Nothing
842 (<*>) = zipAsyncWith id
844instance Streaming ZipAsync where
845 toStream = getZipAsync
846 fromStream = ZipAsync
848instance (MonadAsync m, Num a) => Num (ZipAsync m a) where
849 fromInteger n = pure (fromInteger n)
859instance (MonadAsync m, Fractional a) => Fractional (ZipAsync m a) where
860 fromRational n = pure (fromRational n)
866instance (MonadAsync m, Floating a) => Floating (ZipAsync m a) where
886 logBase = liftA2 logBase
888-------------------------------------------------------------------------------
889-- Type adapting combinators
890-------------------------------------------------------------------------------
892-- | Adapt one streaming type to another.
893adapt :: (Streaming t1, Streaming t2) => t1 m a -> t2 m a
894adapt = fromStream . toStream
896-- | Interpret an ambiguously typed stream as 'StreamT'.
897serially :: StreamT m a -> StreamT m a
900-- | Interpret an ambiguously typed stream as 'InterleavedT'.
901interleaving :: InterleavedT m a -> InterleavedT m a
904-- | Interpret an ambiguously typed stream as 'AsyncT'.
905asyncly :: AsyncT m a -> AsyncT m a
908-- | Interpret an ambiguously typed stream as 'ParallelT'.
909parallely :: ParallelT m a -> ParallelT m a
912-- | Interpret an ambiguously typed stream as 'ZipStream'.
913zipping :: ZipStream m a -> ZipStream m a
916-- | Interpret an ambiguously typed stream as 'ZipAsync'.
917zippingAsync :: ZipAsync m a -> ZipAsync m a
920-------------------------------------------------------------------------------
921-- Running Streams, convenience functions specialized to types
922-------------------------------------------------------------------------------
924-- | Same as @runStreaming . serially@.
925runStreamT :: Monad m => StreamT m a -> m ()
926runStreamT = runStreaming
928-- | Same as @runStreaming . interleaving@.
929runInterleavedT :: Monad m => InterleavedT m a -> m ()
930runInterleavedT = runStreaming
932-- | Same as @runStreaming . asyncly@.
933runAsyncT :: Monad m => AsyncT m a -> m ()
934runAsyncT = runStreaming
936-- | Same as @runStreaming . parallely@.
937runParallelT :: Monad m => ParallelT m a -> m ()
938runParallelT = runStreaming
940-- | Same as @runStreaming . zipping@.
941runZipStream :: Monad m => ZipStream m a -> m ()
942runZipStream = runStreaming
944-- | Same as @runStreaming . zippingAsync@.
945runZipAsync :: Monad m => ZipAsync m a -> m ()
946runZipAsync = runStreaming
948------------------------------------------------------------------------------
949-- Sum Style Composition
950------------------------------------------------------------------------------
954-- | Sequential interleaved composition, in contrast to '<>' this operator
955-- fairly interleaves two streams instead of appending them; yielding one
956-- element from each stream alternately.
959-- main = ('toList' . 'serially' $ (return 1 <> return 2) \<=\> (return 3 <> return 4)) >>= print
965-- This operator corresponds to the 'InterleavedT' style. Unlike '<>', this
966-- operator cannot be used to fold infinite containers since that might
967-- accumulate too many partially drained streams. To be clear, it can combine
968-- infinite streams but not infinite number of streams.
970(<=>) :: Streaming t => t m a -> t m a -> t m a
971m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
973-- | Demand driven concurrent composition. In contrast to '<|>' this operator
974-- concurrently "merges" streams in a left biased manner rather than fairly
975-- interleaving them. It keeps yielding from the stream on the left as long as
976-- it can. If the left stream blocks or cannot keep up with the pace of the
977-- consumer it can concurrently yield from the stream on the right in parallel.
980-- main = ('toList' . 'serially' $ (return 1 <> return 2) \<| (return 3 <> return 4)) >>= print
986-- Unlike '<|>' it can be used to fold infinite containers of streams. This
987-- operator corresponds to the 'AsyncT' type for product style composition.
990(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
991m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
993------------------------------------------------------------------------------
995------------------------------------------------------------------------------
998-- These utilities are designed to pass the first argument as one of the sum
999-- style composition operators (i.e. '<>', '<=>', '<|', '<|>') to conveniently
1000-- fold a container using any style of stream composition.
1002-- | Like the 'Prelude' 'fold' but allows you to specify a binary sum style
1003-- stream composition operator to fold a container of streams.
1005-- @foldWith (<>) $ map return [1..3]@
1006{-# INLINABLE foldWith #-}
1007foldWith :: (Streaming t, Foldable f)
1008 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
1009foldWith f = foldr f nil
1011-- | Like 'foldMap' but allows you to specify a binary sum style composition
1012-- operator to fold a container of streams. Maps a monadic streaming action on
1013-- the container before folding it.
1015-- @foldMapWith (<>) return [1..3]@
1016{-# INLINABLE foldMapWith #-}
1017foldMapWith :: (Streaming t, Foldable f)
1018 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
1019foldMapWith f g = foldr (f . g) nil
1021-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
1022-- monadic streaming function is the last argument.
1023{-# INLINABLE forEachWith #-}
1024forEachWith :: (Streaming t, Foldable f)
1025 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
1026forEachWith 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