git.iko.soy/streamly
on 57fad21 help sign in
streamly/src/Streamly/Streams/StreamK.hs
977 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.StreamK.Type
174-- | Detach a stream from an SVar
175{-# INLINE unShare #-}
176unShare :: IsStream t => t m a -> t m a
177unShare x = mkStream $ \st yld sng stp ->
178 foldStream st yld sng stp 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 yieldk single stop 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 $ \_ yld _ stp ->
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 yld sng stp -> do
244 mayb <- step s
245 case mayb of
246 Nothing -> stp
247 Just (a, b) ->
248 foldStreamShared st yld sng stp $ 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 yieldk single stop 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 yieldk single stop 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 yieldk single stp 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 $ \_ yld sng _ ->
376 let stop = sng acc
377 single a = sng $ step acc a
378 -- XXX this is foldNonEmptyStream
379 yieldk a r = foldStream defState yld sng undefined $
380 go r (step acc a)
381 in foldStream defState yieldk single stop 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 yieldk single stop 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 yieldk single stop 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 yieldk single stop 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 yieldk single stop 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 yieldk single stop 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 $ \_ yld sng stp ->
452 let single _ = sng p
453 yieldk a x = yld p $ go a x
454 in foldStream defState yieldk single stp 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 yieldk single stop 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 yieldk single stop 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 yieldk single (return True) 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 yieldk single (return False) 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 yieldk single stop 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 yieldk single stop 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 yieldk single stop 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 yieldk single stop 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 yieldk single stop 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 yieldk single stop 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 yieldk single stop 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 yieldk single stop 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 yieldk single (return Nothing) 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 yieldk single (return Nothing) 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 yieldk single (return Nothing) 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 yld sng stp ->
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) yld sng stp $
636 go (offset + 1) x
637 in foldStream (adaptState st) yieldk single stp 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 yieldk single stop 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 yld sng stp ->
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) yieldk single stp 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 yld sng stp ->
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 yieldk single stp r
699 in foldStream st yieldk single stp 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 yld sng stp ->
706 let yieldk a r = yld a (go (n1 - 1) r)
707 in if n1 <= 0
708 then stp
709 else foldStream st yieldk sng stp 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 yld sng stp ->
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 yieldk single stp m1
722drop :: IsStream t => Int -> t m a -> t m a
723drop n m = go n m
724 where
725 go n1 m1 = mkStream $ \st yld sng stp ->
726 let single _ = stp
727 yieldk _ r = foldStream st yld sng stp $ go (n1 - 1) r
728 -- Somehow "<=" check performs better than a ">"
729 in if n1 <= 0
730 then foldStream st yld sng stp m1
731 else foldStream st yieldk single stp m1
733{-# INLINE dropWhile #-}
734dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
735dropWhile p m = go m
736 where
737 go m1 = mkStream $ \st yld sng stp ->
738 let single a | p a = stp
739 | otherwise = sng a
740 yieldk a r | p a = foldStream st yieldk single stp r
741 | otherwise = yld a r
742 in foldStream st yieldk single stp m1
744-------------------------------------------------------------------------------
745-- Mapping
746-------------------------------------------------------------------------------
748-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
749-- other stream types to overload it.
750{-# INLINE mapM #-}
751mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
752mapM f m = go m
753 where
754 go m1 = mkStream $ \st yld sng stp ->
755 let single a = f a >>= sng
756 yieldk a r = foldStreamShared st yld sng stp $ f a |: go r
757 in foldStream (adaptState st) yieldk single stp m1
759-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
760-- other stream types to overload it.
761{-# INLINE sequence #-}
762sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
763sequence m = go m
764 where
765 go m1 = mkStream $ \st yld sng stp ->
766 let single ma = ma >>= sng
767 yieldk ma r = foldStreamShared st yld sng stp $ ma |: go r
768 in foldStream (adaptState st) yieldk single stp m1
770-------------------------------------------------------------------------------
771-- Inserting
772-------------------------------------------------------------------------------
774{-# INLINE intersperseM #-}
775intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
776intersperseM a m = prependingStart m
777 where
778 prependingStart m1 = mkStream $ \st yld sng stp ->
779 let yieldk i x = foldStreamShared st yld sng stp $ return i |: go x
780 in foldStream st yieldk sng stp m1
781 go m2 = mkStream $ \st yld sng stp ->
782 let single i = foldStreamShared st yld sng stp $ a |: yield i
783 yieldk i x = foldStreamShared st yld sng stp $ a |: return i |: go x
784 in foldStream st yieldk single stp m2
786{-# INLINE insertBy #-}
787insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a
788insertBy cmp x m = go m
789 where
790 go m1 = mkStream $ \st yld _ _ ->
791 let single a = case cmp x a of
792 GT -> yld a (yield x)
793 _ -> yld x (yield a)
794 stop = yld x nil
795 yieldk a r = case cmp x a of
796 GT -> yld a (go r)
797 _ -> yld x (a `cons` r)
798 in foldStream st yieldk single stop m1
800------------------------------------------------------------------------------
801-- Deleting
802------------------------------------------------------------------------------
804{-# INLINE deleteBy #-}
805deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a
806deleteBy eq x m = go m
807 where
808 go m1 = mkStream $ \st yld sng stp ->
809 let single a = if eq x a then stp else sng a
810 yieldk a r = if eq x a
811 then foldStream st yld sng stp r
812 else yld a (go r)
813 in foldStream st yieldk single stp m1
815-------------------------------------------------------------------------------
816-- Map and Filter
817-------------------------------------------------------------------------------
819{-# INLINE mapMaybe #-}
820mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b
821mapMaybe f m = go m
822 where
823 go m1 = mkStream $ \st yld sng stp ->
824 let single a = case f a of
825 Just b -> sng b
826 Nothing -> stp
827 yieldk a r = case f a of
828 Just b -> yld b $ go r
829 Nothing -> foldStream (adaptState st) yieldk single stp r
830 in foldStream (adaptState st) yieldk single stp m1
832------------------------------------------------------------------------------
833-- Serial Zipping
834------------------------------------------------------------------------------
836-- | Zip two streams serially using a pure zipping function.
837--
838-- @since 0.1.0
839{-# INLINABLE zipWith #-}
840zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
841zipWith f = go
842 where
843 go mx my = mkStream $ \st yld sng stp -> do
844 let merge a ra =
845 let single2 b = sng (f a b)
846 yield2 b rb = yld (f a b) (go ra rb)
847 in foldStream (adaptState st) yield2 single2 stp my
848 let single1 a = merge a nil
849 yield1 = merge
850 foldStream (adaptState st) yield1 single1 stp mx
852-- | Zip two streams serially using a monadic zipping function.
853--
854-- @since 0.1.0
855zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
856zipWithM f m1 m2 = go m1 m2
857 where
858 go mx my = mkStream $ \st yld sng stp -> do
859 let merge a ra =
860 let runIt x = foldStream st yld sng stp x
861 single2 b = f a b >>= sng
862 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)
863 in foldStream (adaptState st) yield2 single2 stp my
864 let single1 a = merge a nil
865 yield1 = merge
866 foldStream (adaptState st) yield1 single1 stp mx
868------------------------------------------------------------------------------
869-- Merging
870------------------------------------------------------------------------------
872{-# INLINE mergeByM #-}
873mergeByM
874 :: (IsStream t, Monad m)
875 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
876mergeByM cmp = go
877 where
878 go mx my = mkStream $ \st yld sng stp -> do
879 let mergeWithY a ra =
880 let stop2 = foldStream st yld sng stp mx
881 single2 b = do
882 r <- cmp a b
883 case r of
884 GT -> yld b (go (a `cons` ra) nil)
885 _ -> yld a (go ra (b `cons` nil))
886 yield2 b rb = do
887 r <- cmp a b
888 case r of
889 GT -> yld b (go (a `cons` ra) rb)
890 _ -> yld a (go ra (b `cons` rb))
891 in foldStream st yield2 single2 stop2 my
892 let stopX = foldStream st yld sng stp my
893 singleX a = mergeWithY a nil
894 yieldX = mergeWithY
895 foldStream st yieldX singleX stopX mx
897{-# INLINABLE mergeBy #-}
898mergeBy
899 :: (IsStream t, Monad m)
900 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
901mergeBy cmp = mergeByM (\a b -> return $ cmp a b)
903------------------------------------------------------------------------------
904-- Transformation comprehensions
905------------------------------------------------------------------------------
907{-# INLINE the #-}
908the :: (Eq a, IsStream t, Monad m) => t m a -> m (Maybe a)
909the m = do
910 r <- uncons m
911 case r of
912 Nothing -> return Nothing
913 Just (h, t) -> go h t
914 where
915 go h m1 =
916 let single a | h == a = return $ Just h
917 | otherwise = return Nothing
918 yieldk a r | h == a = go h r
919 | otherwise = return Nothing
920 in foldStream defState yieldk single (return $ Just h) m1
922-------------------------------------------------------------------------------
923-- Bind utility
924-------------------------------------------------------------------------------
926{-# INLINE bindWith #-}
927bindWith
928 :: IsStream t
929 => (forall c. t m c -> t m c -> t m c)
930 -> t m a
931 -> (a -> t m b)
932 -> t m b
933bindWith par m1 f = go m1
934 where
935 go m =
936 mkStream $ \st yld sng stp ->
937 let foldShared = foldStreamShared st yld sng stp
938 single a = foldShared $ unShare (f a)
939 yieldk a r = foldShared $ unShare (f a) `par` go r
940 in foldStream (adaptState st) yieldk single stp m
942------------------------------------------------------------------------------
943-- Alternative & MonadPlus
944------------------------------------------------------------------------------
946_alt :: Stream m a -> Stream m a -> Stream m a
947_alt m1 m2 = mkStream $ \st yld sng stp ->
948 let stop = foldStream st yld sng stp m2
949 in foldStream st yld sng stop m1
951------------------------------------------------------------------------------
952-- MonadReader
953------------------------------------------------------------------------------
955withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
956withLocal f m =
957 mkStream $ \st yld sng stp ->
958 let single = local f . sng
959 yieldk a r = local f $ yld a (withLocal f r)
960 in foldStream st yieldk single (local f stp) m
962------------------------------------------------------------------------------
963-- MonadError
964------------------------------------------------------------------------------
966{-
967-- XXX handle and test cross thread state transfer
968withCatchError
969 :: MonadError e m
970 => Stream m a -> (e -> Stream m a) -> Stream m a
971withCatchError m h =
972 mkStream $ \_ stp sng yld ->
973 let run x = unStream x Nothing stp sng yieldk
974 handle r = r `catchError` \e -> run $ h e
975 yieldk a r = yld a (withCatchError r h)
976 in handle $ run m
977-}