@@ -47,11 +47,11 @@ module Streamly.Prelude
4747 (
4848 -- * Construction
4949 -- | Primitives to construct or inspect a stream.
50− nil
50+ K.nil
5151 , K.yield
5252 , K.yieldM
53− , cons
54− , (.:)
53+ , K.cons
54+ , (K..:)
5555 , consM
5656 , (|:)
5757
@@ -116,6 +116,7 @@ module Streamly.Prelude
116116 , reverse
117117
118118 -- * Mapping
119+ , Serial.map
119120 , mapM
120121 , mapMaybe
121122 , mapMaybeM
@@ -140,25 +141,24 @@ module Streamly.Prelude
140141 )
141142where
142143
143−import Control.Monad (void)
144−import Control.Monad.IO.Class (MonadIO(..))
145−import Data.Maybe (isJust, fromJust)
146−import Prelude hiding (filter, drop, dropWhile, take,
147− takeWhile, zipWith, foldr, foldl,
148− mapM, mapM_, sequence, all, any,
149− sum, product, elem, notElem,
150− maximum, minimum, head, last,
151− tail, length, null, reverse,
152− iterate)
144+import Control.Monad.IO.Class (MonadIO(..))
145+import Data.Maybe (isJust, fromJust)
146+import Prelude
147+ hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
148+ foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem,
149+ notElem, maximum, minimum, head, last, tail, length, null,
150+ reverse, iterate)
153151import qualified Prelude
154152import qualified System.IO as IO
155153
156−import Streamly.Streams.StreamK (IsStream(..), cons, nil, (.:))
154+import Streamly.SVar (MonadAsync)
155+import Streamly.Streams.StreamK (IsStream(..))
156+import Streamly.Streams.Serial (SerialT)
157+import Streamly.Streams.Zip (zipWith, zipWithM, zipAsyncWith, zipAsyncWithM)
158+
157159import qualified Streamly.Streams.StreamK as K
158160import qualified Streamly.Streams.StreamD as D
159−import Streamly.Streams.Zip
160−import Streamly.Streams.Serial
161−import Streamly.SVar (MonadAsync)
161+import qualified Streamly.Streams.Serial as Serial
162162
163163------------------------------------------------------------------------------
164164-- Construction
@@ -260,7 +260,7 @@ fromListM = fromStream . D.toStreamK . D.fromListM
260260-- @since 0.3.0
261261{-# INLINE fromFoldableM #-}
262262fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
263−fromFoldableM = Prelude.foldr consM nil
263+fromFoldableM = Prelude.foldr consM K.nil
264264
265265-- | Same as 'fromFoldable'.
266266--
@@ -284,7 +284,7 @@ each = K.fromFoldable
284284replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
285285replicateM n m = go n
286286 where
287− go cnt = if cnt <= 0 then nil else m |: go (cnt - 1)
287+ go cnt = if cnt <= 0 then K.nil else m |: go (cnt - 1)
288288
289289-- | Generate a stream by repeatedly executing a monadic action forever.
290290--
@@ -360,14 +360,11 @@ fromHandle h = fromStream go
360360-- @
361361--
362362-- @since 0.1.0
363+{-# INLINE foldr #-}
363364foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
364−foldr step acc m = go (toStream m)
365− where
366− go m1 =
367− let stop = return acc
368− single a = return (step a acc)
369− yieldk a r = go r >>= \b -> return (step a b)
370− in (K.unStream m1) Nothing stop single yieldk
365+-- XXX somehow this definition does not perform well, need to investigate
366+-- foldr step acc m = D.foldr step acc $ D.fromStreamK (toStream m)
367+foldr f = foldrM (\a b -> return (f a b))
371368
372369-- | Lazy right fold with a monadic step function. For example, to fold a
373370-- stream into a list:
@@ -380,13 +377,7 @@ foldr step acc m = go (toStream m)
380377-- @since 0.2.0
381378{-# INLINE foldrM #-}
382379foldrM :: Monad m => (a -> b -> m b) -> b -> SerialT m a -> m b
383−foldrM step acc m = go (toStream m)
384− where
385− go m1 =
386− let stop = return acc
387− single a = step a acc
388− yieldk a r = go r >>= step a
389− in (K.unStream m1) Nothing stop single yieldk
380+foldrM step acc m = D.foldrM step acc $ D.fromStreamK (toStream m)
390381
391382-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
392383-- user supplied extraction function (the third argument) at each step. This is
@@ -397,7 +388,7 @@ foldrM step acc m = go (toStream m)
397388{-# INLINE scanx #-}
398389scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
399390scanx step begin done m =
400− cons (done begin) $ fromStream $ go (toStream m) begin
391+ K.cons (done begin) $ fromStream $ go (toStream m) begin
401392 where
402393 go m1 !acc = K.Stream $ \_ stp sng yld ->
403394 let single a = sng (done $ step acc a)
@@ -430,24 +421,7 @@ scanl' step begin m = scanx step begin id m
430421-- @since 0.2.0
431422{-# INLINE foldx #-}
432423foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
433−foldx step begin done m = get $ go (toStream m) begin
434− where
435− {-# NOINLINE get #-}
436− get m1 =
437− let single = return . done
438− in (K.unStream m1) Nothing undefined single undefined
439−
440− -- Note, this can be implemented by making a recursive call to "go",
441− -- however that is more expensive because of unnecessary recursion
442− -- that cannot be tail call optimized. Unfolding recursion explicitly via
443− -- continuations is much more efficient.
444− go m1 !acc = K.Stream $ \_ _ sng yld ->
445− let stop = sng acc
446− single a = sng $ step acc a
447− yieldk a r =
448− let stream = go r (step acc a)
449− in (K.unStream stream) Nothing undefined sng yld
450− in (K.unStream m1) Nothing stop single yieldk
424+foldx = K.foldx
451425
452426-- |
453427-- @since 0.1.0
@@ -460,20 +434,14 @@ foldl = foldx
460434-- @since 0.2.0
461435{-# INLINE foldl' #-}
462436foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
463−foldl' step begin m = foldx step begin id m
437+foldl' step begin m = D.foldl' step begin $ D.fromStreamK (toStream m)
464438
465439-- XXX replace the recursive "go" with explicit continuations.
466440-- | Like 'foldx', but with a monadic step function.
467441--
468442-- @since 0.2.0
469443foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
470−foldxM step begin done m = go begin (toStream m)
471− where
472− go !acc m1 =
473− let stop = acc >>= done
474− single a = acc >>= \b -> step b a >>= done
475− yieldk a r = acc >>= \b -> go (step b a) r
476− in (K.unStream m1) Nothing stop single yieldk
444+foldxM = K.foldxM
477445
478446-- |
479447-- @since 0.1.0
@@ -485,7 +453,7 @@ foldlM = foldxM
485453--
486454-- @since 0.2.0
487455foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
488−foldlM' step begin m = foldxM step (return begin) return m
456+foldlM' step begin m = D.foldlM' step begin $ D.fromStreamK (toStream m)
489457
490458-- | Decompose a stream into its head and tail. If the stream is empty, returns
491459-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
@@ -495,7 +463,7 @@ foldlM' step begin m = foldxM step (return begin) return m
495463uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
496464uncons m =
497465 let stop = return Nothing
498− single a = return (Just (a, nil))
466+ single a = return (Just (a, K.nil))
499467 yieldk a r = return (Just (a, fromStream r))
500468 in (K.unStream (toStream m)) Nothing stop single yieldk
501469
@@ -518,9 +486,9 @@ toHandle h m = go (toStream m)
518486-- | Convert a stream into a list in the underlying monad.
519487--
520488-- @since 0.1.0
521−{-# INLINABLE toList #-}
489+{-# INLINE toList #-}
522490toList :: Monad m => SerialT m a -> m [a]
523−toList = foldrM (\a xs -> return (a : xs)) []
491+toList m = D.toList $ D.fromStreamK (toStream m)
524492
525493-- | Take first 'n' elements from the stream and discard the rest.
526494--
@@ -644,7 +612,7 @@ head m =
644612tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
645613tail m =
646614 let stop = return Nothing
647− single _ = return $ Just nil
615+ single _ = return $ Just K.nil
648616 yieldk _ r = return $ Just $ fromStream r
649617 in (K.unStream (toStream m)) Nothing stop single yieldk
650618
@@ -653,7 +621,7 @@ tail m =
653621-- @since 0.1.1
654622{-# INLINE last #-}
655623last :: Monad m => SerialT m a -> m (Maybe a)
656−last = foldl (\_ y -> Just y) Nothing id
624+last m = D.last $ D.fromStreamK (toStream m)
657625
658626-- | Determine whether the stream is empty.
659627--
@@ -761,14 +729,14 @@ maximum m = go Nothing (toStream m)
761729-- /Concurrent (do not use with 'parallely' on infinite streams)/
762730--
763731-- @since 0.1.0
764−{-# INLINE mapM #-}
732+{-# INLINE_EARLY mapM #-}
765733mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
766−mapM f m = go (toStream m)
767− where
768− go m1 = fromStream $ K.Stream $ \svr stp sng yld ->
769− let single a = f a >>= sng
770− yieldk a r = K.unStream (toStream (f a |: (go r))) svr stp sng yld
771− in (K.unStream m1) Nothing stp single yieldk
734+mapM = K.mapM
735+
736+{-# RULES "mapM serial" mapM = mapMSerial #-}
737+{-# INLINE mapMSerial #-}
738+mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
739+mapMSerial = Serial.mapM
772740
773741-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
774742-- elements, and return a stream of values extracted from 'Just'.
@@ -802,14 +770,9 @@ mapMaybeM f = fmap fromJust . filter isJust . mapM f
802770-- output of the action.
803771--
804772-- @since 0.1.0
773+{-# INLINE mapM_ #-}
805774mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()
806−mapM_ f m = go (toStream m)
807− where
808− go m1 =
809− let stop = return ()
810− single a = void (f a)
811− yieldk a r = f a >> go r
812− in (K.unStream m1) Nothing stop single yieldk
775+mapM_ f m = D.mapM_ f $ D.fromStreamK (toStream m)
813776
814777-- | Reduce a stream of monadic actions to a stream of the output of those
815778-- actions.