326b055 · streamly · git.iko.soy

git.iko.soy/streamly

Copy combinators from IsStream, absent in the new Stream modules

commit326b05570e2c1f0c5b8ea41fc7428c7a1e76f3db authorAdithya Kumar <adithya@composewell.com> dateFri 21 Oct 2022, 00:06 (4 years ago) parent0fb7970

5 files changed+348−6

compare with master

unifiedsplit

core/src/Streamly/Internal/Data/Stream/Eliminate.hs+11−1 core/src/Streamly/Internal/Data/Stream/Expand.hs+77−3 core/src/Streamly/Internal/Data/Stream/Reduce.hs+113 core/src/Streamly/Internal/Data/Stream/Transform.hs+57−2 src/Streamly/Internal/Data/Stream/Time.hs+90

core/src/Streamly/Internal/Data/Stream/Eliminate.hs+11−1

@@ -40,6 +40,7 @@ module Streamly.Internal.Data.Stream.Eliminate

4040 -- the stream one element at a time, and we have the remaining stream all

4141 -- the time.

4242 , uncons

43+ , init

4344

4445 -- * Right Folds

4546 , foldrM

@@ -87,7 +88,7 @@ import qualified Streamly.Internal.Data.Stream.StreamK as K

8788import Streamly.Internal.Data.Stream.Bottom

8889import Streamly.Internal.Data.Stream.Type

8990

90−import Prelude hiding (foldr, reverse)

91+import Prelude hiding (foldr, init, reverse)

9192

9293-- $setup

9394-- >>> :m

@@ -132,6 +133,15 @@ import Prelude hiding (foldr, reverse)

132133uncons :: Monad m => Stream m a -> m (Maybe (a, Stream m a))

133134uncons m = fmap (fmap (fmap fromStreamK)) $ K.uncons (toStreamK m)

134135

136+-- | Extract all but the last element of the stream, if any.

137+--

138+-- Note: This will end up buffering the entire stream.

139+--

140+-- /Pre-release/

