@@ -246,6 +246,55 @@ compactLEParserD n = ParserD.Parser step initial extract
246246 functionPath =
247247 "Streamly.Internal.Data.Array.Stream.Mut.Foreign.compactLEParserD"
248248
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.
254+--
255+-- /Internal/
256+{-# INLINE_NORMAL compactGEFold #-}
257+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
261+
262+ where
263+
264+ nBytes = n * sizeOf (undefined :: a)
265+
266+ initial =
267+ return
268+ $ if n < 0
269+ then error
270+ $ functionPath
271+ ++ ": the size of arrays ["
272+ ++ show n ++ "] must be a natural number"
273+ else FL.Partial Nothing
274+
275+ step Nothing arr =
276+ return
277+ $ let len = MArray.byteLength arr
278+ in if len >= nBytes
279+ then FL.Done arr
280+ else FL.Partial (Just arr)
281+ step (Just buf) arr = do
282+ let len = MArray.byteLength buf + MArray.byteLength arr
283+ buf1 <-
284+ if MArray.byteCapacity buf < len
285+ then liftIO $ MArray.realloc (max len nBytes) buf
286+ else return buf
287+ buf2 <- MArray.splice buf1 arr
288+ if len >= n
289+ then return $ FL.Done buf2
290+ else return $ FL.Partial (Just buf2)
291+
292+ extract Nothing = return MArray.nil
293+ extract (Just buf) = return buf
294+
295+ functionPath =
296+ "Streamly.Internal.Data.Array.Stream.Mut.Foreign.compactGEFold"
297+
249298-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
250299-- maximum specified size in bytes.
251300--
@@ -268,9 +317,10 @@ compactEQ _n _xs = undefined
268317-- | Like 'compactLE' but generates arrays of size greater than or equal to the
269318-- specified except for the last array in the stream which could be shorter.
270319--
271−-- /Unimplemented/
320+-- /Internal/
272321{-# INLINE compactGE #-}
273−compactGE :: -- (MonadIO m, Storable a) =>
274− Int -> SerialT m (Array a) -> SerialT m (Array a)
275−compactGE _n _xs = undefined
276− -- IsStream.fromStreamD $ D.foldMany (compactGEFold n) (IsStream.toStreamD xs)
322+compactGE ::
323+ (MonadIO m, Storable a)
324+ => Int -> SerialT m (Array a) -> SerialT m (Array a)
325+compactGE n (SerialT xs) =
326+ SerialT $ D.toStreamK $ D.foldMany (compactGEFold n) (D.fromStreamK xs)