1{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE MultiParamTypeClasses #-}
6{-# LANGUAGE RankNTypes #-}
7{-# LANGUAGE UndecidableInstances #-} -- XXX
9#if __GLASGOW_HASKELL__ >= 800
10{-# OPTIONS_GHC -Wno-orphans #-}
13#include "Streams/inline.h"
16-- Module : Streamly.Prelude
17-- Copyright : (c) 2017 Harendra Kumar
20-- Maintainer : harendra.kumar@gmail.com
21-- Stability : experimental
24-- This module is designed to be imported qualified:
27-- import qualified Streamly.Prelude as S
30-- Functions with the suffix @M@ are general functions that work on monadic
31-- arguments. The corresponding functions without the suffix @M@ work on pure
32-- arguments and can in general be derived from their monadic versions but are
33-- provided for convenience and for consistency with other pure APIs in the
36-- Functions having a 'MonadAsync' constraint work concurrently when used with
37-- appropriate stream type combinator. Please be careful to not use 'parallely'
38-- with infinite streams.
40-- Deconstruction and folds accept a 'SerialT' type instead of a polymorphic
41-- type to ensure that streams always have a concrete monomorphic type by
42-- default, reducing type errors. In case you want to use any other type of
43-- stream you can use one of the type combinators provided in the "Streamly"
44-- module to convert the stream type.
46module Streamly.Prelude
49 -- | Primitives to construct a stream.
64 -- ** Specialized Generation
65 -- | Generate a monadic stream from a seed.
73 -- | Transform an input structure into a stream.
91 -- ** Specialized Folds
110 -- | Transform a stream into an output structure of another type.
115 -- ** By folding (scans)
157import Control.Monad.IO.Class (MonadIO(..))
158import Data.Maybe (isJust, fromJust)
160 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
161 foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem,
162 notElem, maximum, minimum, head, last, tail, length, null,
164import qualified Prelude
165import qualified System.IO as IO
167import Streamly.SVar (MonadAsync, defState, rstState)
168import Streamly.Streams.SVar (maxYields)
169import Streamly.Streams.StreamK (IsStream(..))
170import Streamly.Streams.Serial (SerialT)
172import qualified Streamly.Streams.StreamK as K
173import qualified Streamly.Streams.StreamD as D
174import qualified Streamly.Streams.Zip as Z
176#ifdef USE_STREAMK_ONLY
177import qualified Streamly.Streams.StreamK as S
178import qualified Streamly.Streams.Zip as S
180import qualified Streamly.Streams.StreamD as S
183import qualified Streamly.Streams.Serial as Serial
185------------------------------------------------------------------------------
186-- Conversion to and from direct style stream
187------------------------------------------------------------------------------
189-- These definitions are dependent on what is imported as S
190{-# INLINE fromStreamS #-}
191fromStreamS :: (IsStream t, Monad m) => S.Stream m a -> t m a
192fromStreamS = fromStream . S.toStreamK
194{-# INLINE toStreamS #-}
195toStreamS :: (IsStream t, Monad m) => t m a -> S.Stream m a
196toStreamS = S.fromStreamK . toStream
198{-# INLINE fromStreamD #-}
199fromStreamD :: (IsStream t, Monad m) => D.Stream m a -> t m a
200fromStreamD = fromStream . D.toStreamK
202{-# INLINE toStreamD #-}
203toStreamD :: (IsStream t, Monad m) => t m a -> D.Stream m a
204toStreamD = D.fromStreamK . toStream
206------------------------------------------------------------------------------
208------------------------------------------------------------------------------
210-- | Decompose a stream into its head and tail. If the stream is empty, returns
211-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
212-- the head of the stream and @ma@ its tail.
215uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
216uncons m = K.uncons (K.adapt m)
218------------------------------------------------------------------------------
219-- Generation by Unfolding
220------------------------------------------------------------------------------
222-- | Build a stream by unfolding a /pure/ step function starting from a seed.
223-- The step function returns the next element in the stream and the next seed
224-- value. When it is done it returns 'Nothing' and the stream ends. For
231-- else Just (b, b + 1)
232-- in toList $ unfoldr f 0
239{-# INLINE_EARLY unfoldr #-}
240unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
241unfoldr step seed = fromStreamS (S.unfoldr step seed)
242{-# RULES "unfoldr fallback to StreamK" [1]
243 forall a b. S.toStreamK (S.unfoldr a b) = K.unfoldr a b #-}
245-- | Build a stream by unfolding a /monadic/ step function starting from a
246-- seed. The step function returns the next element in the stream and the next
247-- seed value. When it is done it returns 'Nothing' and the stream ends. For
253-- then return Nothing
254-- else print b >> return (Just (b, b + 1))
255-- in runStream $ unfoldrM f 0
263-- When run concurrently, the next unfold step can run concurrently with the
264-- processing of the output of the previous step. Note that more than one step
265-- cannot run concurrently as the next step depends on the output of the
269-- (asyncly $ S.unfoldrM (\\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0)
270-- & S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
276{-# INLINE_EARLY unfoldrM #-}
277unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
280{-# RULES "unfoldrM serial" unfoldrM = unfoldrMSerial #-}
281{-# INLINE_EARLY unfoldrMSerial #-}
282unfoldrMSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> SerialT m a
283unfoldrMSerial step seed = fromStreamS (S.unfoldrM step seed)
285------------------------------------------------------------------------------
286-- Specialized Generation
287------------------------------------------------------------------------------
290yield :: IsStream t => a -> t m a
294yieldM :: (Monad m, IsStream t) => m a -> t m a
297-- | Generate a stream by performing a monadic action @n@ times.
301-- runStream $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
302-- runStream $ asyncly $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
308replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
311 go cnt = if cnt <= 0 then K.nil else m |: go (cnt - 1)
313-- | Generate a stream by repeatedly executing a monadic action forever.
316-- runStream $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
317-- runStream $ asyncly $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
320-- /Concurrent, infinite (do not use with 'parallely')/
323repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
325 where go m = m |: go m
327-- | Iterate a pure function from a seed value, streaming the results forever.
330iterate :: IsStream t => (a -> a) -> a -> t m a
331iterate step = fromStream . go
333 go s = K.cons s (go (step s))
335-- | Iterate a monadic function from a seed value, streaming the results
338-- When run concurrently, the next iteration can run concurrently with the
339-- processing of the previous iteration. Note that more than one iteration
340-- cannot run concurrently as the next iteration depends on the output of the
341-- previous iteration.
344-- runStream $ serially $ S.take 10 $ S.iterateM
345-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0
347-- runStream $ asyncly $ S.take 10 $ S.iterateM
348-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0
354iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a
357 go s = fromStream $ K.Stream $ \svr stp sng yld -> do
359 K.unStream (toStream (return s |: go next)) svr stp sng yld
361------------------------------------------------------------------------------
363------------------------------------------------------------------------------
365-- | Construct a stream from a list containing pure values. This can be more
366-- efficient than 'K.fromFoldable' for lists as it can fuse the list.
369{-# INLINE_EARLY fromList #-}
370fromList :: (Monad m, IsStream t) => [a] -> t m a
371fromList = fromStreamS . S.fromList
372{-# RULES "fromList fallback to StreamK" [1]
373 forall a. S.toStreamK (S.fromList a) = K.fromFoldable a #-}
375-- | Construct a stream from a list containing monadic actions. This can be
376-- more efficient than 'fromFoldableM' especially for serial streams as it can
380{-# INLINE_EARLY fromListM #-}
381fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a
382fromListM = fromStreamD . D.fromListM
383{-# RULES "fromListM fallback to StreamK" [1]
384 forall a. D.toStreamK (D.fromListM a) = fromFoldableM a #-}
386-- | Construct a stream from a 'Foldable' containing monadic actions.
389-- runStream $ serially $ S.fromFoldableM $ replicate 10 (threadDelay 1000000 >> print 1)
390-- runStream $ asyncly $ S.fromFoldableM $ replicate 10 (threadDelay 1000000 >> print 1)
393-- /Concurrent (do not use with 'parallely' on infinite containers)/
396{-# INLINE fromFoldableM #-}
397fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
398fromFoldableM = Prelude.foldr consM K.nil
400-- | Same as 'fromFoldable'.
403{-# DEPRECATED each "Please use fromFoldable instead." #-}
405each :: (IsStream t, Foldable f) => f a -> t m a
408-- | Read lines from an IO Handle into a stream of Strings.
411fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
412fromHandle h = fromStream go
414 go = K.Stream $ \_ stp _ yld -> do
415 eof <- liftIO $ IO.hIsEOF h
419 str <- liftIO $ IO.hGetLine h
422------------------------------------------------------------------------------
423-- Elimination by Folding
424------------------------------------------------------------------------------
426-- | Lazy right fold with a monadic step function. For example, to fold a
427-- stream into a list:
430-- >> runIdentity $ foldrM (\\x xs -> return (x : xs)) [] (serially $ fromFoldable [1,2,3])
436foldrM :: Monad m => (a -> b -> m b) -> b -> SerialT m a -> m b
437foldrM step acc m = S.foldrM step acc $ toStreamS m
439-- | Lazy right associative fold. For example, to fold a stream into a list:
442-- >> runIdentity $ foldr (:) [] (serially $ fromFoldable [1,2,3])
448foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
449-- XXX somehow this definition does not perform well, need to investigate
450-- foldr step acc m = S.foldr step acc $ S.fromStreamK (toStream m)
451foldr f = foldrM (\a b -> return (f a b))
453-- | Strict left fold with an extraction function. Like the standard strict
454-- left fold, but applies a user supplied extraction function (the third
455-- argument) to the folded value at the end. This is designed to work with the
456-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
460foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
465{-# DEPRECATED foldl "Please use foldx instead." #-}
466foldl :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
469-- | Strict left associative fold.
473foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
474foldl' step begin m = S.foldl' step begin $ toStreamS m
476-- XXX replace the recursive "go" with explicit continuations.
477-- | Like 'foldx', but with a monadic step function.
480foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
485{-# DEPRECATED foldlM "Please use foldxM instead." #-}
486foldlM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
489-- | Like 'foldl'' but with a monadic step function.
492foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
493foldlM' step begin m = S.foldlM' step begin $ toStreamS m
495------------------------------------------------------------------------------
497------------------------------------------------------------------------------
499-- | Determine whether the stream is empty.
503null :: Monad m => SerialT m a -> m Bool
506-- | Extract the first element of the stream, if any.
510head :: Monad m => SerialT m a -> m (Maybe a)
513-- | Extract all but the first element of the stream, if any.
517tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
518tail m = K.tail (K.adapt m)
520-- | Extract the last element of the stream, if any.
524last :: Monad m => SerialT m a -> m (Maybe a)
525last m = S.last $ toStreamS m
527-- | Determine whether an element is present in the stream.
531elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
532elem e m = S.elem e (toStreamS m)
534-- | Determine whether an element is not present in the stream.
537{-# INLINE notElem #-}
538notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
539notElem e m = S.notElem e (toStreamS m)
541-- | Determine the length of the stream.
545length :: Monad m => SerialT m a -> m Int
546length = foldl' (\n _ -> n + 1) 0
548-- | Determine whether all elements of a stream satisfy a predicate.
552all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
553all p m = S.all p (toStreamS m)
555-- | Determine whether any of the elements of a stream satisfy a predicate.
559any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
560any p m = S.any p (toStreamS m)
562-- | Determine the sum of all elements of a stream of numbers
566sum :: (Monad m, Num a) => SerialT m a -> m a
569-- | Determine the product of all elements of a stream of numbers
572{-# INLINE product #-}
573product :: (Monad m, Num a) => SerialT m a -> m a
574product = foldl' (*) 1
576-- | Determine the minimum element in a stream.
579{-# INLINE minimum #-}
580minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
581minimum m = S.minimum (toStreamS m)
583-- | Determine the maximum element in a stream.
586{-# INLINE maximum #-}
587maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
588maximum m = S.maximum (toStreamS m)
590------------------------------------------------------------------------------
592------------------------------------------------------------------------------
594-- XXX this can utilize parallel mapping if we implement it as runStream . mapM
595-- | Apply a monadic action to each element of the stream and discard the
596-- output of the action.
600mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()
601mapM_ f m = S.mapM_ f $ toStreamS m
603------------------------------------------------------------------------------
605------------------------------------------------------------------------------
607-- | Convert a stream into a list in the underlying monad.
611toList :: Monad m => SerialT m a -> m [a]
612toList m = S.toList $ toStreamS m
614-- | Write a stream of Strings to an IO Handle.
617toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
618toHandle h m = go (toStream m)
622 single a = liftIO (IO.hPutStrLn h a)
623 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r
624 in (K.unStream m1) defState stop single yieldk
626------------------------------------------------------------------------------
627-- Transformation by Folding (Scans)
628------------------------------------------------------------------------------
630-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
631-- user supplied extraction function (the third argument) at each step. This is
632-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
637scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
642{-# DEPRECATED scan "Please use scanx instead." #-}
643scan :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
646-- | Like 'scanl'' but with a monadic step function.
649{-# INLINE scanlM' #-}
650scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
651scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
653-- | Strict left scan. Like 'foldl'', but returns the folded value at each
654-- step, generating a stream of all intermediate fold results. The first
655-- element of the stream is the user supplied initial value, and the last
656-- element of the stream is the same as the result of 'foldl''.
660scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
661scanl' step = scanlM' (\a b -> return (step a b))
663------------------------------------------------------------------------------
664-- Transformation by Filtering
665------------------------------------------------------------------------------
667-- | Include only those elements that pass a predicate.
671#if __GLASGOW_HASKELL__ != 802
672-- GHC 8.2.2 crashes with this code, when used with "stack"
673filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
674filter p m = fromStreamS $ S.filter p $ toStreamS m
676filter :: IsStream t => (a -> Bool) -> t m a -> t m a
680-- | Same as 'filter' but with a monadic predicate.
683{-# INLINE filterM #-}
684filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
685filterM p m = fromStreamD $ D.filterM p $ toStreamD m
687-- | Take first 'n' elements from the stream and discard the rest.
691take :: (IsStream t, Monad m) => Int -> t m a -> t m a
692take n m = fromStreamS $ S.take n $ toStreamS (maxYields (Just n) m)
694-- | End the stream as soon as the predicate fails on an element.
697{-# INLINE takeWhile #-}
698takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
699takeWhile p m = fromStreamS $ S.takeWhile p $ toStreamS m
701-- | Same as 'takeWhile' but with a monadic predicate.
704{-# INLINE takeWhileM #-}
705takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
706takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
708-- | Discard first 'n' elements from the stream and take the rest.
712drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
713drop n m = fromStreamS $ S.drop n $ toStreamS m
715-- | Drop elements in the stream as long as the predicate succeeds and then
716-- take the rest of the stream.
719{-# INLINE dropWhile #-}
720dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
721dropWhile p m = fromStreamS $ S.dropWhile p $ toStreamS m
723-- | Same as 'dropWhile' but with a monadic predicate.
726{-# INLINE dropWhileM #-}
727dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
728dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
730------------------------------------------------------------------------------
731-- Transformation by Mapping
732------------------------------------------------------------------------------
734-- | Replace each element of the stream with the result of a monadic action
735-- applied on the element.
738-- runStream $ S.replicateM 10 (return 1)
739-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
741-- runStream $ S.replicateM 10 (return 1)
742-- & (asyncly . S.mapM (\\x -> threadDelay 1000000 >> print x))
745-- /Concurrent (do not use with 'parallely' on infinite streams)/
748{-# INLINE_EARLY mapM #-}
749mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
752{-# RULES "mapM serial" mapM = mapMSerial #-}
753{-# INLINE mapMSerial #-}
754mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
755mapMSerial = Serial.mapM
757-- | Reduce a stream of monadic actions to a stream of the output of those
761-- runStream $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
762-- & (serially . S.sequence)
764-- runStream $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
765-- & (asyncly . S.sequence)
768-- /Concurrent (do not use with 'parallely' on infinite streams)/
771{-# INLINE sequence #-}
772sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
775------------------------------------------------------------------------------
776-- Transformation by Map and Filter
777------------------------------------------------------------------------------
779-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
780-- elements, and return a stream of values extracted from 'Just'.
783{-# INLINE mapMaybe #-}
784mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
785mapMaybe f m = fromStreamS $ S.mapMaybe f $ toStreamS m
787-- | Like 'mapMaybe' but maps a monadic function.
789-- /Concurrent (do not use with 'parallely' on infinite streams)/
792{-# INLINE mapMaybeM #-}
793mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
794 => (a -> m (Maybe b)) -> t m a -> t m b
795mapMaybeM f = fmap fromJust . filter isJust . mapM f
797------------------------------------------------------------------------------
798-- Transformation by Reordering
799------------------------------------------------------------------------------
801-- | Returns the elements of the stream in reverse order.
802-- The stream must be finite.
805reverse :: (IsStream t) => t m a -> t m a
806reverse m = fromStream $ go K.nil (toStream m)
808 go rev rest = K.Stream $ \st stp sng yld ->
809 let runIt x = K.unStream x (rstState st) stp sng yld
811 single a = runIt $ a `K.cons` rev
812 yieldk a r = runIt $ go (a `K.cons` rev) r
813 in K.unStream rest (rstState st) stop single yieldk
815------------------------------------------------------------------------------
817------------------------------------------------------------------------------
819-- | Zip two streams serially using a monadic zipping function.
822{-# INLINABLE zipWithM #-}
823zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
824zipWithM f m1 m2 = fromStreamS $ S.zipWithM f (toStreamS m1) (toStreamS m2)
826-- | Zip two streams serially using a pure zipping function.
829{-# INLINABLE zipWith #-}
830zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
831zipWith 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