git.iko.soy/streamly
on master help sign in
Use Either result in parse ops in Chunked stream/Producer (#2145)
10b1ac1Harendra Kumar
core/src/Streamly/Internal/Data/Fold/Chunked.hs+1−1
compare with master
unifiedsplit
@@ -26,7 +26,7 @@
26-- >>> f = ChunkFold.fromFold (Fold.take 7 Fold.toList)
26-- >>> f = ChunkFold.fromFold (Fold.take 7 Fold.toList)
27-- >>> s = Stream.arraysOf 5 $ Stream.fromList "hello world"
27-- >>> s = Stream.arraysOf 5 $ Stream.fromList "hello world"
28-- >>> ArrayStream.runArrayFold f s
28-- >>> ArrayStream.runArrayFold f s
29−-- "hello w"
29+-- Right "hello w"
30--
30--
31module Streamly.Internal.Data.Fold.Chunked
31module Streamly.Internal.Data.Fold.Chunked
32 (
32 (
core/src/Streamly/Internal/Data/Producer/Source.hs+30−24
@@ -34,7 +34,6 @@ where
34#include "inline.hs"
34#include "inline.hs"
35
35
36import Control.Exception (assert)
36import Control.Exception (assert)
37−import Control.Monad.Catch (MonadThrow, throwM)
38import GHC.Exts (SpecConstrAnnotation(..))
37import GHC.Exts (SpecConstrAnnotation(..))
39import GHC.Types (SPEC(..))
38import GHC.Types (SPEC(..))
40import Streamly.Internal.Data.Parser.ParserD (ParseError(..), Step(..))
39import Streamly.Internal.Data.Parser.ParserD (ParseError(..), Step(..))
@@ -112,11 +111,11 @@ newtype List a = List {getList :: [a]}
112
111
113{-# INLINE_NORMAL parseD #-}
112{-# INLINE_NORMAL parseD #-}
114parseD
113parseD
115− :: MonadThrow m
114+ :: Monad m =>
116− => ParserD.Parser a m b
115+ ParserD.Parser a m b
117 -> Producer m (Source s a) a
116 -> Producer m (Source s a) a
118 -> Source s a
117 -> Source s a
119− -> m (b, Source s a)
118+ -> m (Either ParseError b, Source s a)
120parseD
119parseD
121 (ParserD.Parser pstep initial extract)
120 (ParserD.Parser pstep initial extract)
122 (Producer ustep uinject uextract)
121 (Producer ustep uinject uextract)
@@ -127,8 +126,8 @@ parseD
127 ParserD.IPartial s -> do
126 ParserD.IPartial s -> do
128 state <- uinject seed
127 state <- uinject seed
129 go SPEC state (List []) s
128 go SPEC state (List []) s
130− ParserD.IDone b -> return (b, seed)
129+ ParserD.IDone b -> return (Right b, seed)
131− ParserD.IError err -> throwM $ ParseError err
130+ ParserD.IError err -> return (Left (ParseError err), seed)
132
131
133 where
132 where
134
133
@@ -158,8 +157,10 @@ parseD
158 let src0 = Prelude.take n (x:getList buf)
157 let src0 = Prelude.take n (x:getList buf)
159 src = Prelude.reverse src0
158 src = Prelude.reverse src0
160 s1 <- uextract s
159 s1 <- uextract s
161− return (b, unread src s1)
160+ return (Right b, unread src s1)
161+ Error err -> do
162+ s1 <- uextract s
162− Error err -> throwM $ ParseError err
163+ return (Left (ParseError err), unread [x] s1)
163 Skip s -> go SPEC s buf pst
164 Skip s -> go SPEC s buf pst
164 Stop -> goStop buf pst
165 Stop -> goStop buf pst
165
166
@@ -186,8 +187,10 @@ parseD
186 let src0 = Prelude.take n (x:getList buf)
187 let src0 = Prelude.take n (x:getList buf)
187 src = Prelude.reverse src0
188 src = Prelude.reverse src0
188 s1 <- uextract s
189 s1 <- uextract s
189− return (b, unread src s1)
190+ return (Right b, unread src s1)
190− Error err -> throwM $ ParseError err
191+ Error err -> do
192+ s1 <- uextract s
193+ return (Left (ParseError err), unread (x:xs) s1)
191
194
192 -- This is a simplified gobuf
195 -- This is a simplified gobuf
193 goExtract !_ buf (List []) !pst = goStop buf pst
196 goExtract !_ buf (List []) !pst = goStop buf pst
@@ -212,8 +215,9 @@ parseD
212 assert (n <= length (x:getList buf)) (return ())
215 assert (n <= length (x:getList buf)) (return ())
213 let src0 = Prelude.take n (x:getList buf)
216 let src0 = Prelude.take n (x:getList buf)
214 src = Prelude.reverse src0
217 src = Prelude.reverse src0
215− return (b, unread src (source Nothing))
218+ return (Right b, unread src (source Nothing))
216− Error err -> throwM $ ParseError err
219+ Error err ->
220+ return (Left (ParseError err), unread (x:xs) (source Nothing))
217
221
218 -- This is a simplified goExtract
222 -- This is a simplified goExtract
219 {-# INLINE goStop #-}
223 {-# INLINE goStop #-}
@@ -228,25 +232,25 @@ parseD
228 let (src0, buf1) = splitAt n (getList buf)
232 let (src0, buf1) = splitAt n (getList buf)
229 src = Prelude.reverse src0
233 src = Prelude.reverse src0
230 goExtract SPEC (List buf1) (List src) pst1
234 goExtract SPEC (List buf1) (List src) pst1
231− Done 0 b -> return (b, source Nothing)
235+ Done 0 b -> return (Right b, source Nothing)
232 Done n b -> do
236 Done n b -> do
233 assert (n <= length (getList buf)) (return ())
237 assert (n <= length (getList buf)) (return ())
234 let src0 = Prelude.take n (getList buf)
238 let src0 = Prelude.take n (getList buf)
235 src = Prelude.reverse src0
239 src = Prelude.reverse src0
236− return (b, unread src (source Nothing))
240+ return (Right b, unread src (source Nothing))
237− Error err -> throwM $ ParseError err
241+ Error err ->
242+ return (Left (ParseError err), source Nothing)
238
243
239-- | Parse a buffered source using a parser, returning the parsed value and the
244-- | Parse a buffered source using a parser, returning the parsed value and the
240-- remaining source.
245-- remaining source.
241--
246--
242-- /Pre-release/
247-- /Pre-release/
243{-# INLINE [3] parse #-}
248{-# INLINE [3] parse #-}
244−parse
249+parse :: Monad m =>
245− :: MonadThrow m
246− => ParserK.Parser a m b
250+ ParserK.Parser a m b
247 -> Producer m (Source s a) a
251 -> Producer m (Source s a) a
248 -> Source s a
252 -> Source s a
249− -> m (b, Source s a)
253+ -> m (Either ParseError b, Source s a)
250parse = parseD . ParserD.fromParserK
254parse = parseD . ParserD.fromParserK
251
255
252-------------------------------------------------------------------------------
256-------------------------------------------------------------------------------
@@ -254,10 +258,10 @@ parse = parseD . ParserD.fromParserK
254-------------------------------------------------------------------------------
258-------------------------------------------------------------------------------
255
259
256{-# INLINE parseManyD #-}
260{-# INLINE parseManyD #-}
257−parseManyD :: MonadThrow m =>
261+parseManyD :: Monad m =>
258 ParserD.Parser a m b
262 ParserD.Parser a m b
259 -> Producer m (Source x a) a
263 -> Producer m (Source x a) a
260− -> Producer m (Source x a) b
264+ -> Producer m (Source x a) (Either ParseError b)
261parseManyD parser reader = Producer step return return
265parseManyD parser reader = Producer step return return
262
266
263 where
267 where
@@ -268,15 +272,17 @@ parseManyD parser reader = Producer step return return
268 then return Stop
272 then return Stop
269 else do
273 else do
270 (b, s1) <- parseD parser reader src
274 (b, s1) <- parseD parser reader src
275+ case b of
271− return $ Yield b s1
276+ Right b1 -> return $ Yield (Right b1) s1
277+ Left _ -> return Stop
272
278
273-- | Apply a parser repeatedly on a buffered source producer to generate a
279-- | Apply a parser repeatedly on a buffered source producer to generate a
274-- producer of parsed values.
280-- producer of parsed values.
275--
281--
276-- /Pre-release/
282-- /Pre-release/
277{-# INLINE parseMany #-}
283{-# INLINE parseMany #-}
278−parseMany :: MonadThrow m =>
284+parseMany :: Monad m =>
279 ParserK.Parser a m b
285 ParserK.Parser a m b
280 -> Producer m (Source x a) a
286 -> Producer m (Source x a) a
281− -> Producer m (Source x a) b
287+ -> Producer m (Source x a) (Either ParseError b)
282parseMany parser = parseManyD (ParserD.fromParserK parser)
288parseMany parser = parseManyD (ParserD.fromParserK parser)
core/src/Streamly/Internal/Data/Stream/Chunked.hs+112−59
@@ -52,7 +52,6 @@ where
52
52
53import Data.Bifunctor (second)
53import Data.Bifunctor (second)
54import Control.Exception (assert)
54import Control.Exception (assert)
55−import Control.Monad.Catch (MonadThrow, throwM)
56import Control.Monad.IO.Class (MonadIO(..))
55import Control.Monad.IO.Class (MonadIO(..))
57import Data.Word (Word8)
56import Data.Word (Word8)
58import Streamly.Internal.Data.Unboxed (Unbox, peekWith, sizeOf)
57import Streamly.Internal.Data.Unboxed (Unbox, peekWith, sizeOf)
@@ -634,16 +633,16 @@ parseBreakD
634
633
635{-# INLINE_NORMAL parseBreakK #-}
634{-# INLINE_NORMAL parseBreakK #-}
636parseBreakK ::
635parseBreakK ::
637− forall m a b. (MonadIO m, MonadThrow m, Unbox a)
636+ forall m a b. (MonadIO m, Unbox a)
638 => PRD.Parser a m b
637 => PRD.Parser a m b
639 -> K.Stream m (Array.Array a)
638 -> 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))
641parseBreakK (PRD.Parser pstep initial extract) stream = do
640parseBreakK (PRD.Parser pstep initial extract) stream = do
642 res <- initial
641 res <- initial
643 case res of
642 case res of
644 PRD.IPartial s -> go s stream []
643 PRD.IPartial s -> go s stream []
645− PRD.IDone b -> return (b, stream)
644+ PRD.IDone b -> return (Right b, stream)
646− PRD.IError err -> throwM $ ParseError err
645+ PRD.IError err -> return (Left (ParseError err), stream)
647
646
648 where
647 where
649
648
@@ -686,7 +685,7 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do
686 goArray s buf1 st src
685 goArray s buf1 st src
687 PR.Done 0 b -> do
686 PR.Done 0 b -> do
688 let arr = Array contents next end
687 let arr = Array contents next end
689− return (b, K.cons arr st)
688+ return (Right b, K.cons arr st)
690 PR.Done n b -> do
689 PR.Done n b -> do
691 assert (n <= Prelude.length (x:backBuf)) (return ())
690 assert (n <= Prelude.length (x:backBuf)) (return ())
692 let src0 = Prelude.take n (x:backBuf)
691 let src0 = Prelude.take n (x:backBuf)
@@ -695,8 +694,10 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do
695 arr0 = A.fromListN n (Prelude.reverse src0)
694 arr0 = A.fromListN n (Prelude.reverse src0)
696 arr1 = Array contents next end
695 arr1 = Array contents next end
697 str = K.cons arr0 (K.cons arr1 st)
696 str = K.cons arr0 (K.cons arr1 st)
698− return (b, str)
697+ return (Right b, str)
699− PR.Error err -> throwM $ ParseError err
698+ PR.Error err -> do
699+ let str = K.cons (Array contents cur end) stream
700+ return (Left (ParseError err), str)
700
701
701 -- This is a simplified goArray
702 -- This is a simplified goArray
702 goExtract !pst backBuf (Array _ cur end)
703 goExtract !pst backBuf (Array _ cur end)
@@ -726,7 +727,7 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do
726 goExtract s buf1 src
727 goExtract s buf1 src
727 PR.Done 0 b -> do
728 PR.Done 0 b -> do
728 let arr = Array contents next end
729 let arr = Array contents next end
729− return (b, K.fromPure arr)
730+ return (Right b, K.fromPure arr)
730 PR.Done n b -> do
731 PR.Done n b -> do
731 assert (n <= Prelude.length backBuf) (return ())
732 assert (n <= Prelude.length backBuf) (return ())
732 let src0 = Prelude.take n backBuf
733 let src0 = Prelude.take n backBuf
@@ -735,8 +736,10 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do
735 arr0 = A.fromListN n (Prelude.reverse src0)
736 arr0 = A.fromListN n (Prelude.reverse src0)
736 arr1 = Array contents next end
737 arr1 = Array contents next end
737 str = K.cons arr0 (K.fromPure arr1)
738 str = K.cons arr0 (K.fromPure arr1)
738− return (b, str)
739+ return (Right b, str)
739− PR.Error err -> throwM $ ParseError err
740+ PR.Error err -> do
741+ let str = K.fromPure (Array contents cur end)
742+ return (Left (ParseError err), str)
740
743
741 -- This is a simplified goExtract
744 -- This is a simplified goExtract
742 {-# INLINE goStop #-}
745 {-# INLINE goStop #-}
@@ -752,15 +755,16 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do
752 arr = A.fromListN n (Prelude.reverse src0)
755 arr = A.fromListN n (Prelude.reverse src0)
753 goExtract s buf1 arr
756 goExtract s buf1 arr
754 PR.Done 0 b ->
757 PR.Done 0 b ->
755− return (b, K.nil)
758+ return (Right b, K.nil)
756 PR.Done n b -> do
759 PR.Done n b -> do
757 assert (n <= Prelude.length backBuf) (return ())
760 assert (n <= Prelude.length backBuf) (return ())
758 let src0 = Prelude.take n backBuf
761 let src0 = Prelude.take n backBuf
759 -- XXX Use fromListRevN once implemented
762 -- XXX Use fromListRevN once implemented
760 -- arr0 = A.fromListRevN n src0
763 -- arr0 = A.fromListRevN n src0
761 arr0 = A.fromListN n (Prelude.reverse src0)
764 arr0 = A.fromListN n (Prelude.reverse src0)
762− return (b, K.fromPure arr0)
765+ return (Right b, K.fromPure arr0)
763− PR.Error err -> throwM $ ParseError err
766+ PR.Error err ->
767+ return (Left (ParseError err), K.nil)
764
768
765-- | Parse an array stream using the supplied 'Parser'. Returns the parse
769-- | Parse an array stream using the supplied 'Parser'. Returns the parse
766-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.
770-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.
@@ -769,10 +773,10 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do
769--
773--
770{-# INLINE_NORMAL parseBreak #-}
774{-# INLINE_NORMAL parseBreak #-}
771parseBreak ::
775parseBreak ::
772− (MonadIO m, MonadThrow m, Unbox a)
776+ (MonadIO m, Unbox a)
773 => PR.Parser a m b
777 => PR.Parser a m b
774 -> Stream m (A.Array a)
778 -> Stream m (A.Array a)
775− -> m (b, Stream m (A.Array a))
779+ -> m (Either ParseError b, Stream m (A.Array a))
776{-
780{-
777parseBreak p s =
781parseBreak p s =
778 fmap fromStreamD <$> parseBreakD (PRD.fromParserK p) (toStreamD s)
782 fmap fromStreamD <$> parseBreakD (PRD.fromParserK p) (toStreamD s)
@@ -795,10 +799,10 @@ parseBreak p =
795--
799--
796{-# INLINE_NORMAL runArrayParserDBreak #-}
800{-# INLINE_NORMAL runArrayParserDBreak #-}
797runArrayParserDBreak ::
801runArrayParserDBreak ::
798− forall m a b. (MonadIO m, MonadThrow m, Unbox a)
802+ forall m a b. (MonadIO m, Unbox a)
799 => PRD.Parser (Array a) m b
803 => PRD.Parser (Array a) m b
800 -> D.Stream m (Array.Array a)
804 -> 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))
802runArrayParserDBreak
806runArrayParserDBreak
803 (PRD.Parser pstep initial extract)
807 (PRD.Parser pstep initial extract)
804 stream@(D.Stream step state) = do
808 stream@(D.Stream step state) = do
@@ -806,8 +810,8 @@ runArrayParserDBreak
806 res <- initial
810 res <- initial
807 case res of
811 case res of
808 PRD.IPartial s -> go SPEC state (List []) s
812 PRD.IPartial s -> go SPEC state (List []) s
809− PRD.IDone b -> return (b, stream)
813+ PRD.IDone b -> return (Right b, stream)
810− PRD.IError err -> throwM $ ParseError err
814+ PRD.IError err -> return (Left (ParseError err), stream)
811
815
812 where
816 where
813
817
@@ -846,16 +850,19 @@ runArrayParserDBreak
846 let (src0, buf1) = splitAtArrayListRev n (x:getList backBuf)
850 let (src0, buf1) = splitAtArrayListRev n (x:getList backBuf)
847 src = Prelude.reverse src0 ++ xs
851 src = Prelude.reverse src0 ++ xs
848 gobuf SPEC src s (List buf1) pst1
852 gobuf SPEC src s (List buf1) pst1
849− PR.Done 0 b ->
853+ PR.Done 0 b -> do
850− return (b, D.Stream step s)
854+ let str = D.append (D.fromList xs) (D.Stream step s)
855+ return (Right b, str)
851 PR.Done n b -> do
856 PR.Done n b -> do
852 assert
857 assert
853 (n <= sum (map Array.length (x:getList backBuf)))
858 (n <= sum (map Array.length (x:getList backBuf)))
854 (return ())
859 (return ())
855 let src0 = takeArrayListRev n (x:getList backBuf)
860 let src0 = takeArrayListRev n (x:getList backBuf)
856 src = Prelude.reverse src0 ++ xs
861 src = Prelude.reverse src0 ++ xs
857− return (b, D.append (D.fromList src) (D.Stream step s))
862+ return (Right b, D.append (D.fromList src) (D.Stream step s))
858− PR.Error err -> throwM $ ParseError err
863+ PR.Error err -> do
864+ let strm = D.append (D.fromList (x:xs)) (D.Stream step s)
865+ return (Left (ParseError err), strm)
859
866
860 -- This is a simplified gobuf
867 -- This is a simplified gobuf
861 goExtract _ [] backBuf !pst = goStop backBuf pst
868 goExtract _ [] backBuf !pst = goStop backBuf pst
@@ -881,15 +888,16 @@ runArrayParserDBreak
881 src = Prelude.reverse src0 ++ xs
888 src = Prelude.reverse src0 ++ xs
882 goExtract SPEC src (List buf1) pst1
889 goExtract SPEC src (List buf1) pst1
883 PR.Done 0 b ->
890 PR.Done 0 b ->
884− return (b, D.nil)
891+ return (Right b, D.fromList xs)
885 PR.Done n b -> do
892 PR.Done n b -> do
886 assert
893 assert
887 (n <= sum (map Array.length (x:getList backBuf)))
894 (n <= sum (map Array.length (x:getList backBuf)))
888 (return ())
895 (return ())
889 let src0 = takeArrayListRev n (x:getList backBuf)
896 let src0 = takeArrayListRev n (x:getList backBuf)
890 src = Prelude.reverse src0 ++ xs
897 src = Prelude.reverse src0 ++ xs
891− return (b, D.fromList src)
898+ return (Right b, D.fromList src)
892− PR.Error err -> throwM $ ParseError err
899+ PR.Error err ->
900+ return (Left (ParseError err), D.fromList (x:xs))
893
901
894 -- This is a simplified goExtract
902 -- This is a simplified goExtract
895 {-# INLINE goStop #-}
903 {-# INLINE goStop #-}
@@ -906,15 +914,16 @@ runArrayParserDBreak
906 let (src0, buf1) = splitAtArrayListRev n (getList backBuf)
914 let (src0, buf1) = splitAtArrayListRev n (getList backBuf)
907 src = Prelude.reverse src0
915 src = Prelude.reverse src0
908 goExtract SPEC src (List buf1) pst1
916 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)
910 PR.Done n b -> do
918 PR.Done n b -> do
911 assert
919 assert
912 (n <= sum (map Array.length (getList backBuf)))
920 (n <= sum (map Array.length (getList backBuf)))
913 (return ())
921 (return ())
914 let src0 = takeArrayListRev n (getList backBuf)
922 let src0 = takeArrayListRev n (getList backBuf)
915 src = Prelude.reverse src0
923 src = Prelude.reverse src0
916− return (b, D.fromList src)
924+ return (Right b, D.fromList src)
917− PR.Error err -> throwM $ ParseError err
925+ PR.Error err ->
926+ return (Left (ParseError err), D.nil)
918
927
919{-
928{-
920-- | Parse an array stream using the supplied 'Parser'. Returns the parse
929-- | Parse an array stream using the supplied 'Parser'. Returns the parse
@@ -936,8 +945,8 @@ parseArr p s = fmap fromStreamD <$> parseBreakD p (toStreamD s)
936-- /Pre-release/
945-- /Pre-release/
937--
946--
938{-# INLINE runArrayFold #-}
947{-# INLINE runArrayFold #-}
939−runArrayFold :: (MonadIO m, MonadThrow m, Unbox a) =>
948+runArrayFold :: (MonadIO m, Unbox a) =>
940− ChunkFold m a b -> Stream m (A.Array a) -> m b
949+ ChunkFold m a b -> Stream m (A.Array a) -> m (Either ParseError b)
941runArrayFold (ChunkFold p) s = fst <$> runArrayParserDBreak p (toStreamD s)
950runArrayFold (ChunkFold p) s = fst <$> runArrayParserDBreak p (toStreamD s)
942
951
943-- | Like 'fold' but also returns the remaining stream.
952-- | Like 'fold' but also returns the remaining stream.
@@ -945,8 +954,8 @@ runArrayFold (ChunkFold p) s = fst <$> runArrayParserDBreak p (toStreamD s)
945-- /Pre-release/
954-- /Pre-release/
946--
955--
947{-# INLINE runArrayFoldBreak #-}
956{-# INLINE runArrayFoldBreak #-}
948−runArrayFoldBreak :: (MonadIO m, MonadThrow m, Unbox a) =>
957+runArrayFoldBreak :: (MonadIO m, Unbox a) =>
949− ChunkFold m a b -> Stream m (A.Array a) -> m (b, Stream m (A.Array a))
958+ ChunkFold m a b -> Stream m (A.Array a) -> m (Either ParseError b, Stream m (A.Array a))
950runArrayFoldBreak (ChunkFold p) s =
959runArrayFoldBreak (ChunkFold p) s =
951 second fromStreamD <$> runArrayParserDBreak p (toStreamD s)
960 second fromStreamD <$> runArrayParserDBreak p (toStreamD s)
952
961
@@ -963,10 +972,10 @@ data ParseChunksState x inpBuf st pst =
963
972
964{-# INLINE_NORMAL runArrayFoldManyD #-}
973{-# INLINE_NORMAL runArrayFoldManyD #-}
965runArrayFoldManyD
974runArrayFoldManyD
966− :: (MonadThrow m, Unbox a)
975+ :: (Monad m, Unbox a)
967 => ChunkFold m a b
976 => ChunkFold m a b
968 -> D.Stream m (Array a)
977 -> D.Stream m (Array a)
969− -> D.Stream m b
978+ -> D.Stream m (Either ParseError b)
970runArrayFoldManyD
979runArrayFoldManyD
971 (ChunkFold (PRD.Parser pstep initial extract)) (D.Stream step state) =
980 (ChunkFold (PRD.Parser pstep initial extract)) (D.Stream step state) =
972
981
@@ -985,10 +994,14 @@ runArrayFoldManyD
985 case res of
994 case res of
986 PRD.IPartial ps ->
995 PRD.IPartial ps ->
987 return $ D.Skip $ ParseChunksBuf [x] s [] ps
996 return $ D.Skip $ ParseChunksBuf [x] s [] ps
988− PRD.IDone pb ->
997+ PRD.IDone pb -> do
989 let next = ParseChunksInit [x] s
998 let next = ParseChunksInit [x] s
990− in return $ D.Skip $ ParseChunksYield pb next
999+ return $ D.Skip $ ParseChunksYield (Right pb) next
991− PRD.IError err -> throwM $ ParseError err
1000+ PRD.IError err -> do
1001+ let next = ParseChunksInitLeftOver []
1002+ return
1003+ $ D.Skip
1004+ $ ParseChunksYield (Left (ParseError err)) next
992 D.Skip s -> return $ D.Skip $ ParseChunksInit [] s
1005 D.Skip s -> return $ D.Skip $ ParseChunksInit [] s
993 D.Stop -> return D.Stop
1006 D.Stop -> return D.Stop
994
1007
@@ -1000,8 +1013,12 @@ runArrayFoldManyD
1000 return $ D.Skip $ ParseChunksBuf src st [] ps
1013 return $ D.Skip $ ParseChunksBuf src st [] ps
1001 PRD.IDone pb ->
1014 PRD.IDone pb ->
1002 let next = ParseChunksInit src st
1015 let next = ParseChunksInit src st
1003− in return $ D.Skip $ ParseChunksYield pb next
1016+ in return $ D.Skip $ ParseChunksYield (Right pb) next
1004− PRD.IError err -> throwM $ ParseError err
1017+ PRD.IError err -> do
1018+ let next = ParseChunksInitLeftOver []
1019+ return
1020+ $ D.Skip
1021+ $ ParseChunksYield (Left (ParseError err)) next
1005
1022
1006 -- This is a simplified ParseChunksInit
1023 -- This is a simplified ParseChunksInit
1007 stepOuter _ (ParseChunksInitBuf src) = do
1024 stepOuter _ (ParseChunksInitBuf src) = do
@@ -1011,8 +1028,12 @@ runArrayFoldManyD
1011 return $ D.Skip $ ParseChunksExtract src [] ps
1028 return $ D.Skip $ ParseChunksExtract src [] ps
1012 PRD.IDone pb ->
1029 PRD.IDone pb ->
1013 let next = ParseChunksInitBuf src
1030 let next = ParseChunksInitBuf src
1014− in return $ D.Skip $ ParseChunksYield pb next
1031+ in return $ D.Skip $ ParseChunksYield (Right pb) next
1015− PRD.IError err -> throwM $ ParseError err
1032+ PRD.IError err -> do
1033+ let next = ParseChunksInitLeftOver []
1034+ return
1035+ $ D.Skip
1036+ $ ParseChunksYield (Left (ParseError err)) next
1016
1037
1017 -- XXX we just discard any leftover input at the end
1038 -- XXX we just discard any leftover input at the end
1018 stepOuter _ (ParseChunksInitLeftOver _) = return D.Stop
1039 stepOuter _ (ParseChunksInitLeftOver _) = return D.Stop
@@ -1044,16 +1065,23 @@ runArrayFoldManyD
1044 return $ D.Skip $ ParseChunksBuf src s buf1 pst1
1065 return $ D.Skip $ ParseChunksBuf src s buf1 pst1
1045 PR.Done 0 b -> do
1066 PR.Done 0 b -> do
1046 return $ D.Skip $
1067 return $ D.Skip $
1047− ParseChunksYield b (ParseChunksInit [] s)
1068+ ParseChunksYield (Right b) (ParseChunksInit [] s)
1048 PR.Done n b -> do
1069 PR.Done n b -> do
1049 assert
1070 assert
1050 (n <= sum (map Array.length (x:backBuf)))
1071 (n <= sum (map Array.length (x:backBuf)))
1051 (return ())
1072 (return ())
1052 let src0 = takeArrayListRev n (x:backBuf)
1073 let src0 = takeArrayListRev n (x:backBuf)
1053− let src = Prelude.reverse src0
1074+ src = Prelude.reverse src0
1054− return $ D.Skip $
1055− ParseChunksYield b (ParseChunksInit src s)
1075+ next = ParseChunksInit src s
1076+ return
1077+ $ D.Skip
1078+ $ ParseChunksYield (Right b) next
1056− PR.Error err -> throwM $ ParseError err
1079+ PR.Error err -> do
1080+ let next = ParseChunksInitLeftOver []
1081+ return
1082+ $ D.Skip
1083+ $ ParseChunksYield (Left (ParseError err)) next
1084+
1057 D.Skip s -> return $ D.Skip $ ParseChunksStream s backBuf pst
1085 D.Skip s -> return $ D.Skip $ ParseChunksStream s backBuf pst
1058 D.Stop -> return $ D.Skip $ ParseChunksStop backBuf pst
1086 D.Stop -> return $ D.Skip $ ParseChunksStop backBuf pst
1059
1087
@@ -1080,13 +1108,21 @@ runArrayFoldManyD
1080 src = Prelude.reverse src0 ++ xs
1108 src = Prelude.reverse src0 ++ xs
1081 return $ D.Skip $ ParseChunksBuf src s buf1 pst1
1109 return $ D.Skip $ ParseChunksBuf src s buf1 pst1
1082 PR.Done 0 b ->
1110 PR.Done 0 b ->
1111+ return
1112+ $ D.Skip
1083− return $ D.Skip $ ParseChunksYield b (ParseChunksInit xs s)
1113+ $ ParseChunksYield (Right b) (ParseChunksInit xs s)
1084 PR.Done n b -> do
1114 PR.Done n b -> do
1085 assert (n <= sum (map Array.length (x:backBuf))) (return ())
1115 assert (n <= sum (map Array.length (x:backBuf))) (return ())
1086 let src0 = takeArrayListRev n (x:backBuf)
1116 let src0 = takeArrayListRev n (x:backBuf)
1087 src = Prelude.reverse src0 ++ xs
1117 src = Prelude.reverse src0 ++ xs
1118+ return
1119+ $ D.Skip
1088− return $ D.Skip $ ParseChunksYield b (ParseChunksInit src s)
1120+ $ ParseChunksYield (Right b) (ParseChunksInit src s)
1089− PR.Error err -> throwM $ ParseError err
1121+ PR.Error err -> do
1122+ let next = ParseChunksInitLeftOver []
1123+ return
1124+ $ D.Skip
1125+ $ ParseChunksYield (Left (ParseError err)) next
1090
1126
1091 -- This is a simplified ParseChunksBuf
1127 -- This is a simplified ParseChunksBuf
1092 stepOuter _ (ParseChunksExtract [] buf pst) =
1128 stepOuter _ (ParseChunksExtract [] buf pst) =
@@ -1110,13 +1146,22 @@ runArrayFoldManyD
1110 src = Prelude.reverse src0 ++ xs
1146 src = Prelude.reverse src0 ++ xs
1111 return $ D.Skip $ ParseChunksExtract src buf1 pst1
1147 return $ D.Skip $ ParseChunksExtract src buf1 pst1
1112 PR.Done 0 b ->
1148 PR.Done 0 b ->
1149+ return
1150+ $ D.Skip
1113− return $ D.Skip $ ParseChunksYield b (ParseChunksInitBuf xs)
1151+ $ ParseChunksYield (Right b) (ParseChunksInitBuf xs)
1114 PR.Done n b -> do
1152 PR.Done n b -> do
1115 assert (n <= sum (map Array.length (x:backBuf))) (return ())
1153 assert (n <= sum (map Array.length (x:backBuf))) (return ())
1116 let src0 = takeArrayListRev n (x:backBuf)
1154 let src0 = takeArrayListRev n (x:backBuf)
1117 src = Prelude.reverse src0 ++ xs
1155 src = Prelude.reverse src0 ++ xs
1156+ return
1157+ $ D.Skip
1118− return $ D.Skip $ ParseChunksYield b (ParseChunksInitBuf src)
1158+ $ ParseChunksYield (Right b) (ParseChunksInitBuf src)
1119− PR.Error err -> throwM $ ParseError err
1159+ PR.Error err -> do
1160+ let next = ParseChunksInitLeftOver []
1161+ return
1162+ $ D.Skip
1163+ $ ParseChunksYield (Left (ParseError err)) next
1164+
1120
1165
1121 -- This is a simplified ParseChunksExtract
1166 -- This is a simplified ParseChunksExtract
1122 stepOuter _ (ParseChunksStop backBuf pst) = do
1167 stepOuter _ (ParseChunksStop backBuf pst) = do
@@ -1131,13 +1176,21 @@ runArrayFoldManyD
1131 src = Prelude.reverse src0
1176 src = Prelude.reverse src0
1132 return $ D.Skip $ ParseChunksExtract src buf1 pst1
1177 return $ D.Skip $ ParseChunksExtract src buf1 pst1
1133 PR.Done 0 b ->
1178 PR.Done 0 b ->
1179+ return
1180+ $ D.Skip
1134− return $ D.Skip $ ParseChunksYield b (ParseChunksInitLeftOver [])
1181+ $ ParseChunksYield (Right b) (ParseChunksInitLeftOver [])
1135 PR.Done n b -> do
1182 PR.Done n b -> do
1136 assert (n <= sum (map Array.length backBuf)) (return ())
1183 assert (n <= sum (map Array.length backBuf)) (return ())
1137 let src0 = takeArrayListRev n backBuf
1184 let src0 = takeArrayListRev n backBuf
1138 src = Prelude.reverse src0
1185 src = Prelude.reverse src0
1186+ return
1187+ $ D.Skip
1139− return $ D.Skip $ ParseChunksYield b (ParseChunksInitBuf src)
1188+ $ ParseChunksYield (Right b) (ParseChunksInitBuf src)
1140− PR.Error err -> throwM $ ParseError err
1189+ PR.Error err -> do
1190+ let next = ParseChunksInitLeftOver []
1191+ return
1192+ $ D.Skip
1193+ $ ParseChunksYield (Left (ParseError err)) next
1141
1194
1142 stepOuter _ (ParseChunksYield a next) = return $ D.Yield a next
1195 stepOuter _ (ParseChunksYield a next) = return $ D.Yield a next
1143
1196
@@ -1149,8 +1202,8 @@ runArrayFoldManyD
1149-- /Pre-release/
1202-- /Pre-release/
1150{-# INLINE runArrayFoldMany #-}
1203{-# INLINE runArrayFoldMany #-}
1151runArrayFoldMany
1204runArrayFoldMany
1152− :: (MonadThrow m, Unbox a)
1205+ :: (Monad m, Unbox a)
1153 => ChunkFold m a b
1206 => ChunkFold m a b
1154 -> Stream m (Array a)
1207 -> Stream m (Array a)
1155− -> Stream m b
1208+ -> Stream m (Either ParseError b)
1156runArrayFoldMany p m = fromStreamD $ runArrayFoldManyD p (toStreamD m)
1209runArrayFoldMany 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)
7import Test.Hspec (hspec, describe, shouldBe)
7import Test.Hspec (hspec, describe, shouldBe)
8import Test.Hspec.QuickCheck
8import Test.Hspec.QuickCheck
9import Test.QuickCheck (forAll, Property, vectorOf, Gen, Arbitrary (arbitrary))
9import Test.QuickCheck (forAll, Property, vectorOf, Gen, Arbitrary (arbitrary))
10−import Test.QuickCheck.Monadic (monadicIO, run)
10+import Test.QuickCheck.Monadic (assert, monadicIO, run)
11
11
12import qualified Streamly.Internal.Data.Array as Array
12import qualified Streamly.Internal.Data.Array as Array
13import qualified Streamly.Internal.Data.Stream.Chunked as ArrayStream
13import qualified Streamly.Internal.Data.Stream.Chunked as ArrayStream
14import qualified Streamly.Internal.Data.Fold as Fold
14import qualified Streamly.Internal.Data.Fold as Fold
15import qualified Streamly.Internal.Data.Parser as Parser
15import 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
17import qualified Test.Hspec as Hspec
17import qualified Test.Hspec as Hspec
18
18
19import Prelude hiding (sequence)
19import Prelude hiding (sequence)
@@ -36,6 +36,10 @@ chooseAny = MkGen (\r _ -> let (x,_) = random r in x)
36maxTestCount :: Int
36maxTestCount :: Int
37maxTestCount = 100
37maxTestCount = 100
38
38
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+
39parseBreak :: Property
43parseBreak :: Property
40parseBreak = do
44parseBreak = do
41 let len = 200
45 let len = 200
@@ -51,19 +55,21 @@ parseBreak = do
51 monadicIO $ do
55 monadicIO $ do
52 (ls1, str) <-
56 (ls1, str) <-
53 let input =
57 let input =
54− Stream.chunksOf
58+ chunksOf
55 clen (Array.writeN clen) (Stream.fromList ls)
59 clen (Array.writeN clen) (Stream.fromList ls)
56 parser = Parser.fromFold (Fold.take tlen Fold.toList)
60 parser = Parser.fromFold (Fold.take tlen Fold.toList)
57 in run $ ArrayStream.parseBreak parser input
61 in run $ ArrayStream.parseBreak parser input
58− ls2 <- run $ Stream.toList $ ArrayStream.concat str
62+ ls2 <- run $ Stream.fold Fold.toList (ArrayStream.concat str)
63+ case ls1 of
59− listEquals (==) (ls1 ++ ls2) ls
64+ Right x -> listEquals (==) (x ++ ls2) ls
65+ Left _ -> assert False
60
66
61splitOnSuffix :: Word8 -> [Word8] -> [[Word8]] -> IO ()
67splitOnSuffix :: Word8 -> [Word8] -> [[Word8]] -> IO ()
62splitOnSuffix sep inp out = do
68splitOnSuffix sep inp out = do
63 res <-
69 res <-
64− Stream.toList
70+ Stream.fold Fold.toList
65 $ ArrayStream.splitOnSuffix sep
71 $ ArrayStream.splitOnSuffix sep
66− $ Stream.chunksOf 2 (Array.writeN 2) $ Stream.fromList inp
72+ $ chunksOf 2 (Array.writeN 2) $ Stream.fromList inp
67 fmap Array.toList res `shouldBe` out
73 fmap Array.toList res `shouldBe` out
68
74
69-------------------------------------------------------------------------------
75-------------------------------------------------------------------------------
@@ -81,7 +87,7 @@ concatArrayW8 =
81 forAll (vectorOf 10000 (arbitrary :: Gen Word8))
87 forAll (vectorOf 10000 (arbitrary :: Gen Word8))
82 $ \w8List -> do
88 $ \w8List -> do
83 let w8ArrList = Array.fromList . (: []) <$> w8List
89 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
85 w8List `shouldBe` f2
91 w8List `shouldBe` f2
86
92
87
93
test/Streamly/Test/Data/Parser/ParserD.hs+1
@@ -669,6 +669,7 @@ parseUnfold = do
669 xs <- run
669 xs <- run
670 $ toList
670 $ toList
671 $ S.unfoldMany Unfold.fromList
671 $ S.unfoldMany Unfold.fromList
672+ $ S.catRights
672 $ S.unfold streamParser src
673 $ S.unfold streamParser src
673
674
674 listEquals (==) xs ls
675 listEquals (==) xs ls