git.iko.soy/streamly
on 2ba5f01 help sign in
streamly/src/Streamly/Streams/StreamK.hs
1002 lines 32 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-- |
13-- Module : Streamly.Streams.StreamK
14-- Copyright : (c) 2017 Harendra Kumar
15--
16-- License : BSD3
17-- Maintainer : harendra.kumar@gmail.com
18-- Stability : experimental
19-- Portability : GHC
20--
21--
22-- Continuation passing style (CPS) stream implementation. The symbol 'K' below
23-- denotes a function as well as a Kontinuation.
24--
25-- @
26-- import qualified Streamly.Streams.StreamK as K
27-- @
28--
29module Streamly.Streams.StreamK
30 (
31 -- * A class for streams
32 IsStream (..)
33 , adapt
35 -- * The stream type
36 , Stream
38 -- * Construction Primitives
39 , mkStream
40 , nil
41 , cons
42 , (.:)
44 -- * Elimination Primitives
45 , foldStream
46 , foldStreamShared
47 , foldStreamSVar
49 -- * Transformation Primitives
50 , unShare
52 -- * Asynchronous construction
53 , nilK
54 , yieldK
55 , consK
57 -- * Deconstruction
58 , uncons
60 -- * Generation
61 -- ** Unfolds
62 , unfoldr
63 , unfoldrM
65 -- ** Specialized Generation
66 , repeat
67 , replicate
68 , replicateM
70 -- ** Conversions
71 , yield
72 , yieldM
73 , fromFoldable
74 , fromList
75 , fromStreamK
77 -- * Elimination
78 -- ** General Folds
79 , foldr
80 , foldrM
81 , foldr1
82 , foldl'
83 , foldlM'
84 , foldx
85 , foldxM
87 -- ** Specialized Folds
88 , runStream
89 , null
90 , head
91 , tail
92 , init
93 , elem
94 , notElem
95 , all
96 , any
97 , last
98 , minimum
99 , minimumBy
100 , maximum
101 , maximumBy
102 , findIndices
103 , lookup
104 , findM
105 , find
106 , (!!)
108 -- ** Map and Fold
109 , mapM_
111 -- ** Conversions
112 , toList
113 , toStreamK
115 -- * Transformation
116 -- ** By folding (scans)
117 , scanl'
118 , scanx
120 -- ** Filtering
121 , filter
122 , take
123 , takeWhile
124 , drop
125 , dropWhile
127 -- ** Mapping
128 , map
129 , mapM
130 , sequence
132 -- ** Inserting
133 , intersperseM
134 , insertBy
136 -- ** Deleting
137 , deleteBy
139 -- ** Map and Filter
140 , mapMaybe
142 -- ** Zipping
143 , zipWith
144 , zipWithM
146 -- ** Merging
147 , mergeBy
148 , mergeByM
150 -- ** Transformation comprehensions
151 , the
153 -- * Semigroup Style Composition
154 , serial
156 -- * Utilities
157 , consMSerial
158 , bindWith
159 , withLocal
161 -- * Deprecated
162 , Streaming -- deprecated
163 , once -- deprecated
164 )
165where
167import Control.Monad (void)
168import Control.Monad.Reader.Class (MonadReader(..))
169import Prelude
170 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,
171 take, filter, all, any, takeWhile, drop, dropWhile, minimum,
172 maximum, elem, notElem, null, head, tail, init, zipWith, lookup,
173 foldr1, (!!), replicate)
174import qualified Prelude
176import Streamly.SVar
177import Streamly.Streams.StreamKType
179-- | Detach a stream from an SVar
180{-# INLINE unShare #-}
181unShare :: IsStream t => t m a -> t m a
182unShare x = mkStream $ \st stp sng yld ->
183 foldStream st stp sng yld x
185------------------------------------------------------------------------------
186-- Construction
187------------------------------------------------------------------------------
189infixr 5 `cons`
191-- faster than consM because there is no bind.
192-- | Construct a stream by adding a pure value at the head of an existing
193-- stream. For serial streams this is the same as @(return a) \`consM` r@ but
194-- more efficient. For concurrent streams this is not concurrent whereas
195-- 'consM' is concurrent. For example:
196--
197-- @
198-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
199-- [1,2,3]
200-- @
201--
202-- @since 0.1.0
203cons :: IsStream t => a -> t m a -> t m a
204cons a r = mkStream $ \_ _ _ yld -> yld a r
206infixr 5 .:
208-- | Operator equivalent of 'cons'.
209--
210-- @
211-- > toList $ 1 .: 2 .: 3 .: nil
212-- [1,2,3]
213-- @
214--
215-- @since 0.1.1
216(.:) :: IsStream t => a -> t m a -> t m a
217(.:) = cons
219------------------------------------------------------------------------------
220-- Asynchronous construction
221------------------------------------------------------------------------------
223-- | Make an empty stream from a callback function.
224nilK :: IsStream t => (forall r. m r -> m r) -> t m a
225nilK k = mkStream $ \_ stp _ _ -> k stp
227-- | Make a singleton stream from a one shot callback function.
228yieldK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
229yieldK k = mkStream $ \_ _ sng _ -> k sng
231-- | Construct a stream from a callback function.
232consK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a -> t m a
233consK k r = mkStream $ \_ _ _ yld -> k (\x -> yld x r)
235-- XXX consK with concurrent callbacks
236-- XXX Build a stream from a repeating callback function.
238-------------------------------------------------------------------------------
239-- Deconstruction
240-------------------------------------------------------------------------------
242{-# INLINE uncons #-}
243uncons :: (IsStream t, Monad m) => t m a -> m (Maybe (a, t m a))
244uncons m =
245 let stop = return Nothing
246 single a = return (Just (a, nil))
247 yieldk a r = return (Just (a, r))
248 in foldStream defState stop single yieldk m
250-------------------------------------------------------------------------------
251-- Generation
252-------------------------------------------------------------------------------
254{-# INLINE unfoldr #-}
255unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
256unfoldr step = go
257 where
258 go s = mkStream $ \_ stp _ yld ->
259 case step s of
260 Nothing -> stp
261 Just (a, b) -> yld a (go b)
263{-# INLINE unfoldrM #-}
264unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
265unfoldrM step = go
266 where
267 go s = mkStream $ \st stp sng yld -> do
268 mayb <- step s
269 case mayb of
270 Nothing -> stp
271 Just (a, b) ->
272 foldStreamShared st stp sng yld $ return a |: go b
274-------------------------------------------------------------------------------
275-- Special generation
276-------------------------------------------------------------------------------
278yield :: IsStream t => a -> t m a
279yield a = mkStream $ \_ _ single _ -> single a
281-- | Same as yieldM
282--
283-- @since 0.2.0
284{-# DEPRECATED once "Please use yieldM instead." #-}
285{-# INLINE once #-}
286once :: (Monad m, IsStream t) => m a -> t m a
287once = yieldM
289-- |
290-- @
291-- repeatM = fix . cons
292-- repeatM = cycle1 . yield
293-- @
294--
295-- Generate an infinite stream by repeating a pure value.
296--
297-- @since 0.4.0
298repeat :: IsStream t => a -> t m a
299repeat a = let x = cons a x in x
301replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
302replicateM n m = go n
303 where
304 go cnt = if cnt <= 0 then nil else m |: go (cnt - 1)
306replicate :: IsStream t => Int -> a -> t m a
307replicate n a = go n
308 where
309 go cnt = if cnt <= 0 then nil else a `cons` go (cnt - 1)
311-------------------------------------------------------------------------------
312-- Conversions
313-------------------------------------------------------------------------------
315-- |
316-- @
317-- fromFoldable = 'Prelude.foldr' 'cons' 'nil'
318-- @
319--
320-- Construct a stream from a 'Foldable' containing pure values:
321--
322-- @since 0.2.0
323{-# INLINE fromFoldable #-}
324fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
325fromFoldable = Prelude.foldr cons nil
327{-# INLINE fromList #-}
328fromList :: IsStream t => [a] -> t m a
329fromList = fromFoldable
331{-# INLINE fromStreamK #-}
332fromStreamK :: IsStream t => Stream m a -> t m a
333fromStreamK = fromStream
335-------------------------------------------------------------------------------
336-- Elimination by Folding
337-------------------------------------------------------------------------------
339-- | Lazy right associative fold.
340foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b
341foldr step acc m = go m
342 where
343 go m1 =
344 let stop = return acc
345 single a = return (step a acc)
346 yieldk a r = go r >>= \b -> return (step a b)
347 in foldStream defState stop single yieldk m1
349-- | Lazy right fold with a monadic step function.
350{-# INLINE foldrM #-}
351foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b
352foldrM step acc m = go m
353 where
354 go m1 =
355 let stop = return acc
356 single a = step a acc
357 yieldk a r = go r >>= step a
358 in foldStream defState stop single yieldk m1
360{-# INLINE foldr1 #-}
361foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)
362foldr1 step m = do
363 r <- uncons m
364 case r of
365 Nothing -> return Nothing
366 Just (h, t) -> fmap Just (go h t)
367 where
368 go p m1 =
369 let stp = return p
370 single a = return $ step a p
371 yieldk a r = fmap (step p) (go a r)
372 in foldStream defState stp single yieldk m1
374-- | Strict left fold with an extraction function. Like the standard strict
375-- left fold, but applies a user supplied extraction function (the third
376-- argument) to the folded value at the end. This is designed to work with the
377-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
378--
379-- Note that the accumulator is always evaluated including the initial value.
380{-# INLINE foldx #-}
381foldx :: forall t m a b x. (IsStream t, Monad m)
382 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
383foldx step begin done m = get $ go m begin
384 where
385 {-# NOINLINE get #-}
386 get :: t m x -> m b
387 get m1 =
388 -- XXX we are not strictly evaluating the accumulator here. Is this
389 -- okay?
390 let single = return . done
391 -- XXX this is foldSingleton. why foldStreamShared?
392 in foldStreamShared undefined undefined single undefined m1
394 -- Note, this can be implemented by making a recursive call to "go",
395 -- however that is more expensive because of unnecessary recursion
396 -- that cannot be tail call optimized. Unfolding recursion explicitly via
397 -- continuations is much more efficient.
398 go :: t m a -> x -> t m x
399 go m1 !acc = mkStream $ \_ _ sng yld ->
400 let stop = sng acc
401 single a = sng $ step acc a
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
407-- | Strict left associative fold.
408{-# INLINE foldl' #-}
409foldl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> m b
410foldl' step begin = foldx step begin id
412-- XXX replace the recursive "go" with explicit continuations.
413-- | Like 'foldx', but with a monadic step function.
414foldxM :: (IsStream t, Monad m)
415 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b
416foldxM step begin done m = go begin m
417 where
418 go !acc m1 =
419 let stop = acc >>= done
420 single a = acc >>= \b -> step b a >>= done
421 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r
422 in foldStream defState stop single yieldk m1
424-- | Like 'foldl'' but with a monadic step function.
425foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
426foldlM' step begin = foldxM step (return begin) return
428------------------------------------------------------------------------------
429-- Specialized folds
430------------------------------------------------------------------------------
432{-# INLINE runStream #-}
433runStream :: (Monad m, IsStream t) => t m a -> m ()
434runStream = go
435 where
436 go m1 =
437 let stop = return ()
438 single _ = return ()
439 yieldk _ r = go r
440 in foldStream defState stop single yieldk m1
442{-# INLINE null #-}
443null :: (IsStream t, Monad m) => t m a -> m Bool
444null m =
445 let stop = return True
446 single _ = return False
447 yieldk _ _ = return False
448 in foldStream defState stop single yieldk m
450{-# INLINE head #-}
451head :: (IsStream t, Monad m) => t m a -> m (Maybe a)
452head m =
453 let stop = return Nothing
454 single a = return (Just a)
455 yieldk a _ = return (Just a)
456 in foldStream defState stop single yieldk m
458{-# INLINE tail #-}
459tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
460tail m =
461 let stop = return Nothing
462 single _ = return $ Just nil
463 yieldk _ r = return $ Just r
464 in foldStream defState stop single yieldk m
466{-# INLINE init #-}
467init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
468init m = go1 m
469 where
470 go1 m1 = do
471 r <- uncons m1
472 case r of
473 Nothing -> return Nothing
474 Just (h, t) -> return . Just $ go h t
475 go p m1 = mkStream $ \_ stp sng yld ->
476 let single _ = sng p
477 yieldk a x = yld p $ go a x
478 in foldStream defState stp single yieldk m1
480{-# INLINE elem #-}
481elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
482elem e m = go m
483 where
484 go m1 =
485 let stop = return False
486 single a = return (a == e)
487 yieldk a r = if a == e then return True else go r
488 in foldStream defState stop single yieldk m1
490{-# INLINE notElem #-}
491notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
492notElem e m = go m
493 where
494 go m1 =
495 let stop = return True
496 single a = return (a /= e)
497 yieldk a r = if a == e then return False else go r
498 in foldStream defState stop single yieldk m1
500all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
501all p m = go m
502 where
503 go m1 =
504 let single a | p a = return True
505 | otherwise = return False
506 yieldk a r | p a = go r
507 | otherwise = return False
508 in foldStream defState (return True) single yieldk m1
510any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
511any p m = go m
512 where
513 go m1 =
514 let single a | p a = return True
515 | otherwise = return False
516 yieldk a r | p a = return True
517 | otherwise = go r
518 in foldStream defState (return False) single yieldk m1
520-- | Extract the last element of the stream, if any.
521{-# INLINE last #-}
522last :: (IsStream t, Monad m) => t m a -> m (Maybe a)
523last = foldx (\_ y -> Just y) Nothing id
525{-# INLINE minimum #-}
526minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
527minimum m = go Nothing m
528 where
529 go Nothing m1 =
530 let stop = return Nothing
531 single a = return (Just a)
532 yieldk a r = go (Just a) r
533 in foldStream defState stop single yieldk m1
535 go (Just res) m1 =
536 let stop = return (Just res)
537 single a =
538 if res <= a
539 then return (Just res)
540 else return (Just a)
541 yieldk a r =
542 if res <= a
543 then go (Just res) r
544 else go (Just a) r
545 in foldStream defState stop single yieldk m1
547{-# INLINE minimumBy #-}
548minimumBy
549 :: (IsStream t, Monad m)
550 => (a -> a -> Ordering) -> t m a -> m (Maybe a)
551minimumBy cmp m = go Nothing m
552 where
553 go Nothing m1 =
554 let stop = return Nothing
555 single a = return (Just a)
556 yieldk a r = go (Just a) r
557 in foldStream defState stop single yieldk m1
559 go (Just res) m1 =
560 let stop = return (Just res)
561 single a = case cmp res a of
562 GT -> return (Just a)
563 _ -> return (Just res)
564 yieldk a r = case cmp res a of
565 GT -> go (Just a) r
566 _ -> go (Just res) r
567 in foldStream defState stop single yieldk m1
569{-# INLINE maximum #-}
570maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
571maximum m = go Nothing m
572 where
573 go Nothing m1 =
574 let stop = return Nothing
575 single a = return (Just a)
576 yieldk a r = go (Just a) r
577 in foldStream defState stop single yieldk m1
579 go (Just res) m1 =
580 let stop = return (Just res)
581 single a =
582 if res <= a
583 then return (Just a)
584 else return (Just res)
585 yieldk a r =
586 if res <= a
587 then go (Just a) r
588 else go (Just res) r
589 in foldStream defState stop single yieldk m1
591{-# INLINE maximumBy #-}
592maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)
593maximumBy cmp m = go Nothing m
594 where
595 go Nothing m1 =
596 let stop = return Nothing
597 single a = return (Just a)
598 yieldk a r = go (Just a) r
599 in foldStream defState stop single yieldk m1
601 go (Just res) m1 =
602 let stop = return (Just res)
603 single a = case cmp res a of
604 GT -> return (Just res)
605 _ -> return (Just a)
606 yieldk a r = case cmp res a of
607 GT -> go (Just res) r
608 _ -> go (Just a) r
609 in foldStream defState stop single yieldk m1
611{-# INLINE (!!) #-}
612(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)
613m !! i = go i m
614 where
615 go n m1 =
616 let single a | n == 0 = return $ Just a
617 | otherwise = return Nothing
618 yieldk a x | n < 0 = return Nothing
619 | n == 0 = return $ Just a
620 | otherwise = go (n - 1) x
621 in foldStream defState (return Nothing) single yieldk m1
623{-# INLINE lookup #-}
624lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)
625lookup e m = go m
626 where
627 go m1 =
628 let single (a, b) | a == e = return $ Just b
629 | otherwise = return Nothing
630 yieldk (a, b) x | a == e = return $ Just b
631 | otherwise = go x
632 in foldStream defState (return Nothing) single yieldk m1
634{-# INLINE findM #-}
635findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)
636findM p m = go m
637 where
638 go m1 =
639 let single a = do
640 b <- p a
641 if b then return $ Just a else return Nothing
642 yieldk a x = do
643 b <- p a
644 if b then return $ Just a else go x
645 in foldStream defState (return Nothing) single yieldk m1
647{-# INLINE find #-}
648find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)
649find p = findM (return . p)
651{-# INLINE findIndices #-}
652findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int
653findIndices p = go 0
654 where
655 go offset m1 = mkStream $ \st stp sng yld ->
656 let single a | p a = sng offset
657 | otherwise = stp
658 yieldk a x | p a = yld offset $ go (offset + 1) x
659 | otherwise = foldStream (adaptState st) stp sng yld $
660 go (offset + 1) x
661 in foldStream (adaptState st) stp single yieldk m1
663------------------------------------------------------------------------------
664-- Map and Fold
665------------------------------------------------------------------------------
667-- | Apply a monadic action to each element of the stream and discard the
668-- output of the action.
669mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()
670mapM_ f m = go m
671 where
672 go m1 =
673 let stop = return ()
674 single a = void (f a)
675 yieldk a r = f a >> go r
676 in foldStream defState stop single yieldk m1
678------------------------------------------------------------------------------
679-- Converting folds
680------------------------------------------------------------------------------
682{-# INLINABLE toList #-}
683toList :: (IsStream t, Monad m) => t m a -> m [a]
684toList = foldr (:) []
686{-# INLINE toStreamK #-}
687toStreamK :: Stream m a -> Stream m a
688toStreamK = id
690-------------------------------------------------------------------------------
691-- Transformation by folding (Scans)
692-------------------------------------------------------------------------------
694{-# INLINE scanx #-}
695scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
696scanx step begin done m =
697 cons (done begin) $ go m begin
698 where
699 go m1 !acc = mkStream $ \st stp sng yld ->
700 let single a = sng (done $ step acc a)
701 yieldk a r =
702 let s = step acc a
703 in yld (done s) (go r s)
704 in foldStream (adaptState st) stp single yieldk m1
706{-# INLINE scanl' #-}
707scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
708scanl' step begin = scanx step begin id
710-------------------------------------------------------------------------------
711-- Filtering
712-------------------------------------------------------------------------------
714{-# INLINE filter #-}
715filter :: IsStream t => (a -> Bool) -> t m a -> t m a
716filter p m = go m
717 where
718 go m1 = mkStream $ \st stp sng yld ->
719 let single a | p a = sng a
720 | otherwise = stp
721 yieldk a r | p a = yld a (go r)
722 | otherwise = foldStream st stp single yieldk r
723 in foldStream st stp single yieldk m1
725{-# INLINE take #-}
726take :: IsStream t => Int -> t m a -> t m a
727take n m = go n m
728 where
729 go n1 m1 = mkStream $ \st stp sng yld ->
730 let yieldk a r = yld a (go (n1 - 1) r)
731 in if n1 <= 0
732 then stp
733 else foldStream st stp sng yieldk m1
735{-# INLINE takeWhile #-}
736takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
737takeWhile p m = go m
738 where
739 go m1 = mkStream $ \st stp sng yld ->
740 let single a | p a = sng a
741 | otherwise = stp
742 yieldk a r | p a = yld a (go r)
743 | otherwise = stp
744 in foldStream st stp single yieldk m1
746drop :: IsStream t => Int -> t m a -> t m a
747drop n m = mkStream $ \st stp sng yld ->
748 foldStream st stp sng yld $ go n m
749 where
750 go n1 m1 = mkStream $ \st stp sng yld ->
751 let single _ = stp
752 yieldk _ r = foldStream st stp sng yld $ go (n1 - 1) r
753 -- Somehow "<=" check performs better than a ">"
754 in if n1 <= 0
755 then foldStream st stp sng yld m1
756 else foldStream st stp single yieldk m1
758{-# INLINE dropWhile #-}
759dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
760dropWhile p m = go m
761 where
762 go m1 = mkStream $ \st stp sng yld ->
763 let single a | p a = stp
764 | otherwise = sng a
765 yieldk a r | p a = foldStream st stp single yieldk r
766 | otherwise = yld a r
767 in foldStream st stp single yieldk m1
769-------------------------------------------------------------------------------
770-- Mapping
771-------------------------------------------------------------------------------
773-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
774-- other stream types to overload it.
775{-# INLINE mapM #-}
776mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
777mapM f m = go m
778 where
779 go m1 = mkStream $ \st stp sng yld ->
780 let single a = f a >>= sng
781 yieldk a r = foldStreamShared st stp sng yld $ f a |: go r
782 in foldStream (adaptState st) stp single yieldk m1
784-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
785-- other stream types to overload it.
786{-# INLINE sequence #-}
787sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
788sequence m = go m
789 where
790 go m1 = mkStream $ \st stp sng yld ->
791 let single ma = ma >>= sng
792 yieldk ma r = foldStreamShared st stp sng yld $ ma |: go r
793 in foldStream (adaptState st) stp single yieldk m1
795-------------------------------------------------------------------------------
796-- Inserting
797-------------------------------------------------------------------------------
799{-# INLINE intersperseM #-}
800intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
801intersperseM a m = prependingStart m
802 where
803 prependingStart m1 = mkStream $ \st stp sng yld ->
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
811{-# INLINE insertBy #-}
812insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a
813insertBy cmp x m = go m
814 where
815 go m1 = mkStream $ \st _ _ yld ->
816 let single a = case cmp x a of
817 GT -> yld a (yield x)
818 _ -> yld x (yield a)
819 stop = yld x nil
820 yieldk a r = case cmp x a of
821 GT -> yld a (go r)
822 _ -> yld x (a `cons` r)
823 in foldStream st stop single yieldk m1
825------------------------------------------------------------------------------
826-- Deleting
827------------------------------------------------------------------------------
829{-# INLINE deleteBy #-}
830deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a
831deleteBy eq x m = go m
832 where
833 go m1 = mkStream $ \st stp sng yld ->
834 let single a = if eq x a then stp else sng a
835 yieldk a r = if eq x a
836 then foldStream st stp sng yld r
837 else yld a (go r)
838 in foldStream st stp single yieldk m1
840-------------------------------------------------------------------------------
841-- Map and Filter
842-------------------------------------------------------------------------------
844{-# INLINE mapMaybe #-}
845mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b
846mapMaybe f m = go m
847 where
848 go m1 = mkStream $ \st stp sng yld ->
849 let single a = case f a of
850 Just b -> sng b
851 Nothing -> stp
852 yieldk a r = case f a of
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
857------------------------------------------------------------------------------
858-- Serial Zipping
859------------------------------------------------------------------------------
861-- | Zip two streams serially using a pure zipping function.
862--
863-- @since 0.1.0
864{-# INLINABLE zipWith #-}
865zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
866zipWith f = go
867 where
868 go mx my = mkStream $ \st stp sng yld -> do
869 let merge a ra =
870 let single2 b = sng (f a b)
871 yield2 b rb = yld (f a b) (go ra rb)
872 in foldStream (adaptState st) stp single2 yield2 my
873 let single1 a = merge a nil
874 yield1 = merge
875 foldStream (adaptState st) stp single1 yield1 mx
877-- | Zip two streams serially using a monadic zipping function.
878--
879-- @since 0.1.0
880zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
881zipWithM f m1 m2 = go m1 m2
882 where
883 go mx my = mkStream $ \st stp sng yld -> do
884 let merge a ra =
885 let runIt x = foldStream st stp sng yld x
886 single2 b = f a b >>= sng
887 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)
888 in foldStream (adaptState st) stp single2 yield2 my
889 let single1 a = merge a nil
890 yield1 = merge
891 foldStream (adaptState st) stp single1 yield1 mx
893------------------------------------------------------------------------------
894-- Merging
895------------------------------------------------------------------------------
897{-# INLINE mergeByM #-}
898mergeByM
899 :: (IsStream t, Monad m)
900 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
901mergeByM cmp = go
902 where
903 go mx my = mkStream $ \st stp sng yld -> do
904 let mergeWithY a ra =
905 let stop2 = foldStream st stp sng yld mx
906 single2 b = do
907 r <- cmp a b
908 case r of
909 GT -> yld b (go (a `cons` ra) nil)
910 _ -> yld a (go ra (b `cons` nil))
911 yield2 b rb = do
912 r <- cmp a b
913 case r of
914 GT -> yld b (go (a `cons` ra) rb)
915 _ -> yld a (go ra (b `cons` rb))
916 in foldStream st stop2 single2 yield2 my
917 let stopX = foldStream st stp sng yld my
918 singleX a = mergeWithY a nil
919 yieldX = mergeWithY
920 foldStream st stopX singleX yieldX mx
922{-# INLINABLE mergeBy #-}
923mergeBy
924 :: (IsStream t, Monad m)
925 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
926mergeBy cmp = mergeByM (\a b -> return $ cmp a b)
928------------------------------------------------------------------------------
929-- Transformation comprehensions
930------------------------------------------------------------------------------
932{-# INLINE the #-}
933the :: (Eq a, IsStream t, Monad m) => t m a -> m (Maybe a)
934the m = do
935 r <- uncons m
936 case r of
937 Nothing -> return Nothing
938 Just (h, t) -> go h t
939 where
940 go h m1 =
941 let single a | h == a = return $ Just h
942 | otherwise = return Nothing
943 yieldk a r | h == a = go h r
944 | otherwise = return Nothing
945 in foldStream defState (return $ Just h) single yieldk m1
947-------------------------------------------------------------------------------
948-- Bind utility
949-------------------------------------------------------------------------------
951{-# INLINE bindWith #-}
952bindWith
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
958bindWith par m1 f = go m1
959 where
960 go m =
961 mkStream $ \st stp sng yld ->
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
967------------------------------------------------------------------------------
968-- Alternative & MonadPlus
969------------------------------------------------------------------------------
971_alt :: Stream m a -> Stream m a -> Stream m a
972_alt m1 m2 = mkStream $ \st stp sng yld ->
973 let stop = foldStream st stp sng yld m2
974 in foldStream st stop sng yld m1
976------------------------------------------------------------------------------
977-- MonadReader
978------------------------------------------------------------------------------
980withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
981withLocal f m =
982 mkStream $ \st stp sng yld ->
983 let single = local f . sng
984 yieldk a r = local f $ yld a (withLocal f r)
985 in foldStream st (local f stp) single yieldk m
987------------------------------------------------------------------------------
988-- MonadError
989------------------------------------------------------------------------------
991{-
992-- XXX handle and test cross thread state transfer
993withCatchError
994 :: MonadError e m
995 => Stream m a -> (e -> Stream m a) -> Stream m a
996withCatchError m h =
997 mkStream $ \_ stp sng yld ->
998 let run x = unStream x Nothing stp sng yieldk
999 handle r = r `catchError` \e -> run $ h e
1000 yieldk a r = yld a (withCatchError r h)
1001 in handle $ run m