1{-# LANGUAGE ConstraintKinds #-}
2{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE FlexibleInstances #-}
4{-# LANGUAGE GeneralizedNewtypeDeriving#-}
5{-# LANGUAGE MultiParamTypeClasses #-}
6{-# LANGUAGE RankNTypes #-}
7{-# LANGUAGE StandaloneDeriving #-}
8{-# LANGUAGE UndecidableInstances #-} -- XXX
11-- Module : Streamly.Streams
12-- Copyright : (c) 2017 Harendra Kumar
15-- Maintainer : harendra.kumar@gmail.com
16-- Stability : experimental
20module Streamly.Streams
23 , Streaming -- deprecated
44 , runStreaming -- deprecated
60 , StreamT -- deprecated
62 , InterleavedT -- deprecated
64 , AsyncT -- deprecated
67 , ZipStream -- deprecated
73 , interleaving -- deprecated
75 , asyncly -- deprecated
82 , runStreamT -- deprecated
83 , runInterleavedT -- deprecated
84 , runAsyncT -- deprecated
85 , runParallelT -- deprecated
86 , runZipStream -- deprecated
87 , runZipAsync -- deprecated
101import Control.Applicative (Alternative (..), liftA2)
102import Control.Monad (MonadPlus(..), ap)
103import Control.Monad.Base (MonadBase (..))
104import Control.Monad.Catch (MonadThrow)
105import Control.Monad.Error.Class (MonadError(..))
106import Control.Monad.IO.Class (MonadIO(..))
107import Control.Monad.Reader.Class (MonadReader(..))
108import Control.Monad.State.Class (MonadState(..))
109import Control.Monad.Trans.Class (MonadTrans)
110import Data.Semigroup (Semigroup(..))
111import Prelude hiding (zipWith)
112import Streamly.Core ( MonadAsync, Stream(Stream)
113 , SVar, SVarStyle(..)
114 , SVarTag(..), SVarSched(..))
115import qualified Streamly.Core as S
117------------------------------------------------------------------------------
118-- Types that can behave as a Stream
119------------------------------------------------------------------------------
121-- | Class of types that can represent a stream of elements of some type 'a' in
123class IsStream t where
124 toStream :: t m a -> Stream m a
125 fromStream :: Stream m a -> t m a
127-- | Same as 'IsStream'.
128{-# Deprecated Streaming "Please use IsStream instead." #-}
129type Streaming = IsStream
131------------------------------------------------------------------------------
132-- Constructing a stream
133------------------------------------------------------------------------------
135-- | Represesnts an empty stream just like @[]@ represents an empty list.
136nil :: IsStream t => t m a
137nil = fromStream S.nil
141-- | Constructs a stream by adding a pure value at the head of an existing
142-- stream, just like ':' constructs lists. For example:
145-- > let stream = 1 \`cons` 2 \`cons` 3 \`cons` nil
146-- > (toList . serially) stream
149cons :: IsStream t => a -> t m a -> t m a
150cons a r = fromStream $ S.cons a (Just (toStream r))
154-- | Operator equivalent of 'cons' so that you can construct a stream of pure
155-- values more succinctly like this:
158-- > let stream = 1 .: 2 .: 3 .: nil
159-- > (toList . serially) stream
163-- '.:' constructs a stream just like ':' constructs a list.
165-- Also note that another equivalent way of building streams from pure values
169-- > let stream = pure 1 <> pure 2 <> pure 3
170-- > (toList . serially) stream
174-- In the first method we construct a stream by adding one element at a time.
175-- In the second method we first construct singleton streams using 'pure' and
176-- then compose all those streams together using the 'Semigroup' style
177-- composition of streams. The former method is a bit more efficient than the
180(.:) :: IsStream t => a -> t m a -> t m a
183-- | Build a stream from its church encoding. The function passed maps
184-- directly to the underlying representation of the stream type. The second
185-- parameter to the function is the "yield" function yielding a value and the
186-- remaining stream if any otherwise 'Nothing'. The third parameter is to
187-- represent an "empty" stream.
188streamBuild :: IsStream t
189 => (forall r. Maybe (SVar m a)
190 -> (a -> Maybe (t m a) -> m r)
194streamBuild k = fromStream $ Stream $ \sv stp yld ->
195 let yield a Nothing = yld a Nothing
196 yield a (Just r) = yld a (Just (toStream r))
199-- | Build a singleton stream from a callback function.
200fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
201fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)
203-- | Read an SVar to get a stream.
204fromSVar :: (MonadAsync m, IsStream t) => SVar m a -> t m a
205fromSVar sv = fromStream $ S.fromStreamVar sv
207------------------------------------------------------------------------------
208-- Destroying a stream
209------------------------------------------------------------------------------
211-- | Fold a stream using its church encoding. The second argument is the "step"
212-- function consuming an element and the remaining stream, if any. The third
213-- argument is for consuming an "empty" stream that yields nothing.
214streamFold :: IsStream t
215 => Maybe (SVar m a) -> (a -> Maybe (t m a) -> m r) -> m r -> t m a -> m r
216streamFold sv step blank m =
217 let yield a Nothing = step a Nothing
218 yield a (Just x) = step a (Just (fromStream x))
219 in (S.runStream (toStream m)) sv blank yield
221-- | Run a streaming composition, discard the results.
222runStream :: Monad m => SerialT m a -> m ()
223runStream m = go (toStream m)
227 yield _ Nothing = stop
228 yield _ (Just x) = go x
229 in (S.runStream m1) Nothing stop yield
231-- | Same as 'runStream'
232{-# Deprecated runStreaming "Please use runStream instead." #-}
233runStreaming :: (Monad m, IsStream t) => t m a -> m ()
234runStreaming = runStream . adapt
236-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
237-- be read back from the SVar using 'fromSVar'.
238toSVar :: (IsStream t, MonadAsync m) => SVar m a -> t m a -> m ()
239toSVar sv m = S.toStreamVar sv (toStream m)
241------------------------------------------------------------------------------
243------------------------------------------------------------------------------
245-- XXX Get rid of this?
246-- | Make a stream asynchronous, triggers the computation and returns a stream
247-- in the underlying monad representing the output generated by the original
248-- computation. The returned action is exhaustible and must be drained once. If
249-- not drained fully we may have a thread blocked forever and once exhausted it
250-- will always return 'empty'.
252async :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
254 sv <- S.newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)
257------------------------------------------------------------------------------
259------------------------------------------------------------------------------
261-- | The 'Semigroup' instance of 'SerialT' appends two streams sequentially,
262-- yielding all elements from the first stream, and then all elements from the
266-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
272-- The 'Monad' instance runs the /monadic continuation/ for each
273-- element of the stream, serially.
276-- main = 'runStream' . 'serially' $ do
277-- x <- return 1 \<\> return 2
285-- 'SerialT' nests streams serially in a depth first manner.
288-- main = 'runStream' . 'serially' $ do
289-- x <- return 1 \<\> return 2
290-- y <- return 3 \<\> return 4
291-- liftIO $ print (x, y)
300-- This behavior of 'SerialT' is exactly like a list transformer. We call the
301-- monadic code being run for each element of the stream a monadic
302-- continuation. In imperative paradigm we can think of this composition as
303-- nested @for@ loops and the monadic continuation is the body of the loop. The
304-- loop iterates for all elements of the stream.
306-- The 'serially' combinator can be omitted as the default stream type is
308-- Note that serial composition can be used to combine an infinite number of
309-- streams as it explores only one stream at a time.
311newtype SerialT m a = SerialT {getSerialT :: Stream m a}
312 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
314deriving instance MonadAsync m => Alternative (SerialT m)
315deriving instance MonadAsync m => MonadPlus (SerialT m)
316deriving instance (MonadBase b m, Monad m) => MonadBase b (SerialT m)
317deriving instance MonadError e m => MonadError e (SerialT m)
318deriving instance MonadReader r m => MonadReader r (SerialT m)
319deriving instance MonadState s m => MonadState s (SerialT m)
321instance IsStream SerialT where
322 toStream = getSerialT
326{-# Deprecated StreamT "Please use SerialT instead." #-}
327type StreamT = SerialT
329-- XXX The Functor/Applicative/Num instances for all the types are exactly the
330-- same, how can we reduce this boilerplate (use TH)? We cannot derive them
331-- from a single base type because they depend on the Monad instance which is
332-- different for each type.
334------------------------------------------------------------------------------
336------------------------------------------------------------------------------
338-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
339-- Appends two streams sequentially, yielding all elements from the first
340-- stream, and then all elements from the second stream.
342serial :: IsStream t => t m a -> t m a -> t m a
343serial m1 m2 = fromStream $ S.serial (toStream m1) (toStream m2)
345------------------------------------------------------------------------------
347------------------------------------------------------------------------------
349instance Monad m => Monad (SerialT m) where
351 (SerialT (Stream m)) >>= f = SerialT $ Stream $ \_ stp yld ->
352 let run x = (S.runStream x) Nothing stp yld
353 yield a Nothing = run $ toStream (f a)
354 yield a (Just r) = run $ toStream $ f a <> (fromStream r >>= f)
355 in m Nothing stp yield
357------------------------------------------------------------------------------
359------------------------------------------------------------------------------
361instance Monad m => Applicative (SerialT m) where
362 pure a = SerialT $ S.cons a Nothing
365------------------------------------------------------------------------------
367------------------------------------------------------------------------------
369instance Monad m => Functor (SerialT m) where
370 fmap f (SerialT (Stream m)) = SerialT $ Stream $ \_ stp yld ->
371 let yield a Nothing = yld (f a) Nothing
372 yield a (Just r) = yld (f a)
373 (Just (getSerialT (fmap f (SerialT r))))
374 in m Nothing stp yield
376------------------------------------------------------------------------------
378------------------------------------------------------------------------------
380instance (Monad m, Num a) => Num (SerialT m a) where
381 fromInteger n = pure (fromInteger n)
391instance (Monad m, Fractional a) => Fractional (SerialT m a) where
392 fromRational n = pure (fromRational n)
398instance (Monad m, Floating a) => Floating (SerialT m a) where
418 logBase = liftA2 logBase
420------------------------------------------------------------------------------
422------------------------------------------------------------------------------
424-- | The 'Semigroup' instance of 'CoserialT' interleaves two streams,
425-- yielding one element from each stream alternately.
428-- main = ('toList' . 'interleaving' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
434-- Similarly, the 'Monad' instance fairly interleaves the iterations of the
435-- inner and the outer loop, nesting loops in a breadth first manner.
439-- main = 'runStream' . 'interleaving' $ do
440-- x <- return 1 \<\> return 2
441-- y <- return 3 \<\> return 4
442-- liftIO $ print (x, y)
451-- Note that interleaving composition can only combine a finite number of
452-- streams as it needs to retain state for each unfinished stream.
454newtype CoserialT m a = CoserialT {getCoserialT :: Stream m a}
455 deriving (Monoid, MonadTrans, MonadIO, MonadThrow)
457deriving instance MonadAsync m => Alternative (CoserialT m)
458deriving instance MonadAsync m => MonadPlus (CoserialT m)
459deriving instance (MonadBase b m, Monad m) => MonadBase b (CoserialT m)
460deriving instance MonadError e m => MonadError e (CoserialT m)
461deriving instance MonadReader r m => MonadReader r (CoserialT m)
462deriving instance MonadState s m => MonadState s (CoserialT m)
464{-# DEPRECATED InterleavedT "Please use 'CoserialT' instead." #-}
465type InterleavedT = CoserialT
467instance IsStream CoserialT where
468 toStream = getCoserialT
469 fromStream = CoserialT
471------------------------------------------------------------------------------
473------------------------------------------------------------------------------
475-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CoserialT'.
476-- Interleaves two streams, yielding one element from each stream alternately.
477{-# INLINE coserial #-}
478coserial :: IsStream t => t m a -> t m a -> t m a
479coserial m1 m2 = fromStream $ S.coserial (toStream m1) (toStream m2)
481instance Semigroup (CoserialT m a) where
486-- | Same as 'coserial'.
487{-# DEPRECATED (<=>) "Please use 'coserial' instead." #-}
489(<=>) :: IsStream t => t m a -> t m a -> t m a
492------------------------------------------------------------------------------
494------------------------------------------------------------------------------
496instance Monad m => Monad (CoserialT m) where
498 (CoserialT (Stream m)) >>= f = CoserialT $ Stream $ \_ stp yld ->
499 let run x = (S.runStream x) Nothing stp yld
500 yield a Nothing = run $ toStream (f a)
501 yield a (Just r) = run $ toStream $ f a <> (fromStream r >>= f)
502 in m Nothing stp yield
504------------------------------------------------------------------------------
506------------------------------------------------------------------------------
508instance Monad m => Applicative (CoserialT m) where
509 pure a = CoserialT $ S.cons a Nothing
512------------------------------------------------------------------------------
514------------------------------------------------------------------------------
516instance Monad m => Functor (CoserialT m) where
517 fmap f (CoserialT (Stream m)) = CoserialT $ Stream $ \_ stp yld ->
518 let yield a Nothing = yld (f a) Nothing
520 yld (f a) (Just (getCoserialT (fmap f (CoserialT r))))
521 in m Nothing stp yield
523------------------------------------------------------------------------------
525------------------------------------------------------------------------------
527instance (Monad m, Num a) => Num (CoserialT m a) where
528 fromInteger n = pure (fromInteger n)
538instance (Monad m, Fractional a) => Fractional (CoserialT m a) where
539 fromRational n = pure (fromRational n)
545instance (Monad m, Floating a) => Floating (CoserialT m a) where
565 logBase = liftA2 logBase
567------------------------------------------------------------------------------
569------------------------------------------------------------------------------
571-- | Adaptive parallel, in a left to right 'Semigroup' composition it tries to
572-- yield elements from the left stream as long as it can, but it can run the
573-- right stream in parallel if it needs to based on demand. The right stream
574-- can be run if the left stream blocks on IO or cannot produce elements fast
575-- enough for the consumer.
578-- main = ('toList' . 'coparallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
584-- Similarly, the monad instance of 'ParallelT' /may/ run each iteration
585-- concurrently based on demand. More concurrent iterations are started only
586-- if the previous iterations are not able to produce enough output for the
591-- import Control.Concurrent
593-- main = 'runStream' . 'coparallely' $ do
594-- n <- return 3 \<\> return 2 \<\> return 1
596-- threadDelay (n * 1000000)
597-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
600-- ThreadId 40: Delay 1
601-- ThreadId 39: Delay 2
602-- ThreadId 38: Delay 3
605-- All iterations may run in the same thread if they do not block.
607-- Note that this composition can be used to combine infinite number of streams
608-- as it explores only a bounded number of streams at a time.
610newtype CoparallelT m a = CoparallelT {getCoparallelT :: Stream m a}
611 deriving (MonadTrans)
613deriving instance MonadAsync m => Monoid (CoparallelT m a)
614deriving instance MonadAsync m => Alternative (CoparallelT m)
615deriving instance MonadAsync m => MonadPlus (CoparallelT m)
616deriving instance MonadAsync m => MonadIO (CoparallelT m)
617deriving instance MonadAsync m => MonadThrow (CoparallelT m)
618deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (CoparallelT m)
619deriving instance (MonadError e m, MonadAsync m) => MonadError e (CoparallelT m)
620deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (CoparallelT m)
621deriving instance (MonadState s m, MonadAsync m) => MonadState s (CoparallelT m)
623{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}
624type AsyncT = CoparallelT
626instance IsStream CoparallelT where
627 toStream = getCoparallelT
628 fromStream = CoparallelT
630------------------------------------------------------------------------------
632------------------------------------------------------------------------------
634-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CoparallelT', but
635-- polymorphic. Merges two streams possibly concurrently, preferring the
636-- elements from the left one when available.
637{-# INLINE coparallel #-}
638coparallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
639coparallel m1 m2 = fromStream $ S.coparallel (toStream m1) (toStream m2)
641instance MonadAsync m => Semigroup (CoparallelT m a) where
644-- | Same as 'coparallel'.
645{-# DEPRECATED (<|) "Please use 'coparallel' instead." #-}
647(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
650------------------------------------------------------------------------------
652------------------------------------------------------------------------------
654{-# INLINE parbind #-}
656 :: (forall c. Stream m c -> Stream m c -> Stream m c)
660parbind par m f = go m
663 Stream $ \ctx stp yld ->
664 let run x = (S.runStream x) ctx stp yld
665 yield a Nothing = run $ f a
666 yield a (Just r) = run $ f a `par` go r
667 in g Nothing stp yield
669instance MonadAsync m => Monad (CoparallelT m) where
671 (CoparallelT m) >>= f = CoparallelT $ parbind par m g
672 where g x = getCoparallelT (f x)
673 par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)
675------------------------------------------------------------------------------
677------------------------------------------------------------------------------
679instance MonadAsync m => Applicative (CoparallelT m) where
680 pure a = CoparallelT $ S.cons a Nothing
683------------------------------------------------------------------------------
685------------------------------------------------------------------------------
687instance Monad m => Functor (CoparallelT m) where
688 fmap f (CoparallelT (Stream m)) = CoparallelT $ Stream $ \_ stp yld ->
689 let yield a Nothing = yld (f a) Nothing
690 yield a (Just r) = yld (f a) (Just (getCoparallelT (fmap f (CoparallelT r))))
691 in m Nothing stp yield
693------------------------------------------------------------------------------
695------------------------------------------------------------------------------
697instance (MonadAsync m, Num a) => Num (CoparallelT m a) where
698 fromInteger n = pure (fromInteger n)
708instance (MonadAsync m, Fractional a) => Fractional (CoparallelT m a) where
709 fromRational n = pure (fromRational n)
715instance (MonadAsync m, Floating a) => Floating (CoparallelT m a) where
735 logBase = liftA2 logBase
737------------------------------------------------------------------------------
739------------------------------------------------------------------------------
741-- | Round robin concurrent composition.
743-- The Semigroup instance of 'ParallelT' concurrently /merges/ streams in a
744-- round robin fashion, yielding elements from both streams alternately.
747-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
753-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations fairly
754-- concurrently using a round robin scheduling.
758-- import Control.Concurrent
760-- main = 'runStream' . 'parallely' $ do
761-- n <- return 3 \<\> return 2 \<\> return 1
763-- threadDelay (n * 1000000)
764-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
767-- ThreadId 40: Delay 1
768-- ThreadId 39: Delay 2
769-- ThreadId 38: Delay 3
772-- Unlike 'CoparallelT' all iterations are guaranteed to run fairly
773-- concurrently, unconditionally.
775-- Note that round robin composition can only combine a finite number of
776-- streams as it needs to retain state for each unfinished stream.
778newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
779 deriving (MonadTrans)
781deriving instance MonadAsync m => Monoid (ParallelT m a)
782deriving instance MonadAsync m => Alternative (ParallelT m)
783deriving instance MonadAsync m => MonadPlus (ParallelT m)
784deriving instance MonadAsync m => MonadIO (ParallelT m)
785deriving instance MonadAsync m => MonadThrow (ParallelT m)
786deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)
787deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)
788deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)
789deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)
791instance IsStream ParallelT where
792 toStream = getParallelT
793 fromStream = ParallelT
795------------------------------------------------------------------------------
797------------------------------------------------------------------------------
799-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'.
800-- Merges two streams concurrently choosing elements from both fairly.
801{-# INLINE parallel #-}
802parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
803parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
805instance MonadAsync m => Semigroup (ParallelT m a) where
808------------------------------------------------------------------------------
810------------------------------------------------------------------------------
812instance MonadAsync m => Monad (ParallelT m) where
814 (ParallelT m) >>= f = ParallelT $ parbind par m g
815 where g x = getParallelT (f x)
816 par = S.joinStreamVar2 (SVarStyle Conjunction FIFO)
818------------------------------------------------------------------------------
820------------------------------------------------------------------------------
822instance MonadAsync m => Applicative (ParallelT m) where
823 pure a = ParallelT $ S.cons a Nothing
826------------------------------------------------------------------------------
828------------------------------------------------------------------------------
830instance Monad m => Functor (ParallelT m) where
831 fmap f (ParallelT (Stream m)) = ParallelT $ Stream $ \_ stp yld ->
832 let yield a Nothing = yld (f a) Nothing
833 yield a (Just r) = yld (f a)
834 (Just (getParallelT (fmap f (ParallelT r))))
835 in m Nothing stp yield
837------------------------------------------------------------------------------
839------------------------------------------------------------------------------
841instance (MonadAsync m, Num a) => Num (ParallelT m a) where
842 fromInteger n = pure (fromInteger n)
852instance (MonadAsync m, Fractional a) => Fractional (ParallelT m a) where
853 fromRational n = pure (fromRational n)
859instance (MonadAsync m, Floating a) => Floating (ParallelT m a) where
879 logBase = liftA2 logBase
881------------------------------------------------------------------------------
882-- Serially Zipping Streams
883------------------------------------------------------------------------------
885-- | Zip two streams serially using a pure zipping function.
886zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
887zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
889 go mx my = Stream $ \_ stp yld -> do
891 let yield2 b Nothing = yld (f a b) Nothing
892 yield2 b (Just rb) = yld (f a b) (Just (go ra rb))
893 in (S.runStream my) Nothing stp yield2
894 let yield1 a Nothing = merge a S.nil
895 yield1 a (Just ra) = merge a ra
896 (S.runStream mx) Nothing stp yield1
898-- | The applicative instance of 'ZipSerial' zips a number of streams serially
899-- i.e. it produces one element from each stream serially and then zips all
900-- those elements. Note, for convenience we have used the 'zipping' combinator
901-- in the following example instead of using a 'ZipSerial' type annotation.
904-- main = (toList . 'zipping' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
905-- where s1 = fromFoldable [1, 2]
906-- s2 = fromFoldable [3, 4]
907-- s3 = fromFoldable [5, 6]
913-- The 'Semigroup' instance of this type works the same way as that of
916newtype ZipSerial m a = ZipSerial {getZipSerial :: Stream m a}
917 deriving (Semigroup, Monoid)
919{-# Deprecated ZipStream "Please use ZipSerial instead." #-}
920-- | Same as ZipSerial.
921type ZipStream = ZipSerial
923deriving instance MonadAsync m => Alternative (ZipSerial m)
925instance Monad m => Functor (ZipSerial m) where
926 fmap f (ZipSerial (Stream m)) = ZipSerial $ Stream $ \_ stp yld ->
927 let yield a Nothing = yld (f a) Nothing
928 yield a (Just r) = yld (f a)
929 (Just (getZipSerial (fmap f (ZipSerial r))))
930 in m Nothing stp yield
932instance Monad m => Applicative (ZipSerial m) where
933 pure = ZipSerial . S.repeat
936instance IsStream ZipSerial where
937 toStream = getZipSerial
938 fromStream = ZipSerial
940instance (Monad m, Num a) => Num (ZipSerial m a) where
941 fromInteger n = pure (fromInteger n)
951instance (Monad m, Fractional a) => Fractional (ZipSerial m a) where
952 fromRational n = pure (fromRational n)
958instance (Monad m, Floating a) => Floating (ZipSerial m a) where
978 logBase = liftA2 logBase
980------------------------------------------------------------------------------
981-- Parallely Zipping Streams
982------------------------------------------------------------------------------
984-- | Zip two streams asyncly (i.e. both the elements being zipped are generated
985-- concurrently) using a pure zipping function.
986zipAsyncWith :: (IsStream t, MonadAsync m)
987 => (a -> b -> c) -> t m a -> t m b -> t m c
988zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
991 (S.runStream (toStream (zipWith f ma mb))) Nothing stp yld
993-- | Like 'ZipSerial' but zips in parallel, it generates all the elements to
994-- be zipped concurrently.
997-- main = (toList . 'zippingAsync' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
998-- where s1 = fromFoldable [1, 2]
999-- s2 = fromFoldable [3, 4]
1000-- s3 = fromFoldable [5, 6]
1006-- The 'Semigroup' instance of this type works the same way as that of
1009newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}
1010 deriving (Semigroup, Monoid)
1012deriving instance MonadAsync m => Alternative (ZipAsync m)
1014instance Monad m => Functor (ZipAsync m) where
1015 fmap f (ZipAsync (Stream m)) = ZipAsync $ Stream $ \_ stp yld ->
1016 let yield a Nothing = yld (f a) Nothing
1017 yield a (Just r) = yld (f a)
1018 (Just (getZipAsync (fmap f (ZipAsync r))))
1019 in m Nothing stp yield
1021instance MonadAsync m => Applicative (ZipAsync m) where
1022 pure = ZipAsync . S.repeat
1023 (<*>) = zipAsyncWith id
1025instance IsStream ZipAsync where
1026 toStream = getZipAsync
1027 fromStream = ZipAsync
1029instance (MonadAsync m, Num a) => Num (ZipAsync m a) where
1030 fromInteger n = pure (fromInteger n)
1032 negate = fmap negate
1034 signum = fmap signum
1040instance (MonadAsync m, Fractional a) => Fractional (ZipAsync m a) where
1041 fromRational n = pure (fromRational n)
1047instance (MonadAsync m, Floating a) => Floating (ZipAsync m a) where
1067 logBase = liftA2 logBase
1069-------------------------------------------------------------------------------
1070-- Type adapting combinators
1071-------------------------------------------------------------------------------
1073-- | Adapt any specific stream type to any other specific stream type.
1074adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
1075adapt = fromStream . toStream
1077-- | Fix the type of a polymorphic stream as 'SerialT'.
1078serially :: IsStream t => SerialT m a -> t m a
1081-- | Fix the type of a polymorphic stream as 'CoserialT'.
1082coserially :: IsStream t => CoserialT m a -> t m a
1085-- | Same as 'coserially'.
1086{-# DEPRECATED interleaving "Please use coserially instead." #-}
1087interleaving :: IsStream t => CoserialT m a -> t m a
1088interleaving = coserially
1090-- | Fix the type of a polymorphic stream as 'CoparallelT'.
1091coparallely :: IsStream t => CoparallelT m a -> t m a
1094-- | Same as 'coparallely'.
1095{-# DEPRECATED asyncly "Please use coparallely instead." #-}
1096asyncly :: IsStream t => CoparallelT m a -> t m a
1097asyncly = coparallely
1099-- | Fix the type of a polymorphic stream as 'ParallelT'.
1100parallely :: IsStream t => ParallelT m a -> t m a
1103-- | Fix the type of a polymorphic stream as 'ZipSerial'.
1104zipping :: IsStream t => ZipSerial m a -> t m a
1107-- | Fix the type of a polymorphic stream as 'ZipAsync'.
1108zippingAsync :: IsStream t => ZipAsync m a -> t m a
1111-------------------------------------------------------------------------------
1112-- Running Streams, convenience functions specialized to types
1113-------------------------------------------------------------------------------
1115-- | Same as @runStream@.
1116{-# DEPRECATED runStreamT "Please use runStream instead." #-}
1117runStreamT :: Monad m => SerialT m a -> m ()
1118runStreamT = runStream
1120-- | Same as @runStream . coserially@.
1121{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}
1122runInterleavedT :: Monad m => InterleavedT m a -> m ()
1123runInterleavedT = runStream . coserially
1125-- | Same as @runStream . coparallely@.
1126{-# DEPRECATED runAsyncT "Please use 'runStream . coparallely' instead." #-}
1127runAsyncT :: Monad m => CoparallelT m a -> m ()
1128runAsyncT = runStream . coparallely
1130-- | Same as @runStream . parallely@.
1131{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}
1132runParallelT :: Monad m => ParallelT m a -> m ()
1133runParallelT = runStream . parallely
1135-- | Same as @runStream . zipping@.
1136{-# DEPRECATED runZipStream "Please use 'runStream . zipping instead." #-}
1137runZipStream :: Monad m => ZipSerial m a -> m ()
1138runZipStream = runStream . zipping
1140-- | Same as @runStream . zippingAsync@.
1141{-# DEPRECATED runZipAsync "Please use 'runStream . zippingAsync instead." #-}
1142runZipAsync :: Monad m => ZipAsync m a -> m ()
1143runZipAsync = runStream . zippingAsync
1145------------------------------------------------------------------------------
1147------------------------------------------------------------------------------
1150-- These utilities are designed to pass the first argument as one of the sum
1151-- style composition operators (i.e. '<>', '<=>', '<|', '<|>') to conveniently
1152-- fold a container using any style of stream composition.
1154-- | Like the 'Prelude' 'fold' but allows you to specify a binary sum style
1155-- stream composition operator to fold a container of streams.
1157-- @foldWith (<>) $ map return [1..3]@
1158{-# INLINABLE foldWith #-}
1159foldWith :: (IsStream t, Foldable f)
1160 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
1161foldWith f = foldr f nil
1163-- | Like 'foldMap' but allows you to specify a binary sum style composition
1164-- operator to fold a container of streams. Maps a monadic streaming action on
1165-- the container before folding it.
1167-- @foldMapWith (<>) return [1..3]@
1168{-# INLINABLE foldMapWith #-}
1169foldMapWith :: (IsStream t, Foldable f)
1170 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
1171foldMapWith f g = foldr (f . g) nil
1173-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
1174-- monadic streaming function is the last argument.
1175{-# INLINABLE forEachWith #-}
1176forEachWith :: (IsStream t, Foldable f)
1177 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
1178forEachWith 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