@@ -246,6 +246,55 @@ compactLEParserD n = ParserD.Parser step initial extract
246 functionPath =
246 functionPath =
247 "Streamly.Internal.Data.Array.Stream.Mut.Foreign.compactLEParserD"
247 "Streamly.Internal.Data.Array.Stream.Mut.Foreign.compactLEParserD"
249+-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
250+-- minimum specified size. Note that if all the arrays in the stream together
251+-- are smaller than the specified size the resulting array will be smaller than
252+-- the specified size. When we coalesce multiple arrays if the size would exceed
253+-- the specified size we stop coalescing further.
256+{-# INLINE_NORMAL compactGEFold #-}
258+ forall m a. (MonadIO m, Storable a)
259+ => Int -> FL.Fold m (Array a) (Array a)
260+compactGEFold n = Fold step initial extract
264+ nBytes = n * sizeOf (undefined :: a)
271+ ++ ": the size of arrays ["
272+ ++ show n ++ "] must be a natural number"
273+ else FL.Partial Nothing
277+ $ let len = MArray.byteLength arr
280+ else FL.Partial (Just arr)
281+ step (Just buf) arr = do
282+ let len = MArray.byteLength buf + MArray.byteLength arr
284+ if MArray.byteCapacity buf < len
285+ then liftIO $ MArray.realloc (max len nBytes) buf
287+ buf2 <- MArray.splice buf1 arr
289+ then return $ FL.Done buf2
290+ else return $ FL.Partial (Just buf2)
292+ extract Nothing = return MArray.nil
293+ extract (Just buf) = return buf
296+ "Streamly.Internal.Data.Array.Stream.Mut.Foreign.compactGEFold"
249-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
298-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
250-- maximum specified size in bytes.
299-- maximum specified size in bytes.
@@ -268,9 +317,10 @@ compactEQ _n _xs = undefined
268-- | Like 'compactLE' but generates arrays of size greater than or equal to the
317-- | Like 'compactLE' but generates arrays of size greater than or equal to the
269-- specified except for the last array in the stream which could be shorter.
318-- specified except for the last array in the stream which could be shorter.
271−-- /Unimplemented/
320+-- /Internal/
272{-# INLINE compactGE #-}
321{-# INLINE compactGE #-}
273−compactGE :: -- (MonadIO m, Storable a) =>
323+ (MonadIO m, Storable a)
274− Int -> SerialT m (Array a) -> SerialT m (Array a)
324+ => Int -> SerialT m (Array a) -> SerialT m (Array a)
275−compactGE _n _xs = undefined
325+compactGE n (SerialT xs) =
276− -- IsStream.fromStreamD $ D.foldMany (compactGEFold n) (IsStream.toStreamD xs)
326+ SerialT $ D.toStreamK $ D.foldMany (compactGEFold n) (D.fromStreamK xs)