1{-# LANGUAGE BangPatterns #-}
2{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE FlexibleInstances #-}
4{-# LANGUAGE MultiParamTypeClasses #-}
5{-# LANGUAGE RankNTypes #-}
6{-# LANGUAGE UndecidableInstances #-} -- XXX
9-- Module : Streamly.Prelude
10-- Copyright : (c) 2017 Harendra Kumar
13-- Maintainer : harendra.kumar@gmail.com
14-- Stability : experimental
17-- This module is designed to be imported qualified:
20-- import qualified Streamly.Prelude as S
23-- Functions with the suffix @M@ are general functions that work on monadic
24-- arguments. The corresponding functions without the suffix @M@ work on pure
25-- arguments and can in general be derived from their monadic versions but are
26-- provided for convenience and for consistency with other pure APIs in the
29-- Functions having a 'MonadAsync' constraint work concurrently when used with
30-- appropriate stream type combinator. Please be careful to not use 'parallely'
31-- with infinite streams.
33-- Deconstruction and folds accept a 'SerialT' type instead of a polymorphic
34-- type to ensure that streams always have a concrete monomorphic type by
35-- default, reducing type errors. In case you want to use any other type of
36-- stream you can use one of the type combinators provided in the "Streamly"
37-- module to convert the stream type.
39module Streamly.Prelude
42 -- | Primitives to construct or inspect a stream.
49 -- * Generation by Unfolding
53 -- * Special Generation
54 -- | Generate a monadic stream from an input structure, a seed or a
55 -- generation function.
67 -- * Elimination by Folding
131import Control.Monad (void)
132import Control.Monad.IO.Class (MonadIO(..))
133import Data.Semigroup (Semigroup(..))
134import Data.Maybe (isJust, fromJust)
135import Prelude hiding (filter, drop, dropWhile, take,
136 takeWhile, zipWith, foldr, foldl,
137 mapM, mapM_, sequence, all, any,
138 sum, product, elem, notElem,
139 maximum, minimum, head, last,
140 tail, length, null, reverse,
142import qualified Prelude
143import qualified System.IO as IO
145import qualified Streamly.Core as S
146import Streamly.Core (Stream(Stream))
147import Streamly.Streams
149------------------------------------------------------------------------------
151------------------------------------------------------------------------------
153-- | Build a stream by unfolding a /pure/ step function starting from a seed.
154-- The step function returns the next element in the stream and the next seed
155-- value. When it is done it returns 'Nothing' and the stream ends. For
162-- else Just (b, b + 1)
163-- in toList $ unfoldr f 0
170{-# INLINE unfoldr #-}
171unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
172unfoldr step = fromStream . go
174 go s = Stream $ \_ stp _ yld ->
177 Just (a, b) -> yld a (go b)
179-- | Build a stream by unfolding a /monadic/ step function starting from a
180-- seed. The step function returns the next element in the stream and the next
181-- seed value. When it is done it returns 'Nothing' and the stream ends. For
187-- then return Nothing
188-- else print b >> return (Just (b, b + 1))
189-- in runStream $ unfoldrM f 0
197-- When run concurrently, the next unfold step can run concurrently with the
198-- processing of the output of the previous step. Note that more than one step
199-- cannot run concurrently as the next step depends on the output of the
203-- (asyncly $ S.unfoldrM (\\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0)
204-- & S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
210{-# INLINE unfoldrM #-}
211unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
214 go s = fromStream $ Stream $ \svr stp sng yld -> do
219 S.runStream (toStream (return a |: go b)) svr stp sng yld
221-- | Construct a stream from a 'Foldable' containing pure values.
224{-# INLINE fromFoldable #-}
225fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
226fromFoldable = Prelude.foldr cons nil
228-- | Construct a stream from a 'Foldable' containing monadic actions.
231-- runStream $ serially $ S.fromFoldableM $ replicate 10 (threadDelay 1000000 >> print 1)
232-- runStream $ asyncly $ S.fromFoldableM $ replicate 10 (threadDelay 1000000 >> print 1)
235-- /Concurrent (do not use with 'parallely' on infinite containers)/
238{-# INLINE fromFoldableM #-}
239fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
240fromFoldableM = Prelude.foldr consM nil
242-- | Same as 'fromFoldable'.
245{-# DEPRECATED each "Please use fromFoldable instead." #-}
247each :: (IsStream t, Foldable f) => f a -> t m a
250-- | Create a singleton stream by executing a monadic action once. Same as
251-- @m \`consM` nil@ but more efficient.
254-- > toList $ once getLine
260once :: (IsStream t, Monad m) => m a -> t m a
261once = fromStream . S.once
263-- | Generate a stream by performing a monadic action @n@ times.
267-- runStream $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
268-- runStream $ asyncly $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
274replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
277 go cnt = if cnt <= 0 then nil else m |: go (cnt - 1)
279-- | Generate a stream by repeatedly executing a monadic action forever.
282-- runStream $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
283-- runStream $ asyncly $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
286-- /Concurrent, infinite (do not use with 'parallely')/
289repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
291 where go m = m |: go m
293-- | Iterate a pure function from a seed value, streaming the results forever.
296iterate :: IsStream t => (a -> a) -> a -> t m a
297iterate step = fromStream . go
299 go s = S.cons s (go (step s))
301-- | Iterate a monadic function from a seed value, streaming the results
304-- When run concurrently, the next iteration can run concurrently with the
305-- processing of the previous iteration. Note that more than one iteration
306-- cannot run concurrently as the next iteration depends on the output of the
307-- previous iteration.
310-- runStream $ serially $ S.take 10 $ S.iterateM
311-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0
313-- runStream $ asyncly $ S.take 10 $ S.iterateM
314-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0
320iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a
323 go s = fromStream $ Stream $ \svr stp sng yld -> do
325 S.runStream (toStream (return s |: go next)) svr stp sng yld
327-- | Read lines from an IO Handle into a stream of Strings.
330fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
331fromHandle h = fromStream go
333 go = Stream $ \_ stp _ yld -> do
334 eof <- liftIO $ IO.hIsEOF h
338 str <- liftIO $ IO.hGetLine h
341------------------------------------------------------------------------------
343------------------------------------------------------------------------------
345-- | Lazy right associative fold. For example, to fold a stream into a list:
348-- >> runIdentity $ foldr (:) [] (serially $ fromFoldable [1,2,3])
353foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
354foldr step acc m = go (toStream m)
357 let stop = return acc
358 single a = return (step a acc)
359 yield a r = go r >>= \b -> return (step a b)
360 in (S.runStream m1) Nothing stop single yield
362-- | Lazy right fold with a monadic step function. For example, to fold a
363-- stream into a list:
366-- >> runIdentity $ foldrM (\\x xs -> return (x : xs)) [] (serially $ fromFoldable [1,2,3])
372foldrM :: Monad m => (a -> b -> m b) -> b -> SerialT m a -> m b
373foldrM step acc m = go (toStream m)
376 let stop = return acc
377 single a = step a acc
378 yield a r = go r >>= step a
379 in (S.runStream m1) Nothing stop single yield
381-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
382-- user supplied extraction function (the third argument) at each step. This is
383-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
388scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
389scanx step begin done m =
390 cons (done begin) $ fromStream $ go (toStream m) begin
392 go m1 !acc = Stream $ \_ stp sng yld ->
393 let single a = sng (done $ step acc a)
396 in yld (done s) (go r s)
397 in S.runStream m1 Nothing stp single yield
401{-# DEPRECATED scan "Please use scanx instead." #-}
402scan :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
405-- | Strict left scan. Like 'foldl'', but returns the folded value at each
406-- step, generating a stream of all intermediate fold results. The first
407-- element of the stream is the user supplied initial value, and the last
408-- element of the stream is the same as the result of 'foldl''.
412scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
413scanl' step begin m = scanx step begin id m
415-- | Strict left fold with an extraction function. Like the standard strict
416-- left fold, but applies a user supplied extraction function (the third
417-- argument) to the folded value at the end. This is designed to work with the
418-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
422foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
423foldx step begin done m = get $ go (toStream m) begin
427 let single = return . done
428 in (S.runStream m1) Nothing undefined single undefined
430 -- Note, this can be implemented by making a recursive call to "go",
431 -- however that is more expensive because of unnecessary recursion
432 -- that cannot be tail call optimized. Unfolding recursion explicitly via
433 -- continuations is much more efficient.
434 go m1 !acc = Stream $ \_ _ sng yld ->
436 single a = sng $ step acc a
438 let stream = go r (step acc a)
439 in (S.runStream stream) Nothing undefined sng yld
440 in (S.runStream m1) Nothing stop single yield
444{-# DEPRECATED foldl "Please use foldx instead." #-}
445foldl :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
448-- | Strict left associative fold.
452foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
453foldl' step begin m = foldx step begin id m
455-- XXX replace the recursive "go" with explicit continuations.
456-- | Like 'foldx', but with a monadic step function.
459foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
460foldxM step begin done m = go begin (toStream m)
463 let stop = acc >>= done
464 single a = acc >>= \b -> step b a >>= done
465 yield a r = acc >>= \b -> go (step b a) r
466 in (S.runStream m1) Nothing stop single yield
470{-# DEPRECATED foldlM "Please use foldxM instead." #-}
471foldlM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
474-- | Like 'foldl'' but with a monadic step function.
477foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
478foldlM' step begin m = foldxM step (return begin) return m
480-- | Decompose a stream into its head and tail. If the stream is empty, returns
481-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
482-- the head of the stream and @ma@ its tail.
485uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
487 let stop = return Nothing
488 single a = return (Just (a, nil))
489 yield a r = return (Just (a, fromStream r))
490 in (S.runStream (toStream m)) Nothing stop single yield
492-- | Write a stream of Strings to an IO Handle.
495toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
496toHandle h m = go (toStream m)
500 single a = liftIO (IO.hPutStrLn h a)
501 yield a r = liftIO (IO.hPutStrLn h a) >> go r
502 in (S.runStream m1) Nothing stop single yield
504------------------------------------------------------------------------------
506------------------------------------------------------------------------------
508-- | Convert a stream into a list in the underlying monad.
511{-# INLINABLE toList #-}
512toList :: Monad m => SerialT m a -> m [a]
513toList = foldrM (\a xs -> return (a : xs)) []
515-- | Take first 'n' elements from the stream and discard the rest.
519take :: IsStream t => Int -> t m a -> t m a
520take n m = fromStream $ go n (toStream m)
522 go n1 m1 = Stream $ \_ stp sng yld ->
523 let yield a r = yld a (go (n1 - 1) r)
524 in if n1 <= 0 then stp else (S.runStream m1) Nothing stp sng yield
526-- | Include only those elements that pass a predicate.
530filter :: IsStream t => (a -> Bool) -> t m a -> t m a
531filter p m = fromStream $ go (toStream m)
533 go m1 = Stream $ \_ stp sng yld ->
534 let single a | p a = sng a
536 yield a r | p a = yld a (go r)
537 | otherwise = (S.runStream r) Nothing stp single yield
538 in (S.runStream m1) Nothing stp single yield
540-- | End the stream as soon as the predicate fails on an element.
543{-# INLINE takeWhile #-}
544takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
545takeWhile p m = fromStream $ go (toStream m)
547 go m1 = Stream $ \_ stp sng yld ->
548 let single a | p a = sng a
550 yield a r | p a = yld a (go r)
552 in (S.runStream m1) Nothing stp single yield
554-- | Discard first 'n' elements from the stream and take the rest.
557drop :: IsStream t => Int -> t m a -> t m a
558drop n m = fromStream $ go n (toStream m)
560 go n1 m1 = Stream $ \_ stp sng yld ->
562 yield _ r = (S.runStream $ go (n1 - 1) r) Nothing stp sng yld
563 -- Somehow "<=" check performs better than a ">"
565 then (S.runStream m1) Nothing stp sng yld
566 else (S.runStream m1) Nothing stp single yield
568-- | Drop elements in the stream as long as the predicate succeeds and then
569-- take the rest of the stream.
572{-# INLINE dropWhile #-}
573dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
574dropWhile p m = fromStream $ go (toStream m)
576 go m1 = Stream $ \_ stp sng yld ->
577 let single a | p a = stp
579 yield a r | p a = (S.runStream r) Nothing stp single yield
580 | otherwise = yld a r
581 in (S.runStream m1) Nothing stp single yield
583-- | Determine whether all elements of a stream satisfy a predicate.
586all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
587all p m = go (toStream m)
590 let single a | p a = return True
591 | otherwise = return False
592 yield a r | p a = go r
593 | otherwise = return False
594 in (S.runStream m1) Nothing (return True) single yield
596-- | Determine whether any of the elements of a stream satisfy a predicate.
599any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
600any p m = go (toStream m)
603 let single a | p a = return True
604 | otherwise = return False
605 yield a r | p a = return True
607 in (S.runStream m1) Nothing (return False) single yield
609-- | Determine the sum of all elements of a stream of numbers
612sum :: (Monad m, Num a) => SerialT m a -> m a
615-- | Determine the product of all elements of a stream of numbers
618product :: (Monad m, Num a) => SerialT m a -> m a
619product = foldl (*) 1 id
621-- | Extract the first element of the stream, if any.
624head :: Monad m => SerialT m a -> m (Maybe a)
626 let stop = return Nothing
627 single a = return (Just a)
628 yield a _ = return (Just a)
629 in (S.runStream (toStream m)) Nothing stop single yield
631-- | Extract all but the first element of the stream, if any.
634tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
636 let stop = return Nothing
637 single _ = return $ Just nil
638 yield _ r = return $ Just $ fromStream r
639 in (S.runStream (toStream m)) Nothing stop single yield
641-- | Extract the last element of the stream, if any.
645last :: Monad m => SerialT m a -> m (Maybe a)
646last = foldl (\_ y -> Just y) Nothing id
648-- | Determine whether the stream is empty.
651null :: Monad m => SerialT m a -> m Bool
653 let stop = return True
654 single _ = return False
655 yield _ _ = return False
656 in (S.runStream (toStream m)) Nothing stop single yield
658-- | Determine whether an element is present in the stream.
661elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
662elem e m = go (toStream m)
665 let stop = return False
666 single a = return (a == e)
667 yield a r = if a == e then return True else go r
668 in (S.runStream m1) Nothing stop single yield
670-- | Determine whether an element is not present in the stream.
673notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
674notElem e m = go (toStream m)
677 let stop = return True
678 single a = return (a /= e)
679 yield a r = if a == e then return False else go r
680 in (S.runStream m1) Nothing stop single yield
682-- | Determine the length of the stream.
685length :: Monad m => SerialT m a -> m Int
686length = foldl (\n _ -> n + 1) 0 id
688-- | Returns the elements of the stream in reverse order.
689-- The stream must be finite.
692reverse :: (IsStream t) => t m a -> t m a
693reverse m = fromStream $ go S.nil (toStream m)
695 go rev rest = Stream $ \_ stp sng yld ->
696 let run x = S.runStream x Nothing stp sng yld
698 single a = run $ a `S.cons` rev
699 yield a r = run $ go (a `S.cons` rev) r
700 in S.runStream rest Nothing stop single yield
702-- XXX replace the recursive "go" with continuation
703-- | Determine the minimum element in a stream.
706minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
707minimum m = go Nothing (toStream m)
710 let stop = return res
711 single a = return $ min_ a res
712 yield a r = go (min_ a res) r
713 in (S.runStream m1) Nothing stop single yield
715 min_ a res = case res of
717 Just e -> Just $ min a e
719-- XXX replace the recursive "go" with continuation
720-- | Determine the maximum element in a stream.
723maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
724maximum m = go Nothing (toStream m)
727 let stop = return res
728 single a = return $ max_ a res
729 yield a r = go (max_ a res) r
730 in (S.runStream m1) Nothing stop single yield
732 max_ a res = case res of
734 Just e -> Just $ max a e
736------------------------------------------------------------------------------
738------------------------------------------------------------------------------
740-- | Replace each element of the stream with the result of a monadic action
741-- applied on the element.
744-- runStream $ S.replicateM 10 (return 1)
745-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
747-- runStream $ S.replicateM 10 (return 1)
748-- & (asyncly . S.mapM (\\x -> threadDelay 1000000 >> print x))
751-- /Concurrent (do not use with 'parallely' on infinite streams)/
755mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
756mapM f m = go (toStream m)
758 go m1 = fromStream $ Stream $ \svr stp sng yld ->
759 let single a = f a >>= sng
760 yield a r = S.runStream (toStream (f a |: (go r))) svr stp sng yld
761 in (S.runStream m1) Nothing stp single yield
763-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
764-- elements, and return a stream of values extracted from 'Just'.
767{-# INLINE mapMaybe #-}
768mapMaybe :: (IsStream t) => (a -> Maybe b) -> t m a -> t m b
769mapMaybe f m = go (toStream m)
771 go m1 = fromStream $ Stream $ \_ stp sng yld ->
772 let single a = case f a of
775 yield a r = case f a of
776 Just b -> yld b (toStream $ go r)
777 Nothing -> (S.runStream r) Nothing stp single yield
778 in (S.runStream m1) Nothing stp single yield
780-- | Like 'mapMaybe' but maps a monadic function.
782-- /Concurrent (do not use with 'parallely' on infinite streams)/
785{-# INLINE mapMaybeM #-}
786mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
787 => (a -> m (Maybe b)) -> t m a -> t m b
788mapMaybeM f = fmap fromJust . filter isJust . mapM f
790-- XXX this can utilize parallel mapping if we implement it as runStream . mapM
791-- | Apply a monadic action to each element of the stream and discard the
792-- output of the action.
795mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()
796mapM_ f m = go (toStream m)
800 single a = void (f a)
801 yield a r = f a >> go r
802 in (S.runStream m1) Nothing stop single yield
804-- | Reduce a stream of monadic actions to a stream of the output of those
808-- runStream $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
809-- & (serially . S.sequence)
811-- runStream $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
812-- & (asyncly . S.sequence)
815-- /Concurrent (do not use with 'parallely' on infinite streams)/
818sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
819sequence m = go (toStream m)
821 go m1 = fromStream $ Stream $ \svr stp sng yld ->
822 let single ma = ma >>= sng
823 yield ma r = S.runStream (toStream $ ma |: go r) svr stp sng yld
824 in (S.runStream m1) Nothing stp single yield
826------------------------------------------------------------------------------
827-- Serially Zipping Streams
828------------------------------------------------------------------------------
830-- | Zip two streams serially using a pure zipping function.
833zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
834zipWith f m1 m2 = fromStream $ S.zipWith f (toStream m1) (toStream m2)
836-- | Zip two streams serially using a monadic zipping function.
839zipWithM :: IsStream t => (a -> b -> t m c) -> t m a -> t m b -> t m c
840zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
842 go mx my = Stream $ \_ stp sng yld -> do
844 let run x = S.runStream x Nothing stp sng yld
845 single2 b = run $ toStream (f a b)
846 yield2 b rb = run $ toStream (f a b) <> go ra rb
847 in (S.runStream my) Nothing stp single2 yield2
848 let single1 a = merge a S.nil
849 yield1 a ra = merge a ra
850 (S.runStream mx) Nothing stp single1 yield1
852------------------------------------------------------------------------------
853-- Parallely Zipping Streams
854------------------------------------------------------------------------------
856-- | Zip two streams concurrently (i.e. both the elements being zipped are
857-- generated concurrently) using a pure zipping function.
860zipAsyncWith :: (IsStream t, MonadAsync m)
861 => (a -> b -> c) -> t m a -> t m b -> t m c
862zipAsyncWith f m1 m2 =
863 fromStream $ S.zipAsyncWith f (toStream m1) (toStream m2)
865-- | Zip two streams asyncly (i.e. both the elements being zipped are generated
866-- concurrently) using a monadic zipping function.
869zipAsyncWithM :: (IsStream t, MonadAsync m)
870 => (a -> b -> t m c) -> t m a -> t m b -> t m c
871zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do
874 (S.runStream (toStream (zipWithM f ma mb))) Nothing stp sng yld
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