10b1ac1 · streamly · git.iko.soy

git.iko.soy/streamly

Use Either result in parse ops in Chunked stream/Producer (#2145)

Co-authored-by: Ranjeet Kumar Ranjan <ranjeetr@composewell.com>

commit10b1ac12a9567cd6bba11ff53a00fd14a261daba authorHarendra Kumar <harendra@composewell.com> dateTue 6 Dec 2022, 16:52 (4 years ago) parent3f82f00

5 files changed+158−92

compare with master

unifiedsplit

core/src/Streamly/Internal/Data/Fold/Chunked.hs+1−1 core/src/Streamly/Internal/Data/Producer/Source.hs+30−24 core/src/Streamly/Internal/Data/Stream/Chunked.hs+112−59 test/Streamly/Test/Data/Array/Stream.hs+14−8 test/Streamly/Test/Data/Parser/ParserD.hs+1

core/src/Streamly/Internal/Data/Fold/Chunked.hs+1−1

@@ -26,7 +26,7 @@

2626-- >>> f = ChunkFold.fromFold (Fold.take 7 Fold.toList)

2727-- >>> s = Stream.arraysOf 5 $ Stream.fromList "hello world"

2828-- >>> ArrayStream.runArrayFold f s

29−-- "hello w"

29+-- Right "hello w"

3030--

3131module Streamly.Internal.Data.Fold.Chunked

3232 (

core/src/Streamly/Internal/Data/Producer/Source.hs+30−24

@@ -34,7 +34,6 @@ where

3434#include "inline.hs"

3535

3636import Control.Exception (assert)

37−import Control.Monad.Catch (MonadThrow, throwM)

3837import GHC.Exts (SpecConstrAnnotation(..))

3938import GHC.Types (SPEC(..))

4039import Streamly.Internal.Data.Parser.ParserD (ParseError(..), Step(..))

@@ -112,11 +111,11 @@ newtype List a = List {getList :: [a]}

112111

113112{-# INLINE_NORMAL parseD #-}

114113parseD

115− :: MonadThrow m

116− => ParserD.Parser a m b

114+ :: Monad m =>

115+ ParserD.Parser a m b

117116 -> Producer m (Source s a) a

118117 -> Source s a

119− -> m (b, Source s a)

118+ -> m (Either ParseError b, Source s a)

120119parseD

121120 (ParserD.Parser pstep initial extract)

122121 (Producer ustep uinject uextract)

@@ -127,8 +126,8 @@ parseD

127126 ParserD.IPartial s -> do

128127 state <- uinject seed

129128 go SPEC state (List []) s

130− ParserD.IDone b -> return (b, seed)

131− ParserD.IError err -> throwM $ ParseError err

129+ ParserD.IDone b -> return (Right b, seed)

130+ ParserD.IError err -> return (Left (ParseError err), seed)

132131

133132 where

134133

@@ -158,8 +157,10 @@ parseD

158157 let src0 = Prelude.take n (x:getList buf)

159158 src = Prelude.reverse src0

160159 s1 <- uextract s

161− return (b, unread src s1)

162− Error err -> throwM $ ParseError err

160+ return (Right b, unread src s1)

161+ Error err -> do

162+ s1 <- uextract s

163+ return (Left (ParseError err), unread [x] s1)

163164 Skip s -> go SPEC s buf pst

164165 Stop -> goStop buf pst

165166

@@ -186,8 +187,10 @@ parseD

186187 let src0 = Prelude.take n (x:getList buf)

187188 src = Prelude.reverse src0

188189 s1 <- uextract s

189− return (b, unread src s1)

190− Error err -> throwM $ ParseError err

190+ return (Right b, unread src s1)

191+ Error err -> do

192+ s1 <- uextract s

193+ return (Left (ParseError err), unread (x:xs) s1)

191194

192195 -- This is a simplified gobuf

193196 goExtract !_ buf (List []) !pst = goStop buf pst

@@ -212,8 +215,9 @@ parseD

212215 assert (n <= length (x:getList buf)) (return ())

213216 let src0 = Prelude.take n (x:getList buf)

214217 src = Prelude.reverse src0

215− return (b, unread src (source Nothing))

216− Error err -> throwM $ ParseError err

218+ return (Right b, unread src (source Nothing))

219+ Error err ->

220+ return (Left (ParseError err), unread (x:xs) (source Nothing))

217221

218222 -- This is a simplified goExtract

219223 {-# INLINE goStop #-}

@@ -228,25 +232,25 @@ parseD

228232 let (src0, buf1) = splitAt n (getList buf)

229233 src = Prelude.reverse src0

230234 goExtract SPEC (List buf1) (List src) pst1

231− Done 0 b -> return (b, source Nothing)

235+ Done 0 b -> return (Right b, source Nothing)

232236 Done n b -> do

233237 assert (n <= length (getList buf)) (return ())

234238 let src0 = Prelude.take n (getList buf)

235239 src = Prelude.reverse src0

236− return (b, unread src (source Nothing))

237− Error err -> throwM $ ParseError err

240+ return (Right b, unread src (source Nothing))

241+ Error err ->

242+ return (Left (ParseError err), source Nothing)

238243

239244-- | Parse a buffered source using a parser, returning the parsed value and the

240245-- remaining source.

241246--

242247-- /Pre-release/

243248{-# INLINE [3] parse #-}

244−parse

245− :: MonadThrow m

246− => ParserK.Parser a m b

249+parse :: Monad m =>

250+ ParserK.Parser a m b

247251 -> Producer m (Source s a) a

248252 -> Source s a

249− -> m (b, Source s a)

253+ -> m (Either ParseError b, Source s a)

250254parse = parseD . ParserD.fromParserK

251255

252256-------------------------------------------------------------------------------

@@ -254,10 +258,10 @@ parse = parseD . ParserD.fromParserK

254258-------------------------------------------------------------------------------

255259

256260{-# INLINE parseManyD #-}

257−parseManyD :: MonadThrow m =>

261+parseManyD :: Monad m =>

258262 ParserD.Parser a m b

259263 -> Producer m (Source x a) a

260− -> Producer m (Source x a) b

264+ -> Producer m (Source x a) (Either ParseError b)

261265parseManyD parser reader = Producer step return return

262266

263267 where

@@ -268,15 +272,17 @@ parseManyD parser reader = Producer step return return

268272 then return Stop

269273 else do

270274 (b, s1) <- parseD parser reader src

271− return $ Yield b s1

275+ case b of

276+ Right b1 -> return $ Yield (Right b1) s1

277+ Left _ -> return Stop

272278

273279-- | Apply a parser repeatedly on a buffered source producer to generate a

274280-- producer of parsed values.

275281--

276282-- /Pre-release/

277283{-# INLINE parseMany #-}

278−parseMany :: MonadThrow m =>

284+parseMany :: Monad m =>

279285 ParserK.Parser a m b

280286 -> Producer m (Source x a) a

281− -> Producer m (Source x a) b

287+ -> Producer m (Source x a) (Either ParseError b)

282288parseMany parser = parseManyD (ParserD.fromParserK parser)

core/src/Streamly/Internal/Data/Stream/Chunked.hs+112−59

@@ -52,7 +52,6 @@ where

5252

5353import Data.Bifunctor (second)

5454import Control.Exception (assert)

55−import Control.Monad.Catch (MonadThrow, throwM)

5655import Control.Monad.IO.Class (MonadIO(..))

5756import Data.Word (Word8)

5857import Streamly.Internal.Data.Unboxed (Unbox, peekWith, sizeOf)

@@ -634,16 +633,16 @@ parseBreakD

634633

635634{-# INLINE_NORMAL parseBreakK #-}

636635parseBreakK ::

637− forall m a b. (MonadIO m, MonadThrow m, Unbox a)

636+ forall m a b. (MonadIO m, Unbox a)

638637 => PRD.Parser a m b

639638 -> K.Stream m (Array.Array a)

640− -> m (b, K.Stream m (Array.Array a))

639+ -> m (Either ParseError b, K.Stream m (Array.Array a))

641640parseBreakK (PRD.Parser pstep initial extract) stream = do

642641 res <- initial

643642 case res of

644643 PRD.IPartial s -> go s stream []

645− PRD.IDone b -> return (b, stream)

646− PRD.IError err -> throwM $ ParseError err

644+ PRD.IDone b -> return (Right b, stream)

645+ PRD.IError err -> return (Left (ParseError err), stream)

647646

648647 where

649648

@@ -686,7 +685,7 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do

686685 goArray s buf1 st src

687686 PR.Done 0 b -> do

688687 let arr = Array contents next end

689− return (b, K.cons arr st)

688+ return (Right b, K.cons arr st)

690689 PR.Done n b -> do

691690 assert (n <= Prelude.length (x:backBuf)) (return ())

692691 let src0 = Prelude.take n (x:backBuf)

@@ -695,8 +694,10 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do

695694 arr0 = A.fromListN n (Prelude.reverse src0)

696695 arr1 = Array contents next end

697696 str = K.cons arr0 (K.cons arr1 st)

698− return (b, str)

699− PR.Error err -> throwM $ ParseError err

697+ return (Right b, str)

698+ PR.Error err -> do

699+ let str = K.cons (Array contents cur end) stream

700+ return (Left (ParseError err), str)

700701

701702 -- This is a simplified goArray

702703 goExtract !pst backBuf (Array _ cur end)

@@ -726,7 +727,7 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do

726727 goExtract s buf1 src

727728 PR.Done 0 b -> do

728729 let arr = Array contents next end

729− return (b, K.fromPure arr)

730+ return (Right b, K.fromPure arr)

730731 PR.Done n b -> do

731732 assert (n <= Prelude.length backBuf) (return ())

732733 let src0 = Prelude.take n backBuf

@@ -735,8 +736,10 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do

735736 arr0 = A.fromListN n (Prelude.reverse src0)

736737 arr1 = Array contents next end

737738 str = K.cons arr0 (K.fromPure arr1)

738− return (b, str)

739− PR.Error err -> throwM $ ParseError err

739+ return (Right b, str)

740+ PR.Error err -> do

741+ let str = K.fromPure (Array contents cur end)

742+ return (Left (ParseError err), str)

740743

741744 -- This is a simplified goExtract

742745 {-# INLINE goStop #-}

@@ -752,15 +755,16 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do

752755 arr = A.fromListN n (Prelude.reverse src0)

753756 goExtract s buf1 arr

754757 PR.Done 0 b ->

755− return (b, K.nil)

758+ return (Right b, K.nil)

756759 PR.Done n b -> do

757760 assert (n <= Prelude.length backBuf) (return ())

758761 let src0 = Prelude.take n backBuf

759762 -- XXX Use fromListRevN once implemented

760763 -- arr0 = A.fromListRevN n src0

761764 arr0 = A.fromListN n (Prelude.reverse src0)

762− return (b, K.fromPure arr0)

763− PR.Error err -> throwM $ ParseError err

765+ return (Right b, K.fromPure arr0)

766+ PR.Error err ->

767+ return (Left (ParseError err), K.nil)

764768

765769-- | Parse an array stream using the supplied 'Parser'. Returns the parse

766770-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.

@@ -769,10 +773,10 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do

769773--

770774{-# INLINE_NORMAL parseBreak #-}

771775parseBreak ::

772− (MonadIO m, MonadThrow m, Unbox a)

776+ (MonadIO m, Unbox a)

773777 => PR.Parser a m b

774778 -> Stream m (A.Array a)

775− -> m (b, Stream m (A.Array a))

779+ -> m (Either ParseError b, Stream m (A.Array a))

776780{-

777781parseBreak p s =

778782 fmap fromStreamD <$> parseBreakD (PRD.fromParserK p) (toStreamD s)

@@ -795,10 +799,10 @@ parseBreak p =

795799--

796800{-# INLINE_NORMAL runArrayParserDBreak #-}

797801runArrayParserDBreak ::

798− forall m a b. (MonadIO m, MonadThrow m, Unbox a)

802+ forall m a b. (MonadIO m, Unbox a)

799803 => PRD.Parser (Array a) m b

800804 -> D.Stream m (Array.Array a)

801− -> m (b, D.Stream m (Array.Array a))

805+ -> m (Either ParseError b, D.Stream m (Array.Array a))

802806runArrayParserDBreak

803807 (PRD.Parser pstep initial extract)

804808 stream@(D.Stream step state) = do

@@ -806,8 +810,8 @@ runArrayParserDBreak

806810 res <- initial

807811 case res of

808812 PRD.IPartial s -> go SPEC state (List []) s

809− PRD.IDone b -> return (b, stream)

810− PRD.IError err -> throwM $ ParseError err

813+ PRD.IDone b -> return (Right b, stream)

814+ PRD.IError err -> return (Left (ParseError err), stream)

811815

812816 where

813817

@@ -846,16 +850,19 @@ runArrayParserDBreak

846850 let (src0, buf1) = splitAtArrayListRev n (x:getList backBuf)

847851 src = Prelude.reverse src0 ++ xs

848852 gobuf SPEC src s (List buf1) pst1

849− PR.Done 0 b ->

850− return (b, D.Stream step s)

853+ PR.Done 0 b -> do

854+ let str = D.append (D.fromList xs) (D.Stream step s)

855+ return (Right b, str)

851856 PR.Done n b -> do

852857 assert

853858 (n <= sum (map Array.length (x:getList backBuf)))

854859 (return ())

855860 let src0 = takeArrayListRev n (x:getList backBuf)

856861 src = Prelude.reverse src0 ++ xs

857− return (b, D.append (D.fromList src) (D.Stream step s))

858− PR.Error err -> throwM $ ParseError err

862+ return (Right b, D.append (D.fromList src) (D.Stream step s))

863+ PR.Error err -> do

864+ let strm = D.append (D.fromList (x:xs)) (D.Stream step s)

865+ return (Left (ParseError err), strm)

859866

860867 -- This is a simplified gobuf

861868 goExtract _ [] backBuf !pst = goStop backBuf pst

@@ -881,15 +888,16 @@ runArrayParserDBreak

881888 src = Prelude.reverse src0 ++ xs

882889 goExtract SPEC src (List buf1) pst1

883890 PR.Done 0 b ->

884− return (b, D.nil)

891+ return (Right b, D.fromList xs)

885892 PR.Done n b -> do

886893 assert

887894 (n <= sum (map Array.length (x:getList backBuf)))

888895 (return ())

889896 let src0 = takeArrayListRev n (x:getList backBuf)

890897 src = Prelude.reverse src0 ++ xs

891− return (b, D.fromList src)

892− PR.Error err -> throwM $ ParseError err

898+ return (Right b, D.fromList src)

899+ PR.Error err ->

900+ return (Left (ParseError err), D.fromList (x:xs))

893901

894902 -- This is a simplified goExtract

895903 {-# INLINE goStop #-}

@@ -906,15 +914,16 @@ runArrayParserDBreak

906914 let (src0, buf1) = splitAtArrayListRev n (getList backBuf)

907915 src = Prelude.reverse src0

908916 goExtract SPEC src (List buf1) pst1

909− PR.Done 0 b -> return (b, D.nil)

917+ PR.Done 0 b -> return (Right b, D.nil)

910918 PR.Done n b -> do

911919 assert

912920 (n <= sum (map Array.length (getList backBuf)))

913921 (return ())

914922 let src0 = takeArrayListRev n (getList backBuf)

915923 src = Prelude.reverse src0

916− return (b, D.fromList src)

917− PR.Error err -> throwM $ ParseError err

924+ return (Right b, D.fromList src)

925+ PR.Error err ->

926+ return (Left (ParseError err), D.nil)

918927

919928{-

920929-- | Parse an array stream using the supplied 'Parser'. Returns the parse

@@ -936,8 +945,8 @@ parseArr p s = fmap fromStreamD <$> parseBreakD p (toStreamD s)

936945-- /Pre-release/

937946--

938947{-# INLINE runArrayFold #-}

939−runArrayFold :: (MonadIO m, MonadThrow m, Unbox a) =>

940− ChunkFold m a b -> Stream m (A.Array a) -> m b

948+runArrayFold :: (MonadIO m, Unbox a) =>

949+ ChunkFold m a b -> Stream m (A.Array a) -> m (Either ParseError b)

941950runArrayFold (ChunkFold p) s = fst <$> runArrayParserDBreak p (toStreamD s)

942951

943952-- | Like 'fold' but also returns the remaining stream.

@@ -945,8 +954,8 @@ runArrayFold (ChunkFold p) s = fst <$> runArrayParserDBreak p (toStreamD s)

945954-- /Pre-release/

946955--

947956{-# INLINE runArrayFoldBreak #-}

948−runArrayFoldBreak :: (MonadIO m, MonadThrow m, Unbox a) =>

949− ChunkFold m a b -> Stream m (A.Array a) -> m (b, Stream m (A.Array a))

957+runArrayFoldBreak :: (MonadIO m, Unbox a) =>

958+ ChunkFold m a b -> Stream m (A.Array a) -> m (Either ParseError b, Stream m (A.Array a))

950959runArrayFoldBreak (ChunkFold p) s =

951960 second fromStreamD <$> runArrayParserDBreak p (toStreamD s)

952961

@@ -963,10 +972,10 @@ data ParseChunksState x inpBuf st pst =

963972

964973{-# INLINE_NORMAL runArrayFoldManyD #-}

965974runArrayFoldManyD

966− :: (MonadThrow m, Unbox a)

975+ :: (Monad m, Unbox a)

967976 => ChunkFold m a b

968977 -> D.Stream m (Array a)

969− -> D.Stream m b

978+ -> D.Stream m (Either ParseError b)

970979runArrayFoldManyD

971980 (ChunkFold (PRD.Parser pstep initial extract)) (D.Stream step state) =

972981

@@ -985,10 +994,14 @@ runArrayFoldManyD

985994 case res of

986995 PRD.IPartial ps ->

987996 return $ D.Skip $ ParseChunksBuf [x] s [] ps

988− PRD.IDone pb ->

997+ PRD.IDone pb -> do

989998 let next = ParseChunksInit [x] s

990− in return $ D.Skip $ ParseChunksYield pb next

991− PRD.IError err -> throwM $ ParseError err

999+ return $ D.Skip $ ParseChunksYield (Right pb) next

1000+ PRD.IError err -> do

1001+ let next = ParseChunksInitLeftOver []

1002+ return

1003+ $ D.Skip

1004+ $ ParseChunksYield (Left (ParseError err)) next

9921005 D.Skip s -> return $ D.Skip $ ParseChunksInit [] s

9931006 D.Stop -> return D.Stop

9941007

@@ -1000,8 +1013,12 @@ runArrayFoldManyD

10001013 return $ D.Skip $ ParseChunksBuf src st [] ps

10011014 PRD.IDone pb ->

10021015 let next = ParseChunksInit src st

1003− in return $ D.Skip $ ParseChunksYield pb next

1004− PRD.IError err -> throwM $ ParseError err

1016+ in return $ D.Skip $ ParseChunksYield (Right pb) next

1017+ PRD.IError err -> do

1018+ let next = ParseChunksInitLeftOver []

1019+ return

1020+ $ D.Skip

1021+ $ ParseChunksYield (Left (ParseError err)) next

10051022

10061023 -- This is a simplified ParseChunksInit

10071024 stepOuter _ (ParseChunksInitBuf src) = do

@@ -1011,8 +1028,12 @@ runArrayFoldManyD

10111028 return $ D.Skip $ ParseChunksExtract src [] ps

10121029 PRD.IDone pb ->

10131030 let next = ParseChunksInitBuf src

1014− in return $ D.Skip $ ParseChunksYield pb next

1015− PRD.IError err -> throwM $ ParseError err

1031+ in return $ D.Skip $ ParseChunksYield (Right pb) next

1032+ PRD.IError err -> do

1033+ let next = ParseChunksInitLeftOver []

1034+ return

1035+ $ D.Skip

1036+ $ ParseChunksYield (Left (ParseError err)) next

10161037

10171038 -- XXX we just discard any leftover input at the end

10181039 stepOuter _ (ParseChunksInitLeftOver _) = return D.Stop

@@ -1044,16 +1065,23 @@ runArrayFoldManyD

10441065 return $ D.Skip $ ParseChunksBuf src s buf1 pst1

10451066 PR.Done 0 b -> do

10461067 return $ D.Skip $

1047− ParseChunksYield b (ParseChunksInit [] s)

1068+ ParseChunksYield (Right b) (ParseChunksInit [] s)

10481069 PR.Done n b -> do

10491070 assert

10501071 (n <= sum (map Array.length (x:backBuf)))

10511072 (return ())

10521073 let src0 = takeArrayListRev n (x:backBuf)

1053− let src = Prelude.reverse src0

1054− return $ D.Skip $

1055− ParseChunksYield b (ParseChunksInit src s)

1056− PR.Error err -> throwM $ ParseError err

1074+ src = Prelude.reverse src0

1075+ next = ParseChunksInit src s

1076+ return

1077+ $ D.Skip

1078+ $ ParseChunksYield (Right b) next

1079+ PR.Error err -> do

1080+ let next = ParseChunksInitLeftOver []

1081+ return

1082+ $ D.Skip

1083+ $ ParseChunksYield (Left (ParseError err)) next

1084+

10571085 D.Skip s -> return $ D.Skip $ ParseChunksStream s backBuf pst

10581086 D.Stop -> return $ D.Skip $ ParseChunksStop backBuf pst

10591087

@@ -1080,13 +1108,21 @@ runArrayFoldManyD

10801108 src = Prelude.reverse src0 ++ xs

10811109 return $ D.Skip $ ParseChunksBuf src s buf1 pst1

10821110 PR.Done 0 b ->

1083− return $ D.Skip $ ParseChunksYield b (ParseChunksInit xs s)

1111+ return

1112+ $ D.Skip

1113+ $ ParseChunksYield (Right b) (ParseChunksInit xs s)

10841114 PR.Done n b -> do

10851115 assert (n <= sum (map Array.length (x:backBuf))) (return ())

10861116 let src0 = takeArrayListRev n (x:backBuf)

10871117 src = Prelude.reverse src0 ++ xs

1088− return $ D.Skip $ ParseChunksYield b (ParseChunksInit src s)

1089− PR.Error err -> throwM $ ParseError err

1118+ return

1119+ $ D.Skip

1120+ $ ParseChunksYield (Right b) (ParseChunksInit src s)

1121+ PR.Error err -> do

1122+ let next = ParseChunksInitLeftOver []

1123+ return

1124+ $ D.Skip

1125+ $ ParseChunksYield (Left (ParseError err)) next

10901126

10911127 -- This is a simplified ParseChunksBuf

10921128 stepOuter _ (ParseChunksExtract [] buf pst) =

@@ -1110,13 +1146,22 @@ runArrayFoldManyD

11101146 src = Prelude.reverse src0 ++ xs

11111147 return $ D.Skip $ ParseChunksExtract src buf1 pst1

11121148 PR.Done 0 b ->

1113− return $ D.Skip $ ParseChunksYield b (ParseChunksInitBuf xs)

1149+ return

1150+ $ D.Skip

1151+ $ ParseChunksYield (Right b) (ParseChunksInitBuf xs)

11141152 PR.Done n b -> do

11151153 assert (n <= sum (map Array.length (x:backBuf))) (return ())

11161154 let src0 = takeArrayListRev n (x:backBuf)

11171155 src = Prelude.reverse src0 ++ xs

1118− return $ D.Skip $ ParseChunksYield b (ParseChunksInitBuf src)

1119− PR.Error err -> throwM $ ParseError err

1156+ return

1157+ $ D.Skip

1158+ $ ParseChunksYield (Right b) (ParseChunksInitBuf src)

1159+ PR.Error err -> do

1160+ let next = ParseChunksInitLeftOver []

1161+ return

1162+ $ D.Skip

1163+ $ ParseChunksYield (Left (ParseError err)) next

1164+

11201165

11211166 -- This is a simplified ParseChunksExtract

11221167 stepOuter _ (ParseChunksStop backBuf pst) = do

@@ -1131,13 +1176,21 @@ runArrayFoldManyD

11311176 src = Prelude.reverse src0

11321177 return $ D.Skip $ ParseChunksExtract src buf1 pst1

11331178 PR.Done 0 b ->

1134− return $ D.Skip $ ParseChunksYield b (ParseChunksInitLeftOver [])

1179+ return

1180+ $ D.Skip

1181+ $ ParseChunksYield (Right b) (ParseChunksInitLeftOver [])

11351182 PR.Done n b -> do

11361183 assert (n <= sum (map Array.length backBuf)) (return ())

11371184 let src0 = takeArrayListRev n backBuf

11381185 src = Prelude.reverse src0

1139− return $ D.Skip $ ParseChunksYield b (ParseChunksInitBuf src)

1140− PR.Error err -> throwM $ ParseError err

1186+ return

1187+ $ D.Skip

1188+ $ ParseChunksYield (Right b) (ParseChunksInitBuf src)

1189+ PR.Error err -> do

1190+ let next = ParseChunksInitLeftOver []

1191+ return

1192+ $ D.Skip

1193+ $ ParseChunksYield (Left (ParseError err)) next

11411194

11421195 stepOuter _ (ParseChunksYield a next) = return $ D.Yield a next

11431196

@@ -1149,8 +1202,8 @@ runArrayFoldManyD

11491202-- /Pre-release/

11501203{-# INLINE runArrayFoldMany #-}

11511204runArrayFoldMany

1152− :: (MonadThrow m, Unbox a)

1205+ :: (Monad m, Unbox a)

11531206 => ChunkFold m a b

11541207 -> Stream m (Array a)

1155− -> Stream m b

1208+ -> Stream m (Either ParseError b)

11561209runArrayFoldMany p m = fromStreamD $ runArrayFoldManyD p (toStreamD m)

test/Streamly/Test/Data/Array/Stream.hs+14−8

@@ -7,13 +7,13 @@ import Streamly.Test.Common (listEquals, chooseInt)

77import Test.Hspec (hspec, describe, shouldBe)

88import Test.Hspec.QuickCheck

99import Test.QuickCheck (forAll, Property, vectorOf, Gen, Arbitrary (arbitrary))

10−import Test.QuickCheck.Monadic (monadicIO, run)

10+import Test.QuickCheck.Monadic (assert, monadicIO, run)

1111

1212import qualified Streamly.Internal.Data.Array as Array

1313import qualified Streamly.Internal.Data.Stream.Chunked as ArrayStream

1414import qualified Streamly.Internal.Data.Fold as Fold

1515import qualified Streamly.Internal.Data.Parser as Parser

16−import qualified Streamly.Internal.Data.Stream.IsStream as Stream

16+import qualified Streamly.Internal.Data.Stream as Stream

1717import qualified Test.Hspec as Hspec

1818

1919import Prelude hiding (sequence)

@@ -36,6 +36,10 @@ chooseAny = MkGen (\r _ -> let (x,_) = random r in x)

3636maxTestCount :: Int

3737maxTestCount = 100

3838

39+chunksOf :: Monad m

40+ => Int -> Fold.Fold m a b -> Stream.Stream m a -> Stream.Stream m b

41+chunksOf n f = Stream.foldMany (Fold.take n f)

42+

3943parseBreak :: Property

4044parseBreak = do

4145 let len = 200

@@ -51,19 +55,21 @@ parseBreak = do

5155 monadicIO $ do

5256 (ls1, str) <-

5357 let input =

54− Stream.chunksOf

58+ chunksOf

5559 clen (Array.writeN clen) (Stream.fromList ls)

5660 parser = Parser.fromFold (Fold.take tlen Fold.toList)

5761 in run $ ArrayStream.parseBreak parser input

58− ls2 <- run $ Stream.toList $ ArrayStream.concat str

59− listEquals (==) (ls1 ++ ls2) ls

62+ ls2 <- run $ Stream.fold Fold.toList (ArrayStream.concat str)

63+ case ls1 of

64+ Right x -> listEquals (==) (x ++ ls2) ls

65+ Left _ -> assert False

6066

6167splitOnSuffix :: Word8 -> [Word8] -> [[Word8]] -> IO ()

6268splitOnSuffix sep inp out = do

6369 res <-

64− Stream.toList

70+ Stream.fold Fold.toList

6571 $ ArrayStream.splitOnSuffix sep

66− $ Stream.chunksOf 2 (Array.writeN 2) $ Stream.fromList inp

72+ $ chunksOf 2 (Array.writeN 2) $ Stream.fromList inp

6773 fmap Array.toList res `shouldBe` out

6874

6975-------------------------------------------------------------------------------

@@ -81,7 +87,7 @@ concatArrayW8 =

8187 forAll (vectorOf 10000 (arbitrary :: Gen Word8))

8288 $ \w8List -> do

8389 let w8ArrList = Array.fromList . (: []) <$> w8List

84− f2 <- Stream.toList $ ArrayStream.concat $ Stream.fromList w8ArrList

90+ f2 <- Stream.fold Fold.toList $ ArrayStream.concat $ Stream.fromList w8ArrList

8591 w8List `shouldBe` f2

8692

8793

test/Streamly/Test/Data/Parser/ParserD.hs+1

@@ -669,6 +669,7 @@ parseUnfold = do

669669 xs <- run

670670 $ toList

671671 $ S.unfoldMany Unfold.fromList

672+ $ S.catRights

672673 $ S.unfold streamParser src

673674

674675 listEquals (==) xs ls