141+{-# INLINE init #-}

142+init :: Monad m => Stream m a -> m (Maybe (Stream m a))

143+init m = fmap (fmap fromStreamK) $ K.init $ toStreamK m

144+

135145------------------------------------------------------------------------------

136146-- Right Folds

137147------------------------------------------------------------------------------

core/src/Streamly/Internal/Data/Stream/Expand.hs+77−3

@@ -41,6 +41,11 @@ module Streamly.Internal.Data.Stream.Expand

4141 , interleave

4242 , interleaveFst

4343 , interleaveMin

44+ , interleaveFstSuffix2

45+ , interleaveFst2

46+

47+ -- ** Round Robin

48+ , roundrobin

4449

4550 -- ** Merge

4651 , mergeBy

@@ -140,6 +145,7 @@ import Prelude hiding (concat, concatMap, zipWith)

140145-- >>> :m

141146-- >>> import Data.Either (either)

142147-- >>> import Data.IORef

148+-- >>> import Streamly.Internal.Data.Stream (Stream)

143149-- >>> import Prelude hiding (zipWith, concatMap, concat)

144150-- >>> import qualified Streamly.Data.Array.Unboxed as Array

145151-- >>> import qualified Streamly.Internal.Data.Fold as Fold

@@ -227,6 +233,74 @@ interleaveMin :: Stream m a -> Stream m a -> Stream m a

227233interleaveMin s1 s2 =

228234 fromStreamK $ K.interleaveMin (toStreamK s1) (toStreamK s2)

229235

236+-- | Interleaves the outputs of two streams, yielding elements from each stream

237+-- alternately, starting from the first stream. As soon as the first stream

238+-- finishes, the output stops, discarding the remaining part of the second

239+-- stream. In this case, the last element in the resulting stream would be from

240+-- the second stream. If the second stream finishes early then the first stream

241+-- still continues to yield elements until it finishes.

242+--

243+-- >>> :set -XOverloadedStrings

244+-- >>> import Data.Functor.Identity (Identity)

245+-- >>> Stream.interleaveFstSuffix2 "abc" ",,,," :: Stream Identity Char

246+-- fromList "a,b,c,"

247+-- >>> Stream.interleaveFstSuffix2 "abc" "," :: Stream Identity Char

248+-- fromList "a,bc"

249+--

250+-- 'interleaveFstSuffix2' is a dual of 'interleaveFst2'.

251+--

252+-- Do not use at scale in concatMapWith.

253+--

254+-- /Pre-release/

255+{-# INLINE interleaveFstSuffix2 #-}

256+interleaveFstSuffix2 :: Monad m => Stream m b -> Stream m b -> Stream m b

257+interleaveFstSuffix2 m1 m2 =

258+ fromStreamD $ D.interleaveSuffix (toStreamD m1) (toStreamD m2)

259+

260+-- | Interleaves the outputs of two streams, yielding elements from each stream

261+-- alternately, starting from the first stream and ending at the first stream.

262+-- If the second stream is longer than the first, elements from the second

263+-- stream are infixed with elements from the first stream. If the first stream

264+-- is longer then it continues yielding elements even after the second stream

265+-- has finished.

266+--

267+-- >>> :set -XOverloadedStrings

268+-- >>> import Data.Functor.Identity (Identity)

269+-- >>> Stream.interleaveFst2 "abc" ",,,," :: Stream Identity Char

270+-- fromList "a,b,c"

271+-- >>> Stream.interleaveFst2 "abc" "," :: Stream Identity Char

272+-- fromList "a,bc"

273+--

274+-- 'interleaveFst2' is a dual of 'interleaveFstSuffix2'.

275+--

276+-- Do not use at scale in concatMapWith.

277+--

278+-- /Pre-release/

279+{-# INLINE interleaveFst2 #-}

280+interleaveFst2 :: Monad m => Stream m b -> Stream m b -> Stream m b

281+interleaveFst2 m1 m2 =

282+ fromStreamD $ D.interleaveInfix (toStreamD m1) (toStreamD m2)

283+

284+------------------------------------------------------------------------------

285+-- Scheduling

286+------------------------------------------------------------------------------

287+

288+-- | Schedule the execution of two streams in a fair round-robin manner,

289+-- executing each stream once, alternately. Execution of a stream may not

290+-- necessarily result in an output, a stream may chose to @Skip@ producing an

291+-- element until later giving the other stream a chance to run. Therefore, this

292+-- combinator fairly interleaves the execution of two streams rather than

293+-- fairly interleaving the output of the two streams. This can be useful in

294+-- co-operative multitasking without using explicit threads. This can be used

295+-- as an alternative to `async`.

296+--

297+-- Do not use at scale in concatMapWith.

298+--

299+-- /Pre-release/

300+{-# INLINE roundrobin #-}

301+roundrobin :: Monad m => Stream m b -> Stream m b -> Stream m b

302+roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)

303+

230304------------------------------------------------------------------------------

231305-- Merging (sorted streams)

232306------------------------------------------------------------------------------

@@ -361,7 +435,7 @@ unfoldInterleave u m =

361435{-# INLINE unfoldRoundRobin #-}

362436unfoldRoundRobin ::Monad m => Unfold m a b -> Stream m a -> Stream m b

363437unfoldRoundRobin u m =

364− fromStreamD $ D.unfoldManyInterleave u (toStreamD m)

438+ fromStreamD $ D.unfoldManyRoundRobin u (toStreamD m)

365439

366440------------------------------------------------------------------------------

367441-- Combine N Streams - interpose

@@ -407,7 +481,7 @@ interposeSuffix x unf str =

407481-- > unfoldMany unf str =

408482-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())

409483

410−-- | 'interleaveInfix' followed by unfold and concat.

484+-- | 'interleaveFst' followed by unfold and concat.

411485--

412486-- /Pre-release/

413487{-# INLINE gintercalate #-}

@@ -438,7 +512,7 @@ intercalate :: Monad m

438512intercalate unf seed str = fromStreamD $

439513 D.unfoldMany unf $ D.intersperse seed (toStreamD str)

440514

441−-- | 'interleaveSuffix' followed by unfold and concat.

515+-- | 'interleaveFstSuffix2' followed by unfold and concat.

442516--

443517-- /Pre-release/

444518{-# INLINE gintercalateSuffix #-}

core/src/Streamly/Internal/Data/Stream/Reduce.hs+113

@@ -38,6 +38,12 @@ module Streamly.Internal.Data.Stream.Reduce

3838 -- | Element unaware grouping.

3939 , arraysOf

4040

41+ -- ** Splitting

42+ -- XXX Implement these as folds or parsers instead.

43+ , splitOnSuffixSeqAny

44+ , splitOnPrefix

45+ , splitOnAny

46+

4147 -- * Reduce By Parsers

4248 -- ** Generic Parsing

4349 -- | Apply parsers on a stream.

@@ -204,6 +210,113 @@ refoldIterateM :: Monad m =>

204210 Refold m b a b -> m b -> Stream m a -> Stream m b

205211refoldIterateM c i m = fromStreamD $ D.refoldIterateM c i (toStreamD m)

206212

213+------------------------------------------------------------------------------

214+-- Splitting

215+------------------------------------------------------------------------------

216+

217+-- Implement this as a fold or a parser instead.

218+-- This can be implemented easily using Rabin Karp

219+-- | Split post any one of the given patterns.

220+--

221+-- /Unimplemented/

222+{-# INLINE splitOnSuffixSeqAny #-}

223+splitOnSuffixSeqAny :: -- (Monad m, Unboxed a, Integral a) =>

224+ [Array a] -> Fold m a b -> Stream m a -> Stream m b

225+splitOnSuffixSeqAny _subseq _f _m = undefined

226+ -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)

227+

228+-- | Split on a prefixed separator element, dropping the separator. The

229+-- supplied 'Fold' is applied on the split segments.

230+--

231+-- @

232+-- > splitOnPrefix' p xs = Stream.toList $ Stream.splitOnPrefix p (Fold.toList) (Stream.fromList xs)

233+-- > splitOnPrefix' (== '.') ".a.b"

234+-- ["a","b"]

235+-- @

236+--

237+-- An empty stream results in an empty output stream:

238+-- @

239+-- > splitOnPrefix' (== '.') ""

240+-- []

241+-- @

242+--

243+-- An empty segment consisting of only a prefix is folded to the default output

244+-- of the fold:

245+--

246+-- @

247+-- > splitOnPrefix' (== '.') "."

248+-- [""]

249+--

250+-- > splitOnPrefix' (== '.') ".a.b."

251+-- ["a","b",""]

252+--

253+-- > splitOnPrefix' (== '.') ".a..b"

254+-- ["a","","b"]

255+--

256+-- @

257+--

258+-- A prefix is optional at the beginning of the stream:

259+--

260+-- @

261+-- > splitOnPrefix' (== '.') "a"

262+-- ["a"]

263+--

264+-- > splitOnPrefix' (== '.') "a.b"

265+-- ["a","b"]

266+-- @

267+--

268+-- 'splitOnPrefix' is an inverse of 'intercalatePrefix' with a single element:

269+--

270+-- > Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnPrefix (== '.') Fold.toList === id

271+--

272+-- Assuming the input stream does not contain the separator:

273+--

274+-- > Stream.splitOnPrefix (== '.') Fold.toList . Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList === id

275+--

276+-- /Unimplemented/

277+{-# INLINE splitOnPrefix #-}

278+splitOnPrefix :: -- (IsStream t, MonadCatch m) =>

279+ (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b

280+splitOnPrefix _predicate _f = undefined

281+ -- parseMany (Parser.sliceBeginBy predicate f)

282+

283+-- Int list examples for splitOn:

284+--

285+-- >>> splitList [] [1,2,3,3,4]

286+-- > [[1],[2],[3],[3],[4]]

287+--

288+-- >>> splitList [5] [1,2,3,3,4]

289+-- > [[1,2,3,3,4]]

290+--

291+-- >>> splitList [1] [1,2,3,3,4]

292+-- > [[],[2,3,3,4]]

293+--

294+-- >>> splitList [4] [1,2,3,3,4]

295+-- > [[1,2,3,3],[]]

296+--

297+-- >>> splitList [2] [1,2,3,3,4]

298+-- > [[1],[3,3,4]]

299+--

300+-- >>> splitList [3] [1,2,3,3,4]

301+-- > [[1,2],[],[4]]

302+--

303+-- >>> splitList [3,3] [1,2,3,3,4]

304+-- > [[1,2],[4]]

305+--

306+-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]

307+-- > [[],[]]

308+

309+-- This can be implemented easily using Rabin Karp

310+-- | Split on any one of the given patterns.

311+--

312+-- /Unimplemented/

313+--

314+{-# INLINE splitOnAny #-}

315+splitOnAny :: -- (Monad m, Unboxed a, Integral a) =>

316+ [Array a] -> Fold m a b -> Stream m a -> Stream m b

317+splitOnAny _subseq _f _m =

318+ undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)

319+

207320------------------------------------------------------------------------------

208321-- Parsing

209322------------------------------------------------------------------------------

core/src/Streamly/Internal/Data/Stream/Transform.hs+57−2

@@ -63,9 +63,14 @@ module Streamly.Internal.Data.Stream.Transform

6363 , take

6464 , takeWhile

6565 , takeWhileM

66+ , takeWhileLast

67+ , takeWhileAround

6668 , drop

69+ , dropLast

6770 , dropWhile

6871 , dropWhileM

72+ , dropWhileLast

73+ , dropWhileAround

6974

7075 -- * Position Indexing

7176 , indexed

@@ -512,8 +517,6 @@ filterM p m = fromStreamD $ D.filterM p $ toStreamD m

512517--

513518-- Space: @O(1)@

514519--

515−-- See also: 'nubBy'.

516−--

517520-- /Pre-release/

518521--

519522{-# INLINE uniqBy #-}

@@ -598,6 +601,27 @@ takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a

598601-- takeWhileM p = scanMaybe (FL.takingEndByM_ (\x -> not <$> p x))

599602takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m

600603

604+-- | Take all consecutive elements at the end of the stream for which the

605+-- predicate is true.

606+--

607+-- O(n) space, where n is the number elements taken.

608+--

609+-- /Unimplemented/

610+{-# INLINE takeWhileLast #-}

611+takeWhileLast :: -- Monad m =>

612+ (a -> Bool) -> Stream m a -> Stream m a

613+takeWhileLast = undefined -- fromStreamD $ D.takeWhileLast n $ toStreamD m

614+

615+-- | Like 'takeWhile' and 'takeWhileLast' combined.

616+--

617+-- O(n) space, where n is the number elements taken from the end.

618+--

619+-- /Unimplemented/

620+{-# INLINE takeWhileAround #-}

621+takeWhileAround :: -- Monad m =>

622+ (a -> Bool) -> Stream m a -> Stream m a

623+takeWhileAround = undefined -- fromStreamD $ D.takeWhileAround n $ toStreamD m

624+

601625-- | Drop elements in the stream as long as the predicate succeeds and then

602626-- take the rest of the stream.

603627--

@@ -615,6 +639,37 @@ dropWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a

615639-- dropWhileM p = scanMaybe (FL.droppingWhileM p)

616640dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m

617641

642+-- | Drop @n@ elements at the end of the stream.

643+--

644+-- O(n) space, where n is the number elements dropped.

645+--

646+-- /Unimplemented/

647+{-# INLINE dropLast #-}

648+dropLast :: -- Monad m =>

649+ Int -> Stream m a -> Stream m a

650+dropLast = undefined -- fromStreamD $ D.dropLast n $ toStreamD m

651+

652+-- | Drop all consecutive elements at the end of the stream for which the

653+-- predicate is true.

654+--

655+-- O(n) space, where n is the number elements dropped.

656+--

657+-- /Unimplemented/

658+{-# INLINE dropWhileLast #-}

659+dropWhileLast :: -- Monad m =>

660+ (a -> Bool) -> Stream m a -> Stream m a

661+dropWhileLast = undefined -- fromStreamD $ D.dropWhileLast n $ toStreamD m

662+

663+-- | Like 'dropWhile' and 'dropWhileLast' combined.

664+--

665+-- O(n) space, where n is the number elements dropped from the end.

666+--

667+-- /Unimplemented/

668+{-# INLINE dropWhileAround #-}

669+dropWhileAround :: -- Monad m =>

670+ (a -> Bool) -> Stream m a -> Stream m a

671+dropWhileAround = undefined -- fromStreamD $ D.dropWhileAround n $ toStreamD m

672+

618673------------------------------------------------------------------------------

619674-- Inserting Elements

620675------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/Time.hs+90

@@ -11,8 +11,11 @@ module Streamly.Internal.Data.Stream.Time

1111 (

1212 -- Primitives

1313 interjectSuffix

14+ , ticks

1415 , takeInterval

16+ , takeLastInterval

1517 , dropInterval

18+ , dropLastInterval

1619 , intervalsOf

1720 , chunksOfTimeout

1821

@@ -22,6 +25,15 @@ module Streamly.Internal.Data.Stream.Time

2225 , classifySessionsOf

2326 , classifyKeepAliveSessions

2427

28+ -- ** Buffering and Sampling

29+ -- | Evaluate strictly using a buffer of results. When the buffer becomes

30+ -- full we can block, drop the new elements, drop the oldest element and

31+ -- insert the new at the end or keep dropping elements uniformly to match

32+ -- the rate of the consumer.

33+ , sampleOld

34+ , sampleNew

35+ , sampleRate

36+

2537 -- Sampling

2638 , sampleIntervalEnd

2739 , sampleIntervalStart

@@ -108,6 +120,19 @@ interjectSuffix n f xs = parallelFst [xs, repeatM timed]

108120 timed = liftIO (threadDelay (round $ n * 1000000)) >> f

109121 repeatM = Stream.sequence . Stream.repeat

110122

123+-- | Generate ticks at the specified rate. The rate is adaptive, the tick

124+-- generation speed can be increased or decreased at different times to achieve

125+-- the specified rate. The specific behavior for different styles of 'Rate'

126+-- specifications is documented under 'Rate'. The effective maximum rate

127+-- achieved by a stream is governed by the processor speed.

128+--

129+-- /Unimplemented/

130+--

131+{-# INLINE ticks #-}

132+ticks :: -- (MonadAsync m) =>

133+ Rate -> Stream m ()

134+ticks = undefined

135+

111136-- XXX Notes from D.takeByTime (which was removed)

112137-- XXX using getTime in the loop can be pretty expensive especially for

113138-- computations where iterations are lightweight. We have the following

@@ -146,6 +171,16 @@ takeInterval d =

146171 . Stream.takeWhile isNothing

147172 . interjectSuffix d (return Nothing) . fmap Just

148173

174+-- | Take time interval @i@ seconds at the end of the stream.

175+--

176+-- O(n) space, where n is the number elements taken.

177+--

178+-- /Unimplemented/

179+{-# INLINE takeLastInterval #-}

180+takeLastInterval :: -- MonadAsync m =>

181+ Double -> Stream m a -> Stream m a

182+takeLastInterval = undefined

183+

149184-- | @dropInterval duration@ drops stream elements until specified @duration@ in

150185-- seconds has passed. The duration begins when the stream is evaluated for the

151186-- first time. The time duration is checked /after/ generating a stream element,

@@ -167,6 +202,16 @@ dropInterval d =

167202 . Stream.dropWhile isNothing

168203 . interjectSuffix d (return Nothing) . fmap Just

169204

205+-- | Drop time interval @i@ seconds at the end of the stream.

206+--

207+-- O(n) space, where n is the number elements dropped.

208+--

209+-- /Unimplemented/

210+{-# INLINE dropLastInterval #-}

211+dropLastInterval :: -- MonadAsync m =>

212+ Int -> Stream m a -> Stream m a

213+dropLastInterval = undefined

214+

170215-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.

171216-- XXX add this example after fixing the serial stream rate control

172217--

@@ -860,3 +905,48 @@ sampleBurstEnd = sampleBurst True

860905{-# INLINE sampleBurstStart #-}

861906sampleBurstStart :: MonadAsync m => Double -> Stream m a -> Stream m a

862907sampleBurstStart = sampleBurst False

908+

909+------------------------------------------------------------------------------

910+-- Lossy Buffering

911+------------------------------------------------------------------------------

912+

913+-- XXX We could use 'maxBuffer Block/Drop/Rotate/Sample' instead. However we

914+-- may want to have the evaluation rate independent of the sampling rate. To

915+-- support that we can decouple evaluation and sampling in independent stages.

916+-- The sampling stage would strictly evaluate and sample, the evaluation stage

917+-- would control the evaluation rate.

918+

919+-- | Evaluate the input stream continuously and keep only the oldest @n@

920+-- elements in the buffer, discard the new ones when the buffer is full. When

921+-- the output stream is evaluated it consumes the values from the buffer in a

922+-- FIFO manner.

923+--

924+-- /Unimplemented/

925+--

926+{-# INLINE sampleOld #-}

927+sampleOld :: -- MonadAsync m =>

928+ Int -> Stream m a -> Stream m a

929+sampleOld = undefined

930+

931+-- | Evaluate the input stream continuously and keep only the latest @n@

932+-- elements in a ring buffer, keep discarding the older ones to make space for

933+-- the new ones. When the output stream is evaluated it consumes the values

934+-- from the buffer in a FIFO manner.

935+--

936+-- /Unimplemented/

937+--

938+{-# INLINE sampleNew #-}

939+sampleNew :: -- MonadAsync m =>

940+ Int -> Stream m a -> Stream m a

941+sampleNew = undefined

942+

943+-- | Like 'sampleNew' but samples at uniform intervals to match the consumer

944+-- rate. Note that 'sampleNew' leads to non-uniform sampling depending on the

945+-- consumer pattern.

946+--

947+-- /Unimplemented/

948+--

949+{-# INLINE sampleRate #-}

950+sampleRate :: -- MonadAsync m =>

951+ Double -> Stream m a -> Stream m a

952+sampleRate = undefined