git.iko.soy/streamly
on fa9fa09 help sign in
streamly/core/src/Streamly/Internal/Data/Array/Stream.hs
1232 lines 48 kB history raw
1{-# OPTIONS_GHC -Wno-deprecations #-}
2-- |
3-- Module : Streamly.Internal.Data.Array.Stream
4-- Copyright : (c) 2019 Composewell Technologies
5-- License : BSD3-3-Clause
6-- Maintainer : streamly@composewell.com
7-- Portability : GHC
8--
9-- Combinators to efficiently manipulate streams of immutable arrays.
10--
11-- We can either push these in the MutArray module with a "chunks" prefix or
12-- keep this as a separate module and release it.
13--
14module Streamly.Internal.Data.Array.Stream
15 (
16 -- * Creation
17 Array.chunksOf
18 , Array.pinnedChunksOf
19 , Array.bufferChunks
21 -- * Flattening to elements
22 , concat
23 , Array.flattenArrays
24 , concatRev
25 , Array.flattenArraysRev
26 , interpose
27 , interposeSuffix
28 , intercalateSuffix
29 , unlines
31 -- * Elimination
32 -- ** Element Folds
33 -- The byte level foldBreak can work as efficiently as the chunk level. We
34 -- can flatten the stream to byte stream and use that. But if we want the
35 -- remaining stream to be a chunk stream then this could be handy. But it
36 -- could also be implemented using parseBreak.
37 , foldBreak
38 , foldBreakD
39 -- The byte level parseBreak cannot work efficiently. Because the stream
40 -- will have to be a StreamK for backtracking, StreamK at byte level would
41 -- not be efficient.
42 , parseBreak -- StreamK.parseBreakChunks
43 -- , parseBreakD
44 -- , foldManyChunks
45 -- , parseManyChunks
46 , K.parseBreakChunks
47 , K.parseChunks
49 -- ** Array Folds
50 -- XXX Use parseBreakChunks/parseChunks instead
51 -- foldBreak can be implemented using parseBreak. Use StreamK.
52 , runArrayFold
53 , runArrayFoldBreak
54 -- , parseArr
55 , runArrayParserDBreak -- StreamK.parseBreakChunks
56 , runArrayFoldMany
58 , toArray
60 -- * Compaction
61 -- We can use something like foldManyChunks, parseManyChunks with a take
62 -- fold.
63 , lpackArraysChunksOf -- Fold.compactChunks
64 , compact -- rechunk, compactChunks
66 -- * Splitting
67 -- We can use something like foldManyChunks, parseManyChunks with an
68 -- appropriate splitting fold.
69 , splitOn -- Stream.rechunkOn
70 , splitOnSuffix -- Stream.rechunkOnSuffix
71 )
72where
74#include "ArrayMacros.h"
75#include "inline.hs"
77import Data.Bifunctor (second)
78import Control.Exception (assert)
79import Control.Monad.IO.Class (MonadIO(..))
80import Data.Proxy (Proxy(..))
81import Data.Word (Word8)
82import Streamly.Internal.Data.Unbox (Unbox(..))
83import Fusion.Plugin.Types (Fuse(..))
84import GHC.Exts (SpecConstrAnnotation(..))
85import GHC.Types (SPEC(..))
86import Prelude hiding (null, last, (!!), read, concat, unlines)
88import Streamly.Data.Fold (Fold)
89import Streamly.Internal.Data.Array.Type (Array(..))
90import Streamly.Internal.Data.Fold.Chunked (ChunkFold(..))
91import Streamly.Internal.Data.Parser (ParseError(..))
92import Streamly.Internal.Data.Stream (Stream)
93import Streamly.Internal.Data.StreamK (StreamK, fromStream, toStream)
94import Streamly.Internal.Data.SVar.Type (adaptState, defState)
95import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
97import qualified Streamly.Data.Fold as FL
98import qualified Streamly.Internal.Data.Array as A
99import qualified Streamly.Internal.Data.Array as Array
100import qualified Streamly.Internal.Data.MutArray.Type as MA
101import qualified Streamly.Internal.Data.MutArray.Stream as AS
102import qualified Streamly.Internal.Data.Fold.Type as FL (Fold(..), Step(..))
103import qualified Streamly.Internal.Data.Parser as PR
104import qualified Streamly.Internal.Data.Parser as PRD
105 (Parser(..), Initial(..))
106import qualified Streamly.Internal.Data.Stream as D
107import qualified Streamly.Internal.Data.StreamK as K
109-- XXX Since these are immutable arrays MonadIO constraint can be removed from
110-- most places.
112-------------------------------------------------------------------------------
113-- Append
114-------------------------------------------------------------------------------
116-- XXX efficiently compare two streams of arrays. Two streams can have chunks
117-- of different sizes, we can handle that in the stream comparison abstraction.
118-- This could be useful e.g. to fast compare whether two files differ.
120-- | Convert a stream of arrays into a stream of their elements.
121--
122-- Same as the following:
123--
124-- > concat = Stream.unfoldMany Array.read
125--
126-- @since 0.7.0
127{-# INLINE concat #-}
128concat :: (Monad m, Unbox a) => Stream m (Array a) -> Stream m a
129-- concat m = fromStreamD $ A.flattenArrays (toStreamD m)
130-- concat m = fromStreamD $ D.concatMap A.toStreamD (toStreamD m)
131concat = D.unfoldMany A.reader
133-- | Convert a stream of arrays into a stream of their elements reversing the
134-- contents of each array before flattening.
135--
136-- > concatRev = Stream.unfoldMany Array.readerRev
137--
138-- @since 0.7.0
139{-# INLINE concatRev #-}
140concatRev :: (Monad m, Unbox a) => Stream m (Array a) -> Stream m a
141-- concatRev m = fromStreamD $ A.flattenArraysRev (toStreamD m)
142concatRev = D.unfoldMany A.readerRev
144-------------------------------------------------------------------------------
145-- Intersperse and append
146-------------------------------------------------------------------------------
148-- | Flatten a stream of arrays after inserting the given element between
149-- arrays.
150--
151-- /Pre-release/
152{-# INLINE interpose #-}
153interpose :: (Monad m, Unbox a) => a -> Stream m (Array a) -> Stream m a
154interpose x = D.interpose x A.reader
156{-# INLINE intercalateSuffix #-}
157intercalateSuffix :: (Monad m, Unbox a)
158 => Array a -> Stream m (Array a) -> Stream m a
159intercalateSuffix = D.intercalateSuffix A.reader
161-- | Flatten a stream of arrays appending the given element after each
162-- array.
163--
164-- @since 0.7.0
165{-# INLINE interposeSuffix #-}
166interposeSuffix :: (Monad m, Unbox a)
167 => a -> Stream m (Array a) -> Stream m a
168-- interposeSuffix x = fromStreamD . A.unlines x . toStreamD
169interposeSuffix x = D.interposeSuffix x A.reader
171data FlattenState s =
172 OuterLoop s
173 | InnerLoop s !MA.MutableByteArray !Int !Int
175-- XXX This is a special case of interposeSuffix, can be removed.
176-- XXX Remove monadIO constraint
177{-# INLINE_NORMAL unlines #-}
178unlines :: forall m a. (MonadIO m, Unbox a)
179 => a -> D.Stream m (Array a) -> D.Stream m a
180unlines sep (D.Stream step state) = D.Stream step' (OuterLoop state)
181 where
182 {-# INLINE_LATE step' #-}
183 step' gst (OuterLoop st) = do
184 r <- step (adaptState gst) st
185 return $ case r of
186 D.Yield Array{..} s ->
187 D.Skip (InnerLoop s arrContents arrStart arrEnd)
188 D.Skip s -> D.Skip (OuterLoop s)
189 D.Stop -> D.Stop
191 step' _ (InnerLoop st _ p end) | p == end =
192 return $ D.Yield sep $ OuterLoop st
194 step' _ (InnerLoop st contents p end) = do
195 x <- liftIO $ peekByteIndex p contents
196 return $ D.Yield x (InnerLoop st contents (INDEX_NEXT(p,a)) end)
198-------------------------------------------------------------------------------
199-- Compact
200-------------------------------------------------------------------------------
202-- XXX These would not be needed once we implement compactLEFold, see
203-- module Streamly.Internal.Data.Stream.MutChunked
204--
205-- XXX Note that this thaws immutable arrays for appending, that may be
206-- problematic if multiple users do the same thing, however, immutable arrays
207-- would usually have no capacity to append, therefore, a copy will be forced
208-- anyway. Confirm this. We can forcefully trim the array capacity before thaw
209-- to ensure this.
210{-# INLINE_NORMAL packArraysChunksOf #-}
211packArraysChunksOf :: (MonadIO m, Unbox a)
212 => Int -> D.Stream m (Array a) -> D.Stream m (Array a)
213packArraysChunksOf n str =
214 D.map A.unsafeFreeze $ AS.packArraysChunksOf n $ D.map A.unsafeThaw str
216-- XXX instead of writing two different versions of this operation, we should
217-- write it as a pipe.
218--
219-- XXX Confirm that immutable arrays won't be modified.
220{-# INLINE_NORMAL lpackArraysChunksOf #-}
221lpackArraysChunksOf :: (MonadIO m, Unbox a)
222 => Int -> Fold m (Array a) () -> Fold m (Array a) ()
223lpackArraysChunksOf n fld =
224 FL.lmap A.unsafeThaw $ AS.lpackArraysChunksOf n (FL.lmap A.unsafeFreeze fld)
226-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
227-- maximum specified size in bytes.
228--
229-- @since 0.7.0
230{-# INLINE compact #-}
231compact :: (MonadIO m, Unbox a)
232 => Int -> Stream m (Array a) -> Stream m (Array a)
233compact = packArraysChunksOf
235-------------------------------------------------------------------------------
236-- Split
237-------------------------------------------------------------------------------
239data SplitState s arr
240 = Initial s
241 | Buffering s arr
242 | Splitting s arr
243 | Yielding arr (SplitState s arr)
244 | Finishing
246-- | Split a stream of arrays on a given separator byte, dropping the separator
247-- and coalescing all the arrays between two separators into a single array.
248--
249-- @since 0.7.0
250{-# INLINE_NORMAL _splitOn #-}
251_splitOn
252 :: MonadIO m
253 => Word8
254 -> D.Stream m (Array Word8)
255 -> D.Stream m (Array Word8)
256_splitOn byte (D.Stream step state) = D.Stream step' (Initial state)
258 where
260 {-# INLINE_LATE step' #-}
261 step' gst (Initial st) = do
262 r <- step gst st
263 case r of
264 D.Yield arr s -> do
265 (arr1, marr2) <- A.breakOn byte arr
266 return $ case marr2 of
267 Nothing -> D.Skip (Buffering s arr1)
268 Just arr2 -> D.Skip (Yielding arr1 (Splitting s arr2))
269 D.Skip s -> return $ D.Skip (Initial s)
270 D.Stop -> return D.Stop
272 step' gst (Buffering st buf) = do
273 r <- step gst st
274 case r of
275 D.Yield arr s -> do
276 (arr1, marr2) <- A.breakOn byte arr
277 buf' <- A.splice buf arr1
278 return $ case marr2 of
279 Nothing -> D.Skip (Buffering s buf')
280 Just x -> D.Skip (Yielding buf' (Splitting s x))
281 D.Skip s -> return $ D.Skip (Buffering s buf)
282 D.Stop -> return $
283 if A.byteLength buf == 0
284 then D.Stop
285 else D.Skip (Yielding buf Finishing)
287 step' _ (Splitting st buf) = do
288 (arr1, marr2) <- A.breakOn byte buf
289 return $ case marr2 of
290 Nothing -> D.Skip $ Buffering st arr1
291 Just arr2 -> D.Skip $ Yielding arr1 (Splitting st arr2)
293 step' _ (Yielding arr next) = return $ D.Yield arr next
294 step' _ Finishing = return D.Stop
296-- XXX Remove MonadIO constraint.
297-- | Split a stream of arrays on a given separator byte, dropping the separator
298-- and coalescing all the arrays between two separators into a single array.
299--
300-- @since 0.7.0
301{-# INLINE splitOn #-}
302splitOn
303 :: (MonadIO m)
304 => Word8
305 -> Stream m (Array Word8)
306 -> Stream m (Array Word8)
307splitOn byte = D.splitInnerBy (A.breakOn byte) A.splice
309{-# INLINE splitOnSuffix #-}
310splitOnSuffix
311 :: (MonadIO m)
312 => Word8
313 -> Stream m (Array Word8)
314 -> Stream m (Array Word8)
315-- splitOn byte s = fromStreamD $ A.splitOn byte $ toStreamD s
316splitOnSuffix byte = D.splitInnerBySuffix (A.breakOn byte) A.splice
318-------------------------------------------------------------------------------
319-- Elimination - Running folds
320-------------------------------------------------------------------------------
322{-# INLINE_NORMAL foldBreakD #-}
323foldBreakD :: forall m a b. (MonadIO m, Unbox a) =>
324 Fold m a b -> D.Stream m (Array a) -> m (b, D.Stream m (Array a))
325foldBreakD (FL.Fold fstep initial _ final) stream@(D.Stream step state) = do
326 res <- initial
327 case res of
328 FL.Partial fs -> go SPEC state fs
329 FL.Done fb -> return $! (fb, stream)
331 where
333 {-# INLINE go #-}
334 go !_ st !fs = do
335 r <- step defState st
336 case r of
337 D.Yield (Array contents start end) s ->
338 let fp = Tuple' end contents
339 in goArray SPEC s fp start fs
340 D.Skip s -> go SPEC s fs
341 D.Stop -> do
342 b <- final fs
343 return (b, D.nil)
345 goArray !_ s (Tuple' end _) !cur !fs
346 | cur == end = do
347 go SPEC s fs
348 goArray !_ st fp@(Tuple' end contents) !cur !fs = do
349 x <- liftIO $ peekByteIndex cur contents
350 res <- fstep fs x
351 let next = INDEX_NEXT(cur,a)
352 case res of
353 FL.Done b -> do
354 let arr = Array contents next end
355 return $! (b, D.cons arr (D.Stream step st))
356 FL.Partial fs1 -> goArray SPEC st fp next fs1
358{-# INLINE_NORMAL foldBreakK #-}
359foldBreakK :: forall m a b. (MonadIO m, Unbox a) =>
360 Fold m a b -> K.StreamK m (Array a) -> m (b, K.StreamK m (Array a))
361foldBreakK (FL.Fold fstep initial _ final) stream = do
362 res <- initial
363 case res of
364 FL.Partial fs -> go fs stream
365 FL.Done fb -> return (fb, stream)
367 where
369 {-# INLINE go #-}
370 go !fs st = do
371 let stop = (, K.nil) <$> final fs
372 single a = yieldk a K.nil
373 yieldk (Array contents start end) r =
374 let fp = Tuple' end contents
375 in goArray fs r fp start
376 in K.foldStream defState yieldk single stop st
378 goArray !fs st (Tuple' end _) !cur
379 | cur == end = do
380 go fs st
381 goArray !fs st fp@(Tuple' end contents) !cur = do
382 x <- liftIO $ peekByteIndex cur contents
383 res <- fstep fs x
384 let next = INDEX_NEXT(cur,a)
385 case res of
386 FL.Done b -> do
387 let arr = Array contents next end
388 return $! (b, K.cons arr st)
389 FL.Partial fs1 -> goArray fs1 st fp next
391-- | Fold an array stream using the supplied 'Fold'. Returns the fold result
392-- and the unconsumed stream.
393--
394-- > foldBreak f = runArrayFoldBreak (ChunkFold.fromFold f)
395--
396-- /Internal/
397--
398{-# INLINE_NORMAL foldBreak #-}
399foldBreak ::
400 (MonadIO m, Unbox a)
401 => Fold m a b
402 -> StreamK m (A.Array a)
403 -> m (b, StreamK m (A.Array a))
404-- foldBreak f s = fmap fromStreamD <$> foldBreakD f (toStreamD s)
405foldBreak = foldBreakK
406-- If foldBreak performs better than runArrayFoldBreak we can use a rewrite
407-- rule to rewrite runArrayFoldBreak to fold.
408-- foldBreak f = runArrayFoldBreak (ChunkFold.fromFold f)
410-------------------------------------------------------------------------------
411-- Fold to a single Array
412-------------------------------------------------------------------------------
414-- When we have to take an array partially, take the last part of the array.
415{-# INLINE takeArrayListRev #-}
416takeArrayListRev :: forall a. Unbox a => Int -> [Array a] -> [Array a]
417takeArrayListRev = go
419 where
421 go _ [] = []
422 go n _ | n <= 0 = []
423 go n (x:xs) =
424 let len = Array.length x
425 in if n > len
426 then x : go (n - len) xs
427 else if n == len
428 then [x]
429 else let !(Array contents _ end) = x
430 !start = end - (n * SIZE_OF(a))
431 in [Array contents start end]
433-- When we have to take an array partially, take the last part of the array in
434-- the first split.
435{-# INLINE splitAtArrayListRev #-}
436splitAtArrayListRev ::
437 forall a. Unbox a => Int -> [Array a] -> ([Array a],[Array a])
438splitAtArrayListRev n ls
439 | n <= 0 = ([], ls)
440 | otherwise = go n ls
441 where
442 go :: Int -> [Array a] -> ([Array a], [Array a])
443 go _ [] = ([], [])
444 go m (x:xs) =
445 let len = Array.length x
446 (xs', xs'') = go (m - len) xs
447 in if m > len
448 then (x:xs', xs'')
449 else if m == len
450 then ([x],xs)
451 else let !(Array contents start end) = x
452 end1 = end - (m * SIZE_OF(a))
453 arr2 = Array contents start end1
454 arr1 = Array contents end1 end
455 in ([arr1], arr2:xs)
457-------------------------------------------------------------------------------
458-- Fold to a single Array
459-------------------------------------------------------------------------------
461{-
463-- XXX Both of these implementations of splicing seem to perform equally well.
464-- We need to perform benchmarks over a range of sizes though.
466-- CAUTION! length must more than equal to lengths of all the arrays in the
467-- stream.
468{-# INLINE spliceArraysLenUnsafe #-}
469spliceArraysLenUnsafe :: (MonadIO m, Unbox a)
470 => Int -> Stream m (MutArray a) -> m (MutArray a)
471spliceArraysLenUnsafe len buffered = do
472 -- XXX The new array's pinned state should depend on the first element
473 -- of the stream. We should uncons the stream and do the required.
474 arr <- undefined
475 D.foldlM' MA.spliceUnsafe (return arr) buffered
477{-# INLINE _spliceArrays #-}
478_spliceArrays :: (MonadIO m, Unbox a)
479 => Stream m (Array a) -> m (Array a)
480_spliceArrays s = do
481 buffered <- D.foldr K.cons K.nil s
482 len <- K.fold FL.sum (fmap Array.length buffered)
483 -- XXX The new array's pinned state should depend on the first element
484 -- of the stream. We should uncons the stream and do the required.
485 arr <- undefined
486 final <- D.foldlM' writeArr (return arr) (toStream buffered)
487 return $ A.unsafeFreeze final
489 where
491 writeArr dst arr = MA.spliceUnsafe dst (A.unsafeThaw arr)
493{-# INLINE _spliceArraysBuffered #-}
494_spliceArraysBuffered :: (MonadIO m, Unbox a)
495 => Stream m (Array a) -> m (Array a)
496_spliceArraysBuffered s = do
497 buffered <- D.foldr K.cons K.nil s
498 len <- K.fold FL.sum (fmap Array.length buffered)
499 A.unsafeFreeze <$>
500 spliceArraysLenUnsafe len (fmap A.unsafeThaw (toStream buffered))
502-}
504{-# INLINE spliceArraysRealloced #-}
505spliceArraysRealloced :: forall m a. (MonadIO m, Unbox a)
506 => Stream m (Array a) -> m (Array a)
507spliceArraysRealloced s = do
508 res <- D.uncons s
509 case res of
510 Just (a, strm) -> do
511 arr <-
512 D.foldlM'
513 MA.spliceExp
514 (pure (A.unsafeThaw a))
515 (fmap A.unsafeThaw strm)
516 liftIO $ A.unsafeFreeze <$> MA.rightSize arr
517 Nothing -> pure A.nil
519-- XXX This should just be "fold A.write"
520--
521-- | Given a stream of arrays, splice them all together to generate a single
522-- array. The stream must be /finite/.
523--
524-- @since 0.7.0
525{-# INLINE toArray #-}
526toArray :: (MonadIO m, Unbox a) => Stream m (Array a) -> m (Array a)
527toArray = spliceArraysRealloced
528-- spliceArrays = _spliceArraysBuffered
530-- exponentially increasing sizes of the chunks upto the max limit.
531-- XXX this will be easier to implement with parsers/terminating folds
532-- With this we should be able to reduce the number of chunks/allocations.
533-- The reallocation/copy based toArray can also be implemented using this.
534--
535{-
536{-# INLINE toArraysInRange #-}
537toArraysInRange :: (MonadIO m, Unbox a)
538 => Int -> Int -> Fold m (Array a) b -> Fold m a b
539toArraysInRange low high (Fold step initial extract) =
540-}
542{-
543-- | Fold the input to a pure buffered stream (List) of arrays.
544{-# INLINE _toArraysOf #-}
545_toArraysOf :: (MonadIO m, Unbox a)
546 => Int -> Fold m a (Stream Identity (Array a))
547_toArraysOf n = FL.groupsOf n (A.writeNF n) FL.toStream
548-}
550-------------------------------------------------------------------------------
551-- Elimination - running element parsers
552-------------------------------------------------------------------------------
554-- GHC parser does not accept {-# ANN type [] NoSpecConstr #-}, so we need
555-- to make a newtype.
556{-# ANN type List NoSpecConstr #-}
557newtype List a = List {getList :: [a]}
559{-
560-- This can be generalized to any type provided it can be unfolded to a stream
561-- and it can be combined using a semigroup operation.
562--
563-- XXX This should be written using CPS (as parseK) if we want it to scale wrt
564-- to the number of times it can be called on the same stream.
565{-# INLINE_NORMAL parseBreakD #-}
566parseBreakD ::
567 forall m a b. (MonadIO m, MonadThrow m, Unbox a)
568 => PRD.Parser a m b
569 -> D.Stream m (Array.Array a)
570 -> m (b, D.Stream m (Array.Array a))
571parseBreakD
572 (PRD.Parser pstep initial extract) stream@(D.Stream step state) = do
574 res <- initial
575 case res of
576 PRD.IPartial s -> go SPEC state (List []) s
577 PRD.IDone b -> return (b, stream)
578 PRD.IError err -> throwM $ ParseError err
580 where
582 -- "backBuf" contains last few items in the stream that we may have to
583 -- backtrack to.
584 --
585 -- XXX currently we are using a dumb list based approach for backtracking
586 -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
587 -- That will allow us more efficient random back and forth movement.
588 go !_ st backBuf !pst = do
589 r <- step defState st
590 case r of
591 D.Yield (Array contents start end) s ->
592 gobuf SPEC s backBuf
593 (Tuple' end contents) start pst
594 D.Skip s -> go SPEC s backBuf pst
595 D.Stop -> do
596 b <- extract pst
597 return (b, D.nil)
599 -- Use strictness on "cur" to keep it unboxed
600 gobuf !_ s backBuf (Tuple' end _) !cur !pst
601 | cur == end = do
602 go SPEC s backBuf pst
603 gobuf !_ s backBuf fp@(Tuple' end contents) !cur !pst = do
604 x <- liftIO $ peekByteIndex contents cur
605 pRes <- pstep pst x
606 let next = INDEX_NEXT(cur,a)
607 case pRes of
608 PR.Partial 0 pst1 ->
609 gobuf SPEC s (List []) fp next pst1
610 PR.Partial n pst1 -> do
611 assert (n <= Prelude.length (x:getList backBuf)) (return ())
612 let src0 = Prelude.take n (x:getList backBuf)
613 arr0 = A.fromListN n (Prelude.reverse src0)
614 arr1 = Array contents next end
615 src = arr0 <> arr1
616 let !(Array cont1 start end1) = src
617 fp1 = Tuple' end1 cont1
618 gobuf SPEC s (List []) fp1 start pst1
619 PR.Continue 0 pst1 ->
620 gobuf SPEC s (List (x:getList backBuf)) fp next pst1
621 PR.Continue n pst1 -> do
622 assert (n <= Prelude.length (x:getList backBuf)) (return ())
623 let (src0, buf1) = splitAt n (x:getList backBuf)
624 arr0 = A.fromListN n (Prelude.reverse src0)
625 arr1 = Array contents next end
626 src = arr0 <> arr1
627 let !(Array cont1 start end1) = src
628 fp1 = Tuple' end1 cont1
629 gobuf SPEC s (List buf1) fp1 start pst1
630 PR.Done 0 b -> do
631 let arr = Array contents next end
632 return (b, D.cons arr (D.Stream step s))
633 PR.Done n b -> do
634 assert (n <= Prelude.length (x:getList backBuf)) (return ())
635 let src0 = Prelude.take n (x:getList backBuf)
636 -- XXX create the array in reverse instead
637 arr0 = A.fromListN n (Prelude.reverse src0)
638 arr1 = Array contents next end
639 -- XXX Use StreamK to avoid adding arbitrary layers of
640 -- constructors every time.
641 str = D.cons arr0 (D.cons arr1 (D.Stream step s))
642 return (b, str)
643 PR.Error err -> throwM $ ParseError err
644-}
646{-# INLINE_NORMAL parseBreakK #-}
647parseBreakK ::
648 forall m a b. (MonadIO m, Unbox a)
649 => PRD.Parser a m b
650 -> K.StreamK m (Array.Array a)
651 -> m (Either ParseError b, K.StreamK m (Array.Array a))
652parseBreakK (PRD.Parser pstep initial extract) stream = do
653 res <- initial
654 case res of
655 PRD.IPartial s -> go s stream []
656 PRD.IDone b -> return (Right b, stream)
657 PRD.IError err -> return (Left (ParseError err), stream)
659 where
661 -- "backBuf" contains last few items in the stream that we may have to
662 -- backtrack to.
663 --
664 -- XXX currently we are using a dumb list based approach for backtracking
665 -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
666 -- That will allow us more efficient random back and forth movement.
667 go !pst st backBuf = do
668 let stop = goStop pst backBuf -- (, K.nil) <$> extract pst
669 single a = yieldk a K.nil
670 yieldk arr r = goArray pst backBuf r arr
671 in K.foldStream defState yieldk single stop st
673 -- Use strictness on "cur" to keep it unboxed
674 goArray !pst backBuf st (Array _ cur end) | cur == end = go pst st backBuf
675 goArray !pst backBuf st (Array contents cur end) = do
676 x <- liftIO $ peekByteIndex cur contents
677 pRes <- pstep pst x
678 let next = INDEX_NEXT(cur,a)
679 case pRes of
680 PR.Partial 0 s ->
681 goArray s [] st (Array contents next end)
682 PR.Partial n s -> do
683 assert (n <= Prelude.length (x:backBuf)) (return ())
684 let src0 = Prelude.take n (x:backBuf)
685 arr0 = A.fromListN n (Prelude.reverse src0)
686 arr1 = Array contents next end
687 src = arr0 <> arr1
688 goArray s [] st src
689 PR.Continue 0 s ->
690 goArray s (x:backBuf) st (Array contents next end)
691 PR.Continue n s -> do
692 assert (n <= Prelude.length (x:backBuf)) (return ())
693 let (src0, buf1) = splitAt n (x:backBuf)
694 arr0 = A.fromListN n (Prelude.reverse src0)
695 arr1 = Array contents next end
696 src = arr0 <> arr1
697 goArray s buf1 st src
698 PR.Done 0 b -> do
699 let arr = Array contents next end
700 return (Right b, K.cons arr st)
701 PR.Done n b -> do
702 assert (n <= Prelude.length (x:backBuf)) (return ())
703 let src0 = Prelude.take n (x:backBuf)
704 -- XXX Use fromListRevN once implemented
705 -- arr0 = A.fromListRevN n src0
706 arr0 = A.fromListN n (Prelude.reverse src0)
707 arr1 = Array contents next end
708 str = K.cons arr0 (K.cons arr1 st)
709 return (Right b, str)
710 PR.Error err -> do
711 let n = Prelude.length backBuf
712 arr0 = A.fromListN n (Prelude.reverse backBuf)
713 arr1 = Array contents cur end
714 str = K.cons arr0 (K.cons arr1 stream)
715 return (Left (ParseError err), str)
717 -- This is a simplified goArray
718 goExtract !pst backBuf (Array _ cur end)
719 | cur == end = goStop pst backBuf
720 goExtract !pst backBuf (Array contents cur end) = do
721 x <- liftIO $ peekByteIndex cur contents
722 pRes <- pstep pst x
723 let next = INDEX_NEXT(cur,a)
724 case pRes of
725 PR.Partial 0 s ->
726 goExtract s [] (Array contents next end)
727 PR.Partial n s -> do
728 assert (n <= Prelude.length (x:backBuf)) (return ())
729 let src0 = Prelude.take n (x:backBuf)
730 arr0 = A.fromListN n (Prelude.reverse src0)
731 arr1 = Array contents next end
732 src = arr0 <> arr1
733 goExtract s [] src
734 PR.Continue 0 s ->
735 goExtract s backBuf (Array contents next end)
736 PR.Continue n s -> do
737 assert (n <= Prelude.length (x:backBuf)) (return ())
738 let (src0, buf1) = splitAt n (x:backBuf)
739 arr0 = A.fromListN n (Prelude.reverse src0)
740 arr1 = Array contents next end
741 src = arr0 <> arr1
742 goExtract s buf1 src
743 PR.Done 0 b -> do
744 let arr = Array contents next end
745 return (Right b, K.fromPure arr)
746 PR.Done n b -> do
747 assert (n <= Prelude.length backBuf) (return ())
748 let src0 = Prelude.take n backBuf
749 -- XXX Use fromListRevN once implemented
750 -- arr0 = A.fromListRevN n src0
751 arr0 = A.fromListN n (Prelude.reverse src0)
752 arr1 = Array contents next end
753 str = K.cons arr0 (K.fromPure arr1)
754 return (Right b, str)
755 PR.Error err -> do
756 let n = Prelude.length backBuf
757 arr0 = A.fromListN n (Prelude.reverse backBuf)
758 arr1 = Array contents cur end
759 str = K.cons arr0 (K.cons arr1 stream)
760 return (Left (ParseError err), str)
762 -- This is a simplified goExtract
763 {-# INLINE goStop #-}
764 goStop !pst backBuf = do
765 pRes <- extract pst
766 case pRes of
767 PR.Partial _ _ -> error "Bug: parseBreak: Partial in extract"
768 PR.Continue 0 s ->
769 goStop s backBuf
770 PR.Continue n s -> do
771 assert (n <= Prelude.length backBuf) (return ())
772 let (src0, buf1) = splitAt n backBuf
773 arr = A.fromListN n (Prelude.reverse src0)
774 goExtract s buf1 arr
775 PR.Done 0 b ->
776 return (Right b, K.nil)
777 PR.Done n b -> do
778 assert (n <= Prelude.length backBuf) (return ())
779 let src0 = Prelude.take n backBuf
780 -- XXX Use fromListRevN once implemented
781 -- arr0 = A.fromListRevN n src0
782 arr0 = A.fromListN n (Prelude.reverse src0)
783 return (Right b, K.fromPure arr0)
784 PR.Error err -> do
785 let n = Prelude.length backBuf
786 arr0 = A.fromListN n (Prelude.reverse backBuf)
787 return (Left (ParseError err), K.fromPure arr0)
789-- | Parse an array stream using the supplied 'Parser'. Returns the parse
790-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.
791--
792-- /Internal/
793--
794{-# INLINE_NORMAL parseBreak #-}
795parseBreak ::
796 (MonadIO m, Unbox a)
797 => PR.Parser a m b
798 -> StreamK m (A.Array a)
799 -> m (Either ParseError b, StreamK m (A.Array a))
800{-
801parseBreak p s =
802 fmap fromStreamD <$> parseBreakD (PRD.fromParserK p) (toStreamD s)
803-}
804parseBreak = parseBreakK
806-------------------------------------------------------------------------------
807-- Elimination - Running Array Folds and parsers
808-------------------------------------------------------------------------------
810-- | Note that this is not the same as using a @Parser (Array a) m b@ with the
811-- regular "Streamly.Internal.Data.IsStream.parse" function. The regular parse
812-- would consume the input arrays as single unit. This parser parses in the way
813-- as described in the ChunkFold module. The input arrays are treated as @n@
814-- element units and can be consumed partially. The remaining elements are
815-- inserted in the source stream as an array.
816--
817{-# INLINE_NORMAL runArrayParserDBreak #-}
818runArrayParserDBreak ::
819 forall m a b. (MonadIO m, Unbox a)
820 => PRD.Parser (Array a) m b
821 -> D.Stream m (Array.Array a)
822 -> m (Either ParseError b, D.Stream m (Array.Array a))
823runArrayParserDBreak
824 (PRD.Parser pstep initial extract)
825 stream@(D.Stream step state) = do
827 res <- initial
828 case res of
829 PRD.IPartial s -> go SPEC state (List []) s
830 PRD.IDone b -> return (Right b, stream)
831 PRD.IError err -> return (Left (ParseError err), stream)
833 where
835 -- "backBuf" contains last few items in the stream that we may have to
836 -- backtrack to.
837 --
838 -- XXX currently we are using a dumb list based approach for backtracking
839 -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
840 -- That will allow us more efficient random back and forth movement.
841 go _ st backBuf !pst = do
842 r <- step defState st
843 case r of
844 D.Yield x s -> gobuf SPEC [x] s backBuf pst
845 D.Skip s -> go SPEC s backBuf pst
846 D.Stop -> goStop backBuf pst
848 gobuf !_ [] s backBuf !pst = go SPEC s backBuf pst
849 gobuf !_ (x:xs) s backBuf !pst = do
850 pRes <- pstep pst x
851 case pRes of
852 PR.Partial 0 pst1 ->
853 gobuf SPEC xs s (List []) pst1
854 PR.Partial n pst1 -> do
855 assert
856 (n <= sum (map Array.length (x:getList backBuf)))
857 (return ())
858 let src0 = takeArrayListRev n (x:getList backBuf)
859 src = Prelude.reverse src0 ++ xs
860 gobuf SPEC src s (List []) pst1
861 PR.Continue 0 pst1 ->
862 gobuf SPEC xs s (List (x:getList backBuf)) pst1
863 PR.Continue n pst1 -> do
864 assert
865 (n <= sum (map Array.length (x:getList backBuf)))
866 (return ())
867 let (src0, buf1) = splitAtArrayListRev n (x:getList backBuf)
868 src = Prelude.reverse src0 ++ xs
869 gobuf SPEC src s (List buf1) pst1
870 PR.Done 0 b -> do
871 let str = D.append (D.fromList xs) (D.Stream step s)
872 return (Right b, str)
873 PR.Done n b -> do
874 assert
875 (n <= sum (map Array.length (x:getList backBuf)))
876 (return ())
877 let src0 = takeArrayListRev n (x:getList backBuf)
878 src = Prelude.reverse src0 ++ xs
879 return (Right b, D.append (D.fromList src) (D.Stream step s))
880 PR.Error err -> do
881 let src0 = x:getList backBuf
882 src = Prelude.reverse src0 ++ x:xs
883 strm = D.append (D.fromList src) (D.Stream step s)
884 return (Left (ParseError err), strm)
886 -- This is a simplified gobuf
887 goExtract _ [] backBuf !pst = goStop backBuf pst
888 goExtract _ (x:xs) backBuf !pst = do
889 pRes <- pstep pst x
890 case pRes of
891 PR.Partial 0 pst1 ->
892 goExtract SPEC xs (List []) pst1
893 PR.Partial n pst1 -> do
894 assert
895 (n <= sum (map Array.length (x:getList backBuf)))
896 (return ())
897 let src0 = takeArrayListRev n (x:getList backBuf)
898 src = Prelude.reverse src0 ++ xs
899 goExtract SPEC src (List []) pst1
900 PR.Continue 0 pst1 ->
901 goExtract SPEC xs (List (x:getList backBuf)) pst1
902 PR.Continue n pst1 -> do
903 assert
904 (n <= sum (map Array.length (x:getList backBuf)))
905 (return ())
906 let (src0, buf1) = splitAtArrayListRev n (x:getList backBuf)
907 src = Prelude.reverse src0 ++ xs
908 goExtract SPEC src (List buf1) pst1
909 PR.Done 0 b ->
910 return (Right b, D.fromList xs)
911 PR.Done n b -> do
912 assert
913 (n <= sum (map Array.length (x:getList backBuf)))
914 (return ())
915 let src0 = takeArrayListRev n (x:getList backBuf)
916 src = Prelude.reverse src0 ++ xs
917 return (Right b, D.fromList src)
918 PR.Error err -> do
919 let src0 = getList backBuf
920 src = Prelude.reverse src0 ++ x:xs
921 return (Left (ParseError err), D.fromList src)
923 -- This is a simplified goExtract
924 {-# INLINE goStop #-}
925 goStop backBuf pst = do
926 pRes <- extract pst
927 case pRes of
928 PR.Partial _ _ -> error "Bug: runArrayParserDBreak: Partial in extract"
929 PR.Continue 0 pst1 ->
930 goStop backBuf pst1
931 PR.Continue n pst1 -> do
932 assert
933 (n <= sum (map Array.length (getList backBuf)))
934 (return ())
935 let (src0, buf1) = splitAtArrayListRev n (getList backBuf)
936 src = Prelude.reverse src0
937 goExtract SPEC src (List buf1) pst1
938 PR.Done 0 b -> return (Right b, D.nil)
939 PR.Done n b -> do
940 assert
941 (n <= sum (map Array.length (getList backBuf)))
942 (return ())
943 let src0 = takeArrayListRev n (getList backBuf)
944 src = Prelude.reverse src0
945 return (Right b, D.fromList src)
946 PR.Error err -> do
947 let src0 = getList backBuf
948 src = Prelude.reverse src0
949 return (Left (ParseError err), D.fromList src)
951{-
952-- | Parse an array stream using the supplied 'Parser'. Returns the parse
953-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.
954--
955-- /Internal/
956--
957{-# INLINE parseArr #-}
958parseArr ::
959 (MonadIO m, MonadThrow m, Unbox a)
960 => ASF.Parser a m b
961 -> Stream m (A.Array a)
962 -> m (b, Stream m (A.Array a))
963parseArr p s = fmap fromStreamD <$> parseBreakD p (toStreamD s)
964-}
966-- | Fold an array stream using the supplied array stream 'Fold'.
967--
968-- /Pre-release/
969--
970{-# INLINE runArrayFold #-}
971runArrayFold :: (MonadIO m, Unbox a) =>
972 ChunkFold m a b -> StreamK m (A.Array a) -> m (Either ParseError b)
973runArrayFold (ChunkFold p) s = fst <$> runArrayParserDBreak p (toStream s)
975-- | Like 'fold' but also returns the remaining stream.
976--
977-- /Pre-release/
978--
979{-# INLINE runArrayFoldBreak #-}
980runArrayFoldBreak :: (MonadIO m, Unbox a) =>
981 ChunkFold m a b -> StreamK m (A.Array a) -> m (Either ParseError b, StreamK m (A.Array a))
982runArrayFoldBreak (ChunkFold p) s =
983 second fromStream <$> runArrayParserDBreak p (toStream s)
985{-# ANN type ParseChunksState Fuse #-}
986data ParseChunksState x inpBuf st pst =
987 ParseChunksInit inpBuf st
988 | ParseChunksInitBuf inpBuf
989 | ParseChunksInitLeftOver inpBuf
990 | ParseChunksStream st inpBuf !pst
991 | ParseChunksStop inpBuf !pst
992 | ParseChunksBuf inpBuf st inpBuf !pst
993 | ParseChunksExtract inpBuf inpBuf !pst
994 | ParseChunksYield x (ParseChunksState x inpBuf st pst)
996{-# INLINE_NORMAL runArrayFoldManyD #-}
997runArrayFoldManyD
998 :: (Monad m, Unbox a)
999 => ChunkFold m a b
1000 -> D.Stream m (Array a)
1001 -> D.Stream m (Either ParseError b)
1002runArrayFoldManyD
1003 (ChunkFold (PRD.Parser pstep initial extract)) (D.Stream step state) =
1005 D.Stream stepOuter (ParseChunksInit [] state)
1007 where
1009 {-# INLINE_LATE stepOuter #-}
1010 -- Buffer is empty, get the first element from the stream, initialize the
1011 -- fold and then go to stream processing loop.
1012 stepOuter gst (ParseChunksInit [] st) = do
1013 r <- step (adaptState gst) st
1014 case r of
1015 D.Yield x s -> do
1016 res <- initial
1017 case res of
1018 PRD.IPartial ps ->
1019 return $ D.Skip $ ParseChunksBuf [x] s [] ps
1020 PRD.IDone pb -> do
1021 let next = ParseChunksInit [x] s
1022 return $ D.Skip $ ParseChunksYield (Right pb) next
1023 PRD.IError err -> do
1024 let next = ParseChunksInitLeftOver []
1025 return
1026 $ D.Skip
1027 $ ParseChunksYield (Left (ParseError err)) next
1028 D.Skip s -> return $ D.Skip $ ParseChunksInit [] s
1029 D.Stop -> return D.Stop
1031 -- Buffer is not empty, go to buffered processing loop
1032 stepOuter _ (ParseChunksInit src st) = do
1033 res <- initial
1034 case res of
1035 PRD.IPartial ps ->
1036 return $ D.Skip $ ParseChunksBuf src st [] ps
1037 PRD.IDone pb ->
1038 let next = ParseChunksInit src st
1039 in return $ D.Skip $ ParseChunksYield (Right pb) next
1040 PRD.IError err -> do
1041 let next = ParseChunksInitLeftOver []
1042 return
1043 $ D.Skip
1044 $ ParseChunksYield (Left (ParseError err)) next
1046 -- This is a simplified ParseChunksInit
1047 stepOuter _ (ParseChunksInitBuf src) = do
1048 res <- initial
1049 case res of
1050 PRD.IPartial ps ->
1051 return $ D.Skip $ ParseChunksExtract src [] ps
1052 PRD.IDone pb ->
1053 let next = ParseChunksInitBuf src
1054 in return $ D.Skip $ ParseChunksYield (Right pb) next
1055 PRD.IError err -> do
1056 let next = ParseChunksInitLeftOver []
1057 return
1058 $ D.Skip
1059 $ ParseChunksYield (Left (ParseError err)) next
1061 -- XXX we just discard any leftover input at the end
1062 stepOuter _ (ParseChunksInitLeftOver _) = return D.Stop
1064 -- Buffer is empty, process elements from the stream
1065 stepOuter gst (ParseChunksStream st backBuf pst) = do
1066 r <- step (adaptState gst) st
1067 case r of
1068 D.Yield x s -> do
1069 pRes <- pstep pst x
1070 case pRes of
1071 PR.Partial 0 pst1 ->
1072 return $ D.Skip $ ParseChunksStream s [] pst1
1073 PR.Partial n pst1 -> do
1074 assert
1075 (n <= sum (map Array.length (x:backBuf)))
1076 (return ())
1077 let src0 = takeArrayListRev n (x:backBuf)
1078 src = Prelude.reverse src0
1079 return $ D.Skip $ ParseChunksBuf src s [] pst1
1080 PR.Continue 0 pst1 ->
1081 return $ D.Skip $ ParseChunksStream s (x:backBuf) pst1
1082 PR.Continue n pst1 -> do
1083 assert
1084 (n <= sum (map Array.length (x:backBuf)))
1085 (return ())
1086 let (src0, buf1) = splitAtArrayListRev n (x:backBuf)
1087 src = Prelude.reverse src0
1088 return $ D.Skip $ ParseChunksBuf src s buf1 pst1
1089 PR.Done 0 b -> do
1090 return $ D.Skip $
1091 ParseChunksYield (Right b) (ParseChunksInit [] s)
1092 PR.Done n b -> do
1093 assert
1094 (n <= sum (map Array.length (x:backBuf)))
1095 (return ())
1096 let src0 = takeArrayListRev n (x:backBuf)
1097 src = Prelude.reverse src0
1098 next = ParseChunksInit src s
1099 return
1100 $ D.Skip
1101 $ ParseChunksYield (Right b) next
1102 PR.Error err -> do
1103 let next = ParseChunksInitLeftOver []
1104 return
1105 $ D.Skip
1106 $ ParseChunksYield (Left (ParseError err)) next
1108 D.Skip s -> return $ D.Skip $ ParseChunksStream s backBuf pst
1109 D.Stop -> return $ D.Skip $ ParseChunksStop backBuf pst
1111 -- go back to stream processing mode
1112 stepOuter _ (ParseChunksBuf [] s buf pst) =
1113 return $ D.Skip $ ParseChunksStream s buf pst
1115 -- buffered processing loop
1116 stepOuter _ (ParseChunksBuf (x:xs) s backBuf pst) = do
1117 pRes <- pstep pst x
1118 case pRes of
1119 PR.Partial 0 pst1 ->
1120 return $ D.Skip $ ParseChunksBuf xs s [] pst1
1121 PR.Partial n pst1 -> do
1122 assert (n <= sum (map Array.length (x:backBuf))) (return ())
1123 let src0 = takeArrayListRev n (x:backBuf)
1124 src = Prelude.reverse src0 ++ xs
1125 return $ D.Skip $ ParseChunksBuf src s [] pst1
1126 PR.Continue 0 pst1 ->
1127 return $ D.Skip $ ParseChunksBuf xs s (x:backBuf) pst1
1128 PR.Continue n pst1 -> do
1129 assert (n <= sum (map Array.length (x:backBuf))) (return ())
1130 let (src0, buf1) = splitAtArrayListRev n (x:backBuf)
1131 src = Prelude.reverse src0 ++ xs
1132 return $ D.Skip $ ParseChunksBuf src s buf1 pst1
1133 PR.Done 0 b ->
1134 return
1135 $ D.Skip
1136 $ ParseChunksYield (Right b) (ParseChunksInit xs s)
1137 PR.Done n b -> do
1138 assert (n <= sum (map Array.length (x:backBuf))) (return ())
1139 let src0 = takeArrayListRev n (x:backBuf)
1140 src = Prelude.reverse src0 ++ xs
1141 return
1142 $ D.Skip
1143 $ ParseChunksYield (Right b) (ParseChunksInit src s)
1144 PR.Error err -> do
1145 let next = ParseChunksInitLeftOver []
1146 return
1147 $ D.Skip
1148 $ ParseChunksYield (Left (ParseError err)) next
1150 -- This is a simplified ParseChunksBuf
1151 stepOuter _ (ParseChunksExtract [] buf pst) =
1152 return $ D.Skip $ ParseChunksStop buf pst
1154 stepOuter _ (ParseChunksExtract (x:xs) backBuf pst) = do
1155 pRes <- pstep pst x
1156 case pRes of
1157 PR.Partial 0 pst1 ->
1158 return $ D.Skip $ ParseChunksExtract xs [] pst1
1159 PR.Partial n pst1 -> do
1160 assert (n <= sum (map Array.length (x:backBuf))) (return ())
1161 let src0 = takeArrayListRev n (x:backBuf)
1162 src = Prelude.reverse src0 ++ xs
1163 return $ D.Skip $ ParseChunksExtract src [] pst1
1164 PR.Continue 0 pst1 ->
1165 return $ D.Skip $ ParseChunksExtract xs (x:backBuf) pst1
1166 PR.Continue n pst1 -> do
1167 assert (n <= sum (map Array.length (x:backBuf))) (return ())
1168 let (src0, buf1) = splitAtArrayListRev n (x:backBuf)
1169 src = Prelude.reverse src0 ++ xs
1170 return $ D.Skip $ ParseChunksExtract src buf1 pst1
1171 PR.Done 0 b ->
1172 return
1173 $ D.Skip
1174 $ ParseChunksYield (Right b) (ParseChunksInitBuf xs)
1175 PR.Done n b -> do
1176 assert (n <= sum (map Array.length (x:backBuf))) (return ())
1177 let src0 = takeArrayListRev n (x:backBuf)
1178 src = Prelude.reverse src0 ++ xs
1179 return
1180 $ D.Skip
1181 $ ParseChunksYield (Right b) (ParseChunksInitBuf src)
1182 PR.Error err -> do
1183 let next = ParseChunksInitLeftOver []
1184 return
1185 $ D.Skip
1186 $ ParseChunksYield (Left (ParseError err)) next
1189 -- This is a simplified ParseChunksExtract
1190 stepOuter _ (ParseChunksStop backBuf pst) = do
1191 pRes <- extract pst
1192 case pRes of
1193 PR.Partial _ _ -> error "runArrayFoldManyD: Partial in extract"
1194 PR.Continue 0 pst1 ->
1195 return $ D.Skip $ ParseChunksStop backBuf pst1
1196 PR.Continue n pst1 -> do
1197 assert (n <= sum (map Array.length backBuf)) (return ())
1198 let (src0, buf1) = splitAtArrayListRev n backBuf
1199 src = Prelude.reverse src0
1200 return $ D.Skip $ ParseChunksExtract src buf1 pst1
1201 PR.Done 0 b ->
1202 return
1203 $ D.Skip
1204 $ ParseChunksYield (Right b) (ParseChunksInitLeftOver [])
1205 PR.Done n b -> do
1206 assert (n <= sum (map Array.length backBuf)) (return ())
1207 let src0 = takeArrayListRev n backBuf
1208 src = Prelude.reverse src0
1209 return
1210 $ D.Skip
1211 $ ParseChunksYield (Right b) (ParseChunksInitBuf src)
1212 PR.Error err -> do
1213 let next = ParseChunksInitLeftOver []
1214 return
1215 $ D.Skip
1216 $ ParseChunksYield (Left (ParseError err)) next
1218 stepOuter _ (ParseChunksYield a next) = return $ D.Yield a next
1220-- | Apply an 'ChunkFold' repeatedly on an array stream and emit the
1221-- fold outputs in the output stream.
1223-- See "Streamly.Data.Stream.foldMany" for more details.
1225-- /Pre-release/
1226{-# INLINE runArrayFoldMany #-}
1227runArrayFoldMany
1228 :: (Monad m, Unbox a)
1229 => ChunkFold m a b
1230 -> StreamK m (Array a)
1231 -> StreamK m (Either ParseError b)
1232runArrayFoldMany p m = fromStream $ runArrayFoldManyD p (toStream m)