git.iko.soy/streamly
on 24375dc help sign in
streamly/src/Streamly/Streams/StreamK.hs
978 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-- |
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 -- * Deconstruction
53 , uncons
55 -- * Generation
56 -- ** Unfolds
57 , unfoldr
58 , unfoldrM
60 -- ** Specialized Generation
61 , repeat
62 , replicate
63 , replicateM
65 -- ** Conversions
66 , yield
67 , yieldM
68 , fromFoldable
69 , fromList
70 , fromStreamK
72 -- * Elimination
73 -- ** General Folds
74 , foldr
75 , foldrM
76 , foldr1
77 , foldl'
78 , foldlM'
79 , foldx
80 , foldxM
82 -- ** Specialized Folds
83 , runStream
84 , null
85 , head
86 , tail
87 , init
88 , elem
89 , notElem
90 , all
91 , any
92 , last
93 , minimum
94 , minimumBy
95 , maximum
96 , maximumBy
97 , findIndices
98 , lookup
99 , findM
100 , find
101 , (!!)
103 -- ** Map and Fold
104 , mapM_
106 -- ** Conversions
107 , toList
108 , toStreamK
110 -- * Transformation
111 -- ** By folding (scans)
112 , scanl'
113 , scanx
115 -- ** Filtering
116 , filter
117 , take
118 , takeWhile
119 , drop
120 , dropWhile
122 -- ** Mapping
123 , map
124 , mapM
125 , sequence
127 -- ** Inserting
128 , intersperseM
129 , insertBy
131 -- ** Deleting
132 , deleteBy
134 -- ** Map and Filter
135 , mapMaybe
137 -- ** Zipping
138 , zipWith
139 , zipWithM
141 -- ** Merging
142 , mergeBy
143 , mergeByM
145 -- ** Transformation comprehensions
146 , the
148 -- * Semigroup Style Composition
149 , serial
151 -- * Utilities
152 , consMSerial
153 , bindWith
154 , withLocal
156 -- * Deprecated
157 , Streaming -- deprecated
158 , once -- deprecated
159 )
160where
162import Control.Monad (void)
163import Control.Monad.Reader.Class (MonadReader(..))
164import Prelude
165 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,
166 take, filter, all, any, takeWhile, drop, dropWhile, minimum,
167 maximum, elem, notElem, null, head, tail, init, zipWith, lookup,
168 foldr1, (!!), replicate)
169import qualified Prelude
171import Streamly.SVar
172import Streamly.Streams.StreamKType
174-- | Detach a stream from an SVar
175{-# INLINE unShare #-}
176unShare :: IsStream t => t m a -> t m a
177unShare x = mkStream $ \st stp sng yld ->
178 foldStream st stp sng yld x
180------------------------------------------------------------------------------
181-- Construction
182------------------------------------------------------------------------------
184infixr 5 `cons`
186-- faster than consM because there is no bind.
187-- | Construct a stream by adding a pure value at the head of an existing
188-- stream. For serial streams this is the same as @(return a) \`consM` r@ but
189-- more efficient. For concurrent streams this is not concurrent whereas
190-- 'consM' is concurrent. For example:
191--
192-- @
193-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
194-- [1,2,3]
195-- @
196--
197-- @since 0.1.0
198cons :: IsStream t => a -> t m a -> t m a
199cons a r = mkStream $ \_ _ _ yld -> yld a r
201infixr 5 .:
203-- | Operator equivalent of 'cons'.
204--
205-- @
206-- > toList $ 1 .: 2 .: 3 .: nil
207-- [1,2,3]
208-- @
209--
210-- @since 0.1.1
211(.:) :: IsStream t => a -> t m a -> t m a
212(.:) = cons
214-------------------------------------------------------------------------------
215-- Deconstruction
216-------------------------------------------------------------------------------
218{-# INLINE uncons #-}
219uncons :: (IsStream t, Monad m) => t m a -> m (Maybe (a, t m a))
220uncons m =
221 let stop = return Nothing
222 single a = return (Just (a, nil))
223 yieldk a r = return (Just (a, r))
224 in foldStream defState stop single yieldk m
226-------------------------------------------------------------------------------
227-- Generation
228-------------------------------------------------------------------------------
230{-# INLINE unfoldr #-}
231unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
232unfoldr step = go
233 where
234 go s = mkStream $ \_ stp _ yld ->
235 case step s of
236 Nothing -> stp
237 Just (a, b) -> yld a (go b)
239{-# INLINE unfoldrM #-}
240unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
241unfoldrM step = go
242 where
243 go s = mkStream $ \st stp sng yld -> do
244 mayb <- step s
245 case mayb of
246 Nothing -> stp
247 Just (a, b) ->
248 foldStreamShared st stp sng yld $ return a |: go b
250-------------------------------------------------------------------------------
251-- Special generation
252-------------------------------------------------------------------------------
254yield :: IsStream t => a -> t m a
255yield a = mkStream $ \_ _ single _ -> single a
257-- | Same as yieldM
258--
259-- @since 0.2.0
260{-# DEPRECATED once "Please use yieldM instead." #-}
261{-# INLINE once #-}
262once :: (Monad m, IsStream t) => m a -> t m a
263once = yieldM
265-- |
266-- @
267-- repeatM = fix . cons
268-- repeatM = cycle1 . yield
269-- @
270--
271-- Generate an infinite stream by repeating a pure value.
272--
273-- @since 0.4.0
274repeat :: IsStream t => a -> t m a
275repeat a = let x = cons a x in x
277replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
278replicateM n m = go n
279 where
280 go cnt = if cnt <= 0 then nil else m |: go (cnt - 1)
282replicate :: IsStream t => Int -> a -> t m a
283replicate n a = go n
284 where
285 go cnt = if cnt <= 0 then nil else a `cons` go (cnt - 1)
287-------------------------------------------------------------------------------
288-- Conversions
289-------------------------------------------------------------------------------
291-- |
292-- @
293-- fromFoldable = 'Prelude.foldr' 'cons' 'nil'
294-- @
295--
296-- Construct a stream from a 'Foldable' containing pure values:
297--
298-- @since 0.2.0
299{-# INLINE fromFoldable #-}
300fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
301fromFoldable = Prelude.foldr cons nil
303{-# INLINE fromList #-}
304fromList :: IsStream t => [a] -> t m a
305fromList = fromFoldable
307{-# INLINE fromStreamK #-}
308fromStreamK :: IsStream t => Stream m a -> t m a
309fromStreamK = fromStream
311-------------------------------------------------------------------------------
312-- Elimination by Folding
313-------------------------------------------------------------------------------
315-- | Lazy right associative fold.
316foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b
317foldr step acc m = go m
318 where
319 go m1 =
320 let stop = return acc
321 single a = return (step a acc)
322 yieldk a r = go r >>= \b -> return (step a b)
323 in foldStream defState stop single yieldk m1
325-- | Lazy right fold with a monadic step function.
326{-# INLINE foldrM #-}
327foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b
328foldrM step acc m = go m
329 where
330 go m1 =
331 let stop = return acc
332 single a = step a acc
333 yieldk a r = go r >>= step a
334 in foldStream defState stop single yieldk m1
336{-# INLINE foldr1 #-}
337foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)
338foldr1 step m = do
339 r <- uncons m
340 case r of
341 Nothing -> return Nothing
342 Just (h, t) -> fmap Just (go h t)
343 where
344 go p m1 =
345 let stp = return p
346 single a = return $ step a p
347 yieldk a r = fmap (step p) (go a r)
348 in foldStream defState stp single yieldk m1
350-- | Strict left fold with an extraction function. Like the standard strict
351-- left fold, but applies a user supplied extraction function (the third
352-- argument) to the folded value at the end. This is designed to work with the
353-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
354--
355-- Note that the accumulator is always evaluated including the initial value.
356{-# INLINE foldx #-}
357foldx :: forall t m a b x. (IsStream t, Monad m)
358 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
359foldx step begin done m = get $ go m begin
360 where
361 {-# NOINLINE get #-}
362 get :: t m x -> m b
363 get m1 =
364 -- XXX we are not strictly evaluating the accumulator here. Is this
365 -- okay?
366 let single = return . done
367 -- XXX this is foldSingleton. why foldStreamShared?
368 in foldStreamShared undefined undefined single undefined m1
370 -- Note, this can be implemented by making a recursive call to "go",
371 -- however that is more expensive because of unnecessary recursion
372 -- that cannot be tail call optimized. Unfolding recursion explicitly via
373 -- continuations is much more efficient.
374 go :: t m a -> x -> t m x
375 go m1 !acc = mkStream $ \_ _ sng yld ->
376 let stop = sng acc
377 single a = sng $ step acc a
378 -- XXX this is foldNonEmptyStream
379 yieldk a r = foldStream defState undefined sng yld $
380 go r (step acc a)
381 in foldStream defState stop single yieldk m1
383-- | Strict left associative fold.
384{-# INLINE foldl' #-}
385foldl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> m b
386foldl' step begin = foldx step begin id
388-- XXX replace the recursive "go" with explicit continuations.
389-- | Like 'foldx', but with a monadic step function.
390foldxM :: (IsStream t, Monad m)
391 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b
392foldxM step begin done m = go begin m
393 where
394 go !acc m1 =
395 let stop = acc >>= done
396 single a = acc >>= \b -> step b a >>= done
397 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r
398 in foldStream defState stop single yieldk m1
400-- | Like 'foldl'' but with a monadic step function.
401foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
402foldlM' step begin = foldxM step (return begin) return
404------------------------------------------------------------------------------
405-- Specialized folds
406------------------------------------------------------------------------------
408{-# INLINE runStream #-}
409runStream :: (Monad m, IsStream t) => t m a -> m ()
410runStream = go
411 where
412 go m1 =
413 let stop = return ()
414 single _ = return ()
415 yieldk _ r = go r
416 in foldStream defState stop single yieldk m1
418{-# INLINE null #-}
419null :: (IsStream t, Monad m) => t m a -> m Bool
420null m =
421 let stop = return True
422 single _ = return False
423 yieldk _ _ = return False
424 in foldStream defState stop single yieldk m
426{-# INLINE head #-}
427head :: (IsStream t, Monad m) => t m a -> m (Maybe a)
428head m =
429 let stop = return Nothing
430 single a = return (Just a)
431 yieldk a _ = return (Just a)
432 in foldStream defState stop single yieldk m
434{-# INLINE tail #-}
435tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
436tail m =
437 let stop = return Nothing
438 single _ = return $ Just nil
439 yieldk _ r = return $ Just r
440 in foldStream defState stop single yieldk m
442{-# INLINE init #-}
443init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
444init m = go1 m
445 where
446 go1 m1 = do
447 r <- uncons m1
448 case r of
449 Nothing -> return Nothing
450 Just (h, t) -> return . Just $ go h t
451 go p m1 = mkStream $ \_ stp sng yld ->
452 let single _ = sng p
453 yieldk a x = yld p $ go a x
454 in foldStream defState stp single yieldk m1
456{-# INLINE elem #-}
457elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
458elem e m = go m
459 where
460 go m1 =
461 let stop = return False
462 single a = return (a == e)
463 yieldk a r = if a == e then return True else go r
464 in foldStream defState stop single yieldk m1
466{-# INLINE notElem #-}
467notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
468notElem e m = go m
469 where
470 go m1 =
471 let stop = return True
472 single a = return (a /= e)
473 yieldk a r = if a == e then return False else go r
474 in foldStream defState stop single yieldk m1
476all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
477all p m = go m
478 where
479 go m1 =
480 let single a | p a = return True
481 | otherwise = return False
482 yieldk a r | p a = go r
483 | otherwise = return False
484 in foldStream defState (return True) single yieldk m1
486any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
487any p m = go m
488 where
489 go m1 =
490 let single a | p a = return True
491 | otherwise = return False
492 yieldk a r | p a = return True
493 | otherwise = go r
494 in foldStream defState (return False) single yieldk m1
496-- | Extract the last element of the stream, if any.
497{-# INLINE last #-}
498last :: (IsStream t, Monad m) => t m a -> m (Maybe a)
499last = foldx (\_ y -> Just y) Nothing id
501{-# INLINE minimum #-}
502minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
503minimum m = go Nothing m
504 where
505 go Nothing m1 =
506 let stop = return Nothing
507 single a = return (Just a)
508 yieldk a r = go (Just a) r
509 in foldStream defState stop single yieldk m1
511 go (Just res) m1 =
512 let stop = return (Just res)
513 single a =
514 if res <= a
515 then return (Just res)
516 else return (Just a)
517 yieldk a r =
518 if res <= a
519 then go (Just res) r
520 else go (Just a) r
521 in foldStream defState stop single yieldk m1
523{-# INLINE minimumBy #-}
524minimumBy
525 :: (IsStream t, Monad m)
526 => (a -> a -> Ordering) -> t m a -> m (Maybe a)
527minimumBy cmp 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 = case cmp res a of
538 GT -> return (Just a)
539 _ -> return (Just res)
540 yieldk a r = case cmp res a of
541 GT -> go (Just a) r
542 _ -> go (Just res) r
543 in foldStream defState stop single yieldk m1
545{-# INLINE maximum #-}
546maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
547maximum m = go Nothing m
548 where
549 go Nothing m1 =
550 let stop = return Nothing
551 single a = return (Just a)
552 yieldk a r = go (Just a) r
553 in foldStream defState stop single yieldk m1
555 go (Just res) m1 =
556 let stop = return (Just res)
557 single a =
558 if res <= a
559 then return (Just a)
560 else return (Just res)
561 yieldk a r =
562 if res <= a
563 then go (Just a) r
564 else go (Just res) r
565 in foldStream defState stop single yieldk m1
567{-# INLINE maximumBy #-}
568maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)
569maximumBy cmp m = go Nothing m
570 where
571 go Nothing m1 =
572 let stop = return Nothing
573 single a = return (Just a)
574 yieldk a r = go (Just a) r
575 in foldStream defState stop single yieldk m1
577 go (Just res) m1 =
578 let stop = return (Just res)
579 single a = case cmp res a of
580 GT -> return (Just res)
581 _ -> return (Just a)
582 yieldk a r = case cmp res a of
583 GT -> go (Just res) r
584 _ -> go (Just a) r
585 in foldStream defState stop single yieldk m1
587{-# INLINE (!!) #-}
588(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)
589m !! i = go i m
590 where
591 go n m1 =
592 let single a | n == 0 = return $ Just a
593 | otherwise = return Nothing
594 yieldk a x | n < 0 = return Nothing
595 | n == 0 = return $ Just a
596 | otherwise = go (n - 1) x
597 in foldStream defState (return Nothing) single yieldk m1
599{-# INLINE lookup #-}
600lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)
601lookup e m = go m
602 where
603 go m1 =
604 let single (a, b) | a == e = return $ Just b
605 | otherwise = return Nothing
606 yieldk (a, b) x | a == e = return $ Just b
607 | otherwise = go x
608 in foldStream defState (return Nothing) single yieldk m1
610{-# INLINE findM #-}
611findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)
612findM p m = go m
613 where
614 go m1 =
615 let single a = do
616 b <- p a
617 if b then return $ Just a else return Nothing
618 yieldk a x = do
619 b <- p a
620 if b then return $ Just a else go x
621 in foldStream defState (return Nothing) single yieldk m1
623{-# INLINE find #-}
624find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)
625find p = findM (return . p)
627{-# INLINE findIndices #-}
628findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int
629findIndices p = go 0
630 where
631 go offset m1 = mkStream $ \st stp sng yld ->
632 let single a | p a = sng offset
633 | otherwise = stp
634 yieldk a x | p a = yld offset $ go (offset + 1) x
635 | otherwise = foldStream (adaptState st) stp sng yld $
636 go (offset + 1) x
637 in foldStream (adaptState st) stp single yieldk m1
639------------------------------------------------------------------------------
640-- Map and Fold
641------------------------------------------------------------------------------
643-- | Apply a monadic action to each element of the stream and discard the
644-- output of the action.
645mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()
646mapM_ f m = go m
647 where
648 go m1 =
649 let stop = return ()
650 single a = void (f a)
651 yieldk a r = f a >> go r
652 in foldStream defState stop single yieldk m1
654------------------------------------------------------------------------------
655-- Converting folds
656------------------------------------------------------------------------------
658{-# INLINABLE toList #-}
659toList :: (IsStream t, Monad m) => t m a -> m [a]
660toList = foldr (:) []
662{-# INLINE toStreamK #-}
663toStreamK :: Stream m a -> Stream m a
664toStreamK = id
666-------------------------------------------------------------------------------
667-- Transformation by folding (Scans)
668-------------------------------------------------------------------------------
670{-# INLINE scanx #-}
671scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
672scanx step begin done m =
673 cons (done begin) $ go m begin
674 where
675 go m1 !acc = mkStream $ \st stp sng yld ->
676 let single a = sng (done $ step acc a)
677 yieldk a r =
678 let s = step acc a
679 in yld (done s) (go r s)
680 in foldStream (adaptState st) stp single yieldk m1
682{-# INLINE scanl' #-}
683scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
684scanl' step begin = scanx step begin id
686-------------------------------------------------------------------------------
687-- Filtering
688-------------------------------------------------------------------------------
690{-# INLINE filter #-}
691filter :: IsStream t => (a -> Bool) -> t m a -> t m a
692filter p m = go m
693 where
694 go m1 = mkStream $ \st stp sng yld ->
695 let single a | p a = sng a
696 | otherwise = stp
697 yieldk a r | p a = yld a (go r)
698 | otherwise = foldStream st stp single yieldk r
699 in foldStream st stp single yieldk m1
701{-# INLINE take #-}
702take :: IsStream t => Int -> t m a -> t m a
703take n m = go n m
704 where
705 go n1 m1 = mkStream $ \st stp sng yld ->
706 let yieldk a r = yld a (go (n1 - 1) r)
707 in if n1 <= 0
708 then stp
709 else foldStream st stp sng yieldk m1
711{-# INLINE takeWhile #-}
712takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
713takeWhile p m = go m
714 where
715 go m1 = mkStream $ \st stp sng yld ->
716 let single a | p a = sng a
717 | otherwise = stp
718 yieldk a r | p a = yld a (go r)
719 | otherwise = stp
720 in foldStream st stp single yieldk m1
722drop :: IsStream t => Int -> t m a -> t m a
723drop n m = mkStream $ \st stp sng yld ->
724 foldStream st stp sng yld $ go n m
725 where
726 go n1 m1 = mkStream $ \st stp sng yld ->
727 let single _ = stp
728 yieldk _ r = foldStream st stp sng yld $ go (n1 - 1) r
729 -- Somehow "<=" check performs better than a ">"
730 in if n1 <= 0
731 then foldStream st stp sng yld m1
732 else foldStream st stp single yieldk m1
734{-# INLINE dropWhile #-}
735dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
736dropWhile p m = go m
737 where
738 go m1 = mkStream $ \st stp sng yld ->
739 let single a | p a = stp
740 | otherwise = sng a
741 yieldk a r | p a = foldStream st stp single yieldk r
742 | otherwise = yld a r
743 in foldStream st stp single yieldk m1
745-------------------------------------------------------------------------------
746-- Mapping
747-------------------------------------------------------------------------------
749-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
750-- other stream types to overload it.
751{-# INLINE mapM #-}
752mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
753mapM f m = go m
754 where
755 go m1 = mkStream $ \st stp sng yld ->
756 let single a = f a >>= sng
757 yieldk a r = foldStreamShared st stp sng yld $ f a |: go r
758 in foldStream (adaptState st) stp single yieldk m1
760-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
761-- other stream types to overload it.
762{-# INLINE sequence #-}
763sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
764sequence m = go m
765 where
766 go m1 = mkStream $ \st stp sng yld ->
767 let single ma = ma >>= sng
768 yieldk ma r = foldStreamShared st stp sng yld $ ma |: go r
769 in foldStream (adaptState st) stp single yieldk m1
771-------------------------------------------------------------------------------
772-- Inserting
773-------------------------------------------------------------------------------
775{-# INLINE intersperseM #-}
776intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
777intersperseM a m = prependingStart m
778 where
779 prependingStart m1 = mkStream $ \st stp sng yld ->
780 let yieldk i x = foldStreamShared st stp sng yld $ return i |: go x
781 in foldStream st stp sng yieldk m1
782 go m2 = mkStream $ \st stp sng yld ->
783 let single i = foldStreamShared st stp sng yld $ a |: yield i
784 yieldk i x = foldStreamShared st stp sng yld $ a |: return i |: go x
785 in foldStream st stp single yieldk m2
787{-# INLINE insertBy #-}
788insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a
789insertBy cmp x m = go m
790 where
791 go m1 = mkStream $ \st _ _ yld ->
792 let single a = case cmp x a of
793 GT -> yld a (yield x)
794 _ -> yld x (yield a)
795 stop = yld x nil
796 yieldk a r = case cmp x a of
797 GT -> yld a (go r)
798 _ -> yld x (a `cons` r)
799 in foldStream st stop single yieldk m1
801------------------------------------------------------------------------------
802-- Deleting
803------------------------------------------------------------------------------
805{-# INLINE deleteBy #-}
806deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a
807deleteBy eq x m = go m
808 where
809 go m1 = mkStream $ \st stp sng yld ->
810 let single a = if eq x a then stp else sng a
811 yieldk a r = if eq x a
812 then foldStream st stp sng yld r
813 else yld a (go r)
814 in foldStream st stp single yieldk m1
816-------------------------------------------------------------------------------
817-- Map and Filter
818-------------------------------------------------------------------------------
820{-# INLINE mapMaybe #-}
821mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b
822mapMaybe f m = go m
823 where
824 go m1 = mkStream $ \st stp sng yld ->
825 let single a = case f a of
826 Just b -> sng b
827 Nothing -> stp
828 yieldk a r = case f a of
829 Just b -> yld b $ go r
830 Nothing -> foldStream (adaptState st) stp single yieldk r
831 in foldStream (adaptState st) stp single yieldk m1
833------------------------------------------------------------------------------
834-- Serial Zipping
835------------------------------------------------------------------------------
837-- | Zip two streams serially using a pure zipping function.
838--
839-- @since 0.1.0
840{-# INLINABLE zipWith #-}
841zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
842zipWith f = go
843 where
844 go mx my = mkStream $ \st stp sng yld -> do
845 let merge a ra =
846 let single2 b = sng (f a b)
847 yield2 b rb = yld (f a b) (go ra rb)
848 in foldStream (adaptState st) stp single2 yield2 my
849 let single1 a = merge a nil
850 yield1 = merge
851 foldStream (adaptState st) stp single1 yield1 mx
853-- | Zip two streams serially using a monadic zipping function.
854--
855-- @since 0.1.0
856zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
857zipWithM f m1 m2 = go m1 m2
858 where
859 go mx my = mkStream $ \st stp sng yld -> do
860 let merge a ra =
861 let runIt x = foldStream st stp sng yld x
862 single2 b = f a b >>= sng
863 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)
864 in foldStream (adaptState st) stp single2 yield2 my
865 let single1 a = merge a nil
866 yield1 = merge
867 foldStream (adaptState st) stp single1 yield1 mx
869------------------------------------------------------------------------------
870-- Merging
871------------------------------------------------------------------------------
873{-# INLINE mergeByM #-}
874mergeByM
875 :: (IsStream t, Monad m)
876 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
877mergeByM cmp = go
878 where
879 go mx my = mkStream $ \st stp sng yld -> do
880 let mergeWithY a ra =
881 let stop2 = foldStream st stp sng yld mx
882 single2 b = do
883 r <- cmp a b
884 case r of
885 GT -> yld b (go (a `cons` ra) nil)
886 _ -> yld a (go ra (b `cons` nil))
887 yield2 b rb = do
888 r <- cmp a b
889 case r of
890 GT -> yld b (go (a `cons` ra) rb)
891 _ -> yld a (go ra (b `cons` rb))
892 in foldStream st stop2 single2 yield2 my
893 let stopX = foldStream st stp sng yld my
894 singleX a = mergeWithY a nil
895 yieldX = mergeWithY
896 foldStream st stopX singleX yieldX mx
898{-# INLINABLE mergeBy #-}
899mergeBy
900 :: (IsStream t, Monad m)
901 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
902mergeBy cmp = mergeByM (\a b -> return $ cmp a b)
904------------------------------------------------------------------------------
905-- Transformation comprehensions
906------------------------------------------------------------------------------
908{-# INLINE the #-}
909the :: (Eq a, IsStream t, Monad m) => t m a -> m (Maybe a)
910the m = do
911 r <- uncons m
912 case r of
913 Nothing -> return Nothing
914 Just (h, t) -> go h t
915 where
916 go h m1 =
917 let single a | h == a = return $ Just h
918 | otherwise = return Nothing
919 yieldk a r | h == a = go h r
920 | otherwise = return Nothing
921 in foldStream defState (return $ Just h) single yieldk m1
923-------------------------------------------------------------------------------
924-- Bind utility
925-------------------------------------------------------------------------------
927{-# INLINE bindWith #-}
928bindWith
929 :: IsStream t
930 => (forall c. t m c -> t m c -> t m c)
931 -> t m a
932 -> (a -> t m b)
933 -> t m b
934bindWith par m1 f = go m1
935 where
936 go m =
937 mkStream $ \st stp sng yld ->
938 let foldShared = foldStreamShared st stp sng yld
939 single a = foldShared $ unShare (f a)
940 yieldk a r = foldShared $ unShare (f a) `par` go r
941 in foldStream (adaptState st) stp single yieldk m
943------------------------------------------------------------------------------
944-- Alternative & MonadPlus
945------------------------------------------------------------------------------
947_alt :: Stream m a -> Stream m a -> Stream m a
948_alt m1 m2 = mkStream $ \st stp sng yld ->
949 let stop = foldStream st stp sng yld m2
950 in foldStream st stop sng yld m1
952------------------------------------------------------------------------------
953-- MonadReader
954------------------------------------------------------------------------------
956withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
957withLocal f m =
958 mkStream $ \st stp sng yld ->
959 let single = local f . sng
960 yieldk a r = local f $ yld a (withLocal f r)
961 in foldStream st (local f stp) single yieldk m
963------------------------------------------------------------------------------
964-- MonadError
965------------------------------------------------------------------------------
967{-
968-- XXX handle and test cross thread state transfer
969withCatchError
970 :: MonadError e m
971 => Stream m a -> (e -> Stream m a) -> Stream m a
972withCatchError m h =
973 mkStream $ \_ stp sng yld ->
974 let run x = unStream x Nothing stp sng yieldk
975 handle r = r `catchError` \e -> run $ h e
976 yieldk a r = yld a (withCatchError r h)
977 in handle $ run m
978-}