git.iko.soy/streamly
on defafa6 help sign in
streamly/core/src/Streamly/Internal/Data/Array/Stream.hs
725 lines 28 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{-# DEPRECATED "Please use \"Streamly.Internal.Data.Array\" instead." #-}
16 (
17 -- * Creation
18 Array.chunksOf
19 , Array.pinnedChunksOf
20 , Array.bufferChunks
22 -- * Flattening to elements
23 , Array.concat
24 , Array.flattenArrays
25 , Array.concatRev
26 , Array.flattenArraysRev
27 , Array.interpose
28 , Array.interposeSuffix
29 , Array.intercalateSuffix
30 , unlines
32 -- * Elimination
33 -- ** Element Folds
34 -- The byte level foldBreak can work as efficiently as the chunk level. We
35 -- can flatten the stream to byte stream and use that. But if we want the
36 -- remaining stream to be a chunk stream then this could be handy. But it
37 -- could also be implemented using parseBreak.
38 , foldBreak
39 , foldBreakD
40 -- This is chunked parseBreak. A byte level parseBreak cannot work
41 -- efficiently. Because the stream will have to be a StreamK for
42 -- backtracking, StreamK at byte level would not be efficient.
43 -- parseBreak p = K.parseBreakChunks (ParserK.adaptC p)
44 , parseBreak
45 -- , parseBreakD
46 -- , foldManyChunks
47 -- , parseManyChunks
48 , K.parseBreakChunks
49 , K.parseChunks
51 -- ** Array Folds
52 -- XXX Use parseBreakChunks/parseChunks instead
53 -- foldBreak can be implemented using parseBreak. Use StreamK.
54 , runArrayFold
55 , runArrayFoldBreak
56 -- , parseArr
57 , runArrayParserDBreak -- StreamK.parseBreakChunks
58 , runArrayFoldMany
60 , toArray
62 -- * Compaction
63 -- We can use something like foldManyChunks, parseManyChunks with a take
64 -- fold.
65 , lpackArraysChunksOf -- Fold.compactChunks
66 , compact -- rechunk, compactChunks
68 -- * Splitting
69 -- We can use something like foldManyChunks, parseManyChunks with an
70 -- appropriate splitting fold.
71 , splitOn -- Stream.rechunkOn
72 , splitOnSuffix -- Stream.rechunkOnSuffix
73 )
74where
76#include "ArrayMacros.h"
77#include "inline.hs"
79import Data.Bifunctor (second)
80import Control.Exception (assert)
81import Control.Monad.IO.Class (MonadIO(..))
82-- import Data.Bifunctor (first)
83-- import Data.Either (fromRight)
84import Data.Proxy (Proxy(..))
85import Data.Word (Word8)
86import Streamly.Internal.Data.Unbox (Unbox(..))
87import Fusion.Plugin.Types (Fuse(..))
88import GHC.Exts (SpecConstrAnnotation(..))
89import GHC.Types (SPEC(..))
90import Prelude hiding (null, last, (!!), read, concat, unlines)
92import Streamly.Data.Fold (Fold)
93import Streamly.Internal.Data.Array.Type (Array(..))
94import Streamly.Internal.Data.Fold.Chunked (ChunkFold(..))
95import Streamly.Internal.Data.Parser (ParseError(..))
96import Streamly.Internal.Data.Stream (Stream)
97import Streamly.Internal.Data.StreamK (StreamK, fromStream, toStream)
98import Streamly.Internal.Data.SVar.Type (adaptState, defState)
100import qualified Streamly.Internal.Data.Array as A
101import qualified Streamly.Internal.Data.Array as Array
102import qualified Streamly.Internal.Data.Parser as PR
103import qualified Streamly.Internal.Data.Parser as PRD
104 (Parser(..), Initial(..))
105-- import qualified Streamly.Internal.Data.ParserK as ParserK
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-- Intersperse and append
114-------------------------------------------------------------------------------
116{-# INLINE_NORMAL unlines #-}
117unlines :: forall m a. (MonadIO m, Unbox a)
118 => a -> D.Stream m (Array a) -> D.Stream m a
119unlines = Array.interposeSuffix
121-------------------------------------------------------------------------------
122-- Compact
123-------------------------------------------------------------------------------
125-- XXX instead of writing two different versions of this operation, we should
126-- write it as a pipe.
127--
128-- XXX Confirm that immutable arrays won't be modified.
129{-# INLINE_NORMAL lpackArraysChunksOf #-}
130lpackArraysChunksOf :: (MonadIO m, Unbox a)
131 => Int -> Fold m (Array a) () -> Fold m (Array a) ()
132lpackArraysChunksOf = Array.lCompactGE
134-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
135-- maximum specified size in bytes.
136--
137-- @since 0.7.0
138{-# INLINE compact #-}
139compact :: (MonadIO m, Unbox a)
140 => Int -> Stream m (Array a) -> Stream m (Array a)
141compact = Array.compactLE
143-- | Given a stream of arrays, splice them all together to generate a single
144-- array. The stream must be /finite/.
145--
146-- @since 0.7.0
147{-# INLINE toArray #-}
148toArray :: (MonadIO m, Unbox a) => Stream m (Array a) -> m (Array a)
149toArray = Array.fromChunks
151-------------------------------------------------------------------------------
152-- Split
153-------------------------------------------------------------------------------
155-- XXX Remove MonadIO constraint.
156-- | Split a stream of arrays on a given separator byte, dropping the separator
157-- and coalescing all the arrays between two separators into a single array.
158--
159-- @since 0.7.0
160{-# INLINE splitOn #-}
161splitOn
162 :: (MonadIO m)
163 => Word8
164 -> Stream m (Array Word8)
165 -> Stream m (Array Word8)
166splitOn = Array.compactOnByte
168{-# INLINE splitOnSuffix #-}
169splitOnSuffix
170 :: (MonadIO m)
171 => Word8
172 -> Stream m (Array Word8)
173 -> Stream m (Array Word8)
174splitOnSuffix = Array.compactOnByteSuffix
176-------------------------------------------------------------------------------
177-- Elimination - Running folds
178-------------------------------------------------------------------------------
180{-# INLINE_NORMAL foldBreakD #-}
181foldBreakD :: forall m a b. (MonadIO m, Unbox a) =>
182 Fold m a b -> D.Stream m (Array a) -> m (b, D.Stream m (Array a))
183foldBreakD = Array.foldBreakChunks
185-- | Fold an array stream using the supplied 'Fold'. Returns the fold result
186-- and the unconsumed stream.
187--
188-- > foldBreak f = runArrayFoldBreak (ChunkFold.fromFold f)
189--
190-- Instead of using this we can adapt the fold to ParserK and use
191-- parseBreakChunks instead. ParserK allows composing using Monad as well.
192--
193-- @
194-- foldBreak f s =
195-- fmap (first (fromRight undefined))
196-- $ K.parseBreakChunks (ParserK.adaptC (PR.fromFold f)) s
197-- @
198--
199-- We can compare perf and remove this one or define it in terms of that.
200--
201-- /Internal/
202--
203{-# INLINE_NORMAL foldBreak #-}
204foldBreak ::
205 (MonadIO m, Unbox a)
206 => Fold m a b
207 -> StreamK m (A.Array a)
208 -> m (b, StreamK m (A.Array a))
209foldBreak = Array.foldBreakChunksK
210--
211-- foldBreak f s = fmap fromStreamD <$> foldBreakD f (toStreamD s)
212--
213-- foldBreak f s =
214-- fmap (first (fromRight undefined))
215-- $ K.parseBreakChunks (ParserK.adaptC (PR.fromFold f)) s
216--
217-- If foldBreak performs better than runArrayFoldBreak we can use a rewrite
218-- rule to rewrite runArrayFoldBreak to fold.
219-- foldBreak f = runArrayFoldBreak (ChunkFold.fromFold f)
221-------------------------------------------------------------------------------
222-- Elimination - running element parsers
223-------------------------------------------------------------------------------
225-- When we have to take an array partially, take the last part of the array.
226{-# INLINE takeArrayListRev #-}
227takeArrayListRev :: forall a. Unbox a => Int -> [Array a] -> [Array a]
228takeArrayListRev = go
230 where
232 go _ [] = []
233 go n _ | n <= 0 = []
234 go n (x:xs) =
235 let len = Array.length x
236 in if n > len
237 then x : go (n - len) xs
238 else if n == len
239 then [x]
240 else let !(Array contents _ end) = x
241 !start = end - (n * SIZE_OF(a))
242 in [Array contents start end]
244-- When we have to take an array partially, take the last part of the array in
245-- the first split.
246{-# INLINE splitAtArrayListRev #-}
247splitAtArrayListRev ::
248 forall a. Unbox a => Int -> [Array a] -> ([Array a],[Array a])
249splitAtArrayListRev n ls
250 | n <= 0 = ([], ls)
251 | otherwise = go n ls
252 where
253 go :: Int -> [Array a] -> ([Array a], [Array a])
254 go _ [] = ([], [])
255 go m (x:xs) =
256 let len = Array.length x
257 (xs', xs'') = go (m - len) xs
258 in if m > len
259 then (x:xs', xs'')
260 else if m == len
261 then ([x],xs)
262 else let !(Array contents start end) = x
263 end1 = end - (m * SIZE_OF(a))
264 arr2 = Array contents start end1
265 arr1 = Array contents end1 end
266 in ([arr1], arr2:xs)
268-- GHC parser does not accept {-# ANN type [] NoSpecConstr #-}, so we need
269-- to make a newtype.
270{-# ANN type List NoSpecConstr #-}
271newtype List a = List {getList :: [a]}
273-- | Parse an array stream using the supplied 'Parser'. Returns the parse
274-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.
275--
276-- >> parseBreak p = K.parseBreakChunks (ParserK.adaptC p)
277--
278-- This is redundant and we can just use parseBreakChunks, as ParserK can be
279-- composed using Monad. The only advantage of this is that we do not need to
280-- adapt.
281--
282-- We can compare perf and remove this one or define it in terms of that.
283--
284-- /Internal/
285--
286{-# INLINE_NORMAL parseBreak #-}
287parseBreak ::
288 (MonadIO m, Unbox a)
289 => PR.Parser a m b
290 -> StreamK m (A.Array a)
291 -> m (Either ParseError b, StreamK m (A.Array a))
292{-
293parseBreak p s =
294 fmap fromStreamD <$> parseBreakD (PRD.fromParserK p) (toStreamD s)
295-}
296parseBreak = Array.parseBreakChunksK
297-- parseBreak p = K.parseBreakChunks (ParserK.adaptC p)
299-------------------------------------------------------------------------------
300-- Elimination - Running Array Folds and parsers
301-------------------------------------------------------------------------------
303-- | Note that this is not the same as using a @Parser (Array a) m b@ with the
304-- regular "Streamly.Internal.Data.IsStream.parse" function. The regular parse
305-- would consume the input arrays as single unit. This parser parses in the way
306-- as described in the ChunkFold module. The input arrays are treated as @n@
307-- element units and can be consumed partially. The remaining elements are
308-- inserted in the source stream as an array.
309--
310{-# INLINE_NORMAL runArrayParserDBreak #-}
311runArrayParserDBreak ::
312 forall m a b. (MonadIO m, Unbox a)
313 => PRD.Parser (Array a) m b
314 -> D.Stream m (Array.Array a)
315 -> m (Either ParseError b, D.Stream m (Array.Array a))
316runArrayParserDBreak
317 (PRD.Parser pstep initial extract)
318 stream@(D.Stream step state) = do
320 res <- initial
321 case res of
322 PRD.IPartial s -> go SPEC state (List []) s
323 PRD.IDone b -> return (Right b, stream)
324 PRD.IError err -> return (Left (ParseError err), stream)
326 where
328 -- "backBuf" contains last few items in the stream that we may have to
329 -- backtrack to.
330 --
331 -- XXX currently we are using a dumb list based approach for backtracking
332 -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
333 -- That will allow us more efficient random back and forth movement.
334 go _ st backBuf !pst = do
335 r <- step defState st
336 case r of
337 D.Yield x s -> gobuf SPEC [x] s backBuf pst
338 D.Skip s -> go SPEC s backBuf pst
339 D.Stop -> goStop backBuf pst
341 gobuf !_ [] s backBuf !pst = go SPEC s backBuf pst
342 gobuf !_ (x:xs) s backBuf !pst = do
343 pRes <- pstep pst x
344 case pRes of
345 PR.Partial 0 pst1 ->
346 gobuf SPEC xs s (List []) pst1
347 PR.Partial n pst1 -> do
348 assert
349 (n <= sum (map Array.length (x:getList backBuf)))
350 (return ())
351 let src0 = takeArrayListRev n (x:getList backBuf)
352 src = Prelude.reverse src0 ++ xs
353 gobuf SPEC src s (List []) pst1
354 PR.Continue 0 pst1 ->
355 gobuf SPEC xs s (List (x:getList backBuf)) pst1
356 PR.Continue n pst1 -> do
357 assert
358 (n <= sum (map Array.length (x:getList backBuf)))
359 (return ())
360 let (src0, buf1) = splitAtArrayListRev n (x:getList backBuf)
361 src = Prelude.reverse src0 ++ xs
362 gobuf SPEC src s (List buf1) pst1
363 PR.Done 0 b -> do
364 let str = D.append (D.fromList xs) (D.Stream step s)
365 return (Right b, str)
366 PR.Done n b -> do
367 assert
368 (n <= sum (map Array.length (x:getList backBuf)))
369 (return ())
370 let src0 = takeArrayListRev n (x:getList backBuf)
371 src = Prelude.reverse src0 ++ xs
372 return (Right b, D.append (D.fromList src) (D.Stream step s))
373 PR.Error err -> do
374 let src0 = x:getList backBuf
375 src = Prelude.reverse src0 ++ x:xs
376 strm = D.append (D.fromList src) (D.Stream step s)
377 return (Left (ParseError err), strm)
379 -- This is a simplified gobuf
380 goExtract _ [] backBuf !pst = goStop backBuf pst
381 goExtract _ (x:xs) backBuf !pst = do
382 pRes <- pstep pst x
383 case pRes of
384 PR.Partial 0 pst1 ->
385 goExtract SPEC xs (List []) pst1
386 PR.Partial n pst1 -> do
387 assert
388 (n <= sum (map Array.length (x:getList backBuf)))
389 (return ())
390 let src0 = takeArrayListRev n (x:getList backBuf)
391 src = Prelude.reverse src0 ++ xs
392 goExtract SPEC src (List []) pst1
393 PR.Continue 0 pst1 ->
394 goExtract SPEC xs (List (x:getList backBuf)) pst1
395 PR.Continue n pst1 -> do
396 assert
397 (n <= sum (map Array.length (x:getList backBuf)))
398 (return ())
399 let (src0, buf1) = splitAtArrayListRev n (x:getList backBuf)
400 src = Prelude.reverse src0 ++ xs
401 goExtract SPEC src (List buf1) pst1
402 PR.Done 0 b ->
403 return (Right b, D.fromList xs)
404 PR.Done n b -> do
405 assert
406 (n <= sum (map Array.length (x:getList backBuf)))
407 (return ())
408 let src0 = takeArrayListRev n (x:getList backBuf)
409 src = Prelude.reverse src0 ++ xs
410 return (Right b, D.fromList src)
411 PR.Error err -> do
412 let src0 = getList backBuf
413 src = Prelude.reverse src0 ++ x:xs
414 return (Left (ParseError err), D.fromList src)
416 -- This is a simplified goExtract
417 {-# INLINE goStop #-}
418 goStop backBuf pst = do
419 pRes <- extract pst
420 case pRes of
421 PR.Partial _ _ -> error "Bug: runArrayParserDBreak: Partial in extract"
422 PR.Continue 0 pst1 ->
423 goStop backBuf pst1
424 PR.Continue n pst1 -> do
425 assert
426 (n <= sum (map Array.length (getList backBuf)))
427 (return ())
428 let (src0, buf1) = splitAtArrayListRev n (getList backBuf)
429 src = Prelude.reverse src0
430 goExtract SPEC src (List buf1) pst1
431 PR.Done 0 b -> return (Right b, D.nil)
432 PR.Done n b -> do
433 assert
434 (n <= sum (map Array.length (getList backBuf)))
435 (return ())
436 let src0 = takeArrayListRev n (getList backBuf)
437 src = Prelude.reverse src0
438 return (Right b, D.fromList src)
439 PR.Error err -> do
440 let src0 = getList backBuf
441 src = Prelude.reverse src0
442 return (Left (ParseError err), D.fromList src)
444{-
445-- | Parse an array stream using the supplied 'Parser'. Returns the parse
446-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.
447--
448-- /Internal/
449--
450{-# INLINE parseArr #-}
451parseArr ::
452 (MonadIO m, MonadThrow m, Unbox a)
453 => ASF.Parser a m b
454 -> Stream m (A.Array a)
455 -> m (b, Stream m (A.Array a))
456parseArr p s = fmap fromStreamD <$> parseBreakD p (toStreamD s)
457-}
459-- | Fold an array stream using the supplied array stream 'Fold'.
460--
461-- /Pre-release/
462--
463{-# INLINE runArrayFold #-}
464runArrayFold :: (MonadIO m, Unbox a) =>
465 ChunkFold m a b -> StreamK m (A.Array a) -> m (Either ParseError b)
466runArrayFold (ChunkFold p) s = fst <$> runArrayParserDBreak p (toStream s)
468-- | Like 'fold' but also returns the remaining stream.
469--
470-- /Pre-release/
471--
472{-# INLINE runArrayFoldBreak #-}
473runArrayFoldBreak :: (MonadIO m, Unbox a) =>
474 ChunkFold m a b -> StreamK m (A.Array a) -> m (Either ParseError b, StreamK m (A.Array a))
475runArrayFoldBreak (ChunkFold p) s =
476 second fromStream <$> runArrayParserDBreak p (toStream s)
478{-# ANN type ParseChunksState Fuse #-}
479data ParseChunksState x inpBuf st pst =
480 ParseChunksInit inpBuf st
481 | ParseChunksInitBuf inpBuf
482 | ParseChunksInitLeftOver inpBuf
483 | ParseChunksStream st inpBuf !pst
484 | ParseChunksStop inpBuf !pst
485 | ParseChunksBuf inpBuf st inpBuf !pst
486 | ParseChunksExtract inpBuf inpBuf !pst
487 | ParseChunksYield x (ParseChunksState x inpBuf st pst)
489{-# INLINE_NORMAL runArrayFoldManyD #-}
490runArrayFoldManyD
491 :: (Monad m, Unbox a)
492 => ChunkFold m a b
493 -> D.Stream m (Array a)
494 -> D.Stream m (Either ParseError b)
495runArrayFoldManyD
496 (ChunkFold (PRD.Parser pstep initial extract)) (D.Stream step state) =
498 D.Stream stepOuter (ParseChunksInit [] state)
500 where
502 {-# INLINE_LATE stepOuter #-}
503 -- Buffer is empty, get the first element from the stream, initialize the
504 -- fold and then go to stream processing loop.
505 stepOuter gst (ParseChunksInit [] st) = do
506 r <- step (adaptState gst) st
507 case r of
508 D.Yield x s -> do
509 res <- initial
510 case res of
511 PRD.IPartial ps ->
512 return $ D.Skip $ ParseChunksBuf [x] s [] ps
513 PRD.IDone pb -> do
514 let next = ParseChunksInit [x] s
515 return $ D.Skip $ ParseChunksYield (Right pb) next
516 PRD.IError err -> do
517 let next = ParseChunksInitLeftOver []
518 return
519 $ D.Skip
520 $ ParseChunksYield (Left (ParseError err)) next
521 D.Skip s -> return $ D.Skip $ ParseChunksInit [] s
522 D.Stop -> return D.Stop
524 -- Buffer is not empty, go to buffered processing loop
525 stepOuter _ (ParseChunksInit src st) = do
526 res <- initial
527 case res of
528 PRD.IPartial ps ->
529 return $ D.Skip $ ParseChunksBuf src st [] ps
530 PRD.IDone pb ->
531 let next = ParseChunksInit src st
532 in return $ D.Skip $ ParseChunksYield (Right pb) next
533 PRD.IError err -> do
534 let next = ParseChunksInitLeftOver []
535 return
536 $ D.Skip
537 $ ParseChunksYield (Left (ParseError err)) next
539 -- This is a simplified ParseChunksInit
540 stepOuter _ (ParseChunksInitBuf src) = do
541 res <- initial
542 case res of
543 PRD.IPartial ps ->
544 return $ D.Skip $ ParseChunksExtract src [] ps
545 PRD.IDone pb ->
546 let next = ParseChunksInitBuf src
547 in return $ D.Skip $ ParseChunksYield (Right pb) next
548 PRD.IError err -> do
549 let next = ParseChunksInitLeftOver []
550 return
551 $ D.Skip
552 $ ParseChunksYield (Left (ParseError err)) next
554 -- XXX we just discard any leftover input at the end
555 stepOuter _ (ParseChunksInitLeftOver _) = return D.Stop
557 -- Buffer is empty, process elements from the stream
558 stepOuter gst (ParseChunksStream st backBuf pst) = do
559 r <- step (adaptState gst) st
560 case r of
561 D.Yield x s -> do
562 pRes <- pstep pst x
563 case pRes of
564 PR.Partial 0 pst1 ->
565 return $ D.Skip $ ParseChunksStream s [] pst1
566 PR.Partial n pst1 -> do
567 assert
568 (n <= sum (map Array.length (x:backBuf)))
569 (return ())
570 let src0 = takeArrayListRev n (x:backBuf)
571 src = Prelude.reverse src0
572 return $ D.Skip $ ParseChunksBuf src s [] pst1
573 PR.Continue 0 pst1 ->
574 return $ D.Skip $ ParseChunksStream s (x:backBuf) pst1
575 PR.Continue n pst1 -> do
576 assert
577 (n <= sum (map Array.length (x:backBuf)))
578 (return ())
579 let (src0, buf1) = splitAtArrayListRev n (x:backBuf)
580 src = Prelude.reverse src0
581 return $ D.Skip $ ParseChunksBuf src s buf1 pst1
582 PR.Done 0 b -> do
583 return $ D.Skip $
584 ParseChunksYield (Right b) (ParseChunksInit [] s)
585 PR.Done n b -> do
586 assert
587 (n <= sum (map Array.length (x:backBuf)))
588 (return ())
589 let src0 = takeArrayListRev n (x:backBuf)
590 src = Prelude.reverse src0
591 next = ParseChunksInit src s
592 return
593 $ D.Skip
594 $ ParseChunksYield (Right b) next
595 PR.Error err -> do
596 let next = ParseChunksInitLeftOver []
597 return
598 $ D.Skip
599 $ ParseChunksYield (Left (ParseError err)) next
601 D.Skip s -> return $ D.Skip $ ParseChunksStream s backBuf pst
602 D.Stop -> return $ D.Skip $ ParseChunksStop backBuf pst
604 -- go back to stream processing mode
605 stepOuter _ (ParseChunksBuf [] s buf pst) =
606 return $ D.Skip $ ParseChunksStream s buf pst
608 -- buffered processing loop
609 stepOuter _ (ParseChunksBuf (x:xs) s backBuf pst) = do
610 pRes <- pstep pst x
611 case pRes of
612 PR.Partial 0 pst1 ->
613 return $ D.Skip $ ParseChunksBuf xs s [] pst1
614 PR.Partial n pst1 -> do
615 assert (n <= sum (map Array.length (x:backBuf))) (return ())
616 let src0 = takeArrayListRev n (x:backBuf)
617 src = Prelude.reverse src0 ++ xs
618 return $ D.Skip $ ParseChunksBuf src s [] pst1
619 PR.Continue 0 pst1 ->
620 return $ D.Skip $ ParseChunksBuf xs s (x:backBuf) pst1
621 PR.Continue n pst1 -> do
622 assert (n <= sum (map Array.length (x:backBuf))) (return ())
623 let (src0, buf1) = splitAtArrayListRev n (x:backBuf)
624 src = Prelude.reverse src0 ++ xs
625 return $ D.Skip $ ParseChunksBuf src s buf1 pst1
626 PR.Done 0 b ->
627 return
628 $ D.Skip
629 $ ParseChunksYield (Right b) (ParseChunksInit xs s)
630 PR.Done n b -> do
631 assert (n <= sum (map Array.length (x:backBuf))) (return ())
632 let src0 = takeArrayListRev n (x:backBuf)
633 src = Prelude.reverse src0 ++ xs
634 return
635 $ D.Skip
636 $ ParseChunksYield (Right b) (ParseChunksInit src s)
637 PR.Error err -> do
638 let next = ParseChunksInitLeftOver []
639 return
640 $ D.Skip
641 $ ParseChunksYield (Left (ParseError err)) next
643 -- This is a simplified ParseChunksBuf
644 stepOuter _ (ParseChunksExtract [] buf pst) =
645 return $ D.Skip $ ParseChunksStop buf pst
647 stepOuter _ (ParseChunksExtract (x:xs) backBuf pst) = do
648 pRes <- pstep pst x
649 case pRes of
650 PR.Partial 0 pst1 ->
651 return $ D.Skip $ ParseChunksExtract xs [] pst1
652 PR.Partial n pst1 -> do
653 assert (n <= sum (map Array.length (x:backBuf))) (return ())
654 let src0 = takeArrayListRev n (x:backBuf)
655 src = Prelude.reverse src0 ++ xs
656 return $ D.Skip $ ParseChunksExtract src [] pst1
657 PR.Continue 0 pst1 ->
658 return $ D.Skip $ ParseChunksExtract xs (x:backBuf) pst1
659 PR.Continue n pst1 -> do
660 assert (n <= sum (map Array.length (x:backBuf))) (return ())
661 let (src0, buf1) = splitAtArrayListRev n (x:backBuf)
662 src = Prelude.reverse src0 ++ xs
663 return $ D.Skip $ ParseChunksExtract src buf1 pst1
664 PR.Done 0 b ->
665 return
666 $ D.Skip
667 $ ParseChunksYield (Right b) (ParseChunksInitBuf xs)
668 PR.Done n b -> do
669 assert (n <= sum (map Array.length (x:backBuf))) (return ())
670 let src0 = takeArrayListRev n (x:backBuf)
671 src = Prelude.reverse src0 ++ xs
672 return
673 $ D.Skip
674 $ ParseChunksYield (Right b) (ParseChunksInitBuf src)
675 PR.Error err -> do
676 let next = ParseChunksInitLeftOver []
677 return
678 $ D.Skip
679 $ ParseChunksYield (Left (ParseError err)) next
682 -- This is a simplified ParseChunksExtract
683 stepOuter _ (ParseChunksStop backBuf pst) = do
684 pRes <- extract pst
685 case pRes of
686 PR.Partial _ _ -> error "runArrayFoldManyD: Partial in extract"
687 PR.Continue 0 pst1 ->
688 return $ D.Skip $ ParseChunksStop backBuf pst1
689 PR.Continue n pst1 -> do
690 assert (n <= sum (map Array.length backBuf)) (return ())
691 let (src0, buf1) = splitAtArrayListRev n backBuf
692 src = Prelude.reverse src0
693 return $ D.Skip $ ParseChunksExtract src buf1 pst1
694 PR.Done 0 b ->
695 return
696 $ D.Skip
697 $ ParseChunksYield (Right b) (ParseChunksInitLeftOver [])
698 PR.Done n b -> do
699 assert (n <= sum (map Array.length backBuf)) (return ())
700 let src0 = takeArrayListRev n backBuf
701 src = Prelude.reverse src0
702 return
703 $ D.Skip
704 $ ParseChunksYield (Right b) (ParseChunksInitBuf src)
705 PR.Error err -> do
706 let next = ParseChunksInitLeftOver []
707 return
708 $ D.Skip
709 $ ParseChunksYield (Left (ParseError err)) next
711 stepOuter _ (ParseChunksYield a next) = return $ D.Yield a next
713-- | Apply an 'ChunkFold' repeatedly on an array stream and emit the
714-- fold outputs in the output stream.
715--
716-- See "Streamly.Data.Stream.foldMany" for more details.
717--
718-- /Pre-release/
719{-# INLINE runArrayFoldMany #-}
720runArrayFoldMany
721 :: (Monad m, Unbox a)
722 => ChunkFold m a b
723 -> StreamK m (Array a)
724 -> StreamK m (Either ParseError b)
725runArrayFoldMany p m = fromStream $ runArrayFoldManyD p (toStream m)