on 2ea9fb0 help sign in
940 lines 29 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 , foldBreak
27 , foldContinue
29 -- * Running a 'Parser'
30 -- "Streamly.Internal.Data.Parser".
31 , Stream.parse
32 , Stream.parseK
33 , Stream.parseD
34 , Stream.parseBreak
35 , Stream.parseBreakD
37 -- * Stream Deconstruction
38 -- | foldr and foldl do not provide the remaining stream. 'uncons' is more
39 -- general, as it can be used to implement those as well. It allows to use
40 -- the stream one element at a time, and we have the remaining stream all
41 -- the time.
42 , uncons
44 -- * Right Folds
45 , Stream.foldrM
46 , Stream.foldr
48 -- * Left Folds
49 -- Lazy left folds are useful only for reversing the stream
50 , foldlS
51 , foldlT
53 , foldl'
54 , foldl1'
55 , foldlM'
57 -- * Specific Fold Functions
58 -- | Folds as functions of the shape @t m a -> m b@.
59 --
60 -- These functions are good to run individually but they do not compose
61 -- well. Prefer writing folds as the 'Fold' data type. Use folds from
62 -- "Streamly.Internal.Data.Fold" instead of using the functions in this
63 -- section.
64 --
65 -- This section can possibly be removed in future. Are these better in
66 -- some case compared to 'Fold'? When the input stream is in CPS style
67 -- (StreamK) we may want to rewrite the function call to CPS implementation
68 -- of the fold through these definitions. Will that be more efficient for
69 -- StreamK?
71 -- ** Full Folds
73 -- -- ** To Summary (Full Folds)
74 , mapM_
75 , drain
76 , last
77 , length
78 , sum
79 , product
80 , mconcat
82 -- -- ** To Summary (Maybe) (Full Folds)
83 , maximumBy
84 , maximum
85 , minimumBy
86 , minimum
87 , the
89 -- ** Partial Folds
91 -- -- ** To Elements (Partial Folds)
92 , drainN
93 , drainWhile
95 -- -- | Folds that extract selected elements of a stream or their properties.
96 , (!!)
97 , head
98 , headElse
99 , tail
100 , init
101 , findM
102 , find
103 , findIndex
104 , elemIndex
105 , lookup
107 -- -- ** To Boolean (Partial Folds)
108 , null
109 , elem
110 , notElem
111 , all
112 , any
113 , and
114 , or
116 -- -- ** Lazy Folds
117 -- ** To Containers
118 , toList
119 , toListRev
120 , toStream
121 , toStreamRev
123 -- * Concurrent Folds
124 , foldAsync
125 , (|$.)
126 , (|&.)
128 -- * Multi-Stream folds
129 -- Full equivalence
130 , eqBy
131 , cmpBy
133 -- finding subsequences
134 , isPrefixOf
135 , isInfixOf
136 , isSuffixOf
137 , isSubsequenceOf
139 -- trimming sequences
140 , stripPrefix
141 -- , stripInfix
142 , Stream.stripSuffix
144 -- * Deprecated
145 , foldx
146 , foldxM
147 , foldr1
148 , runStream
149 , runN
150 , runWhile
151 , toHandle
152 )
153where
155#include "inline.hs"
157import Control.Monad.IO.Class (MonadIO(..))
158import Control.Monad.Trans.Class (MonadTrans(..))
159import Streamly.Internal.Control.Concurrent (MonadAsync)
160import Streamly.Internal.Data.SVar (defState)
161import Streamly.Internal.Data.Stream.IsStream.Common
162 ( fold, foldBreak, foldContinue, drop, findIndices, reverse, splitOnSeq
163 , take , takeWhile, mkParallel)
164import Streamly.Internal.Data.Stream.IsStream.Type
165 (IsStream, toStreamD, fromStreamD, toStreamD)
166import Streamly.Internal.Data.Stream.Serial (SerialT)
167import Streamly.Internal.Data.Unboxed (Unboxed)
169import qualified Streamly.Data.Array.Unboxed as A
170import qualified Streamly.Data.Stream as Stream
171import qualified Streamly.Internal.Data.Fold as FL
172import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
173import qualified Streamly.Internal.Data.Stream.StreamD as D
174 (foldr1, foldlT, foldlM', mapM_, null, head, headElse, last, elem
175 , notElem, all, any, minimum, minimumBy, maximum, maximumBy, the, lookup
176 , find, findM, toListRev, isPrefixOf, isSubsequenceOf, stripPrefix, (!!))
177import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
178 (uncons, foldlS, tail, init)
179import qualified Streamly.Internal.Data.Stream as Stream
180 (foldr, toStreamK, parseK, parseD, parseBreakD, foldrM, stripSuffix)
181import qualified System.IO as IO
183import Prelude hiding
184 ( drop, take, takeWhile, foldr , foldl, mapM_, sequence, all, any, sum
185 , product, elem, notElem, maximum, minimum, head, last, tail, length
186 , null , reverse, init, and, or, lookup, foldr1, (!!) , splitAt, break
187 , mconcat)
189-- $setup
190-- >>> :m
191-- >>> import Streamly.Prelude (SerialT)
192-- >>> import qualified Streamly.Prelude as Stream
193-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
194-- >>> import qualified Streamly.Internal.Data.Parser as Parser
195-- >>> import qualified Streamly.Internal.Data.Fold as Fold
197------------------------------------------------------------------------------
198-- Deconstruction
199------------------------------------------------------------------------------
201-- | Decompose a stream into its head and tail. If the stream is empty, returns
202-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
203-- the head of the stream and @ma@ its tail.
204--
205-- This can be used to do pretty much anything in an imperative manner, as it
206-- just breaks down the stream into individual elements and we can loop over
207-- them as we deem fit. For example, this can be used to convert a streamly
208-- stream into other stream types.
209--
210-- All the folds in this module can be expressed in terms of 'uncons', however,
211-- this is generally less efficient than specific folds because it takes apart
212-- the stream one element at a time, therefore, does not take adavantage of
213-- stream fusion.
214--
215-- 'foldBreak' is a more general way of consuming a stream piecemeal.
216--
217-- >>> :{
218-- uncons xs = do
219-- r <- Stream.foldBreak Fold.one xs
220-- return $ case r of
221-- (Nothing, _) -> Nothing
222-- (Just h, t) -> Just (h, Stream.fromSerial t)
223-- :}
224--
225-- @since 0.1.0
226{-# INLINE uncons #-}
227uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
228uncons = fmap (fmap (fmap IsStream.fromStream)) . K.uncons . Stream.toStreamK
230------------------------------------------------------------------------------
231-- Right Folds
232------------------------------------------------------------------------------
233-- XXX This seems to be of limited use as it cannot be used to construct
234-- recursive structures and for reduction foldl1' is better.
235--
236-- | Lazy right fold for non-empty streams, using first element as the starting
237-- value. Returns 'Nothing' if the stream is empty.
238--
239-- @since 0.5.0
240{-# INLINE foldr1 #-}
241{-# DEPRECATED foldr1 "Use foldrM instead." #-}
242foldr1 :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
243foldr1 f m = D.foldr1 f (toStreamD m)
245------------------------------------------------------------------------------
246-- Left Folds
247------------------------------------------------------------------------------
249-- | Lazy left fold to a stream.
250{-# INLINE foldlS #-}
251foldlS :: IsStream t => (t m b -> a -> t m b) -> t m b -> t m a -> t m b
252foldlS f z =
253 IsStream.fromStream
254 . K.foldlS
255 (\xs x -> IsStream.toStream $ f (IsStream.fromStream xs) x)
256 (IsStream.toStream z)
257 . IsStream.toStream
259-- | Lazy left fold to a transformer monad.
260--
261-- For example, to reverse a stream:
262--
263-- > D.toList $ D.foldlT (flip D.cons) D.nil $ (D.fromList [1..5] :: SerialT IO Int)
264--
265{-# INLINE foldlT #-}
266foldlT :: (Monad m, IsStream t, Monad (s m), MonadTrans s)
267 => (s m b -> a -> s m b) -> s m b -> t m a -> s m b
268foldlT f z s = D.foldlT f z (toStreamD s)
270-- | Strict left fold with an extraction function. Like the standard strict
271-- left fold, but applies a user supplied extraction function (the third
272-- argument) to the folded value at the end. This is designed to work with the
273-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
274--
275-- @since 0.2.0
276{-# DEPRECATED foldx "Please use foldl' followed by fmap instead." #-}
277{-# INLINE foldx #-}
278foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
279foldx = IsStream.foldlx'
281-- | Left associative/strict push fold. @foldl' reduce initial stream@ invokes
282-- @reduce@ with the accumulator and the next input in the input stream, using
283-- @initial@ as the initial value of the current value of the accumulator. When
284-- the input is exhausted the current value of the accumulator is returned.
285-- Make sure to use a strict data structure for accumulator to not build
286-- unnecessary lazy expressions unless that's what you want. See the previous
287-- section for more details.
288--
289-- @since 0.2.0
290{-# INLINE foldl' #-}
291foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
292foldl' = IsStream.foldl'
294-- | Strict left fold, for non-empty streams, using first element as the
295-- starting value. Returns 'Nothing' if the stream is empty.
296--
297-- @since 0.5.0
298{-# INLINE foldl1' #-}
299foldl1' :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
300foldl1' step m = do
301 r <- uncons m
302 case r of
303 Nothing -> return Nothing
304 Just (h, t) -> do
305 res <- foldl' step h t
306 return $ Just res
308-- | Like 'foldx', but with a monadic step function.
309--
310-- @since 0.2.0
311{-# DEPRECATED foldxM "Please use foldlM' followed by fmap instead." #-}
312{-# INLINE foldxM #-}
313foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
314foldxM = IsStream.foldlMx'
316-- | Like 'foldl'' but with a monadic step function.
317--
318-- /Since: 0.2.0/
319--
320-- /Since: 0.8.0 (signature change)/
321{-# INLINE foldlM' #-}
322foldlM' :: Monad m => (b -> a -> m b) -> m b -> SerialT m a -> m b
323foldlM' step begin = D.foldlM' step begin . IsStream.toStreamD
325------------------------------------------------------------------------------
326-- Running a sink
327------------------------------------------------------------------------------
329{-
330-- | Drain a stream to a 'Sink'.
331{-# INLINE runSink #-}
332runSink :: Monad m => Sink m a -> SerialT m a -> m ()
333runSink = fold . toFold
334-}
335------------------------------------------------------------------------------
336-- Specific Fold Functions
337------------------------------------------------------------------------------
339-- XXX this can utilize parallel mapping if we implement it as drain . mapM
340-- |
341-- > mapM_ = Stream.drain . Stream.mapM
342--
343-- Apply a monadic action to each element of the stream and discard the output
344-- of the action. This is not really a pure transformation operation but a
345-- transformation followed by fold.
346--
347-- @since 0.1.0
348{-# INLINE mapM_ #-}
349mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()
350mapM_ f = D.mapM_ f . IsStream.toStreamD
352-- |
353-- > drain = mapM_ (\_ -> return ())
354-- > drain = Stream.fold Fold.drain
355--
356-- Run a stream, discarding the results. By default it interprets the stream
357-- as 'SerialT', to run other types of streams use the type adapting
358-- combinators for example @Stream.drain . 'fromAsync'@.
359--
360-- @since 0.7.0
361{-# INLINE drain #-}
362drain :: Monad m => SerialT m a -> m ()
363drain = IsStream.drain
365-- |
366-- > drainN n = Stream.drain . Stream.take n
367-- > drainN n = Stream.fold (Fold.take n Fold.drain)
368--
369-- Run maximum up to @n@ iterations of a stream.
370--
371-- @since 0.7.0
372{-# INLINE drainN #-}
373drainN :: Monad m => Int -> SerialT m a -> m ()
374drainN n = drain . take n
376-- |
377-- > runN n = runStream . take n
378--
379-- Run maximum up to @n@ iterations of a stream.
380--
381-- @since 0.6.0
382{-# DEPRECATED runN "Please use \"drainN\" instead" #-}
383{-# INLINE runN #-}
384runN :: Monad m => Int -> SerialT m a -> m ()
385runN = drainN
387-- |
388-- > drainWhile p = Stream.drain . Stream.takeWhile p
389--
390-- Run a stream as long as the predicate holds true.
391--
392-- @since 0.7.0
393{-# INLINE drainWhile #-}
394drainWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
395drainWhile p = drain . takeWhile p
397-- |
398-- > runWhile p = runStream . takeWhile p
399--
400-- Run a stream as long as the predicate holds true.
401--
402-- @since 0.6.0
403{-# DEPRECATED runWhile "Please use \"drainWhile\" instead" #-}
404{-# INLINE runWhile #-}
405runWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
406runWhile = drainWhile
408-- | Run a stream, discarding the results. By default it interprets the stream
409-- as 'SerialT', to run other types of streams use the type adapting
410-- combinators for example @runStream . 'fromAsync'@.
411--
412-- @since 0.2.0
413{-# DEPRECATED runStream "Please use \"drain\" instead" #-}
414{-# INLINE runStream #-}
415runStream :: Monad m => SerialT m a -> m ()
416runStream = drain
418-- | Determine whether the stream is empty.
419--
420-- > null = Stream.fold Fold.null
421--
422-- @since 0.1.1
423{-# INLINE null #-}
424null :: Monad m => SerialT m a -> m Bool
425null = D.null . toStreamD
427-- | Extract the first element of the stream, if any.
428--
429-- > head = (!! 0)
430-- > head = Stream.fold Fold.one
431--
432-- @since 0.1.0
433{-# INLINE head #-}
434head :: Monad m => SerialT m a -> m (Maybe a)
435head = D.head . toStreamD
437-- | Extract the first element of the stream, if any, otherwise use the
438-- supplied default value. It can help avoid one branch in high performance
439-- code.
440--
441-- /Pre-release/
442{-# INLINE headElse #-}
443headElse :: Monad m => a -> SerialT m a -> m a
444headElse x = D.headElse x . toStreamD
446-- |
447-- > tail = fmap (fmap snd) . Stream.uncons
448--
449-- Extract all but the first element of the stream, if any.
450--
451-- @since 0.1.1
452{-# INLINE tail #-}
453tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
454tail m = fmap (fmap IsStream.fromStream) $ K.tail $ Stream.toStreamK m
456-- | Extract all but the last element of the stream, if any.
457--
458-- @since 0.5.0
459{-# INLINE init #-}
460init :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
461init m = fmap (fmap IsStream.fromStream) $ K.init $ Stream.toStreamK m
463-- | Extract the last element of the stream, if any.
464--
465-- > last xs = xs !! (Stream.length xs - 1)
466-- > last = Stream.fold Fold.last
467--
468-- @since 0.1.1
469{-# INLINE last #-}
470last :: Monad m => SerialT m a -> m (Maybe a)
471last m = D.last $ toStreamD m
473-- | Determine whether an element is present in the stream.
474--
475-- > elem = Stream.fold Fold.elem
476--
477-- @since 0.1.0
478{-# INLINE elem #-}
479elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
480elem e m = D.elem e (toStreamD m)
482-- | Determine whether an element is not present in the stream.
483--
484-- > notElem = Stream.fold Fold.length
485--
486-- @since 0.1.0
487{-# INLINE notElem #-}
488notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
489notElem e m = D.notElem e (toStreamD m)
491-- | Determine the length of the stream.
492--
493-- @since 0.1.0
494{-# INLINE length #-}
495length :: Monad m => SerialT m a -> m Int
496length = foldl' (\n _ -> n + 1) 0
498-- | Determine whether all elements of a stream satisfy a predicate.
499--
500-- > all = Stream.fold Fold.all
501--
502-- @since 0.1.0
503{-# INLINE all #-}
504all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
505all p m = D.all p (toStreamD m)
507-- | Determine whether any of the elements of a stream satisfy a predicate.
508--
509-- > any = Stream.fold Fold.any
510--
511-- @since 0.1.0
512{-# INLINE any #-}
513any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
514any p m = D.any p (toStreamD m)
516-- | Determines if all elements of a boolean stream are True.
517--
518-- > and = Stream.fold Fold.and
519--
520-- @since 0.5.0
521{-# INLINE and #-}
522and :: Monad m => SerialT m Bool -> m Bool
523and = all (==True)
525-- | Determines whether at least one element of a boolean stream is True.
526--
527-- > or = Stream.fold Fold.or
528--
529-- @since 0.5.0
530{-# INLINE or #-}
531or :: Monad m => SerialT m Bool -> m Bool
532or = any (==True)
534-- | Determine the sum of all elements of a stream of numbers. Returns @0@ when
535-- the stream is empty. Note that this is not numerically stable for floating
536-- point numbers.
537--
538-- > sum = Stream.fold Fold.sum
539--
540-- @since 0.1.0
541{-# INLINE sum #-}
542sum :: (Monad m, Num a) => SerialT m a -> m a
543sum = foldl' (+) 0
545-- | Determine the product of all elements of a stream of numbers. Returns @1@
546-- when the stream is empty.
547--
548-- > product = Stream.fold Fold.product
549--
550-- @since 0.1.1
551{-# INLINE product #-}
552product :: (Monad m, Num a) => SerialT m a -> m a
553product = foldl' (*) 1
555-- | Fold a stream of monoid elements by appending them.
556--
557-- > mconcat = Stream.fold Fold.mconcat
558--
559-- /Pre-release/
560{-# INLINE mconcat #-}
561mconcat :: (Monad m, Monoid a) => SerialT m a -> m a
562mconcat = Stream.foldr mappend mempty
564-- |
565-- @
566-- minimum = 'minimumBy' compare
567 -- minimum = Stream.fold Fold.minimum
568-- @
569--
570-- Determine the minimum element in a stream.
571--
572-- @since 0.1.0
573{-# INLINE minimum #-}
574minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
575minimum m = D.minimum (toStreamD m)
577-- | Determine the minimum element in a stream using the supplied comparison
578-- function.
579--
580-- > minimumBy = Stream.fold Fold.minimumBy
581--
582-- @since 0.6.0
583{-# INLINE minimumBy #-}
584minimumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
585minimumBy cmp m = D.minimumBy cmp (toStreamD m)
587-- |
588-- @
589-- maximum = 'maximumBy' compare
590-- maximum = Stream.fold Fold.maximum
591-- @
592--
593-- Determine the maximum element in a stream.
594--
595-- @since 0.1.0
596{-# INLINE maximum #-}
597maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
598maximum m = D.maximum (toStreamD m)
600-- | Determine the maximum element in a stream using the supplied comparison
601-- function.
602--
603-- > maximumBy = Stream.fold Fold.maximumBy
604--
605-- @since 0.6.0
606{-# INLINE maximumBy #-}
607maximumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
608maximumBy cmp m = D.maximumBy cmp (toStreamD m)
610-- | Ensures that all the elements of the stream are identical and then returns
611-- that unique element.
612--
613-- @since 0.6.0
614{-# INLINE the #-}
615the :: (Eq a, Monad m) => SerialT m a -> m (Maybe a)
616the m = D.the (toStreamD m)
618------------------------------------------------------------------------------
619-- Searching
620------------------------------------------------------------------------------
622-- | Lookup the element at the given index.
623--
624-- @since 0.6.0
625{-# INLINE (!!) #-}
626(!!) :: Monad m => SerialT m a -> Int -> m (Maybe a)
627m !! i = toStreamD m D.!! i
629-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
630-- first pair where the key equals the given value @a@.
631--
632-- > lookup = snd <$> Stream.find ((==) . fst)
633-- > lookup = Stream.fold Fold.lookup
634--
635-- @since 0.5.0
636{-# INLINE lookup #-}
637lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)
638lookup a m = D.lookup a (toStreamD m)
640-- | Like 'findM' but with a non-monadic predicate.
641--
642-- > find p = findM (return . p)
643-- > find = Stream.fold Fold.find
644--
645-- @since 0.5.0
646{-# INLINE find #-}
647find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)
648find p m = D.find p (toStreamD m)
650-- | Returns the first element that satisfies the given predicate.
651--
652-- > findM = Stream.fold Fold.findM
653--
654-- @since 0.6.0
655{-# INLINE findM #-}
656findM :: Monad m => (a -> m Bool) -> SerialT m a -> m (Maybe a)
657findM p m = D.findM p (toStreamD m)
659-- | Returns the first index that satisfies the given predicate.
660--
661-- > findIndex = Stream.fold Fold.findIndex
662--
663-- @since 0.5.0
664{-# INLINE findIndex #-}
665findIndex :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe Int)
666findIndex p = head . findIndices p
668-- | Returns the first index where a given value is found in the stream.
669--
670-- > elemIndex a = Stream.findIndex (== a)
671--
672-- @since 0.5.0
673{-# INLINE elemIndex #-}
674elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int)
675elemIndex a = findIndex (== a)
677------------------------------------------------------------------------------
678-- To containers
679------------------------------------------------------------------------------
681-- |
682-- @
683-- toList = Stream.foldr (:) []
684-- @
685--
686-- Convert a stream into a list in the underlying monad. The list can be
687-- consumed lazily in a lazy monad (e.g. 'Identity'). In a strict monad (e.g.
688-- IO) the whole list is generated and buffered before it can be consumed.
689--
690-- /Warning!/ working on large lists accumulated as buffers in memory could be
691-- very inefficient, consider using "Streamly.Array" instead.
692--
693-- @since 0.1.0
694{-# INLINE toList #-}
695toList :: Monad m => SerialT m a -> m [a]
696toList = IsStream.toList
698-- |
699-- @
700-- toListRev = Stream.foldl' (flip (:)) []
701-- @
702--
703-- Convert a stream into a list in reverse order in the underlying monad.
704--
705-- /Warning!/ working on large lists accumulated as buffers in memory could be
706-- very inefficient, consider using "Streamly.Array" instead.
707--
708-- /Pre-release/
709{-# INLINE toListRev #-}
710toListRev :: Monad m => SerialT m a -> m [a]
711toListRev = D.toListRev . toStreamD
713-- |
714-- @
715-- toHandle h = D.mapM_ $ hPutStrLn h
716-- @
717--
718-- Write a stream of Strings to an IO Handle.
719--
720-- @since 0.1.0
721{-# DEPRECATED toHandle
722 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
723toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
724toHandle h = go
725 where
726 go m1 =
727 let stop = return ()
728 single a = liftIO (IO.hPutStrLn h a)
729 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r
730 in IsStream.foldStream defState yieldk single stop m1
732-- | Convert a stream to a pure stream.
733--
734-- @
735-- toStream = Stream.foldr Stream.cons Stream.nil
736-- @
737--
738-- /Pre-release/
739--
740{-# INLINE toStream #-}
741toStream :: Monad m => SerialT m a -> m (SerialT n a)
742toStream = Stream.foldr IsStream.cons IsStream.nil
744-- | Convert a stream to a pure stream in reverse order.
745--
746-- @
747-- toStreamRev = Stream.foldl' (flip Stream.cons) Stream.nil
748-- @
749--
750-- /Pre-release/
751--
752{-# INLINE toStreamRev #-}
753toStreamRev :: Monad m => SerialT m a -> m (SerialT n a)
754toStreamRev = foldl' (flip IsStream.cons) IsStream.nil
756------------------------------------------------------------------------------
757-- Concurrent Application
758------------------------------------------------------------------------------
760-- | Parallel fold application operator; applies a fold function @t m a -> m b@
761-- to a stream @t m a@ concurrently; The the input stream is evaluated
762-- asynchronously in an independent thread yielding elements to a buffer and
763-- the folding action runs in another thread consuming the input from the
764-- buffer.
765--
766-- If you read the signature as @(t m a -> m b) -> (t m a -> m b)@ you can look
767-- at it as a transformation that converts a fold function to a buffered
768-- concurrent fold function.
769--
770-- The @.@ at the end of the operator is a mnemonic for termination of the
771-- stream.
772--
773-- In the example below, each stage introduces a delay of 1 sec but output is
774-- printed every second because both stages are concurrent.
775--
776-- >>> import Control.Concurrent (threadDelay)
777-- >>> import Streamly.Prelude ((|$.))
778-- >>> :{
779-- Stream.foldlM' (\_ a -> threadDelay 1000000 >> print a) (return ())
780-- |$. Stream.replicateM 3 (threadDelay 1000000 >> return 1)
781-- :}
782-- 1
783-- 1
784-- 1
785--
786-- /Concurrent/
787--
788-- /Since: 0.3.0 ("Streamly")/
789--
790-- @since 0.8.0
791{-# INLINE (|$.) #-}
792(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> (t m a -> m b)
793-- (|$.) f = f . Async.mkAsync
794(|$.) f = f . mkParallel
796infixr 0 |$.
798-- | Same as '|$.'.
799--
800-- /Internal/
801--
802{-# INLINE foldAsync #-}
803foldAsync :: (IsStream t, MonadAsync m) => (t m a -> m b) -> (t m a -> m b)
804foldAsync = (|$.)
806-- | Same as '|$.' but with arguments reversed.
807--
808-- > (|&.) = flip (|$.)
809--
810-- /Concurrent/
811--
812-- /Since: 0.3.0 ("Streamly")/
813--
814-- @since 0.8.0
815{-# INLINE (|&.) #-}
816(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
817x |&. f = f |$. x
819infixl 1 |&.
821------------------------------------------------------------------------------
822-- Multi-stream folds
823------------------------------------------------------------------------------
825-- | Returns 'True' if the first stream is the same as or a prefix of the
826-- second. A stream is a prefix of itself.
827--
828-- >>> Stream.isPrefixOf (Stream.fromList "hello") (Stream.fromList "hello" :: SerialT IO Char)
829-- True
830--
831-- @since 0.6.0
832{-# INLINE isPrefixOf #-}
833isPrefixOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
834isPrefixOf m1 m2 = D.isPrefixOf (toStreamD m1) (toStreamD m2)
836-- | Returns 'True' if the first stream is an infix of the second. A stream is
837-- considered an infix of itself.
838--
839-- > Stream.isInfixOf (Stream.fromList "hello") (Stream.fromList "hello" :: SerialT IO Char)
840-- True
841--
842-- Space: @O(n)@ worst case where @n@ is the length of the infix.
843--
844-- /Pre-release/
845--
846-- /Requires 'Storable' constraint/
847--
848{-# INLINE isInfixOf #-}
849isInfixOf :: (MonadIO m, Eq a, Enum a, Unboxed a)
850 => SerialT m a -> SerialT m a -> m Bool
851isInfixOf infx stream = do
852 arr <- fold A.write infx
853 -- XXX can use breakOnSeq instead (when available)
854 r <- null $ drop 1 $ splitOnSeq arr FL.drain stream
855 return (not r)
857-- Note: isPrefixOf uses the prefix stream only once. In contrast, isSuffixOf
858-- may use the suffix stream many times. To run in optimal memory we do not
859-- want to buffer the suffix stream in memory therefore we need an ability to
860-- clone (or consume it multiple times) the suffix stream without any side
861-- effects so that multiple potential suffix matches can proceed in parallel
862-- without buffering the suffix stream. For example, we may create the suffix
863-- stream from a file handle, however, if we evaluate the stream multiple
864-- times, once for each match, we will need a different file handle each time
865-- which may exhaust the file descriptors. Instead, we want to share the same
866-- underlying file descriptor, use pread on it to generate the stream and clone
867-- the stream for each match. Therefore the suffix stream should be built in
868-- such a way that it can be consumed multiple times without any problems.
870-- XXX Can be implemented with better space/time complexity.
871-- Space: @O(n)@ worst case where @n@ is the length of the suffix.
873-- | Returns 'True' if the first stream is a suffix of the second. A stream is
874-- considered a suffix of itself.
875--
876-- >>> Stream.isSuffixOf (Stream.fromList "hello") (Stream.fromList "hello" :: SerialT IO Char)
877-- True
878--
879-- Space: @O(n)@, buffers entire input stream and the suffix.
880--
881-- /Pre-release/
882--
883-- /Suboptimal/ - Help wanted.
884--
885{-# INLINE isSuffixOf #-}
886isSuffixOf :: (Monad m, Eq a) => SerialT m a -> SerialT m a -> m Bool
887isSuffixOf suffix stream = reverse suffix `isPrefixOf` reverse stream
889-- | Returns 'True' if all the elements of the first stream occur, in order, in
890-- the second stream. The elements do not have to occur consecutively. A stream
891-- is a subsequence of itself.
892--
893-- >>> Stream.isSubsequenceOf (Stream.fromList "hlo") (Stream.fromList "hello" :: SerialT IO Char)
894-- True
895--
896-- @since 0.6.0
897{-# INLINE isSubsequenceOf #-}
898isSubsequenceOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
899isSubsequenceOf m1 m2 = D.isSubsequenceOf (toStreamD m1) (toStreamD m2)
901-- Note: If we want to return a Maybe value to know whether the
902-- suffix/infix was present or not along with the stripped stream then
903-- we need to buffer the whole input stream.
904--
905-- | @stripPrefix prefix stream@ strips @prefix@ from @stream@ if it is a
906-- prefix of stream. Returns 'Nothing' if the stream does not start with the
907-- given prefix, stripped stream otherwise. Returns @Just nil@ when the prefix
908-- is the same as the stream.
909--
910-- See also "Streamly.Internal.Data.Stream.IsStream.Nesting.dropPrefix".
911--
912-- Space: @O(1)@
913--
914-- @since 0.6.0
915{-# INLINE stripPrefix #-}
916stripPrefix
917 :: (Eq a, IsStream t, Monad m)
918 => t m a -> t m a -> m (Maybe (t m a))
919stripPrefix m1 m2 = fmap fromStreamD <$>
920 D.stripPrefix (toStreamD m1) (toStreamD m2)
922------------------------------------------------------------------------------
923-- Comparison
924------------------------------------------------------------------------------
926-- | Compare two streams for equality using an equality function.
927--
928-- @since 0.6.0
929{-# INLINABLE eqBy #-}
930eqBy :: (IsStream t, Monad m) => (a -> b -> Bool) -> t m a -> t m b -> m Bool
931eqBy = IsStream.eqBy
933-- | Compare two streams lexicographically using a comparison function.
934--
935-- @since 0.6.0
936{-# INLINABLE cmpBy #-}
937cmpBy
938 :: (IsStream t, Monad m)
939 => (a -> b -> Ordering) -> t m a -> t m b -> m Ordering
940cmpBy = IsStream.cmpBy