a5a9419 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Implement stream split ops in terms of folds

a5a9419Harendra Kumar4 years ago

src/Streamly/Internal/Data/Stream/IsStream/Common.hs+29

compare with master

unifiedsplit

@@ -32,6 +32,7 @@ module Streamly.Internal.Data.Stream.IsStream.Common

32 , postscanlMAfter'

33 , postscanlM'

34 , smapM

35+ , foldManyPost

35 -- $smapM_Notes

36 -- $smapM_Notes

36 , take

37 , take

37 , takeWhile

38 , takeWhile

@@ -641,6 +642,27 @@ concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)

641concatM :: (IsStream t, Monad m) => m (t m a) -> t m a

642concatM :: (IsStream t, Monad m) => m (t m a) -> t m a

642concatM generator = concatMapM (\() -> generator) (fromPure ())

643concatM generator = concatMapM (\() -> generator) (fromPure ())

643

644

645+-- | Like 'foldMany' but appends empty fold output if the fold and stream

646+-- termination aligns:

647+--

648+-- >>> f = Fold.take 2 Fold.sum

649+-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList []

650+-- [0]

651+-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..9]

652+-- [3,7,11,15,9]

653+-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..10]

654+-- [3,7,11,15,19,0]

655+--

656+-- /Pre-release/

657+--

658+{-# INLINE foldManyPost #-}

659+foldManyPost

660+ :: (IsStream t, Monad m)

661+ => Fold m a b

662+ -> t m a

663+ -> t m b

664+foldManyPost f m = fromStreamD $ D.foldManyPost f (toStreamD m)

665+

644------------------------------------------------------------------------------

666------------------------------------------------------------------------------

645-- Split stream and fold

667-- Split stream and fold

646------------------------------------------------------------------------------

668------------------------------------------------------------------------------

@@ -692,6 +714,8 @@ concatM generator = concatMapM (\() -> generator) (fromPure ())

692--

714--

693-- > splitOnSeq . intercalate == id

715-- > splitOnSeq . intercalate == id

694--

716--

717+-- >>> splitOnSeq pat f = Stream.foldManyPost (Fold.takeEndBySeq_ pat f)

718+--

695-- /Pre-release/

719-- /Pre-release/

696

720

697-- XXX We can use a polymorphic vector implemented by Array# to represent the

721-- XXX We can use a polymorphic vector implemented by Array# to represent the

@@ -705,6 +729,11 @@ splitOnSeq

705 => Array a -> Fold m a b -> t m a -> t m b

729 => Array a -> Fold m a b -> t m a -> t m b

706splitOnSeq patt f m =

730splitOnSeq patt f m =

707 IsStream.fromStreamD $ D.splitOnSeq patt f (IsStream.toStreamD m)

731 IsStream.fromStreamD $ D.splitOnSeq patt f (IsStream.toStreamD m)

732+-- XXX This may have a problem if the stream terminates and we start extracting

733+-- the fold and then the fold terminates before consuming all the buffered

734+-- input. We have no way to supply the buffered input back to the driver.

735+-- splitOnSeq patt f =

736+ -- foldManyPost (Fold.takeEndBySeq_ patt f)

708

737

709------------------------------------------------------------------------------

738------------------------------------------------------------------------------

710-- Zipping

739-- Zipping

src/Streamly/Internal/Data/Stream/IsStream/Reduce.hs+12−21

@@ -170,6 +170,7 @@ import Streamly.Internal.Data.Stream.IsStream.Common

170 , intersperseM

171 , map

172 , scanlMAfter'

173+ , foldManyPost

173 , splitOnSeq

174 , splitOnSeq

174 , fromPure)

175 , fromPure)

175import Streamly.Internal.Data.Stream.IsStream.Type

176import Streamly.Internal.Data.Stream.IsStream.Type

@@ -251,27 +252,6 @@ dropSuffix = error "Not implemented yet!"

251--

252--

252-- XXX We need takeGE/takeBetween to implement "some" using "many".

