@@ -6,6 +6,7 @@
66{-# LANGUAGE InstanceSigs #-}
77{-# LANGUAGE MultiParamTypeClasses #-}
88{-# LANGUAGE RankNTypes #-}
9+{-# LANGUAGE ScopedTypeVariables #-}
910{-# LANGUAGE UndecidableInstances #-} -- XXX
1011
1112-- |
@@ -33,17 +34,21 @@ module Streamly.Streams.StreamK
3334
3435 -- * The stream type
3536 , Stream
36− , unStream
37− , unStreamShared
38− , unShare
39− , runStreamSVar
4037
41− -- * Construction
38+ -- * Construction Primitives
4239 , mkStream
4340 , nil
4441 , cons
4542 , (.:)
4643
44+ -- * Elimination Primitives
45+ , foldStream
46+ , foldStreamShared
47+ , foldStreamSVar
48+
49+ -- * Transformation Primitives
50+ , unShare
51+
4752 -- * Asynchronous construction
4853 , nilK
4954 , yieldK
@@ -71,7 +76,6 @@ module Streamly.Streams.StreamK
7176
7277 -- * Elimination
7378 -- ** General Folds
74− , foldStream
7579 , foldr
7680 , foldrM
7781 , foldr1
@@ -174,9 +178,9 @@ import Streamly.Streams.StreamKType
174178
175179-- | Detach a stream from an SVar
176180{-# INLINE unShare #-}
177−unShare :: Stream m a -> Stream m a
181+unShare :: IsStream t => t m a -> t m a
178182unShare x = mkStream $ \st stp sng yld ->
179− unStream x st stp sng yld
183+ foldStream st stp sng yld x
180184
181185------------------------------------------------------------------------------
182186-- Construction
@@ -197,7 +201,7 @@ infixr 5 `cons`
197201--
198202-- @since 0.1.0
199203cons :: IsStream t => a -> t m a -> t m a
200−cons a r = fromStream $ mkStream $ \_ _ _ yld -> yld a (toStream r)
204+cons a r = mkStream $ \_ _ _ yld -> yld a r
201205
202206infixr 5 .:
203207
@@ -218,15 +222,15 @@ infixr 5 .:
218222
219223-- | Make an empty stream from a callback function.
220224nilK :: IsStream t => (forall r. m r -> m r) -> t m a
221−nilK k = fromStream $ mkStream $ \_ stp _ _ -> k stp
225+nilK k = mkStream $ \_ stp _ _ -> k stp
222226
223227-- | Make a singleton stream from a one shot callback function.
224228yieldK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
225−yieldK k = fromStream $ mkStream $ \_ _ sng _ -> k sng
229+yieldK k = mkStream $ \_ _ sng _ -> k sng
226230
227231-- | Construct a stream from a callback function.
228232consK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a -> t m a
229−consK k r = fromStream $ mkStream $ \_ _ _ yld -> k (\x -> yld x (toStream r))
233+consK k r = mkStream $ \_ _ _ yld -> k (\x -> yld x r)
230234
231235-- XXX consK with concurrent callbacks
232236-- XXX Build a stream from a repeating callback function.
@@ -240,8 +244,8 @@ uncons :: (IsStream t, Monad m) => t m a -> m (Maybe (a, t m a))
240244uncons m =
241245 let stop = return Nothing
242246 single a = return (Just (a, nil))
243− yieldk a r = return (Just (a, fromStream r))
244− in unStream (toStream m) defState stop single yieldk
247+ yieldk a r = return (Just (a, r))
248+ in foldStream defState stop single yieldk m
245249
246250-------------------------------------------------------------------------------
247251-- Generation
@@ -249,7 +253,7 @@ uncons m =
249253
250254{-# INLINE unfoldr #-}
251255unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
252−unfoldr step = fromStream . go
256+unfoldr step = go
253257 where
254258 go s = mkStream $ \_ stp _ yld ->
255259 case step s of
@@ -260,19 +264,19 @@ unfoldr step = fromStream . go
260264unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
261265unfoldrM step = go
262266 where
263− go s = fromStream $ mkStream $ \svr stp sng yld -> do
267+ go s = mkStream $ \st stp sng yld -> do
264268 mayb <- step s
265269 case mayb of
266270 Nothing -> stp
267271 Just (a, b) ->
268− unStreamShared (toStream (return a |: go b)) svr stp sng yld
272+ foldStreamShared st stp sng yld $ return a |: go b
269273
270274-------------------------------------------------------------------------------
271275-- Special generation
272276-------------------------------------------------------------------------------
273277
274278yield :: IsStream t => a -> t m a
275−yield a = fromStream $ mkStream $ \_ _ single _ -> single a
279+yield a = mkStream $ \_ _ single _ -> single a
276280
277281-- | Same as yieldM
278282--
@@ -325,8 +329,8 @@ fromList :: IsStream t => [a] -> t m a
325329fromList = fromFoldable
326330
327331{-# INLINE fromStreamK #-}
328−fromStreamK :: Stream m a -> Stream m a
329−fromStreamK = id
332+fromStreamK :: IsStream t => Stream m a -> t m a
333+fromStreamK = fromStream
330334
331335-------------------------------------------------------------------------------
332336-- Elimination by Folding
@@ -334,24 +338,24 @@ fromStreamK = id
334338
335339-- | Lazy right associative fold.
336340foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b
337−foldr step acc m = go (toStream m)
341+foldr step acc m = go m
338342 where
339343 go m1 =
340344 let stop = return acc
341345 single a = return (step a acc)
342346 yieldk a r = go r >>= \b -> return (step a b)
343− in unStream m1 defState stop single yieldk
347+ in foldStream defState stop single yieldk m1
344348
345349-- | Lazy right fold with a monadic step function.
346350{-# INLINE foldrM #-}
347351foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b
348−foldrM step acc m = go (toStream m)
352+foldrM step acc m = go m
349353 where
350354 go m1 =
351355 let stop = return acc
352356 single a = step a acc
353357 yieldk a r = go r >>= step a
354− in unStream m1 defState stop single yieldk
358+ in foldStream defState stop single yieldk m1
355359
356360{-# INLINE foldr1 #-}
357361foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)
@@ -359,13 +363,13 @@ foldr1 step m = do
359363 r <- uncons m
360364 case r of
361365 Nothing -> return Nothing
362− Just (h, t) -> fmap Just (go h (toStream t))
366+ Just (h, t) -> fmap Just (go h t)
363367 where
364368 go p m1 =
365369 let stp = return p
366370 single a = return $ step a p
367371 yieldk a r = fmap (step p) (go a r)
368− in unStream m1 defState stp single yieldk
372+ in foldStream defState stp single yieldk m1
369373
370374-- | Strict left fold with an extraction function. Like the standard strict
371375-- left fold, but applies a user supplied extraction function (the third
@@ -374,28 +378,31 @@ foldr1 step m = do
374378--
375379-- Note that the accumulator is always evaluated including the initial value.
376380{-# INLINE foldx #-}
377−foldx :: (IsStream t, Monad m)
381+foldx :: forall t m a b x. (IsStream t, Monad m)
378382 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
379−foldx step begin done m = get $ go (toStream m) begin
383+foldx step begin done m = get $ go m begin
380384 where
381385 {-# NOINLINE get #-}
386+ get :: t m x -> m b
382387 get m1 =
383388 -- XXX we are not strictly evaluating the accumulator here. Is this
384389 -- okay?
385390 let single = return . done
386− in unStreamShared m1 undefined undefined single undefined
391+ -- XXX this is foldSingleton. why foldStreamShared?
392+ in foldStreamShared undefined undefined single undefined m1
387393
388394 -- Note, this can be implemented by making a recursive call to "go",
389395 -- however that is more expensive because of unnecessary recursion
390396 -- that cannot be tail call optimized. Unfolding recursion explicitly via
391397 -- continuations is much more efficient.
398+ go :: t m a -> x -> t m x
392399 go m1 !acc = mkStream $ \_ _ sng yld ->
393400 let stop = sng acc
394401 single a = sng $ step acc a
395− yieldk a r =
396− let stream = go r (step acc a)
397− in unStream stream defState undefined sng yld
398− in unStream m1 defState stop single yieldk
402+ -- XXX this is foldNonEmptyStream
403+ yieldk a r = foldStream defState undefined sng yld $
404+ go r (step acc a)
405+ in foldStream defState stop single yieldk m1
399406
400407-- | Strict left associative fold.
401408{-# INLINE foldl' #-}
@@ -406,13 +413,13 @@ foldl' step begin = foldx step begin id
406413-- | Like 'foldx', but with a monadic step function.
407414foldxM :: (IsStream t, Monad m)
408415 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b
409−foldxM step begin done m = go begin (toStream m)
416+foldxM step begin done m = go begin m
410417 where
411418 go !acc m1 =
412419 let stop = acc >>= done
413420 single a = acc >>= \b -> step b a >>= done
414421 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r
415− in unStream m1 defState stop single yieldk
422+ in foldStream defState stop single yieldk m1
416423
417424-- | Like 'foldl'' but with a monadic step function.
418425foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
@@ -424,13 +431,13 @@ foldlM' step begin = foldxM step (return begin) return
424431
425432{-# INLINE runStream #-}
426433runStream :: (Monad m, IsStream t) => t m a -> m ()
427−runStream m = go (toStream m)
434+runStream = go
428435 where
429436 go m1 =
430437 let stop = return ()
431438 single _ = return ()
432− yieldk _ r = go (toStream r)
433− in unStream m1 defState stop single yieldk
439+ yieldk _ r = go r
440+ in foldStream defState stop single yieldk m1
434441
435442{-# INLINE null #-}
436443null :: (IsStream t, Monad m) => t m a -> m Bool
@@ -438,7 +445,7 @@ null m =
438445 let stop = return True
439446 single _ = return False
440447 yieldk _ _ = return False
441− in unStream (toStream m) defState stop single yieldk
448+ in foldStream defState stop single yieldk m
442449
443450{-# INLINE head #-}
444451head :: (IsStream t, Monad m) => t m a -> m (Maybe a)
@@ -446,69 +453,69 @@ head m =
446453 let stop = return Nothing
447454 single a = return (Just a)
448455 yieldk a _ = return (Just a)
449− in unStream (toStream m) defState stop single yieldk
456+ in foldStream defState stop single yieldk m
450457
451458{-# INLINE tail #-}
452459tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
453460tail m =
454461 let stop = return Nothing
455462 single _ = return $ Just nil
456− yieldk _ r = return $ Just $ fromStream r
457− in unStream (toStream m) defState stop single yieldk
463+ yieldk _ r = return $ Just r
464+ in foldStream defState stop single yieldk m
458465
459466{-# INLINE init #-}
460467init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
461−init m = go1 (toStream m)
468+init m = go1 m
462469 where
463470 go1 m1 = do
464471 r <- uncons m1
465472 case r of
466473 Nothing -> return Nothing
467− Just (h, t) -> return . Just . fromStream $ go h t
474+ Just (h, t) -> return . Just $ go h t
468475 go p m1 = mkStream $ \_ stp sng yld ->
469476 let single _ = sng p
470477 yieldk a x = yld p $ go a x
471− in unStream m1 defState stp single yieldk
478+ in foldStream defState stp single yieldk m1
472479
473480{-# INLINE elem #-}
474481elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
475−elem e m = go (toStream m)
482+elem e m = go m
476483 where
477484 go m1 =
478485 let stop = return False
479486 single a = return (a == e)
480487 yieldk a r = if a == e then return True else go r
481− in unStream m1 defState stop single yieldk
488+ in foldStream defState stop single yieldk m1
482489
483490{-# INLINE notElem #-}
484491notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
485−notElem e m = go (toStream m)
492+notElem e m = go m
486493 where
487494 go m1 =
488495 let stop = return True
489496 single a = return (a /= e)
490497 yieldk a r = if a == e then return False else go r
491− in unStream m1 defState stop single yieldk
498+ in foldStream defState stop single yieldk m1
492499
493500all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
494−all p m = go (toStream m)
501+all p m = go m
495502 where
496503 go m1 =
497504 let single a | p a = return True
498505 | otherwise = return False
499506 yieldk a r | p a = go r
500507 | otherwise = return False
501− in unStream m1 defState (return True) single yieldk
508+ in foldStream defState (return True) single yieldk m1
502509
503510any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
504−any p m = go (toStream m)
511+any p m = go m
505512 where
506513 go m1 =
507514 let single a | p a = return True
508515 | otherwise = return False
509516 yieldk a r | p a = return True
510517 | otherwise = go r
511− in unStream m1 defState (return False) single yieldk
518+ in foldStream defState (return False) single yieldk m1
512519
513520-- | Extract the last element of the stream, if any.
514521{-# INLINE last #-}
@@ -517,13 +524,13 @@ last = foldx (\_ y -> Just y) Nothing id
517524
518525{-# INLINE minimum #-}
519526minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
520−minimum m = go Nothing (toStream m)
527+minimum m = go Nothing m
521528 where
522529 go Nothing m1 =
523530 let stop = return Nothing
524531 single a = return (Just a)
525532 yieldk a r = go (Just a) r
526− in unStream m1 defState stop single yieldk
533+ in foldStream defState stop single yieldk m1
527534
528535 go (Just res) m1 =
529536 let stop = return (Just res)
@@ -535,19 +542,19 @@ minimum m = go Nothing (toStream m)
535542 if res <= a
536543 then go (Just res) r
537544 else go (Just a) r
538− in unStream m1 defState stop single yieldk
545+ in foldStream defState stop single yieldk m1
539546
540547{-# INLINE minimumBy #-}
541548minimumBy
542549 :: (IsStream t, Monad m)
543550 => (a -> a -> Ordering) -> t m a -> m (Maybe a)
544−minimumBy cmp m = go Nothing (toStream m)
551+minimumBy cmp m = go Nothing m
545552 where
546553 go Nothing m1 =
547554 let stop = return Nothing
548555 single a = return (Just a)
549556 yieldk a r = go (Just a) r
550− in unStream m1 defState stop single yieldk
557+ in foldStream defState stop single yieldk m1
551558
552559 go (Just res) m1 =
553560 let stop = return (Just res)
@@ -557,17 +564,17 @@ minimumBy cmp m = go Nothing (toStream m)
557564 yieldk a r = case cmp res a of
558565 GT -> go (Just a) r
559566 _ -> go (Just res) r
560− in unStream m1 defState stop single yieldk
567+ in foldStream defState stop single yieldk m1
561568
562569{-# INLINE maximum #-}
563570maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
564−maximum m = go Nothing (toStream m)
571+maximum m = go Nothing m
565572 where
566573 go Nothing m1 =
567574 let stop = return Nothing
568575 single a = return (Just a)
569576 yieldk a r = go (Just a) r
570− in unStream m1 defState stop single yieldk
577+ in foldStream defState stop single yieldk m1
571578
572579 go (Just res) m1 =
573580 let stop = return (Just res)
@@ -579,17 +586,17 @@ maximum m = go Nothing (toStream m)
579586 if res <= a
580587 then go (Just a) r
581588 else go (Just res) r
582− in unStream m1 defState stop single yieldk
589+ in foldStream defState stop single yieldk m1
583590
584591{-# INLINE maximumBy #-}
585592maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)
586−maximumBy cmp m = go Nothing (toStream m)
593+maximumBy cmp m = go Nothing m
587594 where
588595 go Nothing m1 =
589596 let stop = return Nothing
590597 single a = return (Just a)
591598 yieldk a r = go (Just a) r
592− in unStream m1 defState stop single yieldk
599+ in foldStream defState stop single yieldk m1
593600
594601 go (Just res) m1 =
595602 let stop = return (Just res)
@@ -599,11 +606,11 @@ maximumBy cmp m = go Nothing (toStream m)
599606 yieldk a r = case cmp res a of
600607 GT -> go (Just res) r
601608 _ -> go (Just a) r
602− in unStream m1 defState stop single yieldk
609+ in foldStream defState stop single yieldk m1
603610
604611{-# INLINE (!!) #-}
605612(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)
606−m !! i = go i (toStream m)
613+m !! i = go i m
607614 where
608615 go n m1 =
609616 let single a | n == 0 = return $ Just a
@@ -611,22 +618,22 @@ m !! i = go i (toStream m)
611618 yieldk a x | n < 0 = return Nothing
612619 | n == 0 = return $ Just a
613620 | otherwise = go (n - 1) x
614− in unStream m1 defState (return Nothing) single yieldk
621+ in foldStream defState (return Nothing) single yieldk m1
615622
616623{-# INLINE lookup #-}
617624lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)
618−lookup e m = go (toStream m)
625+lookup e m = go m
619626 where
620627 go m1 =
621628 let single (a, b) | a == e = return $ Just b
622629 | otherwise = return Nothing
623630 yieldk (a, b) x | a == e = return $ Just b
624631 | otherwise = go x
625− in unStream m1 defState (return Nothing) single yieldk
632+ in foldStream defState (return Nothing) single yieldk m1
626633
627634{-# INLINE findM #-}
628635findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)
629−findM p m = go (toStream m)
636+findM p m = go m
630637 where
631638 go m1 =
632639 let single a = do
@@ -635,7 +642,7 @@ findM p m = go (toStream m)
635642 yieldk a x = do
636643 b <- p a
637644 if b then return $ Just a else go x
638− in unStream m1 defState (return Nothing) single yieldk
645+ in foldStream defState (return Nothing) single yieldk m1
639646
640647{-# INLINE find #-}
641648find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)
@@ -643,14 +650,15 @@ find p = findM (return . p)
643650
644651{-# INLINE findIndices #-}
645652findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int
646−findIndices p = fromStream . go 0 . toStream
653+findIndices p = go 0
647654 where
648655 go offset m1 = mkStream $ \st stp sng yld ->
649656 let single a | p a = sng offset
650657 | otherwise = stp
651658 yieldk a x | p a = yld offset $ go (offset + 1) x
652− | otherwise = unStream (go (offset + 1) x) st stp sng yld
653− in unStream m1 st stp single yieldk
659+ | otherwise = foldStream (adaptState st) stp sng yld $
660+ go (offset + 1) x
661+ in foldStream (adaptState st) stp single yieldk m1
654662
655663------------------------------------------------------------------------------
656664-- Map and Fold
@@ -659,13 +667,13 @@ findIndices p = fromStream . go 0 . toStream
659667-- | Apply a monadic action to each element of the stream and discard the
660668-- output of the action.
661669mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()
662−mapM_ f m = go (toStream m)
670+mapM_ f m = go m
663671 where
664672 go m1 =
665673 let stop = return ()
666674 single a = void (f a)
667675 yieldk a r = f a >> go r
668− in unStream m1 defState stop single yieldk
676+ in foldStream defState stop single yieldk m1
669677
670678------------------------------------------------------------------------------
671679-- Converting folds
@@ -686,14 +694,14 @@ toStreamK = id
686694{-# INLINE scanx #-}
687695scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
688696scanx step begin done m =
689− cons (done begin) $ fromStream $ go (toStream m) begin
697+ cons (done begin) $ go m begin
690698 where
691699 go m1 !acc = mkStream $ \st stp sng yld ->
692700 let single a = sng (done $ step acc a)
693701 yieldk a r =
694702 let s = step acc a
695703 in yld (done s) (go r s)
696− in unStream m1 st stp single yieldk
704+ in foldStream (adaptState st) stp single yieldk m1
697705
698706{-# INLINE scanl' #-}
699707scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
@@ -705,58 +713,58 @@ scanl' step begin = scanx step begin id
705713
706714{-# INLINE filter #-}
707715filter :: IsStream t => (a -> Bool) -> t m a -> t m a
708−filter p m = fromStream $ go (toStream m)
716+filter p m = go m
709717 where
710718 go m1 = mkStream $ \st stp sng yld ->
711719 let single a | p a = sng a
712720 | otherwise = stp
713721 yieldk a r | p a = yld a (go r)
714− | otherwise = unStream r st stp single yieldk
715− in unStream m1 st stp single yieldk
722+ | otherwise = foldStream st stp single yieldk r
723+ in foldStream st stp single yieldk m1
716724
717725{-# INLINE take #-}
718726take :: IsStream t => Int -> t m a -> t m a
719−take n m = fromStream $ go n (toStream m)
727+take n m = go n m
720728 where
721729 go n1 m1 = mkStream $ \st stp sng yld ->
722730 let yieldk a r = yld a (go (n1 - 1) r)
723731 in if n1 <= 0
724732 then stp
725− else unStream m1 st stp sng yieldk
733+ else foldStream st stp sng yieldk m1
726734
727735{-# INLINE takeWhile #-}
728736takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
729−takeWhile p m = fromStream $ go (toStream m)
737+takeWhile p m = go m
730738 where
731739 go m1 = mkStream $ \st stp sng yld ->
732740 let single a | p a = sng a
733741 | otherwise = stp
734742 yieldk a r | p a = yld a (go r)
735743 | otherwise = stp
736− in unStream m1 st stp single yieldk
744+ in foldStream st stp single yieldk m1
737745
738746drop :: IsStream t => Int -> t m a -> t m a
739−drop n m = fromStream $ mkStream $ \st stp sng yld ->
740− unStream (go n (toStream m)) st stp sng yld
747+drop n m = mkStream $ \st stp sng yld ->
748+ foldStream st stp sng yld $ go n m
741749 where
742750 go n1 m1 = mkStream $ \st stp sng yld ->
743751 let single _ = stp
744− yieldk _ r = (unStream $ go (n1 - 1) r) st stp sng yld
752+ yieldk _ r = foldStream st stp sng yld $ go (n1 - 1) r
745753 -- Somehow "<=" check performs better than a ">"
746754 in if n1 <= 0
747− then unStream m1 st stp sng yld
748− else unStream m1 st stp single yieldk
755+ then foldStream st stp sng yld m1
756+ else foldStream st stp single yieldk m1
749757
750758{-# INLINE dropWhile #-}
751759dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
752−dropWhile p m = fromStream $ go (toStream m)
760+dropWhile p m = go m
753761 where
754762 go m1 = mkStream $ \st stp sng yld ->
755763 let single a | p a = stp
756764 | otherwise = sng a
757− yieldk a r | p a = unStream r st stp single yieldk
765+ yieldk a r | p a = foldStream st stp single yieldk r
758766 | otherwise = yld a r
759− in unStream m1 st stp single yieldk
767+ in foldStream st stp single yieldk m1
760768
761769-------------------------------------------------------------------------------
762770-- Mapping
@@ -766,23 +774,23 @@ dropWhile p m = fromStream $ go (toStream m)
766774-- other stream types to overload it.
767775{-# INLINE mapM #-}
768776mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
769−mapM f m = go (toStream m)
777+mapM f m = go m
770778 where
771− go m1 = fromStream $ mkStream $ \st stp sng yld ->
779+ go m1 = mkStream $ \st stp sng yld ->
772780 let single a = f a >>= sng
773− yieldk a r = unStreamShared (toStream (f a |: go r)) st stp sng yld
774− in unStream m1 st stp single yieldk
781+ yieldk a r = foldStreamShared st stp sng yld $ f a |: go r
782+ in foldStream (adaptState st) stp single yieldk m1
775783
776784-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
777785-- other stream types to overload it.
778786{-# INLINE sequence #-}
779787sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
780−sequence m = go (toStream m)
788+sequence m = go m
781789 where
782− go m1 = fromStream $ mkStream $ \st stp sng yld ->
790+ go m1 = mkStream $ \st stp sng yld ->
783791 let single ma = ma >>= sng
784− yieldk ma r = unStreamShared (toStream $ ma |: go r) st stp sng yld
785− in unStream m1 st stp single yieldk
792+ yieldk ma r = foldStreamShared st stp sng yld $ ma |: go r
793+ in foldStream (adaptState st) stp single yieldk m1
786794
787795-------------------------------------------------------------------------------
788796-- Inserting
@@ -790,19 +798,19 @@ sequence m = go (toStream m)
790798
791799{-# INLINE intersperseM #-}
792800intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
793−intersperseM a m = fromStream $ prependingStart (toStream m)
801+intersperseM a m = prependingStart m
794802 where
795803 prependingStart m1 = mkStream $ \st stp sng yld ->
796− let yieldk i x = unStreamShared (return i |: go x) st stp sng yld
797− in unStream m1 st stp sng yieldk
798− go m2 = fromStream $ mkStream $ \st stp sng yld ->
799− let single i = unStreamShared (a |: yield i) st stp sng yld
800− yieldk i x = unStreamShared (a |: return i |: go x) st stp sng yld
801− in unStream m2 st stp single yieldk
804+ let yieldk i x = foldStreamShared st stp sng yld $ return i |: go x
805+ in foldStream st stp sng yieldk m1
806+ go m2 = mkStream $ \st stp sng yld ->
807+ let single i = foldStreamShared st stp sng yld $ a |: yield i
808+ yieldk i x = foldStreamShared st stp sng yld $ a |: return i |: go x
809+ in foldStream st stp single yieldk m2
802810
803811{-# INLINE insertBy #-}
804812insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a
805−insertBy cmp x m = fromStream $ go (toStream m)
813+insertBy cmp x m = go m
806814 where
807815 go m1 = mkStream $ \st _ _ yld ->
808816 let single a = case cmp x a of
@@ -812,7 +820,7 @@ insertBy cmp x m = fromStream $ go (toStream m)
812820 yieldk a r = case cmp x a of
813821 GT -> yld a (go r)
814822 _ -> yld x (a `cons` r)
815− in unStream m1 st stop single yieldk
823+ in foldStream st stop single yieldk m1
816824
817825------------------------------------------------------------------------------
818826-- Deleting
@@ -820,14 +828,14 @@ insertBy cmp x m = fromStream $ go (toStream m)
820828
821829{-# INLINE deleteBy #-}
822830deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a
823−deleteBy eq x m = fromStream $ go (toStream m)
831+deleteBy eq x m = go m
824832 where
825833 go m1 = mkStream $ \st stp sng yld ->
826834 let single a = if eq x a then stp else sng a
827835 yieldk a r = if eq x a
828− then unStream r st stp sng yld
836+ then foldStream st stp sng yld r
829837 else yld a (go r)
830− in unStream m1 st stp single yieldk
838+ in foldStream st stp single yieldk m1
831839
832840-------------------------------------------------------------------------------
833841-- Map and Filter
@@ -835,70 +843,66 @@ deleteBy eq x m = fromStream $ go (toStream m)
835843
836844{-# INLINE mapMaybe #-}
837845mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b
838−mapMaybe f m = go (toStream m)
846+mapMaybe f m = go m
839847 where
840− go m1 = fromStream $ mkStream $ \st stp sng yld ->
848+ go m1 = mkStream $ \st stp sng yld ->
841849 let single a = case f a of
842850 Just b -> sng b
843851 Nothing -> stp
844852 yieldk a r = case f a of
845− Just b -> yld b (toStream $ go r)
846− Nothing -> unStream r st stp single yieldk
847− in unStream m1 st stp single yieldk
853+ Just b -> yld b $ go r
854+ Nothing -> foldStream (adaptState st) stp single yieldk r
855+ in foldStream (adaptState st) stp single yieldk m1
848856
849857------------------------------------------------------------------------------
850858-- Serial Zipping
851859------------------------------------------------------------------------------
852860
853−{-# INLINE zipWithS #-}
854−zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
855−zipWithS f = go
861+-- | Zip two streams serially using a pure zipping function.
862+--
863+-- @since 0.1.0
864+{-# INLINABLE zipWith #-}
865+zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
866+zipWith f = go
856867 where
857868 go mx my = mkStream $ \st stp sng yld -> do
858869 let merge a ra =
859870 let single2 b = sng (f a b)
860871 yield2 b rb = yld (f a b) (go ra rb)
861− in unStream my st stp single2 yield2
872+ in foldStream (adaptState st) stp single2 yield2 my
862873 let single1 a = merge a nil
863874 yield1 = merge
864− unStream mx st stp single1 yield1
865−
866−-- | Zip two streams serially using a pure zipping function.
867−--
868−-- @since 0.1.0
869−{-# INLINABLE zipWith #-}
870−zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
871−zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)
875+ foldStream (adaptState st) stp single1 yield1 mx
872876
873877-- | Zip two streams serially using a monadic zipping function.
874878--
875879-- @since 0.1.0
876880zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
877−zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
881+zipWithM f m1 m2 = go m1 m2
878882 where
879883 go mx my = mkStream $ \st stp sng yld -> do
880884 let merge a ra =
881− let runIt x = unStream x st stp sng yld
885+ let runIt x = foldStream st stp sng yld x
882886 single2 b = f a b >>= sng
883887 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)
884− in unStream my st stp single2 yield2
888+ in foldStream (adaptState st) stp single2 yield2 my
885889 let single1 a = merge a nil
886890 yield1 = merge
887− unStream mx st stp single1 yield1
891+ foldStream (adaptState st) stp single1 yield1 mx
888892
889893------------------------------------------------------------------------------
890894-- Merging
891895------------------------------------------------------------------------------
892896
893−{-# INLINE mergeByS #-}
894−mergeByS
895− :: Monad m
896− => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a
897−mergeByS cmp = go
897+{-# INLINE mergeByM #-}
898+mergeByM
899+ :: (IsStream t, Monad m)
900+ => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
901+mergeByM cmp = go
898902 where
899903 go mx my = mkStream $ \st stp sng yld -> do
900904 let mergeWithY a ra =
901− let stop2 = unStream mx st stp sng yld
905+ let stop2 = foldStream st stp sng yld mx
902906 single2 b = do
903907 r <- cmp a b
904908 case r of
@@ -909,17 +913,11 @@ mergeByS cmp = go
909913 case r of
910914 GT -> yld b (go (a `cons` ra) rb)
911915 _ -> yld a (go ra (b `cons` rb))
912− in unStream my st stop2 single2 yield2
913− let stopX = unStream my st stp sng yld
916+ in foldStream st stop2 single2 yield2 my
917+ let stopX = foldStream st stp sng yld my
914918 singleX a = mergeWithY a nil
915919 yieldX = mergeWithY
916− unStream mx st stopX singleX yieldX
917−
918−{-# INLINABLE mergeByM #-}
919−mergeByM
920− :: (IsStream t, Monad m)
921− => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
922−mergeByM f m1 m2 = fromStream $ mergeByS f (toStream m1) (toStream m2)
920+ foldStream st stopX singleX yieldX mx
923921
924922{-# INLINABLE mergeBy #-}
925923mergeBy
@@ -937,14 +935,14 @@ the m = do
937935 r <- uncons m
938936 case r of
939937 Nothing -> return Nothing
940− Just (h, t) -> go h (toStream t)
938+ Just (h, t) -> go h t
941939 where
942940 go h m1 =
943941 let single a | h == a = return $ Just h
944942 | otherwise = return Nothing
945943 yieldk a r | h == a = go h r
946944 | otherwise = return Nothing
947− in unStream m1 defState (return $ Just h) single yieldk
945+ in foldStream defState (return $ Just h) single yieldk m1
948946
949947-------------------------------------------------------------------------------
950948-- Bind utility
@@ -952,20 +950,19 @@ the m = do
952950
953951{-# INLINE bindWith #-}
954952bindWith
955− :: (forall c. Stream m c -> Stream m c -> Stream m c)
956− -> Stream m a
957− -> (a -> Stream m b)
958− -> Stream m b
953+ :: IsStream t
954+ => (forall c. t m c -> t m c -> t m c)
955+ -> t m a
956+ -> (a -> t m b)
957+ -> t m b
959958bindWith par m1 f = go m1
960959 where
961960 go m =
962961 mkStream $ \st stp sng yld ->
963− let runShared x = unStreamShared x st stp sng yld
964− runIsolated x = unStream x st stp sng yld
965−
966− single a = runIsolated $ f a
967− yieldk a r = runShared $ unShare (f a) `par` go r
968− in unStream m st stp single yieldk
962+ let foldShared = foldStreamShared st stp sng yld
963+ single a = foldShared $ unShare (f a)
964+ yieldk a r = foldShared $ unShare (f a) `par` go r
965+ in foldStream (adaptState st) stp single yieldk m
969966
970967------------------------------------------------------------------------------
971968-- Alternative & MonadPlus
@@ -973,8 +970,8 @@ bindWith par m1 f = go m1
973970
974971_alt :: Stream m a -> Stream m a -> Stream m a
975972_alt m1 m2 = mkStream $ \st stp sng yld ->
976− let stop = unStream m2 st stp sng yld
977− in unStream m1 st stop sng yld
973+ let stop = foldStream st stp sng yld m2
974+ in foldStream st stop sng yld m1
978975
979976------------------------------------------------------------------------------
980977-- MonadReader
@@ -985,7 +982,7 @@ withLocal f m =
985982 mkStream $ \st stp sng yld ->
986983 let single = local f . sng
987984 yieldk a r = local f $ yld a (withLocal f r)
988− in unStream m st (local f stp) single yieldk
985+ in foldStream st (local f stp) single yieldk m
989986
990987------------------------------------------------------------------------------
991988-- MonadError
@@ -1003,4 +1000,3 @@ withCatchError m h =
10031000 yieldk a r = yld a (withCatchError r h)
10041001 in handle $ run m
10051002-}
1006−