on f2b2aa4 help sign in
1021 lines 31 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.IsStream.Eliminate
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9-- This module contains functions ending in the shape:
10--
11-- @
12-- t m a -> m b
13-- @
14--
15-- We call them stream folding functions, they reduce a stream @t m a@ to a
16-- monadic value @m b@.
18module Streamly.Internal.Data.Stream.IsStream.Eliminate
19 (
20 -- * Running Examples
21 -- $setup
23 -- * Running a 'Fold'
24 -- See "Streamly.Internal.Data.Fold".
25 fold
26 , fold_
28 -- * Running a 'Parser'
29 -- "Streamly.Internal.Data.Parser".
30 , parse
31 , parseK
32 , parseD
33 , parse_
34 , parseD_
36 -- * Stream Deconstruction
37 -- | foldr and foldl do not provide the remaining stream. 'uncons' is more
38 -- general, as it can be used to implement those as well. It allows to use
39 -- the stream one element at a time, and we have the remaining stream all
40 -- the time.
41 , uncons
43 -- * Right Folds
44 , foldrM
45 , foldr
47 -- * Left Folds
48 , foldl'
49 , foldl1'
50 , foldlM'
52 -- * Specific Fold Functions
53 -- | Folds as functions of the shape @t m a -> m b@.
54 --
55 -- These functions are good to run individually but they do not compose
56 -- well. Prefer writing folds as the 'Fold' data type. Use folds from
57 -- "Streamly.Internal.Data.Fold" instead of using the functions in this
58 -- section.
59 --
60 -- This section can possibly be removed in future. Are these better in
61 -- some case compared to 'Fold'? When the input stream is in CPS style
62 -- (StreamK) we may want to rewrite the function call to CPS implementation
63 -- of the fold through these definitions. Will that be more efficient for
64 -- StreamK?
66 -- ** Full Folds
68 -- -- ** To Summary (Full Folds)
69 , mapM_
70 , drain
71 , last
72 , length
73 , sum
74 , product
75 , mconcat
77 -- -- ** To Summary (Maybe) (Full Folds)
78 , maximumBy
79 , maximum
80 , minimumBy
81 , minimum
82 , the
84 -- ** Partial Folds
86 -- -- ** To Elements (Partial Folds)
87 , drainN
88 , drainWhile
90 -- -- | Folds that extract selected elements of a stream or their properties.
91 , (!!)
92 , head
93 , headElse
94 , tail
95 , init
96 , findM
97 , find
98 , findIndex
99 , elemIndex
100 , lookup
102 -- -- ** To Boolean (Partial Folds)
103 , null
104 , elem
105 , notElem
106 , all
107 , any
108 , and
109 , or
111 -- -- ** Lazy Folds
112 -- ** To Containers
113 , toList
114 , toListRev
115 , toStream
116 , toStreamRev
118 -- * Concurrent Folds
119 , foldAsync
120 , (|$.)
121 , (|&.)
123 -- * Multi-Stream folds
124 -- Full equivalence
125 , eqBy
126 , cmpBy
128 -- finding subsequences
129 , isPrefixOf
130 , isInfixOf
131 , isSuffixOf
132 , isSubsequenceOf
134 -- trimming sequences
135 , stripPrefix
136 -- , stripInfix
137 , stripSuffix
139 -- * Deprecated
140 , foldx
141 , foldxM
142 , foldr1
143 , runStream
144 , runN
145 , runWhile
146 , toHandle
147 )
148where
150#include "inline.hs"
152import Control.Monad.Catch (MonadThrow)
153import Control.Monad.IO.Class (MonadIO(..))
154import Data.Functor.Identity (Identity (..))
155import Foreign.Storable (Storable)
156import Streamly.Internal.Data.Parser (Parser (..))
157import Streamly.Internal.Data.SVar (MonadAsync, defState)
158import Streamly.Internal.Data.Stream.IsStream.Common
159 ( fold, fold_, drop, findIndices, reverse, splitOnSeq, take
160 , takeWhile)
161import Streamly.Internal.Data.Stream.Prelude (toStreamS)
162import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
163import Streamly.Internal.Data.Stream.StreamK (IsStream)
164import Streamly.Internal.Data.Stream.Serial (SerialT)
166import qualified Streamly.Internal.Data.Array.Foreign.Type as A
167import qualified Streamly.Internal.Data.Fold as FL
168import qualified Streamly.Internal.Data.Stream.Parallel as Par
169import qualified Streamly.Internal.Data.Stream.Prelude as P
170import qualified Streamly.Internal.Data.Stream.StreamD as D
171import qualified Streamly.Internal.Data.Stream.StreamK as K
172import qualified Streamly.Internal.Data.Parser.ParserD as PRD
173import qualified Streamly.Internal.Data.Parser.ParserK.Type as PRK
174import qualified System.IO as IO
175#ifdef USE_STREAMK_ONLY
176import qualified Streamly.Internal.Data.Stream.StreamK as S
177#else
178import qualified Streamly.Internal.Data.Stream.StreamD as S
179#endif
181import Prelude hiding
182 ( drop, take, takeWhile, foldr , foldl, mapM_, sequence, all, any, sum
183 , product, elem, notElem, maximum, minimum, head, last, tail, length
184 , null , reverse, init, and, or, lookup, foldr1, (!!) , splitAt, break
185 , mconcat)
187-- $setup
188-- >>> :m
189-- >>> import Streamly.Prelude (SerialT)
190-- >>> import qualified Streamly.Prelude as Stream
191-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
192-- >>> import qualified Streamly.Internal.Data.Parser as Parser
193-- >>> import qualified Streamly.Data.Fold as Fold
195------------------------------------------------------------------------------
196-- Deconstruction
197------------------------------------------------------------------------------
199-- | Decompose a stream into its head and tail. If the stream is empty, returns
200-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
201-- the head of the stream and @ma@ its tail.
202--
203-- This is a brute force primitive. Avoid using it as long as possible, use it
204-- when no other combinator can do the job. This can be used to do pretty much
205-- anything in an imperative manner, as it just breaks down the stream into
206-- individual elements and we can loop over them as we deem fit. For example,
207-- this can be used to convert a streamly stream into other stream types.
208--
209-- All the folds in this module can be expressed in terms of 'uncons', however
210-- the specific implementations are generally more efficient.
211--
212-- @since 0.1.0
213{-# INLINE uncons #-}
214uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
215uncons m = K.uncons (K.adapt m)
217------------------------------------------------------------------------------
218-- Right Folds
219------------------------------------------------------------------------------
221-- | Right associative/lazy pull fold. @foldrM build final stream@ constructs
222-- an output structure using the step function @build@. @build@ is invoked with
223-- the next input element and the remaining (lazy) tail of the output
224-- structure. It builds a lazy output expression using the two. When the "tail
225-- structure" in the output expression is evaluated it calls @build@ again thus
226-- lazily consuming the input @stream@ until either the output expression built
227-- by @build@ is free of the "tail" or the input is exhausted in which case
228-- @final@ is used as the terminating case for the output structure. For more
229-- details see the description in the previous section.
230--
231-- Example, determine if any element is 'odd' in a stream:
232--
233-- >>> Stream.foldrM (\x xs -> if odd x then return True else xs) (return False) $ Stream.fromList (2:4:5:undefined)
234-- True
235--
236-- /Since: 0.7.0 (signature changed)/
237--
238-- /Since: 0.2.0 (signature changed)/
239--
240-- /Since: 0.1.0/
241{-# INLINE foldrM #-}
242foldrM :: Monad m => (a -> m b -> m b) -> m b -> SerialT m a -> m b
243foldrM = P.foldrM
245-- | Right fold, lazy for lazy monads and pure streams, and strict for strict
246-- monads.
247--
248-- Please avoid using this routine in strict monads like IO unless you need a
249-- strict right fold. This is provided only for use in lazy monads (e.g.
250-- Identity) or pure streams. Note that with this signature it is not possible
251-- to implement a lazy foldr when the monad @m@ is strict. In that case it
252-- would be strict in its accumulator and therefore would necessarily consume
253-- all its input.
254--
255-- @since 0.1.0
256{-# INLINE foldr #-}
257foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
258foldr = P.foldr
260-- XXX This seems to be of limited use as it cannot be used to construct
261-- recursive structures and for reduction foldl1' is better.
262--
263-- | Lazy right fold for non-empty streams, using first element as the starting
264-- value. Returns 'Nothing' if the stream is empty.
265--
266-- @since 0.5.0
267{-# INLINE foldr1 #-}
268{-# DEPRECATED foldr1 "Use foldrM instead." #-}
269foldr1 :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
270foldr1 f m = S.foldr1 f (toStreamS m)
272------------------------------------------------------------------------------
273-- Left Folds
274------------------------------------------------------------------------------
276-- | Strict left fold with an extraction function. Like the standard strict
277-- left fold, but applies a user supplied extraction function (the third
278-- argument) to the folded value at the end. This is designed to work with the
279-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
280--
281-- @since 0.2.0
282{-# DEPRECATED foldx "Please use foldl' followed by fmap instead." #-}
283{-# INLINE foldx #-}
284foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
285foldx = P.foldlx'
287-- | Left associative/strict push fold. @foldl' reduce initial stream@ invokes
288-- @reduce@ with the accumulator and the next input in the input stream, using
289-- @initial@ as the initial value of the current value of the accumulator. When
290-- the input is exhausted the current value of the accumulator is returned.
291-- Make sure to use a strict data structure for accumulator to not build
292-- unnecessary lazy expressions unless that's what you want. See the previous
293-- section for more details.
294--
295-- @since 0.2.0
296{-# INLINE foldl' #-}
297foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
298foldl' = P.foldl'
300-- | Strict left fold, for non-empty streams, using first element as the
301-- starting value. Returns 'Nothing' if the stream is empty.
302--
303-- @since 0.5.0
304{-# INLINE foldl1' #-}
305foldl1' :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
306foldl1' step m = do
307 r <- uncons m
308 case r of
309 Nothing -> return Nothing
310 Just (h, t) -> do
311 res <- foldl' step h t
312 return $ Just res
314-- | Like 'foldx', but with a monadic step function.
315--
316-- @since 0.2.0
317{-# DEPRECATED foldxM "Please use foldlM' followed by fmap instead." #-}
318{-# INLINE foldxM #-}
319foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
320foldxM = P.foldlMx'
322-- | Like 'foldl'' but with a monadic step function.
323--
324-- /Since: 0.2.0/
325--
326-- /Since: 0.8.0 (signature change)/
327{-# INLINE foldlM' #-}
328foldlM' :: Monad m => (b -> a -> m b) -> m b -> SerialT m a -> m b
329foldlM' step begin m = S.foldlM' step begin $ toStreamS m
331------------------------------------------------------------------------------
332-- Running a sink
333------------------------------------------------------------------------------
335{-
336-- | Drain a stream to a 'Sink'.
337{-# INLINE runSink #-}
338runSink :: Monad m => Sink m a -> SerialT m a -> m ()
339runSink = fold . toFold
340-}
342------------------------------------------------------------------------------
343-- Running a Parser
344------------------------------------------------------------------------------
346-- | Parse a stream using the supplied ParserD 'PRD.Parser'.
347--
348-- /Internal/
349--
350{-# INLINE_NORMAL parseD #-}
351parseD :: MonadThrow m => PRD.Parser m a b -> SerialT m a -> m b
352parseD p = D.parse p . toStreamD
354-- | Parse a stream using the supplied ParserK 'PRK.Parser'.
355--
356-- /Internal/
357{-# INLINE parseK #-}
358parseK :: MonadThrow m => PRK.Parser m a b -> SerialT m a -> m b
359parseK = parse
361-- | Parse a stream using the supplied 'Parser'.
362--
363-- Unlike folds, parsers may not always result in a valid output, they may
364-- result in an error. For example:
365--
366-- >>> Stream.parse (Parser.takeEQ 1 Fold.drain) Stream.nil
367-- *** Exception: ParseError "takeEQ: Expecting exactly 1 elements, input terminated on 0"
368--
369-- Note:
370--
371-- @
372-- fold f = Stream.parse (Parser.fromFold f)
373-- @
374--
375-- @parse p@ is not the same as @head . parseMany p@ on an empty stream.
376--
377-- /Pre-release/
378--
379{-# INLINE [3] parse #-}
380parse :: MonadThrow m => Parser m a b -> SerialT m a -> m b
381parse = parseD . PRK.fromParserK
383{-# INLINE_NORMAL parseD_ #-}
384parseD_ :: MonadThrow m => PRD.Parser m a b -> SerialT m a -> m (b, SerialT m a)
385parseD_ parser strm = do
386 (b, strmD) <- D.parse_ parser (toStreamD strm)
387 return $! (b, fromStreamD strmD)
389-- | Parse a stream using the supplied 'Parser'.
390--
391-- /Internal/
392--
393{-# INLINE [3] parse_ #-}
394parse_ :: MonadThrow m => Parser m a b -> SerialT m a -> m (b, SerialT m a)
395parse_ = parseD_ . PRK.fromParserK
397------------------------------------------------------------------------------
398-- Specific Fold Functions
399------------------------------------------------------------------------------
401-- XXX this can utilize parallel mapping if we implement it as drain . mapM
402-- |
403-- > mapM_ = Stream.drain . Stream.mapM
404--
405-- Apply a monadic action to each element of the stream and discard the output
406-- of the action. This is not really a pure transformation operation but a
407-- transformation followed by fold.
408--
409-- @since 0.1.0
410{-# INLINE mapM_ #-}
411mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()
412mapM_ f m = S.mapM_ f $ toStreamS m
414-- |
415-- > drain = mapM_ (\_ -> return ())
416-- > drain = Stream.fold Fold.drain
417--
418-- Run a stream, discarding the results. By default it interprets the stream
419-- as 'SerialT', to run other types of streams use the type adapting
420-- combinators for example @Stream.drain . 'fromAsync'@.
421--
422-- @since 0.7.0
423{-# INLINE drain #-}
424drain :: Monad m => SerialT m a -> m ()
425drain = P.drain
427-- |
428-- > drainN n = Stream.drain . Stream.take n
429-- > drainN n = Stream.fold (Fold.take n Fold.drain)
430--
431-- Run maximum up to @n@ iterations of a stream.
432--
433-- @since 0.7.0
434{-# INLINE drainN #-}
435drainN :: Monad m => Int -> SerialT m a -> m ()
436drainN n = drain . take n
438-- |
439-- > runN n = runStream . take n
440--
441-- Run maximum up to @n@ iterations of a stream.
442--
443-- @since 0.6.0
444{-# DEPRECATED runN "Please use \"drainN\" instead" #-}
445{-# INLINE runN #-}
446runN :: Monad m => Int -> SerialT m a -> m ()
447runN = drainN
449-- |
450-- > drainWhile p = Stream.drain . Stream.takeWhile p
451--
452-- Run a stream as long as the predicate holds true.
453--
454-- @since 0.7.0
455{-# INLINE drainWhile #-}
456drainWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
457drainWhile p = drain . takeWhile p
459-- |
460-- > runWhile p = runStream . takeWhile p
461--
462-- Run a stream as long as the predicate holds true.
463--
464-- @since 0.6.0
465{-# DEPRECATED runWhile "Please use \"drainWhile\" instead" #-}
466{-# INLINE runWhile #-}
467runWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
468runWhile = drainWhile
470-- | Run a stream, discarding the results. By default it interprets the stream
471-- as 'SerialT', to run other types of streams use the type adapting
472-- combinators for example @runStream . 'fromAsync'@.
473--
474-- @since 0.2.0
475{-# DEPRECATED runStream "Please use \"drain\" instead" #-}
476{-# INLINE runStream #-}
477runStream :: Monad m => SerialT m a -> m ()
478runStream = drain
480-- | Determine whether the stream is empty.
481--
482-- > null = Stream.fold Fold.null
483--
484-- @since 0.1.1
485{-# INLINE null #-}
486null :: Monad m => SerialT m a -> m Bool
487null = S.null . toStreamS
489-- | Extract the first element of the stream, if any.
490--
491-- > head = (!! 0)
492-- > head = Stream.fold Fold.head
493--
494-- @since 0.1.0
495{-# INLINE head #-}
496head :: Monad m => SerialT m a -> m (Maybe a)
497head = S.head . toStreamS
499-- | Extract the first element of the stream, if any, otherwise use the
500-- supplied default value. It can help avoid one branch in high performance
501-- code.
502--
503-- /Pre-release/
504{-# INLINE headElse #-}
505headElse :: Monad m => a -> SerialT m a -> m a
506headElse x = D.headElse x . toStreamD
508-- |
509-- > tail = fmap (fmap snd) . Stream.uncons
510--
511-- Extract all but the first element of the stream, if any.
512--
513-- @since 0.1.1
514{-# INLINE tail #-}
515tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
516tail m = K.tail (K.adapt m)
518-- | Extract all but the last element of the stream, if any.
519--
520-- @since 0.5.0
521{-# INLINE init #-}
522init :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
523init m = K.init (K.adapt m)
525-- | Extract the last element of the stream, if any.
526--
527-- > last xs = xs !! (Stream.length xs - 1)
528-- > last = Stream.fold Fold.last
529--
530-- @since 0.1.1
531{-# INLINE last #-}
532last :: Monad m => SerialT m a -> m (Maybe a)
533last m = S.last $ toStreamS m
535-- | Determine whether an element is present in the stream.
536--
537-- > elem = Stream.fold Fold.elem
538--
539-- @since 0.1.0
540{-# INLINE elem #-}
541elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
542elem e m = S.elem e (toStreamS m)
544-- | Determine whether an element is not present in the stream.
545--
546-- > notElem = Stream.fold Fold.length
547--
548-- @since 0.1.0
549{-# INLINE notElem #-}
550notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
551notElem e m = S.notElem e (toStreamS m)
553-- | Determine the length of the stream.
554--
555-- @since 0.1.0
556{-# INLINE length #-}
557length :: Monad m => SerialT m a -> m Int
558length = foldl' (\n _ -> n + 1) 0
560-- | Determine whether all elements of a stream satisfy a predicate.
561--
562-- > all = Stream.fold Fold.all
563--
564-- @since 0.1.0
565{-# INLINE all #-}
566all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
567all p m = S.all p (toStreamS m)
569-- | Determine whether any of the elements of a stream satisfy a predicate.
570--
571-- > any = Stream.fold Fold.any
572--
573-- @since 0.1.0
574{-# INLINE any #-}
575any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
576any p m = S.any p (toStreamS m)
578-- | Determines if all elements of a boolean stream are True.
579--
580-- > and = Stream.fold Fold.and
581--
582-- @since 0.5.0
583{-# INLINE and #-}
584and :: Monad m => SerialT m Bool -> m Bool
585and = all (==True)
587-- | Determines whether at least one element of a boolean stream is True.
588--
589-- > or = Stream.fold Fold.or
590--
591-- @since 0.5.0
592{-# INLINE or #-}
593or :: Monad m => SerialT m Bool -> m Bool
594or = any (==True)
596-- | Determine the sum of all elements of a stream of numbers. Returns @0@ when
597-- the stream is empty. Note that this is not numerically stable for floating
598-- point numbers.
599--
600-- > sum = Stream.fold Fold.sum
601--
602-- @since 0.1.0
603{-# INLINE sum #-}
604sum :: (Monad m, Num a) => SerialT m a -> m a
605sum = foldl' (+) 0
607-- | Determine the product of all elements of a stream of numbers. Returns @1@
608-- when the stream is empty.
609--
610-- > product = Stream.fold Fold.product
611--
612-- @since 0.1.1
613{-# INLINE product #-}
614product :: (Monad m, Num a) => SerialT m a -> m a
615product = foldl' (*) 1
617-- | Fold a stream of monoid elements by appending them.
618--
619-- > mconcat = Stream.fold Fold.mconcat
620--
621-- /Pre-release/
622{-# INLINE mconcat #-}
623mconcat :: (Monad m, Monoid a) => SerialT m a -> m a
624mconcat = foldr mappend mempty
626-- |
627-- @
628-- minimum = 'minimumBy' compare
629 -- minimum = Stream.fold Fold.minimum
630-- @
631--
632-- Determine the minimum element in a stream.
633--
634-- @since 0.1.0
635{-# INLINE minimum #-}
636minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
637minimum m = S.minimum (toStreamS m)
639-- | Determine the minimum element in a stream using the supplied comparison
640-- function.
641--
642-- > minimumBy = Stream.fold Fold.minimumBy
643--
644-- @since 0.6.0
645{-# INLINE minimumBy #-}
646minimumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
647minimumBy cmp m = S.minimumBy cmp (toStreamS m)
649-- |
650-- @
651-- maximum = 'maximumBy' compare
652-- maximum = Stream.fold Fold.maximum
653-- @
654--
655-- Determine the maximum element in a stream.
656--
657-- @since 0.1.0
658{-# INLINE maximum #-}
659maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
660maximum = P.maximum
662-- | Determine the maximum element in a stream using the supplied comparison
663-- function.
664--
665-- > maximumBy = Stream.fold Fold.maximumBy
666--
667-- @since 0.6.0
668{-# INLINE maximumBy #-}
669maximumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
670maximumBy cmp m = S.maximumBy cmp (toStreamS m)
672-- | Ensures that all the elements of the stream are identical and then returns
673-- that unique element.
674--
675-- @since 0.6.0
676{-# INLINE the #-}
677the :: (Eq a, Monad m) => SerialT m a -> m (Maybe a)
678the m = S.the (toStreamS m)
680------------------------------------------------------------------------------
681-- Searching
682------------------------------------------------------------------------------
684-- | Lookup the element at the given index.
685--
686-- @since 0.6.0
687{-# INLINE (!!) #-}
688(!!) :: Monad m => SerialT m a -> Int -> m (Maybe a)
689m !! i = toStreamS m S.!! i
691-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
692-- first pair where the key equals the given value @a@.
693--
694-- > lookup = snd <$> Stream.find ((==) . fst)
695-- > lookup = Stream.fold Fold.lookup
696--
697-- @since 0.5.0
698{-# INLINE lookup #-}
699lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)
700lookup a m = S.lookup a (toStreamS m)
702-- | Like 'findM' but with a non-monadic predicate.
703--
704-- > find p = findM (return . p)
705-- > find = Stream.fold Fold.find
706--
707-- @since 0.5.0
708{-# INLINE find #-}
709find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)
710find p m = S.find p (toStreamS m)
712-- | Returns the first element that satisfies the given predicate.
713--
714-- > findM = Stream.fold Fold.findM
715--
716-- @since 0.6.0
717{-# INLINE findM #-}
718findM :: Monad m => (a -> m Bool) -> SerialT m a -> m (Maybe a)
719findM p m = S.findM p (toStreamS m)
721-- | Returns the first index that satisfies the given predicate.
722--
723-- > findIndex = Stream.fold Fold.findIndex
724--
725-- @since 0.5.0
726{-# INLINE findIndex #-}
727findIndex :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe Int)
728findIndex p = head . findIndices p
730-- | Returns the first index where a given value is found in the stream.
731--
732-- > elemIndex a = Stream.findIndex (== a)
733--
734-- @since 0.5.0
735{-# INLINE elemIndex #-}
736elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int)
737elemIndex a = findIndex (== a)
739------------------------------------------------------------------------------
740-- To containers
741------------------------------------------------------------------------------
743-- |
744-- @
745-- toList = Stream.foldr (:) []
746-- @
747--
748-- Convert a stream into a list in the underlying monad. The list can be
749-- consumed lazily in a lazy monad (e.g. 'Identity'). In a strict monad (e.g.
750-- IO) the whole list is generated and buffered before it can be consumed.
751--
752-- /Warning!/ working on large lists accumulated as buffers in memory could be
753-- very inefficient, consider using "Streamly.Array" instead.
754--
755-- @since 0.1.0
756{-# INLINE toList #-}
757toList :: Monad m => SerialT m a -> m [a]
758toList = P.toList
760-- |
761-- @
762-- toListRev = Stream.foldl' (flip (:)) []
763-- @
764--
765-- Convert a stream into a list in reverse order in the underlying monad.
766--
767-- /Warning!/ working on large lists accumulated as buffers in memory could be
768-- very inefficient, consider using "Streamly.Array" instead.
769--
770-- /Pre-release/
771{-# INLINE toListRev #-}
772toListRev :: Monad m => SerialT m a -> m [a]
773toListRev = D.toListRev . toStreamD
775-- |
776-- @
777-- toHandle h = S.mapM_ $ hPutStrLn h
778-- @
779--
780-- Write a stream of Strings to an IO Handle.
781--
782-- @since 0.1.0
783{-# DEPRECATED toHandle
784 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
785toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
786toHandle h = go
787 where
788 go m1 =
789 let stop = return ()
790 single a = liftIO (IO.hPutStrLn h a)
791 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r
792 in K.foldStream defState yieldk single stop m1
794-- | Convert a stream to a pure stream.
795--
796-- @
797-- toStream = Stream.foldr Stream.cons Stream.nil
798-- @
799--
800-- /Pre-release/
801--
802{-# INLINE toStream #-}
803toStream :: Monad m => SerialT m a -> m (SerialT Identity a)
804toStream = foldr K.cons K.nil
806-- | Convert a stream to a pure stream in reverse order.
807--
808-- @
809-- toStreamRev = Stream.foldl' (flip Stream.cons) Stream.nil
810-- @
811--
812-- /Pre-release/
813--
814{-# INLINE toStreamRev #-}
815toStreamRev :: Monad m => SerialT m a -> m (SerialT Identity a)
816toStreamRev = foldl' (flip K.cons) K.nil
818------------------------------------------------------------------------------
819-- Concurrent Application
820------------------------------------------------------------------------------
822-- | Parallel fold application operator; applies a fold function @t m a -> m b@
823-- to a stream @t m a@ concurrently; The the input stream is evaluated
824-- asynchronously in an independent thread yielding elements to a buffer and
825-- the folding action runs in another thread consuming the input from the
826-- buffer.
827--
828-- If you read the signature as @(t m a -> m b) -> (t m a -> m b)@ you can look
829-- at it as a transformation that converts a fold function to a buffered
830-- concurrent fold function.
831--
832-- The @.@ at the end of the operator is a mnemonic for termination of the
833-- stream.
834--
835-- In the example below, each stage introduces a delay of 1 sec but output is
836-- printed every second because both stages are concurrent.
837--
838-- >>> import Control.Concurrent (threadDelay)
839-- >>> import Streamly.Prelude ((|$.))
840-- >>> :{
841-- Stream.foldlM' (\_ a -> threadDelay 1000000 >> print a) (return ())
842-- |$. Stream.replicateM 3 (threadDelay 1000000 >> return 1)
843-- :}
844-- 1
845-- 1
846-- 1
847--
848-- /Concurrent/
849--
850-- /Since: 0.3.0 ("Streamly")/
851--
852-- @since 0.8.0
853{-# INLINE (|$.) #-}
854(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> (t m a -> m b)
855-- (|$.) f = f . Async.mkAsync
856(|$.) f = f . Par.mkParallel
858infixr 0 |$.
860-- | Same as '|$.'.
861--
862-- /Internal/
863--
864{-# INLINE foldAsync #-}
865foldAsync :: (IsStream t, MonadAsync m) => (t m a -> m b) -> (t m a -> m b)
866foldAsync = (|$.)
868-- | Same as '|$.' but with arguments reversed.
869--
870-- > (|&.) = flip (|$.)
871--
872-- /Concurrent/
873--
874-- /Since: 0.3.0 ("Streamly")/
875--
876-- @since 0.8.0
877{-# INLINE (|&.) #-}
878(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
879x |&. f = f |$. x
881infixl 1 |&.
883------------------------------------------------------------------------------
884-- Multi-stream folds
885------------------------------------------------------------------------------
887-- | Returns 'True' if the first stream is the same as or a prefix of the
888-- second. A stream is a prefix of itself.
889--
890-- >>> Stream.isPrefixOf (Stream.fromList "hello") (Stream.fromList "hello" :: SerialT IO Char)
891-- True
892--
893-- @since 0.6.0
894{-# INLINE isPrefixOf #-}
895isPrefixOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
896isPrefixOf m1 m2 = D.isPrefixOf (toStreamD m1) (toStreamD m2)
898-- | Returns 'True' if the first stream is an infix of the second. A stream is
899-- considered an infix of itself.
900--
901-- > Stream.isInfixOf (Stream.fromList "hello") (Stream.fromList "hello" :: SerialT IO Char)
902-- True
903--
904-- Space: @O(n)@ worst case where @n@ is the length of the infix.
905--
906-- /Pre-release/
907--
908-- /Requires 'Storable' constraint/
909--
910{-# INLINE isInfixOf #-}
911isInfixOf :: (MonadIO m, Eq a, Enum a, Storable a)
912 => SerialT m a -> SerialT m a -> m Bool
913isInfixOf infx stream = do
914 arr <- fold A.write infx
915 -- XXX can use breakOnSeq instead (when available)
916 r <- null $ drop 1 $ splitOnSeq arr FL.drain stream
917 return (not r)
919-- Note: isPrefixOf uses the prefix stream only once. In contrast, isSuffixOf
920-- may use the suffix stream many times. To run in optimal memory we do not
921-- want to buffer the suffix stream in memory therefore we need an ability to
922-- clone (or consume it multiple times) the suffix stream without any side
923-- effects so that multiple potential suffix matches can proceed in parallel
924-- without buffering the suffix stream. For example, we may create the suffix
925-- stream from a file handle, however, if we evaluate the stream multiple
926-- times, once for each match, we will need a different file handle each time
927-- which may exhaust the file descriptors. Instead, we want to share the same
928-- underlying file descriptor, use pread on it to generate the stream and clone
929-- the stream for each match. Therefore the suffix stream should be built in
930-- such a way that it can be consumed multiple times without any problems.
932-- XXX Can be implemented with better space/time complexity.
933-- Space: @O(n)@ worst case where @n@ is the length of the suffix.
935-- | Returns 'True' if the first stream is a suffix of the second. A stream is
936-- considered a suffix of itself.
937--
938-- >>> Stream.isSuffixOf (Stream.fromList "hello") (Stream.fromList "hello" :: SerialT IO Char)
939-- True
940--
941-- Space: @O(n)@, buffers entire input stream and the suffix.
942--
943-- /Pre-release/
944--
945-- /Suboptimal/ - Help wanted.
946--
947{-# INLINE isSuffixOf #-}
948isSuffixOf :: (Monad m, Eq a) => SerialT m a -> SerialT m a -> m Bool
949isSuffixOf suffix stream = reverse suffix `isPrefixOf` reverse stream
951-- | Returns 'True' if all the elements of the first stream occur, in order, in
952-- the second stream. The elements do not have to occur consecutively. A stream
953-- is a subsequence of itself.
954--
955-- >>> Stream.isSubsequenceOf (Stream.fromList "hlo") (Stream.fromList "hello" :: SerialT IO Char)
956-- True
957--
958-- @since 0.6.0
959{-# INLINE isSubsequenceOf #-}
960isSubsequenceOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
961isSubsequenceOf m1 m2 = D.isSubsequenceOf (toStreamD m1) (toStreamD m2)
963-- Note: If we want to return a Maybe value to know whether the
964-- suffix/infix was present or not along with the stripped stream then
965-- we need to buffer the whole input stream.
966--
967-- | @stripPrefix prefix stream@ strips @prefix@ from @stream@ if it is a
968-- prefix of stream. Returns 'Nothing' if the stream does not start with the
969-- given prefix, stripped stream otherwise. Returns @Just nil@ when the prefix
970-- is the same as the stream.
971--
972-- See also "Streamly.Internal.Data.Stream.IsStream.Nesting.dropPrefix".
973--
974-- Space: @O(1)@
975--
976-- @since 0.6.0
977{-# INLINE stripPrefix #-}
978stripPrefix
979 :: (Eq a, IsStream t, Monad m)
980 => t m a -> t m a -> m (Maybe (t m a))
981stripPrefix m1 m2 = fmap fromStreamD <$>
982 D.stripPrefix (toStreamD m1) (toStreamD m2)
984-- | Drops the given suffix from a stream. Returns 'Nothing' if the stream does
985-- not end with the given suffix. Returns @Just nil@ when the suffix is the
986-- same as the stream.
987--
988-- It may be more efficient to convert the stream to an Array and use
989-- stripSuffix on that especially if the elements have a Storable or Prim
990-- instance.
991--
992-- See also "Streamly.Internal.Data.Stream.IsStream.Nesting.dropSuffix".
993--
994-- Space: @O(n)@, buffers the entire input stream as well as the suffix
995--
996-- /Pre-release/
997{-# INLINE stripSuffix #-}
998stripSuffix
999 :: (Monad m, Eq a)
1000 => SerialT m a -> SerialT m a -> m (Maybe (SerialT m a))
1001stripSuffix m1 m2 = fmap reverse <$> stripPrefix (reverse m1) (reverse m2)
1003------------------------------------------------------------------------------
1004-- Comparison
1005------------------------------------------------------------------------------
1007-- | Compare two streams for equality using an equality function.
1009-- @since 0.6.0
1010{-# INLINABLE eqBy #-}
1011eqBy :: (IsStream t, Monad m) => (a -> b -> Bool) -> t m a -> t m b -> m Bool
1012eqBy = P.eqBy
1014-- | Compare two streams lexicographically using a comparison function.
1016-- @since 0.6.0
1017{-# INLINABLE cmpBy #-}
1018cmpBy
1019 :: (IsStream t, Monad m)
1020 => (a -> b -> Ordering) -> t m a -> t m b -> m Ordering
1021cmpBy = P.cmpBy