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
56import Control.Concurrent (threadDelay)
57import Control.Monad.IO.Class (MonadIO(..))
58import Foreign.Storable (Storable)
59import Streamly.Internal.Control.Concurrent (MonadAsync)
60import Streamly.Internal.Data.Array.Foreign.Type (Array)
61import Streamly.Internal.Data.Fold.Type (Fold (..))
62import Streamly.Internal.Data.Stream.IsStream.Combinators (maxYields)
63import Streamly.Internal.Data.Stream.Prelude (fromStreamS, toStreamS)
64import Streamly.Internal.Data.Stream.Serial (SerialT)
65import Streamly.Internal.Data.Stream.StreamD.Type (fromStreamD, toStreamD)
66import Streamly.Internal.Data.Stream.StreamK.Type (IsStream())
67import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64, addToAbsTime64)
69import qualified Streamly.Internal.Data.Stream.Parallel as Par
70import qualified Streamly.Internal.Data.Stream.Serial as Serial
71import qualified Streamly.Internal.Data.Stream.StreamK as K (repeatM)
72import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
73import qualified Streamly.Internal.Data.Stream.StreamD as D
74#ifdef USE_STREAMK_ONLY
75import qualified Streamly.Internal.Data.Stream.StreamK as S
77import qualified Streamly.Internal.Data.Stream.StreamD as S
80import Prelude hiding (take, takeWhile, drop, reverse, concatMap)
85-- >>> import Control.Concurrent (threadDelay)
86-- >>> import Control.Monad (join)
87-- >>> import Control.Monad.Trans.Class (lift)
88-- >>> import Data.Function (fix, (&))
89-- >>> import Data.Semigroup (cycle1)
90-- >>> import Prelude hiding (take, takeWhile, drop, reverse)
91-- >>> import System.IO.Unsafe (unsafePerformIO)
92-- >>> import qualified Streamly.Prelude as Stream
93-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
94-- >>> import qualified Streamly.Data.Array.Foreign as Array
95-- >>> import qualified Streamly.Data.Fold as Fold
96-- >>> import qualified Streamly.Data.Unfold as Unfold
98------------------------------------------------------------------------------
100------------------------------------------------------------------------------
102-- Faster than fromEffect because there is no bind.
106-- fromPure a = a \`cons` nil
109-- Create a singleton stream from a pure value.
111-- The following holds in monadic streams, but not in Zip streams:
115-- fromPure = fromEffect . pure
118-- In Zip applicative streams 'fromPure' is not the same as 'pure' because in that
119-- case 'pure' is equivalent to 'repeat' instead. 'fromPure' and 'pure' are
120-- equally efficient, in other cases 'fromPure' may be slightly more efficient
121-- than the other equivalent definitions.
123-- /Since: 0.8.0 (Renamed yield to fromPure)/
125{-# INLINE fromPure #-}
126fromPure :: IsStream t => a -> t m a
129-- | Same as 'fromPure'
132{-# DEPRECATED yield "Please use fromPure instead." #-}
134yield :: IsStream t => a -> t m a
139-- fromEffect m = m \`consM` nil
142-- Create a singleton stream from a monadic action.
145-- > Stream.toList $ Stream.fromEffect getLine
150-- /Since: 0.8.0 (Renamed yieldM to fromEffect)/
152{-# INLINE fromEffect #-}
153fromEffect :: (Monad m, IsStream t) => m a -> t m a
154fromEffect = K.fromEffect
156-- | Same as 'fromEffect'
159{-# DEPRECATED yieldM "Please use fromEffect instead." #-}
161yieldM :: (Monad m, IsStream t) => m a -> t m a
164-- >>> repeatM = fix . consM
165-- >>> repeatM = cycle1 . fromEffect
167-- Generate a stream by repeatedly executing a monadic action forever.
171-- Stream.repeatM (threadDelay 1000000 >> print 1)
177-- /Concurrent, infinite (do not use with 'fromParallel')/
180{-# INLINE_EARLY repeatM #-}
181repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
184{-# RULES "repeatM serial" repeatM = repeatMSerial #-}
185{-# INLINE repeatMSerial #-}
186repeatMSerial :: MonadAsync m => m a -> SerialT m a
187repeatMSerial = fromStreamS . S.repeatM
189------------------------------------------------------------------------------
190-- Generation - Time related
191------------------------------------------------------------------------------
193-- | @timesWith g@ returns a stream of time value tuples. The first component
194-- of the tuple is an absolute time reference (epoch) denoting the start of the
195-- stream and the second component is a time relative to the reference.
197-- The argument @g@ specifies the granularity of the relative time in seconds.
198-- A lower granularity clock gives higher precision but is more expensive in
199-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
201-- >>> import Control.Concurrent (threadDelay)
202-- >>> import Streamly.Internal.Data.Stream.IsStream.Common as Stream (timesWith)
203-- >>> Stream.mapM_ (\x -> print x >> threadDelay 1000000) $ Stream.take 3 $ Stream.timesWith 0.01
204-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
205-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
206-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
208-- Note: This API is not safe on 32-bit machines.
212{-# INLINE timesWith #-}
213timesWith :: (IsStream t, MonadAsync m) => Double -> t m (AbsTime, RelTime64)
214timesWith g = fromStreamD $ D.times g
216-- | @absTimesWith g@ returns a stream of absolute timestamps using a clock of
217-- granularity @g@ specified in seconds. A low granularity clock is more
218-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
221-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ absTimesWith 0.01
222-- AbsTime (TimeSpec {sec = ..., nsec = ...})
223-- AbsTime (TimeSpec {sec = ..., nsec = ...})
224-- AbsTime (TimeSpec {sec = ..., nsec = ...})
226-- Note: This API is not safe on 32-bit machines.
230{-# INLINE absTimesWith #-}
231absTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
232 => Double -> t m AbsTime
233absTimesWith = fmap (uncurry addToAbsTime64) . timesWith
235-- | @relTimesWith g@ returns a stream of relative time values starting from 0,
236-- using a clock of granularity @g@ specified in seconds. A low granularity
237-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
238-- ms is treated as 1 ms.
240-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimesWith 0.01
241-- RelTime64 (NanoSecond64 ...)
242-- RelTime64 (NanoSecond64 ...)
243-- RelTime64 (NanoSecond64 ...)
245-- Note: This API is not safe on 32-bit machines.
249{-# INLINE relTimesWith #-}
250relTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
251 => Double -> t m RelTime64
252relTimesWith = fmap snd . timesWith
254------------------------------------------------------------------------------
255-- Elimination - Running a Fold
256------------------------------------------------------------------------------
258-- | Fold a stream using the supplied left 'Fold' and reducing the resulting
259-- expression strictly at each step. The behavior is similar to 'foldl''. A
260-- 'Fold' can terminate early without consuming the full stream. See the
261-- documentation of individual 'Fold's for termination behavior.
263-- >>> Stream.fold Fold.sum (Stream.enumerateFromTo 1 100)
266-- Folds never fail, therefore, they produce a default value even when no input
267-- is provided. It means we can always fold an empty stream and get a valid
268-- result. For example:
270-- >>> Stream.fold Fold.sum Stream.nil
273-- However, 'foldMany' on an empty stream results in an empty stream.
274-- Therefore, @Stream.fold f@ is not the same as @Stream.head . Stream.foldMany
277-- @fold f = Stream.parse (Parser.fromFold f)@
281fold :: Monad m => Fold m a b -> SerialT m a -> m b
283 (b, _) <- fold_ fl strm
287fold_ :: Monad m => Fold m a b -> SerialT m a -> m (b, SerialT m a)
289 (b, str) <- D.fold_ fl $ D.toStreamD strm
290 return $! (b, D.fromStreamD str)
292------------------------------------------------------------------------------
294------------------------------------------------------------------------------
296-- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
297-- that it provides an additional @done@ function to be applied on the
298-- accumulator when the stream stops. The result of @done@ is also emitted in
301-- This function can be used to allocate a resource in the beginning of the
302-- scan and release it when the stream ends or to flush the internal state of
303-- the scan at the end.
307{-# INLINE scanlMAfter' #-}
308scanlMAfter' :: (IsStream t, Monad m)
309 => (b -> a -> m b) -> m b -> (b -> m b) -> t m a -> t m b
310scanlMAfter' step initial done stream =
311 fromStreamD $ D.scanlMAfter' step initial done $ toStreamD stream
313-- XXX this needs to be concurrent
314-- | Like 'postscanl'' but with a monadic step function and a monadic seed.
316-- >>> postscanlM' f z xs = Stream.drop 1 $ Stream.scanlM' f z xs
320-- /Since: 0.8.0 (signature change)/
321{-# INLINE postscanlM' #-}
322postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
323postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
327-- The stateful step function can be simplified to @(s -> a -> m b)@ to provide
328-- a read-only environment. However, that would just be 'mapM'.
330-- The initial action could be @m (s, Maybe b)@, and we can also add a final
331-- action @s -> m (Maybe b)@. This can be used to get pre/post scan like
332-- functionality and also to flush the state in the end like scanlMAfter'.
333-- We can also use it along with a fusible version of bracket to get
334-- scanlMAfter' like functionality. See issue #677.
336-- This can be further generalized to a type similar to Fold/Parser, giving it
337-- filtering and parsing capability as well (this is in fact equivalent to
340-- smapM :: (s -> a -> m (Step s b)) -> m s -> t m a -> t m b
343-- | A stateful 'mapM', equivalent to a left scan, more like mapAccumL.
344-- Hopefully, this is a better alternative to @scan@. Separation of state from
345-- the output makes it easier to think in terms of a shared state, and also
346-- makes it easier to keep the state fully strict and the output lazy.
348-- See also: 'scanlM''
353smapM :: (IsStream t, Monad m) =>
358smapM step initial stream =
359 -- XXX implement this directly instead of using scanlM'
360 -- Once we have postscanlM' with monadic initial we can use this code
361 -- let r = postscanlM'
362 -- (\(s, _) a -> step s a)
363 -- (fmap (,undefined) initial)
366 (\(s, _) a -> step s a)
367 (fmap (,undefined) initial)
371------------------------------------------------------------------------------
372-- Transformation - Trimming
373------------------------------------------------------------------------------
375-- | Take first 'n' elements from the stream and discard the rest.
379take :: (IsStream t, Monad m) => Int -> t m a -> t m a
380take n m = fromStreamS $ S.take n $ toStreamS
381 (maxYields (Just (fromIntegral n)) m)
383-- | End the stream as soon as the predicate fails on an element.
386{-# INLINE takeWhile #-}
387takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
388takeWhile p m = fromStreamS $ S.takeWhile p $ toStreamS m
390-- | Discard first 'n' elements from the stream and take the rest.
394drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
395drop n m = fromStreamS $ S.drop n $ toStreamS m
397------------------------------------------------------------------------------
399------------------------------------------------------------------------------
401-- | Find all the indices where the element in the stream satisfies the given
404-- > findIndices = fold Fold.findIndices
407{-# INLINE findIndices #-}
408findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
409findIndices p m = fromStreamS $ S.findIndices p (toStreamS m)
411------------------------------------------------------------------------------
412-- Transformation by Inserting
413------------------------------------------------------------------------------
415-- intersperseM = intersperseBySpan 1
417-- | Insert an effect and its output before consuming an element of a stream
418-- except the first one.
420-- >>> Stream.toList $ Stream.trace putChar $ Stream.intersperseM (putChar '.' >> return ',') $ Stream.fromList "hello"
421-- h.,e.,l.,l.,o"h,e,l,l,o"
424{-# INLINE intersperseM #-}
425intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
426intersperseM m = fromStreamS . S.intersperseM m . toStreamS
428-- | Intersperse a monadic action into the input stream after every @n@
432-- > import Control.Concurrent (threadDelay)
433-- > Stream.drain $ Stream.interjectSuffix 1 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
438{-# INLINE interjectSuffix #-}
440 :: (IsStream t, MonadAsync m)
441 => Double -> m a -> t m a -> t m a
442interjectSuffix n f xs = xs `Par.parallelFst` repeatM timed
443 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
445------------------------------------------------------------------------------
446-- Transformation by Reordering
447------------------------------------------------------------------------------
449-- XXX Use a compact region list to temporarily store the list, in both reverse
450-- as well as in reverse'.
452-- /Note:/ 'reverse'' is much faster than this, use that when performance
455-- | Returns the elements of the stream in reverse order. The stream must be
456-- finite. Note that this necessarily buffers the entire stream in memory.
458-- >>> reverse = Stream.foldlT (flip Stream.cons) Stream.nil
460-- /Since 0.7.0 (Monad m constraint)/
463{-# INLINE reverse #-}
464reverse :: (IsStream t, Monad m) => t m a -> t m a
465reverse s = fromStreamS $ S.reverse $ toStreamS s
467-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
470{-# INLINE reverse' #-}
471reverse' :: (IsStream t, MonadIO m, Storable a) => t m a -> t m a
472reverse' s = fromStreamD $ D.reverse' $ toStreamD s
474------------------------------------------------------------------------------
475-- Combine streams and flatten
476------------------------------------------------------------------------------
478-- | Map a stream producing monadic function on each element of the stream
479-- and then flatten the results into a single stream. Since the stream
480-- generation function is monadic, unlike 'concatMap', it can produce an
481-- effect at the beginning of each iteration of the inner loop.
484{-# INLINE concatMapM #-}
485concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b
486concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
488-- | Map a stream producing function on each element of the stream and then
489-- flatten the results into a single stream.
491-- >>> concatMap f = Stream.concatMapM (return . f)
492-- >>> concatMap f = Stream.concatMapWith Stream.serial f
493-- >>> concatMap f = Stream.concat . Stream.map f
494-- >>> concatMap f = Stream.unfoldMany (Unfold.lmap f Unfold.fromStream)
497{-# INLINE concatMap #-}
498concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
499concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
501-- | Given a stream value in the underlying monad, lift and join the underlying
502-- monad with the stream monad.
504-- >>> concatM = Stream.concat . Stream.fromEffect
505-- >>> concatM = Stream.concat . lift -- requires (MonadTrans t)
506-- >>> concatM = join . lift -- requires (MonadTrans t, Monad (t m))
508-- See also: 'concat', 'sequence'
512{-# INLINE concatM #-}
513concatM :: (IsStream t, Monad m) => m (t m a) -> t m a
514concatM generator = concatMapM (\() -> generator) (fromPure ())
516------------------------------------------------------------------------------
517-- Split stream and fold
518------------------------------------------------------------------------------
521-- > intercalate . splitOnSeq == id
522-- prop> (\f -> unsafePerformIO . Stream.toList . f . Stream.fromList) (Stream.intercalate Unfold.fromList " " . Stream.splitOnSeq (Array.fromList " ") Fold.toList) xs == xs
524-- | Like 'splitOn' but the separator is a sequence of elements instead of a
527-- For illustration, let's define a function that operates on pure lists:
529-- >>> splitOnSeq' pat xs = Stream.toList $ Stream.splitOnSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
531-- >>> splitOnSeq' "" "hello"
532-- ["h","e","l","l","o"]
534-- >>> splitOnSeq' "hello" ""
537-- >>> splitOnSeq' "hello" "hello"
540-- >>> splitOnSeq' "x" "hello"
543-- >>> splitOnSeq' "h" "hello"
546-- >>> splitOnSeq' "o" "hello"
549-- >>> splitOnSeq' "e" "hello"
552-- >>> splitOnSeq' "l" "hello"
555-- >>> splitOnSeq' "ll" "hello"
558-- 'splitOnSeq' is an inverse of 'intercalate'. The following law always holds:
560-- > intercalate . splitOnSeq == id
562-- The following law holds when the separator is non-empty and contains none of
563-- the elements present in the input lists:
565-- > splitOnSeq . intercalate == id
569-- XXX We can use a polymorphic vector implemented by Array# to represent the
570-- sequence, that way we can avoid the Storable constraint. If we still need
571-- Storable Array for performance, we can use a separate splitOnArray API for
572-- that. We can also have an API where the sequence itself is a lazy stream, so
573-- that we can search files in files for example.
574{-# INLINE splitOnSeq #-}
576 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
577 => Array a -> Fold m a b -> t m a -> t m b
578splitOnSeq patt f m = D.fromStreamD $ D.splitOnSeq patt f (D.toStreamD m)
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