@@ -83,6 +83,7 @@ module Streamly.Internal.Data.Stream.StreamK
8383 , foldlMx'
8484 , fold
8585 , foldBreak
86+ , parseBreak
8687
8788 -- ** Specialized Folds
8889 , drain
@@ -174,17 +175,21 @@ module Streamly.Internal.Data.Stream.StreamK
174175 )
175176where
176177
178+import Control.Exception (assert)
179+import Control.Monad.Catch (MonadThrow, throwM)
177180import Control.Monad.Trans.Class (MonadTrans(lift))
178181import Control.Monad (void, join)
179182import Streamly.Internal.Data.SVar.Type (adaptState, defState)
180183
181184import qualified Streamly.Internal.Data.Fold.Type as FL
185+import qualified Streamly.Internal.Data.Parser.ParserD.Type as PR
186+import qualified Prelude
182187
183188import Prelude
184189 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,
185190 take, filter, all, any, takeWhile, drop, dropWhile, minimum,
186191 maximum, elem, notElem, null, head, tail, init, zipWith, lookup,
187− foldr1, (!!), replicate, reverse, concatMap, iterate)
192+ foldr1, (!!), replicate, reverse, concatMap, iterate, splitAt)
188193
189194import Streamly.Internal.Data.Stream.StreamK.Type
190195
@@ -979,3 +984,95 @@ withCatchError m h =
979984 yieldk a r = yld a (withCatchError r h)
980985 in handle $ run m
981986-}
987+
988+-------------------------------------------------------------------------------
989+-- Parsing
990+-------------------------------------------------------------------------------
991+
992+-- Inlined definition.
993+{-# INLINE splitAt #-}
994+splitAt :: Int -> [a] -> ([a],[a])
995+splitAt n ls
996+ | n <= 0 = ([], ls)
997+ | otherwise = splitAt' n ls
998+ where
999+ splitAt' :: Int -> [a] -> ([a], [a])
1000+ splitAt' _ [] = ([], [])
1001+ splitAt' 1 (x:xs) = ([x], xs)
1002+ splitAt' m (x:xs) = (x:xs', xs'')
1003+ where
1004+ (xs', xs'') = splitAt' (m - 1) xs
1005+
1006+-- | Run a 'Parser' over a stream and return rest of the Stream.
1007+{-# INLINE_NORMAL parseBreak #-}
1008+parseBreak
1009+ :: MonadThrow m
1010+ => PR.Parser m a b
1011+ -> Stream m a
1012+ -> m (b, Stream m a)
1013+parseBreak (PR.Parser pstep initial extract) stream = do
1014+ res <- initial
1015+ case res of
1016+ PR.IPartial s -> goStream stream [] s
1017+ PR.IDone b -> return (b, stream)
1018+ PR.IError err -> throwM $ PR.ParseError err
1019+
1020+ where
1021+
1022+ -- "buf" contains last few items in the stream that we may have to
1023+ -- backtrack to.
1024+ --
1025+ -- XXX currently we are using a dumb list based approach for backtracking
1026+ -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
1027+ -- That will allow us more efficient random back and forth movement.
1028+ goStream st buf !pst =
1029+ let stop = do
1030+ b <- extract pst
1031+ return (b, fromList buf)
1032+ single x = yieldk x nil
1033+ yieldk x r = do
1034+ res <- pstep pst x
1035+ case res of
1036+ PR.Partial 0 s -> goStream r [] s
1037+ PR.Partial n s -> do
1038+ assert (n <= length (x:buf)) (return ())
1039+ let src0 = Prelude.take n (x:buf)
1040+ src = Prelude.reverse src0
1041+ goBuf r [] src s
1042+ PR.Continue 0 s -> goStream r (x:buf) s
1043+ PR.Continue n s -> do
1044+ assert (n <= length (x:buf)) (return ())
1045+ let (src0, buf1) = splitAt n (x:buf)
1046+ src = Prelude.reverse src0
1047+ goBuf r buf1 src s
1048+ PR.Done 0 b -> return (b, r)
1049+ PR.Done n b -> do
1050+ assert (n <= length (x:buf)) (return ())
1051+ let src0 = Prelude.take n (x:buf)
1052+ src = Prelude.reverse src0
1053+ return (b, serial (fromList src) r)
1054+ PR.Error err -> throwM $ PR.ParseError err
1055+ in foldStream defState yieldk single stop st
1056+
1057+ goBuf st buf [] !pst = goStream st buf pst
1058+ goBuf st buf (x:xs) !pst = do
1059+ pRes <- pstep pst x
1060+ case pRes of
1061+ PR.Partial 0 s -> goBuf st [] xs s
1062+ PR.Partial n s -> do
1063+ assert (n <= length (x:buf)) (return ())
1064+ let src0 = Prelude.take n (x:buf)
1065+ src = Prelude.reverse src0 ++ xs
1066+ goBuf st [] src s
1067+ PR.Continue 0 s -> goBuf st (x:buf) xs s
1068+ PR.Continue n s -> do
1069+ assert (n <= length (x:buf)) (return ())
1070+ let (src0, buf1) = splitAt n (x:buf)
1071+ src = Prelude.reverse src0 ++ xs
1072+ goBuf st buf1 src s
1073+ PR.Done n b -> do
1074+ assert (n <= length (x:buf)) (return ())
1075+ let src0 = Prelude.take n (x:buf)
1076+ src = Prelude.reverse src0
1077+ return (b, serial (fromList src) st)
1078+ PR.Error err -> throwM $ PR.ParseError err