git.iko.soy/streamly
on 9c8a750 help sign in
streamly/src/Streamly/Streams/StreamK.hs
999 lines 31 kB history raw
1{-# LANGUAGE BangPatterns #-}
2{-# LANGUAGE CPP #-}
3{-# LANGUAGE ConstraintKinds #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE FlexibleInstances #-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE RankNTypes #-}
9{-# LANGUAGE ScopedTypeVariables #-}
10{-# LANGUAGE UndecidableInstances #-} -- XXX
12#include "inline.hs"
14-- |
15-- Module : Streamly.Streams.StreamK
16-- Copyright : (c) 2017 Harendra Kumar
17--
18-- License : BSD3
19-- Maintainer : harendra.kumar@gmail.com
20-- Stability : experimental
21-- Portability : GHC
22--
23--
24-- Continuation passing style (CPS) stream implementation. The symbol 'K' below
25-- denotes a function as well as a Kontinuation.
26--
27-- @
28-- import qualified Streamly.Streams.StreamK as K
29-- @
30--
31module Streamly.Streams.StreamK
32 (
33 -- * A class for streams
34 IsStream (..)
35 , adapt
37 -- * The stream type
38 , Stream
40 -- * Construction Primitives
41 , mkStream
42 , nil
43 , cons
44 , (.:)
46 -- * Elimination Primitives
47 , foldStream
48 , foldStreamShared
49 , foldStreamSVar
51 -- * Transformation Primitives
52 , unShare
54 -- * Deconstruction
55 , uncons
57 -- * Generation
58 -- ** Unfolds
59 , unfoldr
60 , unfoldrM
62 -- ** Specialized Generation
63 , repeat
64 , replicate
65 , replicateM
67 -- ** Conversions
68 , yield
69 , yieldM
70 , fromFoldable
71 , fromList
72 , fromStreamK
74 -- * Elimination
75 -- ** General Folds
76 , foldr
77 , foldrM
78 , foldr1
79 , foldl'
80 , foldlM'
81 , foldx
82 , foldxM
84 -- ** Specialized Folds
85 , runStream
86 , null
87 , head
88 , tail
89 , init
90 , elem
91 , notElem
92 , all
93 , any
94 , last
95 , minimum
96 , minimumBy
97 , maximum
98 , maximumBy
99 , findIndices
100 , lookup
101 , findM
102 , find
103 , (!!)
105 -- ** Map and Fold
106 , mapM_
108 -- ** Conversions
109 , toList
110 , toStreamK
112 -- * Transformation
113 -- ** By folding (scans)
114 , scanl'
115 , scanx
117 -- ** Filtering
118 , filter
119 , take
120 , takeWhile
121 , drop
122 , dropWhile
124 -- ** Mapping
125 , map
126 , mapM
127 , sequence
129 -- ** Inserting
130 , intersperseM
131 , insertBy
133 -- ** Deleting
134 , deleteBy
136 -- ** Map and Filter
137 , mapMaybe
139 -- ** Zipping
140 , zipWith
141 , zipWithM
143 -- ** Merging
144 , mergeBy
145 , mergeByM
147 -- ** Transformation comprehensions
148 , the
150 -- * Semigroup Style Composition
151 , serial
153 -- * Utilities
154 , consMSerial
155 , bindWith
156 , withLocal
158 -- * Deprecated
159 , Streaming -- deprecated
160 , once -- deprecated
161 )
162where
164import Control.Monad (void)
165import Control.Monad.Reader.Class (MonadReader(..))
166import Prelude
167 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,
168 take, filter, all, any, takeWhile, drop, dropWhile, minimum,
169 maximum, elem, notElem, null, head, tail, init, zipWith, lookup,
170 foldr1, (!!), replicate)
171import qualified Prelude
173import Streamly.SVar
174import Streamly.Streams.StreamK.Type
176-- | Detach a stream from an SVar
177{-# INLINE unShare #-}
178unShare :: IsStream t => t m a -> t m a
179unShare x = mkStream $ \st yld sng stp ->
180 foldStream st yld sng stp x
182------------------------------------------------------------------------------
183-- Construction
184------------------------------------------------------------------------------
186infixr 5 `cons`
188-- faster than consM because there is no bind.
189-- | Construct a stream by adding a pure value at the head of an existing
190-- stream. For serial streams this is the same as @(return a) \`consM` r@ but
191-- more efficient. For concurrent streams this is not concurrent whereas
192-- 'consM' is concurrent. For example:
193--
194-- @
195-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
196-- [1,2,3]
197-- @
198--
199-- @since 0.1.0
200cons :: IsStream t => a -> t m a -> t m a
201cons a r = mkStream $ \_ yld _ _ -> yld a r
203infixr 5 .:
205-- | Operator equivalent of 'cons'.
206--
207-- @
208-- > toList $ 1 .: 2 .: 3 .: nil
209-- [1,2,3]
210-- @
211--
212-- @since 0.1.1
213(.:) :: IsStream t => a -> t m a -> t m a
214(.:) = cons
216-------------------------------------------------------------------------------
217-- Deconstruction
218-------------------------------------------------------------------------------
220{-# INLINE uncons #-}
221uncons :: (IsStream t, Monad m) => t m a -> m (Maybe (a, t m a))
222uncons m =
223 let stop = return Nothing
224 single a = return (Just (a, nil))
225 yieldk a r = return (Just (a, r))
226 in foldStream defState yieldk single stop m
228-------------------------------------------------------------------------------
229-- Generation
230-------------------------------------------------------------------------------
232{-# INLINE_NORMAL build #-}
233build :: IsStream t => forall a. (forall b. (a -> b -> b) -> b -> b) -> t m a
234build g = g cons nil
236{-# INLINE_NORMAL _augment #-}
237_augment
238 :: IsStream t
239 => forall a. (forall b. (a -> b -> b) -> b -> b) -> t m a -> t m a
240_augment g xs = g cons xs
242{-# INLINE_NORMAL _buildM #-}
243_buildM
244 :: (IsStream t, MonadAsync m)
245 => forall a. ((m a -> t m a -> t m a) -> t m a -> t m a) -> t m a
246_buildM g = g consM nil
248{-# INLINE unfoldr #-}
249unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
250unfoldr step b0 = build $ \cns nl ->
251 let go s =
252 case step s of
253 Just (a, b) -> a `cns` go b
254 Nothing -> nl
255 in go b0
257{-# INLINE unfoldrM #-}
258unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
259unfoldrM step = go
260 where
261 go s = mkStream $ \st yld sng stp -> do
262 mayb <- step s
263 case mayb of
264 Nothing -> stp
265 Just (a, b) ->
266 foldStreamShared st yld sng stp $ return a |: go b
268-------------------------------------------------------------------------------
269-- Special generation
270-------------------------------------------------------------------------------
272yield :: IsStream t => a -> t m a
273yield a = mkStream $ \_ _ single _ -> single a
275-- | Same as yieldM
276--
277-- @since 0.2.0
278{-# DEPRECATED once "Please use yieldM instead." #-}
279{-# INLINE once #-}
280once :: (Monad m, IsStream t) => m a -> t m a
281once = yieldM
283-- |
284-- @
285-- repeatM = fix . cons
286-- repeatM = cycle1 . yield
287-- @
288--
289-- Generate an infinite stream by repeating a pure value.
290--
291-- @since 0.4.0
292repeat :: IsStream t => a -> t m a
293repeat a = let x = cons a x in x
295replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
296replicateM n m = go n
297 where
298 go cnt = if cnt <= 0 then nil else m |: go (cnt - 1)
300replicate :: IsStream t => Int -> a -> t m a
301replicate n a = go n
302 where
303 go cnt = if cnt <= 0 then nil else a `cons` go (cnt - 1)
305-------------------------------------------------------------------------------
306-- Conversions
307-------------------------------------------------------------------------------
309-- |
310-- @
311-- fromFoldable = 'Prelude.foldr' 'cons' 'nil'
312-- @
313--
314-- Construct a stream from a 'Foldable' containing pure values:
315--
316-- @since 0.2.0
317{-# INLINE fromFoldable #-}
318fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
319fromFoldable = Prelude.foldr cons nil
321{-# INLINE fromList #-}
322fromList :: IsStream t => [a] -> t m a
323fromList = fromFoldable
325{-# INLINE fromStreamK #-}
326fromStreamK :: IsStream t => Stream m a -> t m a
327fromStreamK = fromStream
329-------------------------------------------------------------------------------
330-- Elimination by Folding
331-------------------------------------------------------------------------------
333-- | Lazy right associative fold.
334foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b
335foldr step acc m = go m
336 where
337 go m1 =
338 let stop = return acc
339 single a = return (step a acc)
340 yieldk a r = go r >>= \b -> return (step a b)
341 in foldStream defState yieldk single stop m1
343-- | Lazy right fold with a monadic step function.
344{-# INLINE foldrM #-}
345foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b
346foldrM step acc m = go m
347 where
348 go m1 =
349 let stop = return acc
350 single a = step a acc
351 yieldk a r = go r >>= step a
352 in foldStream defState yieldk single stop m1
354{-# INLINE foldr1 #-}
355foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)
356foldr1 step m = do
357 r <- uncons m
358 case r of
359 Nothing -> return Nothing
360 Just (h, t) -> fmap Just (go h t)
361 where
362 go p m1 =
363 let stp = return p
364 single a = return $ step a p
365 yieldk a r = fmap (step p) (go a r)
366 in foldStream defState yieldk single stp m1
368-- | Strict left fold with an extraction function. Like the standard strict
369-- left fold, but applies a user supplied extraction function (the third
370-- argument) to the folded value at the end. This is designed to work with the
371-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
372--
373-- Note that the accumulator is always evaluated including the initial value.
374{-# INLINE foldx #-}
375foldx :: forall t m a b x. (IsStream t, Monad m)
376 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
377foldx step begin done m = get $ go m begin
378 where
379 {-# NOINLINE get #-}
380 get :: t m x -> m b
381 get m1 =
382 -- XXX we are not strictly evaluating the accumulator here. Is this
383 -- okay?
384 let single = return . done
385 -- XXX this is foldSingleton. why foldStreamShared?
386 in foldStreamShared undefined undefined single undefined m1
388 -- Note, this can be implemented by making a recursive call to "go",
389 -- however that is more expensive because of unnecessary recursion
390 -- that cannot be tail call optimized. Unfolding recursion explicitly via
391 -- continuations is much more efficient.
392 go :: t m a -> x -> t m x
393 go m1 !acc = mkStream $ \_ yld sng _ ->
394 let stop = sng acc
395 single a = sng $ step acc a
396 -- XXX this is foldNonEmptyStream
397 yieldk a r = foldStream defState yld sng undefined $
398 go r (step acc a)
399 in foldStream defState yieldk single stop m1
401-- | Strict left associative fold.
402{-# INLINE foldl' #-}
403foldl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> m b
404foldl' step begin = foldx step begin id
406-- XXX replace the recursive "go" with explicit continuations.
407-- | Like 'foldx', but with a monadic step function.
408foldxM :: (IsStream t, Monad m)
409 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b
410foldxM step begin done m = go begin m
411 where
412 go !acc m1 =
413 let stop = acc >>= done
414 single a = acc >>= \b -> step b a >>= done
415 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r
416 in foldStream defState yieldk single stop m1
418-- | Like 'foldl'' but with a monadic step function.
419foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
420foldlM' step begin = foldxM step (return begin) return
422------------------------------------------------------------------------------
423-- Specialized folds
424------------------------------------------------------------------------------
426-- |
427-- > runStream = foldl' (\_ _ -> ()) ()
428-- > runStream = mapM_ (\_ -> return ())
429{-# INLINE runStream #-}
430runStream :: (Monad m, IsStream t) => t m a -> m ()
431runStream = go
432 where
433 go m1 =
434 let stop = return ()
435 single _ = return ()
436 yieldk _ r = go r
437 in foldStream defState yieldk single stop m1
439{-# INLINE null #-}
440null :: (IsStream t, Monad m) => t m a -> m Bool
441null m =
442 let stop = return True
443 single _ = return False
444 yieldk _ _ = return False
445 in foldStream defState yieldk single stop m
447{-# INLINE head #-}
448head :: (IsStream t, Monad m) => t m a -> m (Maybe a)
449head m =
450 let stop = return Nothing
451 single a = return (Just a)
452 yieldk a _ = return (Just a)
453 in foldStream defState yieldk single stop m
455{-# INLINE tail #-}
456tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
457tail m =
458 let stop = return Nothing
459 single _ = return $ Just nil
460 yieldk _ r = return $ Just r
461 in foldStream defState yieldk single stop m
463{-# INLINE init #-}
464init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
465init m = go1 m
466 where
467 go1 m1 = do
468 r <- uncons m1
469 case r of
470 Nothing -> return Nothing
471 Just (h, t) -> return . Just $ go h t
472 go p m1 = mkStream $ \_ yld sng stp ->
473 let single _ = sng p
474 yieldk a x = yld p $ go a x
475 in foldStream defState yieldk single stp m1
477{-# INLINE elem #-}
478elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
479elem e m = go m
480 where
481 go m1 =
482 let stop = return False
483 single a = return (a == e)
484 yieldk a r = if a == e then return True else go r
485 in foldStream defState yieldk single stop m1
487{-# INLINE notElem #-}
488notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
489notElem e m = go m
490 where
491 go m1 =
492 let stop = return True
493 single a = return (a /= e)
494 yieldk a r = if a == e then return False else go r
495 in foldStream defState yieldk single stop m1
497all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
498all p m = go m
499 where
500 go m1 =
501 let single a | p a = return True
502 | otherwise = return False
503 yieldk a r | p a = go r
504 | otherwise = return False
505 in foldStream defState yieldk single (return True) m1
507any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
508any p m = go m
509 where
510 go m1 =
511 let single a | p a = return True
512 | otherwise = return False
513 yieldk a r | p a = return True
514 | otherwise = go r
515 in foldStream defState yieldk single (return False) m1
517-- | Extract the last element of the stream, if any.
518{-# INLINE last #-}
519last :: (IsStream t, Monad m) => t m a -> m (Maybe a)
520last = foldx (\_ y -> Just y) Nothing id
522{-# INLINE minimum #-}
523minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
524minimum m = go Nothing m
525 where
526 go Nothing m1 =
527 let stop = return Nothing
528 single a = return (Just a)
529 yieldk a r = go (Just a) r
530 in foldStream defState yieldk single stop m1
532 go (Just res) m1 =
533 let stop = return (Just res)
534 single a =
535 if res <= a
536 then return (Just res)
537 else return (Just a)
538 yieldk a r =
539 if res <= a
540 then go (Just res) r
541 else go (Just a) r
542 in foldStream defState yieldk single stop m1
544{-# INLINE minimumBy #-}
545minimumBy
546 :: (IsStream t, Monad m)
547 => (a -> a -> Ordering) -> t m a -> m (Maybe a)
548minimumBy cmp m = go Nothing m
549 where
550 go Nothing m1 =
551 let stop = return Nothing
552 single a = return (Just a)
553 yieldk a r = go (Just a) r
554 in foldStream defState yieldk single stop m1
556 go (Just res) m1 =
557 let stop = return (Just res)
558 single a = case cmp res a of
559 GT -> return (Just a)
560 _ -> return (Just res)
561 yieldk a r = case cmp res a of
562 GT -> go (Just a) r
563 _ -> go (Just res) r
564 in foldStream defState yieldk single stop m1
566{-# INLINE maximum #-}
567maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
568maximum m = go Nothing m
569 where
570 go Nothing m1 =
571 let stop = return Nothing
572 single a = return (Just a)
573 yieldk a r = go (Just a) r
574 in foldStream defState yieldk single stop m1
576 go (Just res) m1 =
577 let stop = return (Just res)
578 single a =
579 if res <= a
580 then return (Just a)
581 else return (Just res)
582 yieldk a r =
583 if res <= a
584 then go (Just a) r
585 else go (Just res) r
586 in foldStream defState yieldk single stop m1
588{-# INLINE maximumBy #-}
589maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)
590maximumBy cmp m = go Nothing m
591 where
592 go Nothing m1 =
593 let stop = return Nothing
594 single a = return (Just a)
595 yieldk a r = go (Just a) r
596 in foldStream defState yieldk single stop m1
598 go (Just res) m1 =
599 let stop = return (Just res)
600 single a = case cmp res a of
601 GT -> return (Just res)
602 _ -> return (Just a)
603 yieldk a r = case cmp res a of
604 GT -> go (Just res) r
605 _ -> go (Just a) r
606 in foldStream defState yieldk single stop m1
608{-# INLINE (!!) #-}
609(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)
610m !! i = go i m
611 where
612 go n m1 =
613 let single a | n == 0 = return $ Just a
614 | otherwise = return Nothing
615 yieldk a x | n < 0 = return Nothing
616 | n == 0 = return $ Just a
617 | otherwise = go (n - 1) x
618 in foldStream defState yieldk single (return Nothing) m1
620{-# INLINE lookup #-}
621lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)
622lookup e m = go m
623 where
624 go m1 =
625 let single (a, b) | a == e = return $ Just b
626 | otherwise = return Nothing
627 yieldk (a, b) x | a == e = return $ Just b
628 | otherwise = go x
629 in foldStream defState yieldk single (return Nothing) m1
631{-# INLINE findM #-}
632findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)
633findM p m = go m
634 where
635 go m1 =
636 let single a = do
637 b <- p a
638 if b then return $ Just a else return Nothing
639 yieldk a x = do
640 b <- p a
641 if b then return $ Just a else go x
642 in foldStream defState yieldk single (return Nothing) m1
644{-# INLINE find #-}
645find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)
646find p = findM (return . p)
648{-# INLINE findIndices #-}
649findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int
650findIndices p = go 0
651 where
652 go offset m1 = mkStream $ \st yld sng stp ->
653 let single a | p a = sng offset
654 | otherwise = stp
655 yieldk a x | p a = yld offset $ go (offset + 1) x
656 | otherwise = foldStream (adaptState st) yld sng stp $
657 go (offset + 1) x
658 in foldStream (adaptState st) yieldk single stp m1
660------------------------------------------------------------------------------
661-- Map and Fold
662------------------------------------------------------------------------------
664-- | Apply a monadic action to each element of the stream and discard the
665-- output of the action.
666{-# INLINE mapM_ #-}
667mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()
668mapM_ f m = go m
669 where
670 go m1 =
671 let stop = return ()
672 single a = void (f a)
673 yieldk a r = f a >> go r
674 in foldStream defState yieldk single stop m1
676------------------------------------------------------------------------------
677-- Converting folds
678------------------------------------------------------------------------------
680{-# INLINABLE toList #-}
681toList :: (IsStream t, Monad m) => t m a -> m [a]
682toList = foldr (:) []
684{-# INLINE toStreamK #-}
685toStreamK :: Stream m a -> Stream m a
686toStreamK = id
688-------------------------------------------------------------------------------
689-- Transformation by folding (Scans)
690-------------------------------------------------------------------------------
692{-# INLINE scanx #-}
693scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
694scanx step begin done m =
695 cons (done begin) $ go m begin
696 where
697 go m1 !acc = mkStream $ \st yld sng stp ->
698 let single a = sng (done $ step acc a)
699 yieldk a r =
700 let s = step acc a
701 in yld (done s) (go r s)
702 in foldStream (adaptState st) yieldk single stp m1
704{-# INLINE scanl' #-}
705scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
706scanl' step begin = scanx step begin id
708-------------------------------------------------------------------------------
709-- Filtering
710-------------------------------------------------------------------------------
712{-# INLINE filter #-}
713filter :: IsStream t => (a -> Bool) -> t m a -> t m a
714filter p m = go m
715 where
716 go m1 = mkStream $ \st yld sng stp ->
717 let single a | p a = sng a
718 | otherwise = stp
719 yieldk a r | p a = yld a (go r)
720 | otherwise = foldStream st yieldk single stp r
721 in foldStream st yieldk single stp m1
723{-# INLINE take #-}
724take :: IsStream t => Int -> t m a -> t m a
725take n m = go n m
726 where
727 go n1 m1 = mkStream $ \st yld sng stp ->
728 let yieldk a r = yld a (go (n1 - 1) r)
729 in if n1 <= 0
730 then stp
731 else foldStream st yieldk sng stp m1
733{-# INLINE takeWhile #-}
734takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
735takeWhile p m = go m
736 where
737 go m1 = mkStream $ \st yld sng stp ->
738 let single a | p a = sng a
739 | otherwise = stp
740 yieldk a r | p a = yld a (go r)
741 | otherwise = stp
742 in foldStream st yieldk single stp m1
744drop :: IsStream t => Int -> t m a -> t m a
745drop n m = go n m
746 where
747 go n1 m1 = mkStream $ \st yld sng stp ->
748 let single _ = stp
749 yieldk _ r = foldStream st yld sng stp $ go (n1 - 1) r
750 -- Somehow "<=" check performs better than a ">"
751 in if n1 <= 0
752 then foldStream st yld sng stp m1
753 else foldStream st yieldk single stp m1
755{-# INLINE dropWhile #-}
756dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
757dropWhile p m = go m
758 where
759 go m1 = mkStream $ \st yld sng stp ->
760 let single a | p a = stp
761 | otherwise = sng a
762 yieldk a r | p a = foldStream st yieldk single stp r
763 | otherwise = yld a r
764 in foldStream st yieldk single stp m1
766-------------------------------------------------------------------------------
767-- Mapping
768-------------------------------------------------------------------------------
770-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
771-- other stream types to overload it.
772{-# INLINE mapM #-}
773mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
774mapM f m = go m
775 where
776 go m1 = mkStream $ \st yld sng stp ->
777 let single a = f a >>= sng
778 yieldk a r = foldStreamShared st yld sng stp $ f a |: go r
779 in foldStream (adaptState st) yieldk single stp m1
781-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
782-- other stream types to overload it.
783{-# INLINE sequence #-}
784sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
785sequence m = go m
786 where
787 go m1 = mkStream $ \st yld sng stp ->
788 let single ma = ma >>= sng
789 yieldk ma r = foldStreamShared st yld sng stp $ ma |: go r
790 in foldStream (adaptState st) yieldk single stp m1
792-------------------------------------------------------------------------------
793-- Inserting
794-------------------------------------------------------------------------------
796{-# INLINE intersperseM #-}
797intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
798intersperseM a m = prependingStart m
799 where
800 prependingStart m1 = mkStream $ \st yld sng stp ->
801 let yieldk i x = foldStreamShared st yld sng stp $ return i |: go x
802 in foldStream st yieldk sng stp m1
803 go m2 = mkStream $ \st yld sng stp ->
804 let single i = foldStreamShared st yld sng stp $ a |: yield i
805 yieldk i x = foldStreamShared st yld sng stp $ a |: return i |: go x
806 in foldStream st yieldk single stp m2
808{-# INLINE insertBy #-}
809insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a
810insertBy cmp x m = go m
811 where
812 go m1 = mkStream $ \st yld _ _ ->
813 let single a = case cmp x a of
814 GT -> yld a (yield x)
815 _ -> yld x (yield a)
816 stop = yld x nil
817 yieldk a r = case cmp x a of
818 GT -> yld a (go r)
819 _ -> yld x (a `cons` r)
820 in foldStream st yieldk single stop m1
822------------------------------------------------------------------------------
823-- Deleting
824------------------------------------------------------------------------------
826{-# INLINE deleteBy #-}
827deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a
828deleteBy eq x m = go m
829 where
830 go m1 = mkStream $ \st yld sng stp ->
831 let single a = if eq x a then stp else sng a
832 yieldk a r = if eq x a
833 then foldStream st yld sng stp r
834 else yld a (go r)
835 in foldStream st yieldk single stp m1
837-------------------------------------------------------------------------------
838-- Map and Filter
839-------------------------------------------------------------------------------
841{-# INLINE mapMaybe #-}
842mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b
843mapMaybe f m = go m
844 where
845 go m1 = mkStream $ \st yld sng stp ->
846 let single a = case f a of
847 Just b -> sng b
848 Nothing -> stp
849 yieldk a r = case f a of
850 Just b -> yld b $ go r
851 Nothing -> foldStream (adaptState st) yieldk single stp r
852 in foldStream (adaptState st) yieldk single stp m1
854------------------------------------------------------------------------------
855-- Serial Zipping
856------------------------------------------------------------------------------
858-- | Zip two streams serially using a pure zipping function.
859--
860-- @since 0.1.0
861{-# INLINABLE zipWith #-}
862zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
863zipWith f = go
864 where
865 go mx my = mkStream $ \st yld sng stp -> do
866 let merge a ra =
867 let single2 b = sng (f a b)
868 yield2 b rb = yld (f a b) (go ra rb)
869 in foldStream (adaptState st) yield2 single2 stp my
870 let single1 a = merge a nil
871 yield1 = merge
872 foldStream (adaptState st) yield1 single1 stp mx
874-- | Zip two streams serially using a monadic zipping function.
875--
876-- @since 0.1.0
877zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
878zipWithM f m1 m2 = go m1 m2
879 where
880 go mx my = mkStream $ \st yld sng stp -> do
881 let merge a ra =
882 let runIt x = foldStream st yld sng stp x
883 single2 b = f a b >>= sng
884 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)
885 in foldStream (adaptState st) yield2 single2 stp my
886 let single1 a = merge a nil
887 yield1 = merge
888 foldStream (adaptState st) yield1 single1 stp mx
890------------------------------------------------------------------------------
891-- Merging
892------------------------------------------------------------------------------
894{-# INLINE mergeByM #-}
895mergeByM
896 :: (IsStream t, Monad m)
897 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
898mergeByM cmp = go
899 where
900 go mx my = mkStream $ \st yld sng stp -> do
901 let mergeWithY a ra =
902 let stop2 = foldStream st yld sng stp mx
903 single2 b = do
904 r <- cmp a b
905 case r of
906 GT -> yld b (go (a `cons` ra) nil)
907 _ -> yld a (go ra (b `cons` nil))
908 yield2 b rb = do
909 r <- cmp a b
910 case r of
911 GT -> yld b (go (a `cons` ra) rb)
912 _ -> yld a (go ra (b `cons` rb))
913 in foldStream st yield2 single2 stop2 my
914 let stopX = foldStream st yld sng stp my
915 singleX a = mergeWithY a nil
916 yieldX = mergeWithY
917 foldStream st yieldX singleX stopX mx
919{-# INLINABLE mergeBy #-}
920mergeBy
921 :: (IsStream t, Monad m)
922 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
923mergeBy cmp = mergeByM (\a b -> return $ cmp a b)
925------------------------------------------------------------------------------
926-- Transformation comprehensions
927------------------------------------------------------------------------------
929{-# INLINE the #-}
930the :: (Eq a, IsStream t, Monad m) => t m a -> m (Maybe a)
931the m = do
932 r <- uncons m
933 case r of
934 Nothing -> return Nothing
935 Just (h, t) -> go h t
936 where
937 go h m1 =
938 let single a | h == a = return $ Just h
939 | otherwise = return Nothing
940 yieldk a r | h == a = go h r
941 | otherwise = return Nothing
942 in foldStream defState yieldk single (return $ Just h) m1
944-------------------------------------------------------------------------------
945-- Bind utility
946-------------------------------------------------------------------------------
948{-# INLINE bindWith #-}
949bindWith
950 :: IsStream t
951 => (forall c. t m c -> t m c -> t m c)
952 -> t m a
953 -> (a -> t m b)
954 -> t m b
955bindWith par m1 f = go m1
956 where
957 go m =
958 mkStream $ \st yld sng stp ->
959 let foldShared = foldStreamShared st yld sng stp
960 single a = foldShared $ unShare (f a)
961 yieldk a r = foldShared $ unShare (f a) `par` go r
962 in foldStream (adaptState st) yieldk single stp m
964------------------------------------------------------------------------------
965-- Alternative & MonadPlus
966------------------------------------------------------------------------------
968_alt :: Stream m a -> Stream m a -> Stream m a
969_alt m1 m2 = mkStream $ \st yld sng stp ->
970 let stop = foldStream st yld sng stp m2
971 in foldStream st yld sng stop m1
973------------------------------------------------------------------------------
974-- MonadReader
975------------------------------------------------------------------------------
977withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
978withLocal f m =
979 mkStream $ \st yld sng stp ->
980 let single = local f . sng
981 yieldk a r = local f $ yld a (withLocal f r)
982 in foldStream st yieldk single (local f stp) m
984------------------------------------------------------------------------------
985-- MonadError
986------------------------------------------------------------------------------
988{-
989-- XXX handle and test cross thread state transfer
990withCatchError
991 :: MonadError e m
992 => Stream m a -> (e -> Stream m a) -> Stream m a
993withCatchError m h =
994 mkStream $ \_ stp sng yld ->
995 let run x = unStream x Nothing stp sng yieldk
996 handle r = r `catchError` \e -> run $ h e
997 yieldk a r = yld a (withCatchError r h)
998 in handle $ run m
999-}