1{-# OPTIONS_GHC -Wno-deprecations #-}
2{-# OPTIONS_GHC -Wno-orphans #-}
5-- Module : Streamly.Internal.Data.Stream.IsStream.Common
6-- Copyright : (c) 2017 Composewell Technologies
7-- License : BSD-3-Clause
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
12-- Bottom level IsStream module that can be used by all other upper level
15module Streamly.Internal.Data.Stream.IsStream.Common {-# DEPRECATED "Please use \"Streamly.Data.Stream.*\" instead." #-}
71import Control.Concurrent (threadDelay)
72import Control.Monad.IO.Class (MonadIO(..))
73import Foreign.Storable (Storable)
74import Streamly.Internal.Control.Concurrent (MonadAsync)
75import Streamly.Internal.Data.Array.Type (Array)
76import Streamly.Internal.Data.Fold.Type (Fold (..))
77import Streamly.Internal.Data.Stream.IsStream.Combinators (maxYields)
78import Streamly.Internal.Data.Stream.IsStream.Type
79 (IsStream(..), fromStreamD, toStreamD)
80import Streamly.Internal.Data.Stream.Serial (SerialT)
81import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64, addToAbsTime64)
82import Streamly.Internal.System.IO (defaultChunkSize)
83import Streamly.Internal.Data.Unboxed (Unbox)
85import qualified Streamly.Data.Stream as Stream (fold, foldBreak)
86import qualified Streamly.Internal.Data.Array.Type as A
87import qualified Streamly.Internal.Data.Stream.Async as Async
88import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
89import qualified Streamly.Internal.Data.Stream.Parallel as Par
90import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
91 (fromPure, fromEffect, repeatMWith, reverse)
92import qualified Streamly.Internal.Data.Stream.StreamD as D
93 (repeatM, times, foldContinue, map, scanlMAfter', postscanlMAfter'
94 , postscanlM', take, takeWhile, takeEndBy, drop, findIndices
95 , fromStreamK, toStreamK, concatMapM, concatMap, foldManyPost, splitOnSeq
96 , zipWithM, zipWith, intersperseM, reverse)
98import Prelude hiding (take, takeWhile, drop, reverse, concatMap, map, zipWith)
103-- >>> :set -fno-warn-deprecations
104-- >>> import Control.Concurrent (threadDelay)
105-- >>> import Control.Monad (join)
106-- >>> import Control.Monad.Trans.Class (lift)
107-- >>> import Data.Function (fix, (&))
108-- >>> import Data.Semigroup (cycle1)
109-- >>> import Prelude hiding (take, takeWhile, drop, reverse)
110-- >>> import System.IO.Unsafe (unsafePerformIO)
111-- >>> import qualified Streamly.Prelude as Stream
112-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
113-- >>> import qualified Streamly.Data.Array as Array
114-- >>> import qualified Streamly.Data.Fold as Fold
115-- >>> import qualified Streamly.Internal.Data.Fold as Fold
116-- >>> import qualified Streamly.Data.Unfold as Unfold
118------------------------------------------------------------------------------
120------------------------------------------------------------------------------
122-- Faster than fromEffect because there is no bind.
126-- fromPure a = a \`cons` nil
129-- Create a singleton stream from a pure value.
131-- The following holds in monadic streams, but not in Zip streams:
135-- fromPure = fromEffect . pure
138-- In Zip applicative streams 'fromPure' is not the same as 'pure' because in that
139-- case 'pure' is equivalent to 'repeat' instead. 'fromPure' and 'pure' are
140-- equally efficient, in other cases 'fromPure' may be slightly more efficient
141-- than the other equivalent definitions.
143-- /Since: 0.8.0 (Renamed yield to fromPure)/
145{-# INLINE fromPure #-}
146fromPure :: IsStream t => a -> t m a
147fromPure = fromStream . K.fromPure
149-- | Same as 'fromPure'
152{-# DEPRECATED yield "Please use fromPure instead." #-}
154yield :: IsStream t => a -> t m a
159-- fromEffect m = m \`consM` nil
162-- Create a singleton stream from a monadic action.
165-- > Stream.toList $ Stream.fromEffect getLine
170-- /Since: 0.8.0 (Renamed yieldM to fromEffect)/
172{-# INLINE fromEffect #-}
173fromEffect :: (Monad m, IsStream t) => m a -> t m a
174fromEffect = fromStream . K.fromEffect
176-- | Same as 'fromEffect'
179{-# DEPRECATED yieldM "Please use fromEffect instead." #-}
181yieldM :: (Monad m, IsStream t) => m a -> t m a
184-- >>> repeatM = fix . consM
185-- >>> repeatM = cycle1 . fromEffect
187-- Generate a stream by repeatedly executing a monadic action forever.
191-- Stream.repeatM (threadDelay 1000000 >> print 1)
197-- /Concurrent, infinite (do not use with 'fromParallel')/
200{-# INLINE_EARLY repeatM #-}
201repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
202repeatM = K.repeatMWith IsStream.consM
204{-# RULES "repeatM serial" repeatM = repeatMSerial #-}
205{-# INLINE repeatMSerial #-}
206repeatMSerial :: MonadAsync m => m a -> SerialT m a
207repeatMSerial = fromStreamD . D.repeatM
209------------------------------------------------------------------------------
210-- Generation - Time related
211------------------------------------------------------------------------------
213-- | @timesWith g@ returns a stream of time value tuples. The first component
214-- of the tuple is an absolute time reference (epoch) denoting the start of the
215-- stream and the second component is a time relative to the reference.
217-- The argument @g@ specifies the granularity of the relative time in seconds.
218-- A lower granularity clock gives higher precision but is more expensive in
219-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
221-- >>> import Control.Concurrent (threadDelay)
222-- >>> import Streamly.Internal.Data.Stream.IsStream.Common as Stream (timesWith)
223-- >>> Stream.mapM_ (\x -> print x >> threadDelay 1000000) $ Stream.take 3 $ Stream.timesWith 0.01
224-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
225-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
226-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
228-- Note: This API is not safe on 32-bit machines.
232{-# INLINE timesWith #-}
233timesWith :: (IsStream t, MonadAsync m) => Double -> t m (AbsTime, RelTime64)
234timesWith g = fromStreamD $ D.times g
236-- | @absTimesWith g@ returns a stream of absolute timestamps using a clock of
237-- granularity @g@ specified in seconds. A low granularity clock is more
238-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
241-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ absTimesWith 0.01
242-- AbsTime (TimeSpec {sec = ..., nsec = ...})
243-- AbsTime (TimeSpec {sec = ..., nsec = ...})
244-- AbsTime (TimeSpec {sec = ..., nsec = ...})
246-- Note: This API is not safe on 32-bit machines.
250{-# INLINE absTimesWith #-}
251absTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
252 => Double -> t m AbsTime
253absTimesWith = fmap (uncurry addToAbsTime64) . timesWith
255-- | @relTimesWith g@ returns a stream of relative time values starting from 0,
256-- using a clock of granularity @g@ specified in seconds. A low granularity
257-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
258-- ms is treated as 1 ms.
260-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimesWith 0.01
261-- RelTime64 (NanoSecond64 ...)
262-- RelTime64 (NanoSecond64 ...)
263-- RelTime64 (NanoSecond64 ...)
265-- Note: This API is not safe on 32-bit machines.
269{-# INLINE relTimesWith #-}
270relTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
271 => Double -> t m RelTime64
272relTimesWith = fmap snd . timesWith
274------------------------------------------------------------------------------
275-- Elimination - Running a Fold
276------------------------------------------------------------------------------
278-- | We can create higher order folds using 'foldContinue'. We can fold a
279-- number of streams to a given fold efficiently with full stream fusion. For
280-- example, to fold a list of streams on the same sum fold:
282-- >>> concatFold = Prelude.foldl Stream.foldContinue Fold.sum
284-- >>> fold f = Fold.extractM . Stream.foldContinue f
287{-# INLINE foldContinue #-}
288foldContinue :: Monad m => Fold m a b -> SerialT m a -> Fold m a b
289foldContinue f s = D.foldContinue f $ IsStream.toStreamD s
291------------------------------------------------------------------------------
293------------------------------------------------------------------------------
303-- > D.toList $ D.map (+1) $ D.fromList [1,2,3]
309map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
310map f = fromStreamD . D.map f . toStreamD
312-- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
313-- that it provides an additional @done@ function to be applied on the
314-- accumulator when the stream stops. The result of @done@ is also emitted in
317-- This function can be used to allocate a resource in the beginning of the
318-- scan and release it when the stream ends or to flush the internal state of
319-- the scan at the end.
323{-# INLINE scanlMAfter' #-}
324scanlMAfter' :: (IsStream t, Monad m)
325 => (b -> a -> m b) -> m b -> (b -> m b) -> t m a -> t m b
326scanlMAfter' step initial done stream =
327 fromStreamD $ D.scanlMAfter' step initial done $ toStreamD stream
329{-# INLINE postscanlMAfter' #-}
330postscanlMAfter' :: (IsStream t, Monad m)
331 => (b -> a -> m b) -> m b -> (b -> m b) -> t m a -> t m b
332postscanlMAfter' step initial done stream =
333 fromStreamD $ D.postscanlMAfter' step initial done $ toStreamD stream
335-- XXX this needs to be concurrent
336-- | Like 'postscanl'' but with a monadic step function and a monadic seed.
338-- >>> postscanlM' f z xs = Stream.drop 1 $ Stream.scanlM' f z xs
342-- /Since: 0.8.0 (signature change)/
343{-# INLINE postscanlM' #-}
344postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
345postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
349-- The stateful step function can be simplified to @(s -> a -> m b)@ to provide
350-- a read-only environment. However, that would just be 'mapM'.
352-- The initial action could be @m (s, Maybe b)@, and we can also add a final
353-- action @s -> m (Maybe b)@. This can be used to get pre/post scan like
354-- functionality and also to flush the state in the end like scanlMAfter'.
355-- We can also use it along with a fusible version of bracket to get
356-- scanlMAfter' like functionality. See issue #677.
358-- This can be further generalized to a type similar to Fold/Parser, giving it
359-- filtering and parsing capability as well (this is in fact equivalent to
362-- smapM :: (s -> a -> m (Step s b)) -> m s -> t m a -> t m b
365-- | A stateful 'mapM', equivalent to a left scan, more like mapAccumL.
366-- Hopefully, this is a better alternative to @scan@. Separation of state from
367-- the output makes it easier to think in terms of a shared state, and also
368-- makes it easier to keep the state fully strict and the output lazy.
370-- See also: 'scanlM''
375smapM :: (IsStream t, Monad m) =>
380smapM step initial stream =
381 -- XXX implement this directly instead of using scanlM'
382 -- Once we have postscanlM' with monadic initial we can use this code
383 -- let r = postscanlM'
384 -- (\(s, _) a -> step s a)
385 -- (fmap (,undefined) initial)
388 (\(s, _) a -> step s a)
389 (fmap (,undefined) initial)
393------------------------------------------------------------------------------
394-- Transformation - Trimming
395------------------------------------------------------------------------------
397-- | Take first 'n' elements from the stream and discard the rest.
401take :: (IsStream t, Monad m) => Int -> t m a -> t m a
402take n m = fromStreamD $ D.take n $ toStreamD
403 (maxYields (Just (fromIntegral n)) m)
405-- | End the stream as soon as the predicate fails on an element.
408{-# INLINE takeWhile #-}
409takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
410takeWhile p m = fromStreamD $ D.takeWhile p $ toStreamD m
412{-# INLINE takeEndBy #-}
413takeEndBy :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
414takeEndBy p m = fromStreamD $ D.takeEndBy p $ toStreamD m
416-- | Discard first 'n' elements from the stream and take the rest.
420drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
421drop n m = fromStreamD $ D.drop n $ toStreamD m
423------------------------------------------------------------------------------
425------------------------------------------------------------------------------
427-- | Find all the indices where the element in the stream satisfies the given
430-- > findIndices = fold Fold.findIndices
433{-# INLINE findIndices #-}
434findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
435findIndices p m = fromStreamD $ D.findIndices p (toStreamD m)
437------------------------------------------------------------------------------
438-- Transformation by Inserting
439------------------------------------------------------------------------------
441-- intersperseM = intersperseBySpan 1
443-- | Insert an effect and its output before consuming an element of a stream
444-- except the first one.
446-- >>> Stream.toList $ Stream.trace putChar $ Stream.intersperseM (putChar '.' >> return ',') $ Stream.fromList "hello"
447-- h.,e.,l.,l.,o"h,e,l,l,o"
449-- Be careful about the order of effects. In the above example we used trace
450-- after the intersperse, if we use it before the intersperse the output would
451-- be he.l.l.o."h,e,l,l,o".
453-- >>> Stream.toList $ Stream.intersperseM (putChar '.' >> return ',') $ Stream.trace putChar $ Stream.fromList "hello"
454-- he.l.l.o."h,e,l,l,o"
457{-# INLINE intersperseM #-}
458intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
459intersperseM m = fromStreamD . D.intersperseM m . toStreamD
461-- | Intersperse a monadic action into the input stream after every @n@
465-- > import Control.Concurrent (threadDelay)
466-- > Stream.drain $ Stream.interjectSuffix 1 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
471{-# INLINE interjectSuffix #-}
473 :: (IsStream t, MonadAsync m)
474 => Double -> m a -> t m a -> t m a
475interjectSuffix n f xs = xs `parallelFst` repeatM timed
476 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
478------------------------------------------------------------------------------
479-- Transformation by Reordering
480------------------------------------------------------------------------------
482-- XXX Use a compact region list to temporarily store the list, in both reverse
483-- as well as in reverse'.
485-- /Note:/ 'reverse'' is much faster than this, use that when performance
488-- | Returns the elements of the stream in reverse order. The stream must be
489-- finite. Note that this necessarily buffers the entire stream in memory.
491-- >>> reverse = Stream.foldlT (flip Stream.cons) Stream.nil
493-- /Since 0.7.0 (Monad m constraint)/
496{-# INLINE reverse #-}
497reverse :: (IsStream t, Monad m) => t m a -> t m a
498reverse s = fromStreamD $ D.reverse $ toStreamD s
500-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
503{-# INLINE reverse' #-}
504reverse' :: (IsStream t, MonadIO m, Unbox a) => t m a -> t m a
505-- reverse' s = fromStreamD $ D.reverse' $ toStreamD s
508 . A.flattenArraysRev -- unfoldMany A.readRev
512 . A.arraysOf defaultChunkSize
515------------------------------------------------------------------------------
516-- Concurrent Transformations and Combining
517------------------------------------------------------------------------------
519-- | Make the stream producer and consumer run concurrently by introducing a
520-- buffer between them. The producer thread evaluates the input stream until
521-- the buffer fills, it terminates if the buffer is full and a worker thread is
522-- kicked off again to evaluate the remaining stream when there is space in the
523-- buffer. The consumer consumes the stream lazily from the buffer.
525-- /Since: 0.2.0 (Streamly)/
529{-# INLINE_NORMAL mkAsync #-}
530mkAsync :: (IsStream t, MonadAsync m) => t m a -> t m a
531mkAsync = fromStreamD . Async.mkAsyncD . toStreamD
533-- Compare with mkAsync. mkAsync uses an Async style SVar whereas this uses a
534-- parallel style SVar for evaluation. Currently, parallel style cannot use
535-- rate control whereas Async style can use rate control. In async style SVar
536-- the worker thread terminates when the buffer is full whereas in Parallel
539-- | Make the stream producer and consumer run concurrently by introducing a
540-- buffer between them. The producer thread evaluates the input stream until
541-- the buffer fills, it blocks if the buffer is full until there is space in
542-- the buffer. The consumer consumes the stream lazily from the buffer.
544-- @mkParallel = IsStream.fromStreamD . mkParallelD . IsStream.toStreamD@
548{-# INLINE_NORMAL mkParallel #-}
549mkParallel :: (IsStream t, MonadAsync m) => t m a -> t m a
550mkParallel = fromStreamD . Par.mkParallelD . toStreamD
552-- | Like `parallel` but stops the output as soon as the first stream stops.
555{-# INLINE parallelFst #-}
556parallelFst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
557parallelFst m1 m2 = fromStream $ Par.parallelFstK (toStream m1) (toStream m2)
559------------------------------------------------------------------------------
560-- Combine streams and flatten
561------------------------------------------------------------------------------
563-- | Map a stream producing monadic function on each element of the stream
564-- and then flatten the results into a single stream. Since the stream
565-- generation function is monadic, unlike 'concatMap', it can produce an
566-- effect at the beginning of each iteration of the inner loop.
569{-# INLINE concatMapM #-}
570concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b
571concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
573-- | Map a stream producing function on each element of the stream and then
574-- flatten the results into a single stream.
576-- >>> concatMap f = Stream.concatMapM (return . f)
577-- >>> concatMap f = Stream.concatMapWith Stream.serial f
578-- >>> concatMap f = Stream.concat . Stream.map f
581{-# INLINE concatMap #-}
582concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
583concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
585-- | Given a stream value in the underlying monad, lift and join the underlying
586-- monad with the stream monad.
588-- >>> concatM = Stream.concat . Stream.fromEffect
589-- >>> concatM = Stream.concat . lift -- requires (MonadTrans t)
590-- >>> concatM = join . lift -- requires (MonadTrans t, Monad (t m))
592-- See also: 'concat', 'sequence'
596{-# INLINE concatM #-}
597concatM :: (IsStream t, Monad m) => m (t m a) -> t m a
598concatM generator = concatMapM (\() -> generator) (fromPure ())
600-- | Like 'foldMany' but appends empty fold output if the fold and stream
601-- termination aligns:
603-- >>> f = Fold.take 2 Fold.sum
604-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList []
606-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..9]
608-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..10]
613{-# INLINE foldManyPost #-}
615 :: (IsStream t, Monad m)
619foldManyPost f m = fromStreamD $ D.foldManyPost f (toStreamD m)
621------------------------------------------------------------------------------
622-- Split stream and fold
623------------------------------------------------------------------------------
626-- > intercalate . splitOnSeq == id
627-- prop> (\f -> unsafePerformIO . Stream.toList . f . Stream.fromList) (Stream.intercalate Unfold.fromList " " . Stream.splitOnSeq (Array.fromList " ") Fold.toList) xs == xs
629-- | Like 'splitOn' but the separator is a sequence of elements instead of a
632-- For illustration, let's define a function that operates on pure lists:
634-- >>> splitOnSeq' pat xs = Stream.toList $ Stream.splitOnSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
636-- >>> splitOnSeq' "" "hello"
637-- ["h","e","l","l","o"]
639-- >>> splitOnSeq' "hello" ""
642-- >>> splitOnSeq' "hello" "hello"
645-- >>> splitOnSeq' "x" "hello"
648-- >>> splitOnSeq' "h" "hello"
651-- >>> splitOnSeq' "o" "hello"
654-- >>> splitOnSeq' "e" "hello"
657-- >>> splitOnSeq' "l" "hello"
660-- >>> splitOnSeq' "ll" "hello"
663-- 'splitOnSeq' is an inverse of 'intercalate'. The following law always holds:
665-- > intercalate . splitOnSeq == id
667-- The following law holds when the separator is non-empty and contains none of
668-- the elements present in the input lists:
670-- > splitOnSeq . intercalate == id
672-- >>> splitOnSeq pat f = Stream.foldManyPost (Fold.takeEndBySeq_ pat f)
676-- XXX We can use a polymorphic vector implemented by Array# to represent the
677-- sequence, that way we can avoid the Storable constraint. If we still need
678-- Storable Array for performance, we can use a separate splitOnArray API for
679-- that. We can also have an API where the sequence itself is a lazy stream, so
680-- that we can search files in files for example.
681{-# INLINE splitOnSeq #-}
683 :: (IsStream t, MonadIO m, Storable a, Unbox a, Enum a, Eq a)
684 => Array a -> Fold m a b -> t m a -> t m b
686 IsStream.fromStreamD $ D.splitOnSeq patt f (IsStream.toStreamD m)
687-- XXX This may have a problem if the stream terminates and we start extracting
688-- the fold and then the fold terminates before consuming all the buffered
689-- input. We have no way to supply the buffered input back to the driver.
690-- splitOnSeq patt f =
691 -- foldManyPost (Fold.takeEndBySeq_ patt f)
693------------------------------------------------------------------------------
695------------------------------------------------------------------------------
697-- | Like 'zipWith' but using a monadic zipping function.
700{-# INLINE zipWithM #-}
701zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
704 $ D.zipWithM f (IsStream.toStreamD m1) (IsStream.toStreamD m2)
706-- | Stream @a@ is evaluated first, followed by stream @b@, the resulting
707-- elements @a@ and @b@ are then zipped using the supplied zip function and the
708-- result @c@ is yielded to the consumer.
710-- If stream @a@ or stream @b@ ends, the zipped stream ends. If stream @b@ ends
711-- first, the element @a@ from previous evaluation of stream @a@ is discarded.
714-- > D.toList $ D.zipWith (+) (D.fromList [1,2,3]) (D.fromList [4,5,6])
719{-# INLINE zipWith #-}
720zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
723 $ D.zipWith f (IsStream.toStreamD m1) (IsStream.toStreamD m2)
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list