253-- XXX We need takeGE/takeBetween to implement "some" using "many".

253

254

254−-- | Like 'foldMany' but appends empty fold output if the fold and stream

255−-- termination aligns:

256−--

257−-- >>> f = Fold.take 2 Fold.sum

258−-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList []

259−-- [0]

260−-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..9]

261−-- [3,7,11,15,9]

262−-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..10]

263−-- [3,7,11,15,19,0]

264−--

265−-- /Pre-release/

266−--

267−{-# INLINE foldManyPost #-}

268−foldManyPost

269− :: (IsStream t, Monad m)

270− => Fold m a b

271− -> t m a

272− -> t m b

273−foldManyPost f m = fromStreamD $ D.foldManyPost f (toStreamD m)

274−

275-- | Apply a 'Fold' repeatedly on a stream and emit the fold outputs in the

255-- | Apply a 'Fold' repeatedly on a stream and emit the fold outputs in the

276-- output stream.

256-- output stream.

277--

257--

@@ -889,6 +869,8 @@ splitBySeq patt f m =

889--

869--

890-- > splitSuffixOn . intercalateSuffix == id

870-- > splitSuffixOn . intercalateSuffix == id

891--

871--

872+-- >>> splitOnSuffixSeq pat f = Stream.foldMany (Fold.takeEndBySeq_ pat f)

873+--

892-- /Pre-release/

874-- /Pre-release/

893{-# INLINE splitOnSuffixSeq #-}

875{-# INLINE splitOnSuffixSeq #-}

894splitOnSuffixSeq

876splitOnSuffixSeq

@@ -896,6 +878,11 @@ splitOnSuffixSeq

896 => Array a -> Fold m a b -> t m a -> t m b

878 => Array a -> Fold m a b -> t m a -> t m b

897splitOnSuffixSeq patt f m =

879splitOnSuffixSeq patt f m =

898 fromStreamD $ D.splitOnSuffixSeq False patt f (toStreamD m)

880 fromStreamD $ D.splitOnSuffixSeq False patt f (toStreamD m)

881+-- XXX This may have a problem if the stream terminates and we start extracting

882+-- the fold and then the fold terminates before consuming all the buffered

883+-- input. We have no way to supply the buffered input back to the driver.

884+-- splitOnSuffixSeq patt f =

885+-- foldMany (FL.takeEndBySeq_ patt f)

899

886

900-- | Like 'splitOn' but drops any empty splits.

887-- | Like 'splitOn' but drops any empty splits.

901--

888--

@@ -934,6 +921,8 @@ wordsOn _subseq _f _m =

934-- >>> splitWithSuffixSeq' "." "a..b.."

921-- >>> splitWithSuffixSeq' "." "a..b.."

935-- ["a.",".","b.","."]

922-- ["a.",".","b.","."]

936--

923--

924+-- >>> splitWithSuffixSeq pat f = Stream.foldMany (Fold.takeEndBySeq pat f)

925+--

937-- /Pre-release/

926-- /Pre-release/

938{-# INLINE splitWithSuffixSeq #-}

927{-# INLINE splitWithSuffixSeq #-}

939splitWithSuffixSeq

928splitWithSuffixSeq

@@ -941,6 +930,8 @@ splitWithSuffixSeq

941 => Array a -> Fold m a b -> t m a -> t m b

930 => Array a -> Fold m a b -> t m a -> t m b

942splitWithSuffixSeq patt f m =

931splitWithSuffixSeq patt f m =

943 fromStreamD $ D.splitOnSuffixSeq True patt f (toStreamD m)

932 fromStreamD $ D.splitOnSuffixSeq True patt f (toStreamD m)

933+-- splitWithSuffixSeq patt f =

934+-- foldMany (FL.takeEndBySeq patt f)

944

935

945-- This can be implemented easily using Rabin Karp

936-- This can be implemented easily using Rabin Karp

946-- | Split post any one of the given patterns.

937-- | Split post any one of the given patterns.