1{-# OPTIONS_GHC -Wno-orphans #-}
4-- Module : Streamly.Internal.Data.Stream.IsStream.Common
5-- Copyright : (c) 2017 Composewell Technologies
6-- License : BSD-3-Clause
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
11-- Bottom level IsStream module that can be used by all other upper level
14module Streamly.Internal.Data.Stream.IsStream.Common
66import Control.Concurrent (threadDelay)
67import Control.Monad.IO.Class (MonadIO(..))
68import Foreign.Storable (Storable)
69import Streamly.Internal.Control.Concurrent (MonadAsync)
70import Streamly.Internal.Data.Array.Foreign.Type (Array)
71import Streamly.Internal.Data.Fold.Type (Fold (..))
72import Streamly.Internal.Data.Stream.IsStream.Combinators (maxYields)
73import Streamly.Internal.Data.Stream.IsStream.Type
74 (IsStream(..), fromStreamD, toStreamD, fromStreamS, toStreamS)
75import Streamly.Internal.Data.Stream.Serial (SerialT)
76import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64, addToAbsTime64)
78import qualified Streamly.Internal.Data.Array.Foreign.Type as A
79import qualified Streamly.Internal.Data.Stream.Async as Async
80import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
81import qualified Streamly.Internal.Data.Stream.Parallel as Par
82import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
83import qualified Streamly.Internal.Data.Stream.StreamD as D
84#ifdef USE_STREAMK_ONLY
85import qualified Streamly.Internal.Data.Stream.StreamK as S
87import qualified Streamly.Internal.Data.Stream.StreamD as S
89import Streamly.Internal.System.IO (defaultChunkSize)
91import Prelude hiding (take, takeWhile, drop, reverse, concatMap, map, zipWith)
96-- >>> import Control.Concurrent (threadDelay)
97-- >>> import Control.Monad (join)
98-- >>> import Control.Monad.Trans.Class (lift)
99-- >>> import Data.Function (fix, (&))
100-- >>> import Data.Semigroup (cycle1)
101-- >>> import Prelude hiding (take, takeWhile, drop, reverse)
102-- >>> import System.IO.Unsafe (unsafePerformIO)
103-- >>> import qualified Streamly.Prelude as Stream
104-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
105-- >>> import qualified Streamly.Data.Array.Foreign as Array
106-- >>> import qualified Streamly.Data.Fold as Fold
107-- >>> import qualified Streamly.Data.Unfold as Unfold
109------------------------------------------------------------------------------
111------------------------------------------------------------------------------
113-- Faster than fromEffect because there is no bind.
117-- fromPure a = a \`cons` nil
120-- Create a singleton stream from a pure value.
122-- The following holds in monadic streams, but not in Zip streams:
126-- fromPure = fromEffect . pure
129-- In Zip applicative streams 'fromPure' is not the same as 'pure' because in that
130-- case 'pure' is equivalent to 'repeat' instead. 'fromPure' and 'pure' are
131-- equally efficient, in other cases 'fromPure' may be slightly more efficient
132-- than the other equivalent definitions.
134-- /Since: 0.8.0 (Renamed yield to fromPure)/
136{-# INLINE fromPure #-}
137fromPure :: IsStream t => a -> t m a
138fromPure = fromStream . K.fromPure
140-- | Same as 'fromPure'
143{-# DEPRECATED yield "Please use fromPure instead." #-}
145yield :: IsStream t => a -> t m a
150-- fromEffect m = m \`consM` nil
153-- Create a singleton stream from a monadic action.
156-- > Stream.toList $ Stream.fromEffect getLine
161-- /Since: 0.8.0 (Renamed yieldM to fromEffect)/
163{-# INLINE fromEffect #-}
164fromEffect :: (Monad m, IsStream t) => m a -> t m a
165fromEffect = fromStream . K.fromEffect
167-- | Same as 'fromEffect'
170{-# DEPRECATED yieldM "Please use fromEffect instead." #-}
172yieldM :: (Monad m, IsStream t) => m a -> t m a
175-- >>> repeatM = fix . consM
176-- >>> repeatM = cycle1 . fromEffect
178-- Generate a stream by repeatedly executing a monadic action forever.
182-- Stream.repeatM (threadDelay 1000000 >> print 1)
188-- /Concurrent, infinite (do not use with 'fromParallel')/
191{-# INLINE_EARLY repeatM #-}
192repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
193repeatM = K.repeatMWith IsStream.consM
195{-# RULES "repeatM serial" repeatM = repeatMSerial #-}
196{-# INLINE repeatMSerial #-}
197repeatMSerial :: MonadAsync m => m a -> SerialT m a
198repeatMSerial = fromStreamS . S.repeatM
200------------------------------------------------------------------------------
201-- Generation - Time related
202------------------------------------------------------------------------------
204-- | @timesWith g@ returns a stream of time value tuples. The first component
205-- of the tuple is an absolute time reference (epoch) denoting the start of the
206-- stream and the second component is a time relative to the reference.
208-- The argument @g@ specifies the granularity of the relative time in seconds.
209-- A lower granularity clock gives higher precision but is more expensive in
210-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
212-- >>> import Control.Concurrent (threadDelay)
213-- >>> import Streamly.Internal.Data.Stream.IsStream.Common as Stream (timesWith)
214-- >>> Stream.mapM_ (\x -> print x >> threadDelay 1000000) $ Stream.take 3 $ Stream.timesWith 0.01
215-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
216-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
217-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
219-- Note: This API is not safe on 32-bit machines.
223{-# INLINE timesWith #-}
224timesWith :: (IsStream t, MonadAsync m) => Double -> t m (AbsTime, RelTime64)
225timesWith g = fromStreamD $ D.times g
227-- | @absTimesWith g@ returns a stream of absolute timestamps using a clock of
228-- granularity @g@ specified in seconds. A low granularity clock is more
229-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
232-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ absTimesWith 0.01
233-- AbsTime (TimeSpec {sec = ..., nsec = ...})
234-- AbsTime (TimeSpec {sec = ..., nsec = ...})
235-- AbsTime (TimeSpec {sec = ..., nsec = ...})
237-- Note: This API is not safe on 32-bit machines.
241{-# INLINE absTimesWith #-}
242absTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
243 => Double -> t m AbsTime
244absTimesWith = fmap (uncurry addToAbsTime64) . timesWith
246-- | @relTimesWith g@ returns a stream of relative time values starting from 0,
247-- using a clock of granularity @g@ specified in seconds. A low granularity
248-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
249-- ms is treated as 1 ms.
251-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimesWith 0.01
252-- RelTime64 (NanoSecond64 ...)
253-- RelTime64 (NanoSecond64 ...)
254-- RelTime64 (NanoSecond64 ...)
256-- Note: This API is not safe on 32-bit machines.
260{-# INLINE relTimesWith #-}
261relTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
262 => Double -> t m RelTime64
263relTimesWith = fmap snd . timesWith
265------------------------------------------------------------------------------
266-- Elimination - Running a Fold
267------------------------------------------------------------------------------
269-- | Fold a stream using the supplied left 'Fold' and reducing the resulting
270-- expression strictly at each step. The behavior is similar to 'foldl''. A
271-- 'Fold' can terminate early without consuming the full stream. See the
272-- documentation of individual 'Fold's for termination behavior.
274-- >>> Stream.fold Fold.sum (Stream.enumerateFromTo 1 100)
277-- Folds never fail, therefore, they produce a default value even when no input
278-- is provided. It means we can always fold an empty stream and get a valid
279-- result. For example:
281-- >>> Stream.fold Fold.sum Stream.nil
284-- However, 'foldMany' on an empty stream results in an empty stream.
285-- Therefore, @Stream.fold f@ is not the same as @Stream.head . Stream.foldMany
288-- @fold f = Stream.parse (Parser.fromFold f)@
292fold :: Monad m => Fold m a b -> SerialT m a -> m b
294 (b, _) <- fold_ fl strm
298fold_ :: Monad m => Fold m a b -> SerialT m a -> m (b, SerialT m a)
300 (b, str) <- D.fold_ fl $ IsStream.toStreamD strm
301 return $! (b, IsStream.fromStreamD str)
303------------------------------------------------------------------------------
305------------------------------------------------------------------------------
315-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
321map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
322map f = fromStreamD . D.map f . toStreamD
324-- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
325-- that it provides an additional @done@ function to be applied on the
326-- accumulator when the stream stops. The result of @done@ is also emitted in
329-- This function can be used to allocate a resource in the beginning of the
330-- scan and release it when the stream ends or to flush the internal state of
331-- the scan at the end.
335{-# INLINE scanlMAfter' #-}
336scanlMAfter' :: (IsStream t, Monad m)
337 => (b -> a -> m b) -> m b -> (b -> m b) -> t m a -> t m b
338scanlMAfter' step initial done stream =
339 fromStreamD $ D.scanlMAfter' step initial done $ toStreamD stream
341-- XXX this needs to be concurrent
342-- | Like 'postscanl'' but with a monadic step function and a monadic seed.
344-- >>> postscanlM' f z xs = Stream.drop 1 $ Stream.scanlM' f z xs
348-- /Since: 0.8.0 (signature change)/
349{-# INLINE postscanlM' #-}
350postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
351postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
355-- The stateful step function can be simplified to @(s -> a -> m b)@ to provide
356-- a read-only environment. However, that would just be 'mapM'.
358-- The initial action could be @m (s, Maybe b)@, and we can also add a final
359-- action @s -> m (Maybe b)@. This can be used to get pre/post scan like
360-- functionality and also to flush the state in the end like scanlMAfter'.
361-- We can also use it along with a fusible version of bracket to get
362-- scanlMAfter' like functionality. See issue #677.
364-- This can be further generalized to a type similar to Fold/Parser, giving it
365-- filtering and parsing capability as well (this is in fact equivalent to
368-- smapM :: (s -> a -> m (Step s b)) -> m s -> t m a -> t m b
371-- | A stateful 'mapM', equivalent to a left scan, more like mapAccumL.
372-- Hopefully, this is a better alternative to @scan@. Separation of state from
373-- the output makes it easier to think in terms of a shared state, and also
374-- makes it easier to keep the state fully strict and the output lazy.
376-- See also: 'scanlM''
381smapM :: (IsStream t, Monad m) =>
386smapM step initial stream =
387 -- XXX implement this directly instead of using scanlM'
388 -- Once we have postscanlM' with monadic initial we can use this code
389 -- let r = postscanlM'
390 -- (\(s, _) a -> step s a)
391 -- (fmap (,undefined) initial)
394 (\(s, _) a -> step s a)
395 (fmap (,undefined) initial)
399------------------------------------------------------------------------------
400-- Transformation - Trimming
401------------------------------------------------------------------------------
403-- | Take first 'n' elements from the stream and discard the rest.
407take :: (IsStream t, Monad m) => Int -> t m a -> t m a
408take n m = fromStreamS $ S.take n $ toStreamS
409 (maxYields (Just (fromIntegral n)) m)
411-- | End the stream as soon as the predicate fails on an element.
414{-# INLINE takeWhile #-}
415takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
416takeWhile p m = fromStreamS $ S.takeWhile p $ toStreamS m
418-- | Discard first 'n' elements from the stream and take the rest.
422drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
423drop n m = fromStreamS $ S.drop n $ toStreamS m
425------------------------------------------------------------------------------
427------------------------------------------------------------------------------
429-- | Find all the indices where the element in the stream satisfies the given
432-- > findIndices = fold Fold.findIndices
435{-# INLINE findIndices #-}
436findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
437findIndices p m = fromStreamS $ S.findIndices p (toStreamS m)
439------------------------------------------------------------------------------
440-- Transformation by Inserting
441------------------------------------------------------------------------------
443-- intersperseM = intersperseBySpan 1
445-- | Insert an effect and its output before consuming an element of a stream
446-- except the first one.
448-- >>> Stream.toList $ Stream.trace putChar $ Stream.intersperseM (putChar '.' >> return ',') $ Stream.fromList "hello"
449-- h.,e.,l.,l.,o"h,e,l,l,o"
452{-# INLINE intersperseM #-}
453intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
454intersperseM m = fromStreamS . S.intersperseM m . toStreamS
456-- | Intersperse a monadic action into the input stream after every @n@
460-- > import Control.Concurrent (threadDelay)
461-- > Stream.drain $ Stream.interjectSuffix 1 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
466{-# INLINE interjectSuffix #-}
468 :: (IsStream t, MonadAsync m)
469 => Double -> m a -> t m a -> t m a
470interjectSuffix n f xs = xs `parallelFst` repeatM timed
471 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
473------------------------------------------------------------------------------
474-- Transformation by Reordering
475------------------------------------------------------------------------------
477-- XXX Use a compact region list to temporarily store the list, in both reverse
478-- as well as in reverse'.
480-- /Note:/ 'reverse'' is much faster than this, use that when performance
483-- | Returns the elements of the stream in reverse order. The stream must be
484-- finite. Note that this necessarily buffers the entire stream in memory.
486-- >>> reverse = Stream.foldlT (flip Stream.cons) Stream.nil
488-- /Since 0.7.0 (Monad m constraint)/
491{-# INLINE reverse #-}
492reverse :: (IsStream t, Monad m) => t m a -> t m a
493reverse s = fromStreamS $ S.reverse $ toStreamS s
495-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
498{-# INLINE reverse' #-}
499reverse' :: (IsStream t, MonadIO m, Storable a) => t m a -> t m a
500-- reverse' s = fromStreamD $ D.reverse' $ toStreamD s
503 . A.flattenArraysRev -- unfoldMany A.readRev
507 . A.arraysOf defaultChunkSize
510------------------------------------------------------------------------------
511-- Concurrent Transformations and Combining
512------------------------------------------------------------------------------
514-- | Make the stream producer and consumer run concurrently by introducing a
515-- buffer between them. The producer thread evaluates the input stream until
516-- the buffer fills, it terminates if the buffer is full and a worker thread is
517-- kicked off again to evaluate the remaining stream when there is space in the
518-- buffer. The consumer consumes the stream lazily from the buffer.
520-- /Since: 0.2.0 (Streamly)/
524{-# INLINE_NORMAL mkAsync #-}
525mkAsync :: (IsStream t, MonadAsync m) => t m a -> t m a
526mkAsync = fromStreamD . Async.mkAsyncD . toStreamD
528-- Compare with mkAsync. mkAsync uses an Async style SVar whereas this uses a
529-- parallel style SVar for evaluation. Currently, parallel style cannot use
530-- rate control whereas Async style can use rate control. In async style SVar
531-- the worker thread terminates when the buffer is full whereas in Parallel
534-- | Make the stream producer and consumer run concurrently by introducing a
535-- buffer between them. The producer thread evaluates the input stream until
536-- the buffer fills, it blocks if the buffer is full until there is space in
537-- the buffer. The consumer consumes the stream lazily from the buffer.
539-- @mkParallel = IsStream.fromStreamD . mkParallelD . IsStream.toStreamD@
543{-# INLINE_NORMAL mkParallel #-}
544mkParallel :: (IsStream t, MonadAsync m) => t m a -> t m a
545mkParallel = fromStreamD . Par.mkParallelD . toStreamD
547-- | Like `parallel` but stops the output as soon as the first stream stops.
550{-# INLINE parallelFst #-}
551parallelFst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
552parallelFst m1 m2 = fromStream $ Par.parallelFstK (toStream m1) (toStream m2)
554------------------------------------------------------------------------------
555-- Combine streams and flatten
556------------------------------------------------------------------------------
558-- | Map a stream producing monadic function on each element of the stream
559-- and then flatten the results into a single stream. Since the stream
560-- generation function is monadic, unlike 'concatMap', it can produce an
561-- effect at the beginning of each iteration of the inner loop.
564{-# INLINE concatMapM #-}
565concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b
566concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
568-- | Map a stream producing function on each element of the stream and then
569-- flatten the results into a single stream.
571-- >>> concatMap f = Stream.concatMapM (return . f)
572-- >>> concatMap f = Stream.concatMapWith Stream.serial f
573-- >>> concatMap f = Stream.concat . Stream.map f
574-- >>> concatMap f = Stream.unfoldMany (Unfold.lmap f Unfold.fromStream)
577{-# INLINE concatMap #-}
578concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
579concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
581-- | Given a stream value in the underlying monad, lift and join the underlying
582-- monad with the stream monad.
584-- >>> concatM = Stream.concat . Stream.fromEffect
585-- >>> concatM = Stream.concat . lift -- requires (MonadTrans t)
586-- >>> concatM = join . lift -- requires (MonadTrans t, Monad (t m))
588-- See also: 'concat', 'sequence'
592{-# INLINE concatM #-}
593concatM :: (IsStream t, Monad m) => m (t m a) -> t m a
594concatM generator = concatMapM (\() -> generator) (fromPure ())
596------------------------------------------------------------------------------
597-- Split stream and fold
598------------------------------------------------------------------------------
601-- > intercalate . splitOnSeq == id
602-- prop> (\f -> unsafePerformIO . Stream.toList . f . Stream.fromList) (Stream.intercalate Unfold.fromList " " . Stream.splitOnSeq (Array.fromList " ") Fold.toList) xs == xs
604-- | Like 'splitOn' but the separator is a sequence of elements instead of a
607-- For illustration, let's define a function that operates on pure lists:
609-- >>> splitOnSeq' pat xs = Stream.toList $ Stream.splitOnSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
611-- >>> splitOnSeq' "" "hello"
612-- ["h","e","l","l","o"]
614-- >>> splitOnSeq' "hello" ""
617-- >>> splitOnSeq' "hello" "hello"
620-- >>> splitOnSeq' "x" "hello"
623-- >>> splitOnSeq' "h" "hello"
626-- >>> splitOnSeq' "o" "hello"
629-- >>> splitOnSeq' "e" "hello"
632-- >>> splitOnSeq' "l" "hello"
635-- >>> splitOnSeq' "ll" "hello"
638-- 'splitOnSeq' is an inverse of 'intercalate'. The following law always holds:
640-- > intercalate . splitOnSeq == id
642-- The following law holds when the separator is non-empty and contains none of
643-- the elements present in the input lists:
645-- > splitOnSeq . intercalate == id
649-- XXX We can use a polymorphic vector implemented by Array# to represent the
650-- sequence, that way we can avoid the Storable constraint. If we still need
651-- Storable Array for performance, we can use a separate splitOnArray API for
652-- that. We can also have an API where the sequence itself is a lazy stream, so
653-- that we can search files in files for example.
654{-# INLINE splitOnSeq #-}
656 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
657 => Array a -> Fold m a b -> t m a -> t m b
659 IsStream.fromStreamD $ D.splitOnSeq patt f (IsStream.toStreamD m)
661------------------------------------------------------------------------------
663------------------------------------------------------------------------------
665-- | Like 'zipWith' but using a monadic zipping function.
668{-# INLINABLE zipWithM #-}
669zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
672 $ S.zipWithM f (IsStream.toStreamS m1) (IsStream.toStreamS m2)
674-- | Stream @a@ is evaluated first, followed by stream @b@, the resulting
675-- elements @a@ and @b@ are then zipped using the supplied zip function and the
676-- result @c@ is yielded to the consumer.
678-- If stream @a@ or stream @b@ ends, the zipped stream ends. If stream @b@ ends
679-- first, the element @a@ from previous evaluation of stream @a@ is discarded.
682-- > S.toList $ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])
687{-# INLINABLE zipWith #-}
688zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
691 $ S.zipWith f (IsStream.toStreamS m1) (IsStream.toStreamS 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