2{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE FlexibleInstances #-}
4{-# LANGUAGE MultiParamTypeClasses #-}
5{-# LANGUAGE RankNTypes #-}
6{-# LANGUAGE UndecidableInstances #-} -- XXX
8#if __GLASGOW_HASKELL__ >= 800
9{-# OPTIONS_GHC -Wno-orphans #-}
12#include "Streams/inline.h"
15-- Module : Streamly.Prelude
16-- Copyright : (c) 2017 Harendra Kumar
19-- Maintainer : harendra.kumar@gmail.com
20-- Stability : experimental
23-- This module is designed to be imported qualified:
26-- import qualified Streamly.Prelude as S
29-- Functions with the suffix @M@ are general functions that work on monadic
30-- arguments. The corresponding functions without the suffix @M@ work on pure
31-- arguments and can in general be derived from their monadic versions but are
32-- provided for convenience and for consistency with other pure APIs in the
35-- Functions having a 'MonadAsync' constraint work concurrently when used with
36-- appropriate stream type combinator. Please be careful to not use 'parallely'
37-- with infinite streams.
39-- Deconstruction and folds accept a 'SerialT' type instead of a polymorphic
40-- type to ensure that streams always have a concrete monomorphic type by
41-- default, reducing type errors. In case you want to use any other type of
42-- stream you can use one of the type combinators provided in the "Streamly"
43-- module to convert the stream type.
45module Streamly.Prelude
48 -- | All other stream construction and generation combinators described
49 -- later, and even more custom combinators can be expressed in terms of
50 -- these primitives. However, the special versions provided in this module
51 -- can be more efficient in some situations.
54 -- | Primitives to construct a stream from pure values or monadic actions.
64 -- | You can construct streams by appending or merging existing streams.
65 -- When constructing streams from streams, '<>' and 'mempty' are the
66 -- intuitive equivalents of 'K.cons' and 'K.nil', respectively. These
67 -- primitives can be very useful when constructing your own custom stream
68 -- combinators. Also see the variants of '<>' defined in the "Streamly"
69 -- module. Note that appending streams is inexpensive, it is much more
70 -- efficient than appending lists.
73 -- ** Unfold and Iterate
74 -- | Note that the generative steps of unfold and iterate are inherently
75 -- serial as the next step depends on the result of the previous step.
76 -- However, consumption of the result from the previous step can happen in
77 -- parallel with the generation by the next step.
83 -- ** Replicate and Repeat
84 -- | Generate a monadic stream from a seed value or function. Note that
85 -- these functions can generate a stream fully concurrently as, unlike
86 -- unfolds, there is no dependency between steps, therefore, an unbounded
87 -- number of steps can run concurrently. All of these can be expressed in
88 -- terms of 'K.cons' and 'K.nil' primitives.
94 -- | Convert an input structure, container or source into a stream. All of
95 -- these can be expressed in terms of primitives.
108 -- | All the folds can be implemented in terms of 'uncons', however the
109 -- specific implementations provided here are generally more efficient.
110 -- Folds are inherently serial as each step needs to use the result of
111 -- the previous step.
121 -- ** Specialized Folds
122 -- | These folds can be expressed in terms of the general fold routines but
123 -- the special versions here can be more efficient in many cases.
125 -- Filtering folds: extract parts of the stream
131 -- Conditional folds: may terminate early based on a condition
144 -- Full folds - need to go through all elements
152 -- | Convert or divert a stream into an output structure, container or
158 -- | One to one transformations, each element in the input stream is
159 -- transformed into a corresponding element in the output stream.
160 -- Therefore, the length of the stream and the ordering of elements in the
161 -- stream remains unchanged after the transformation.
164 -- | Scan is a transformation by continuously folding the result with the
165 -- next element of the stream. This is the generalized way to transform a
166 -- stream carrying state from previous transformation steps, other forms of
167 -- transformation like map can be expressed in terms of this.
173 -- | Map is a special form of scan where no state is carried from one step
181 -- * Filtering and Insertion
182 -- | Adding or removing elements from the stream thus changing the length
201 -- * Hybrid Operations
209 -- ** Scan and filter
228import Control.Monad.IO.Class (MonadIO(..))
229import Data.Maybe (isJust, fromJust)
231 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
232 foldl, mapM, mapM_, sequence, all, any, sum, product, elem,
233 notElem, maximum, minimum, head, last, tail, length, null,
234 reverse, iterate, init, and, or, lookup, foldr1)
235import qualified Prelude
236import qualified System.IO as IO
238import Streamly.SVar (MonadAsync, defState, rstState)
239import Streamly.Streams.SVar (maxYields)
240import Streamly.Streams.StreamK (IsStream(..))
241import Streamly.Streams.Serial (SerialT)
243import qualified Streamly.Streams.StreamK as K
244import qualified Streamly.Streams.StreamD as D
245import qualified Streamly.Streams.Zip as Z
247#ifdef USE_STREAMK_ONLY
248import qualified Streamly.Streams.StreamK as S
249import qualified Streamly.Streams.Zip as S
251import qualified Streamly.Streams.StreamD as S
254import qualified Streamly.Streams.Serial as Serial
256------------------------------------------------------------------------------
257-- Conversion to and from direct style stream
258------------------------------------------------------------------------------
260-- These definitions are dependent on what is imported as S
261{-# INLINE fromStreamS #-}
262fromStreamS :: (IsStream t, Monad m) => S.Stream m a -> t m a
263fromStreamS = fromStream . S.toStreamK
265{-# INLINE toStreamS #-}
266toStreamS :: (IsStream t, Monad m) => t m a -> S.Stream m a
267toStreamS = S.fromStreamK . toStream
269{-# INLINE fromStreamD #-}
270fromStreamD :: (IsStream t, Monad m) => D.Stream m a -> t m a
271fromStreamD = fromStream . D.toStreamK
273{-# INLINE toStreamD #-}
274toStreamD :: (IsStream t, Monad m) => t m a -> D.Stream m a
275toStreamD = D.fromStreamK . toStream
277------------------------------------------------------------------------------
279------------------------------------------------------------------------------
281-- | Decompose a stream into its head and tail. If the stream is empty, returns
282-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
283-- the head of the stream and @ma@ its tail.
286uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
287uncons m = K.uncons (K.adapt m)
289------------------------------------------------------------------------------
290-- Generation by Unfolding
291------------------------------------------------------------------------------
293-- | Build a stream by unfolding a /pure/ step function starting from a seed.
294-- The step function returns the next element in the stream and the next seed
295-- value. When it is done it returns 'Nothing' and the stream ends. For
302-- else Just (b, b + 1)
303-- in toList $ unfoldr f 0
309-- unfoldr can be expressed in terms of 'yield' and '<>' as follows:
315-- Just (a, b) -> 'yield' a '<>' (unfoldr step b)
319{-# INLINE_EARLY unfoldr #-}
320unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
321unfoldr step seed = fromStreamS (S.unfoldr step seed)
322{-# RULES "unfoldr fallback to StreamK" [1]
323 forall a b. S.toStreamK (S.unfoldr a b) = K.unfoldr a b #-}
325-- | Build a stream by unfolding a /monadic/ step function starting from a
326-- seed. The step function returns the next element in the stream and the next
327-- seed value. When it is done it returns 'Nothing' and the stream ends. For
333-- then return Nothing
334-- else print b >> return (Just (b, b + 1))
335-- in runStream $ unfoldrM f 0
343-- When run concurrently, the next unfold step can run concurrently with the
344-- processing of the output of the previous step. Note that more than one step
345-- cannot run concurrently as the next step depends on the output of the
349-- (asyncly $ S.unfoldrM (\\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0)
350-- & S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
356{-# INLINE_EARLY unfoldrM #-}
357unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
360{-# RULES "unfoldrM serial" unfoldrM = unfoldrMSerial #-}
361{-# INLINE_EARLY unfoldrMSerial #-}
362unfoldrMSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> SerialT m a
363unfoldrMSerial step seed = fromStreamS (S.unfoldrM step seed)
365------------------------------------------------------------------------------
366-- Specialized Generation
367------------------------------------------------------------------------------
369-- Faster than yieldM because there is no bind.
371-- | Create a singleton stream from a pure value. Same as @a `cons` nil@ but
372-- slighly more efficient. Note that in monadic streams, 'yield' is the same
373-- as 'pure' or 'return', however, in Zip applicative streams it is not the
374-- same as 'pure' because in that case 'pure' is equivalent to 'repeat'
375-- instead. In all other stream types, 'yield' is the same as @yieldM . pure@
376-- but more efficient.
380yield :: IsStream t => a -> t m a
383-- | Create a singleton stream from a monadic action. Same as @m \`consM` nil@
384-- but more efficient.
387-- > toList $ yieldM getLine
394yieldM :: (Monad m, IsStream t) => m a -> t m a
397-- | Generate a stream by performing a monadic action @n@ times. Can be
398-- expressed as @stimes n (yieldM m)@.
402-- runStream $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
403-- runStream $ asyncly $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
409replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
412 go cnt = if cnt <= 0 then K.nil else m |: go (cnt - 1)
414-- | Generate a stream by repeatedly executing a monadic action forever. Can be
415-- expressed as @cycle1 . yieldM@.
418-- runStream $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
419-- runStream $ asyncly $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
422-- /Concurrent, infinite (do not use with 'parallely')/
425repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
427 where go m = m |: go m
429-- | Iterate a pure function from a seed value, streaming the results forever.
432iterate :: IsStream t => (a -> a) -> a -> t m a
433iterate step = fromStream . go
435 go s = K.cons s (go (step s))
437-- | Iterate a monadic function from a seed value, streaming the results
440-- When run concurrently, the next iteration can run concurrently with the
441-- processing of the previous iteration. Note that more than one iteration
442-- cannot run concurrently as the next iteration depends on the output of the
443-- previous iteration.
446-- runStream $ serially $ S.take 10 $ S.iterateM
447-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0
449-- runStream $ asyncly $ S.take 10 $ S.iterateM
450-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0
456iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a
459 go s = fromStream $ K.Stream $ \svr stp sng yld -> do
461 K.unStream (toStream (return s |: go next)) svr stp sng yld
463------------------------------------------------------------------------------
465------------------------------------------------------------------------------
467-- | Construct a stream from a list containing pure values. More efficient list
468-- specific implementation of 'K.fromFoldable' as it works well with fusion
472{-# INLINE_EARLY fromList #-}
473fromList :: (Monad m, IsStream t) => [a] -> t m a
474fromList = fromStreamS . S.fromList
475{-# RULES "fromList fallback to StreamK" [1]
476 forall a. S.toStreamK (S.fromList a) = K.fromFoldable a #-}
478-- | Construct a stream from a list containing monadic actions. More efficient
479-- list specific implementation of 'fromFoldableM' especially for serial
480-- streams as it works well with fusion optimization.
483{-# INLINE_EARLY fromListM #-}
484fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a
485fromListM = fromStreamD . D.fromListM
486{-# RULES "fromListM fallback to StreamK" [1]
487 forall a. D.toStreamK (D.fromListM a) = fromFoldableM a #-}
489-- | Construct a stream from a 'Foldable' containing monadic actions. Same as
490-- @'Prelude.foldr' 'consM' 'K.nil'@.
493-- runStream $ serially $ S.fromFoldableM $ replicate 10 (threadDelay 1000000 >> print 1)
494-- runStream $ asyncly $ S.fromFoldableM $ replicate 10 (threadDelay 1000000 >> print 1)
497-- /Concurrent (do not use with 'parallely' on infinite containers)/
500{-# INLINE fromFoldableM #-}
501fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
502fromFoldableM = Prelude.foldr consM K.nil
504-- | Same as 'fromFoldable'.
507{-# DEPRECATED each "Please use fromFoldable instead." #-}
509each :: (IsStream t, Foldable f) => f a -> t m a
512-- | Read lines from an IO Handle into a stream of Strings.
515fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
516fromHandle h = fromStream go
518 go = K.Stream $ \_ stp _ yld -> do
519 eof <- liftIO $ IO.hIsEOF h
523 str <- liftIO $ IO.hGetLine h
526------------------------------------------------------------------------------
527-- Elimination by Folding
528------------------------------------------------------------------------------
530-- | Lazy right fold with a monadic step function. For example, to fold a
531-- stream into a list:
534-- >> runIdentity $ foldrM (\\x xs -> return (x : xs)) [] (serially $ fromFoldable [1,2,3])
540foldrM :: Monad m => (a -> b -> m b) -> b -> SerialT m a -> m b
541foldrM step acc m = S.foldrM step acc $ toStreamS m
543-- | Lazy right associative fold. For example, to fold a stream into a list:
546-- >> runIdentity $ foldr (:) [] (serially $ fromFoldable [1,2,3])
552foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
553-- XXX somehow this definition does not perform well, need to investigate
554-- foldr step acc m = S.foldr step acc $ S.fromStreamK (toStream m)
555foldr f = foldrM (\a b -> return (f a b))
557-- | Right fold, for non-empty streams, using first element as the starting
558-- value. Returns 'Nothing' if the stream is empty.
562foldr1 :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
565-- | Strict left fold with an extraction function. Like the standard strict
566-- left fold, but applies a user supplied extraction function (the third
567-- argument) to the folded value at the end. This is designed to work with the
568-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
572foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
577{-# DEPRECATED foldl "Please use foldx instead." #-}
578foldl :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
581-- | Strict left associative fold.
585foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
586foldl' step begin m = S.foldl' step begin $ toStreamS m
588-- | Strict left fold, for non-empty streams, using first element as the
589-- starting value. Returns 'Nothing' if the stream is empty.
592foldl1' :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
596 Nothing -> return Nothing
598 res <- foldl' step h t
601-- XXX replace the recursive "go" with explicit continuations.
602-- | Like 'foldx', but with a monadic step function.
605foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
610{-# DEPRECATED foldlM "Please use foldxM instead." #-}
611foldlM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
614-- | Like 'foldl'' but with a monadic step function.
617foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
618foldlM' step begin m = S.foldlM' step begin $ toStreamS m
620------------------------------------------------------------------------------
622------------------------------------------------------------------------------
624-- | Determine whether the stream is empty.
628null :: Monad m => SerialT m a -> m Bool
631-- | Extract the first element of the stream, if any.
635head :: Monad m => SerialT m a -> m (Maybe a)
638-- | Extract all but the first element of the stream, if any.
642tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
643tail m = K.tail (K.adapt m)
645-- | Extract all but the last element of the stream, if any.
649init :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
650init m = K.init (K.adapt m)
652-- | Extract the last element of the stream, if any.
656last :: Monad m => SerialT m a -> m (Maybe a)
657last m = S.last $ toStreamS m
659-- | Determine whether an element is present in the stream.
663elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
664elem e m = S.elem e (toStreamS m)
666-- | Determine whether an element is not present in the stream.
669{-# INLINE notElem #-}
670notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
671notElem e m = S.notElem e (toStreamS m)
673-- | Determine the length of the stream.
677length :: Monad m => SerialT m a -> m Int
678length = foldl' (\n _ -> n + 1) 0
680-- | Determine whether all elements of a stream satisfy a predicate.
684all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
685all p m = S.all p (toStreamS m)
687-- | Determine whether any of the elements of a stream satisfy a predicate.
691any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
692any p m = S.any p (toStreamS m)
694-- | Determines if all elements of a boolean stream are True.
698and :: Monad m => SerialT m Bool -> m Bool
701-- | Determines wheter at least one element of a boolean stream is True.
705or :: Monad m => SerialT m Bool -> m Bool
708-- | Determine the sum of all elements of a stream of numbers
712sum :: (Monad m, Num a) => SerialT m a -> m a
715-- | Determine the product of all elements of a stream of numbers
718{-# INLINE product #-}
719product :: (Monad m, Num a) => SerialT m a -> m a
720product = foldl' (*) 1
722-- | Determine the minimum element in a stream.
725{-# INLINE minimum #-}
726minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
727minimum m = S.minimum (toStreamS m)
729-- | Determine the maximum element in a stream.
732{-# INLINE maximum #-}
733maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
734maximum m = S.maximum (toStreamS m)
736-- | Looks the given key up, treating the given stream as an association list.
740lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)
743-- | Returns the first element of the stream satisfying the given predicate,
748find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)
751-- | Find all the indices where the element in the stream satisfies the given
755{-# INLINE findIndices #-}
756findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int
757findIndices = K.findIndices
759-- | Gives the index of the first stream element satisfying the given
763{-# INLINE findIndex #-}
764findIndex :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe Int)
765findIndex p = head . findIndices p
767-- | Find all the indices where the value of the element in the stream is equal
768-- to the given value.
771{-# INLINE elemIndices #-}
772elemIndices :: (IsStream t, Eq a) => a -> t m a -> t m Int
773elemIndices a = findIndices (==a)
775-- | Gives the first index of an element in the stream, which equals the given.
778{-# INLINE elemIndex #-}
779elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int)
780elemIndex a = findIndex (==a)
782------------------------------------------------------------------------------
784------------------------------------------------------------------------------
786-- XXX this can utilize parallel mapping if we implement it as runStream . mapM
787-- | Apply a monadic action to each element of the stream and discard the
788-- output of the action.
792mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()
793mapM_ f m = S.mapM_ f $ toStreamS m
795------------------------------------------------------------------------------
797------------------------------------------------------------------------------
799-- | Convert a stream into a list in the underlying monad.
803toList :: Monad m => SerialT m a -> m [a]
804toList m = S.toList $ toStreamS m
806-- | Write a stream of Strings to an IO Handle.
809toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
810toHandle h m = go (toStream m)
814 single a = liftIO (IO.hPutStrLn h a)
815 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r
816 in K.unStream m1 defState stop single yieldk
818------------------------------------------------------------------------------
819-- Transformation by Folding (Scans)
820------------------------------------------------------------------------------
822-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
823-- user supplied extraction function (the third argument) at each step. This is
824-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
829scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
834{-# DEPRECATED scan "Please use scanx instead." #-}
835scan :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
838-- | Like 'scanl'' but with a monadic step function.
841{-# INLINE scanlM' #-}
842scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
843scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
845-- | Strict left scan. Like 'foldl'', but returns the folded value at each
846-- step, generating a stream of all intermediate fold results. The first
847-- element of the stream is the user supplied initial value, and the last
848-- element of the stream is the same as the result of 'foldl''.
852scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
853scanl' step = scanlM' (\a b -> return (step a b))
855------------------------------------------------------------------------------
856-- Transformation by Filtering
857------------------------------------------------------------------------------
859-- | Include only those elements that pass a predicate.
863#if __GLASGOW_HASKELL__ != 802
864-- GHC 8.2.2 crashes with this code, when used with "stack"
865filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
866filter p m = fromStreamS $ S.filter p $ toStreamS m
868filter :: IsStream t => (a -> Bool) -> t m a -> t m a
872-- | Same as 'filter' but with a monadic predicate.
875{-# INLINE filterM #-}
876filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
877filterM p m = fromStreamD $ D.filterM p $ toStreamD m
879-- | Take first 'n' elements from the stream and discard the rest.
883take :: (IsStream t, Monad m) => Int -> t m a -> t m a
884take n m = fromStreamS $ S.take n $ toStreamS
885 (maxYields (Just (fromIntegral n)) m)
887-- | End the stream as soon as the predicate fails on an element.
890{-# INLINE takeWhile #-}
891takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
892takeWhile p m = fromStreamS $ S.takeWhile p $ toStreamS m
894-- | Same as 'takeWhile' but with a monadic predicate.
897{-# INLINE takeWhileM #-}
898takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
899takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
901-- | Discard first 'n' elements from the stream and take the rest.
905drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
906drop n m = fromStreamS $ S.drop n $ toStreamS m
908-- | Drop elements in the stream as long as the predicate succeeds and then
909-- take the rest of the stream.
912{-# INLINE dropWhile #-}
913dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
914dropWhile p m = fromStreamS $ S.dropWhile p $ toStreamS m
916-- | Same as 'dropWhile' but with a monadic predicate.
919{-# INLINE dropWhileM #-}
920dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
921dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
923------------------------------------------------------------------------------
924-- Transformation by Mapping
925------------------------------------------------------------------------------
927-- | Replace each element of the stream with the result of a monadic action
928-- applied on the element.
931-- runStream $ S.replicateM 10 (return 1)
932-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
934-- runStream $ S.replicateM 10 (return 1)
935-- & (asyncly . S.mapM (\\x -> threadDelay 1000000 >> print x))
938-- /Concurrent (do not use with 'parallely' on infinite streams)/
941{-# INLINE_EARLY mapM #-}
942mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
945{-# RULES "mapM serial" mapM = mapMSerial #-}
946{-# INLINE mapMSerial #-}
947mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
948mapMSerial = Serial.mapM
950-- | Reduce a stream of monadic actions to a stream of the output of those
954-- runStream $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
955-- & (serially . S.sequence)
957-- runStream $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
958-- & (asyncly . S.sequence)
961-- /Concurrent (do not use with 'parallely' on infinite streams)/
964{-# INLINE sequence #-}
965sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
968------------------------------------------------------------------------------
969-- Transformation by Map and Filter
970------------------------------------------------------------------------------
972-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
973-- elements, and return a stream of values extracted from 'Just'.
976{-# INLINE mapMaybe #-}
977mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
978mapMaybe f m = fromStreamS $ S.mapMaybe f $ toStreamS m
980-- | Like 'mapMaybe' but maps a monadic function.
982-- /Concurrent (do not use with 'parallely' on infinite streams)/
985{-# INLINE mapMaybeM #-}
986mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
987 => (a -> m (Maybe b)) -> t m a -> t m b
988mapMaybeM f = fmap fromJust . filter isJust . mapM f
990------------------------------------------------------------------------------
991-- Transformation by Reordering
992------------------------------------------------------------------------------
994-- XXX to scale this we need to use a slab allocated array backed
995-- representation for temporary storage.
997-- | Returns the elements of the stream in reverse order.
998-- The stream must be finite.
1001reverse :: (IsStream t) => t m a -> t m a
1002reverse m = fromStream $ go K.nil (toStream m)
1004 go rev rest = K.Stream $ \st stp sng yld ->
1005 let runIt x = K.unStream x (rstState st) stp sng yld
1007 single a = runIt $ a `K.cons` rev
1008 yieldk a r = runIt $ go (a `K.cons` rev) r
1009 in K.unStream rest (rstState st) stop single yieldk
1011------------------------------------------------------------------------------
1012-- Transformation by Inserting
1013------------------------------------------------------------------------------
1015-- | Generate a stream by performing the monadic action inbetween all elements
1016-- of the given stream.
1019{-# INLINE intersperseM #-}
1020intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
1021intersperseM = K.intersperseM
1023------------------------------------------------------------------------------
1025------------------------------------------------------------------------------
1027-- | Zip two streams serially using a monadic zipping function.
1030{-# INLINABLE zipWithM #-}
1031zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
1032zipWithM f m1 m2 = fromStreamS $ S.zipWithM f (toStreamS m1) (toStreamS m2)
1034-- | Zip two streams serially using a pure zipping function.
1037{-# INLINABLE zipWith #-}
1038zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
1039zipWith f m1 m2 = fromStreamS $ S.zipWith f (toStreamS m1) (toStreamS m2)
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