@@ -19,13 +19,13 @@ module Streamly.Internal.Data.Array.Stream
1919 , Array.bufferChunks
2020
2121 -- * Flattening to elements
22− , concat
22+ , Array.concat
2323 , Array.flattenArrays
24− , concatRev
24+ , Array.concatRev
2525 , Array.flattenArraysRev
26− , interpose
27− , interposeSuffix
28− , intercalateSuffix
26+ , Array.interpose
27+ , Array.interposeSuffix
28+ , Array.intercalateSuffix
2929 , unlines
3030
3131 -- * Elimination
@@ -90,7 +90,6 @@ import Prelude hiding (null, last, (!!), read, concat, unlines)
9090
9191import Streamly.Data.Fold (Fold)
9292import Streamly.Internal.Data.Array.Type (Array(..))
93−import Streamly.Internal.Data.MutByteArray.Type (MutByteArray)
9493import Streamly.Internal.Data.Fold.Chunked (ChunkFold(..))
9594import Streamly.Internal.Data.Parser (ParseError(..))
9695import Streamly.Internal.Data.Stream (Stream)
@@ -98,11 +97,8 @@ import Streamly.Internal.Data.StreamK (StreamK, fromStream, toStream)
9897import Streamly.Internal.Data.SVar.Type (adaptState, defState)
9998import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
10099
101−import qualified Streamly.Data.Fold as FL
102100import qualified Streamly.Internal.Data.Array as A
103101import qualified Streamly.Internal.Data.Array as Array
104−import qualified Streamly.Internal.Data.MutArray.Type as MA
105−import qualified Streamly.Internal.Data.MutArray.Stream as AS
106102import qualified Streamly.Internal.Data.Fold.Type as FL (Fold(..), Step(..))
107103import qualified Streamly.Internal.Data.Parser as PR
108104import qualified Streamly.Internal.Data.Parser as PRD
@@ -114,110 +110,19 @@ import qualified Streamly.Internal.Data.StreamK as K
114110-- XXX Since these are immutable arrays MonadIO constraint can be removed from
115111-- most places.
116112
117−-------------------------------------------------------------------------------
118−-- Append
119−-------------------------------------------------------------------------------
120−
121−-- XXX efficiently compare two streams of arrays. Two streams can have chunks
122−-- of different sizes, we can handle that in the stream comparison abstraction.
123−-- This could be useful e.g. to fast compare whether two files differ.
124−
125−-- | Convert a stream of arrays into a stream of their elements.
126−--
127−-- Same as the following:
128−--
129−-- > concat = Stream.unfoldMany Array.read
130−--
131−-- @since 0.7.0
132−{-# INLINE concat #-}
133−concat :: (Monad m, Unbox a) => Stream m (Array a) -> Stream m a
134−-- concat m = fromStreamD $ A.flattenArrays (toStreamD m)
135−-- concat m = fromStreamD $ D.concatMap A.toStreamD (toStreamD m)
136−concat = D.unfoldMany A.reader
137−
138−-- | Convert a stream of arrays into a stream of their elements reversing the
139−-- contents of each array before flattening.
140−--
141−-- > concatRev = Stream.unfoldMany Array.readerRev
142−--
143−-- @since 0.7.0
144−{-# INLINE concatRev #-}
145−concatRev :: (Monad m, Unbox a) => Stream m (Array a) -> Stream m a
146−-- concatRev m = fromStreamD $ A.flattenArraysRev (toStreamD m)
147−concatRev = D.unfoldMany A.readerRev
148−
149113-------------------------------------------------------------------------------
150114-- Intersperse and append
151115-------------------------------------------------------------------------------
152116
153−-- | Flatten a stream of arrays after inserting the given element between
154−-- arrays.
155−--
156−-- /Pre-release/
157−{-# INLINE interpose #-}
158−interpose :: (Monad m, Unbox a) => a -> Stream m (Array a) -> Stream m a
159−interpose x = D.interpose x A.reader
160−
161−{-# INLINE intercalateSuffix #-}
162−intercalateSuffix :: (Monad m, Unbox a)
163− => Array a -> Stream m (Array a) -> Stream m a
164−intercalateSuffix = D.intercalateSuffix A.reader
165−
166−-- | Flatten a stream of arrays appending the given element after each
167−-- array.
168−--
169−-- @since 0.7.0
170−{-# INLINE interposeSuffix #-}
171−interposeSuffix :: (Monad m, Unbox a)
172− => a -> Stream m (Array a) -> Stream m a
173−-- interposeSuffix x = fromStreamD . A.unlines x . toStreamD
174−interposeSuffix x = D.interposeSuffix x A.reader
175−
176−data FlattenState s =
177− OuterLoop s
178− | InnerLoop s !MutByteArray !Int !Int
179−
180−-- XXX This is a special case of interposeSuffix, can be removed.
181−-- XXX Remove monadIO constraint
182117{-# INLINE_NORMAL unlines #-}
183118unlines :: forall m a. (MonadIO m, Unbox a)
184119 => a -> D.Stream m (Array a) -> D.Stream m a
185−unlines sep (D.Stream step state) = D.Stream step' (OuterLoop state)
186− where
187− {-# INLINE_LATE step' #-}
188− step' gst (OuterLoop st) = do
189− r <- step (adaptState gst) st
190− return $ case r of
191− D.Yield Array{..} s ->
192− D.Skip (InnerLoop s arrContents arrStart arrEnd)
193− D.Skip s -> D.Skip (OuterLoop s)
194− D.Stop -> D.Stop
195−
196− step' _ (InnerLoop st _ p end) | p == end =
197− return $ D.Yield sep $ OuterLoop st
198−
199− step' _ (InnerLoop st contents p end) = do
200− x <- liftIO $ peekByteIndex p contents
201− return $ D.Yield x (InnerLoop st contents (INDEX_NEXT(p,a)) end)
120+unlines = Array.interposeSuffix
202121
203122-------------------------------------------------------------------------------
204123-- Compact
205124-------------------------------------------------------------------------------
206125
207−-- XXX These would not be needed once we implement compactLEFold, see
208−-- module Streamly.Internal.Data.Stream.MutChunked
209−--
210−-- XXX Note that this thaws immutable arrays for appending, that may be
211−-- problematic if multiple users do the same thing, however, immutable arrays
212−-- would usually have no capacity to append, therefore, a copy will be forced
213−-- anyway. Confirm this. We can forcefully trim the array capacity before thaw
214−-- to ensure this.
215−{-# INLINE_NORMAL packArraysChunksOf #-}
216−packArraysChunksOf :: (MonadIO m, Unbox a)
217− => Int -> D.Stream m (Array a) -> D.Stream m (Array a)
218−packArraysChunksOf n str =
219− D.map A.unsafeFreeze $ AS.packArraysChunksOf n $ D.map A.unsafeThaw str
220−
221126-- XXX instead of writing two different versions of this operation, we should
222127-- write it as a pipe.
223128--
@@ -225,8 +130,7 @@ packArraysChunksOf n str =
225130{-# INLINE_NORMAL lpackArraysChunksOf #-}
226131lpackArraysChunksOf :: (MonadIO m, Unbox a)
227132 => Int -> Fold m (Array a) () -> Fold m (Array a) ()
228−lpackArraysChunksOf n fld =
229− FL.lmap A.unsafeThaw $ AS.lpackArraysChunksOf n (FL.lmap A.unsafeFreeze fld)
133+lpackArraysChunksOf = Array.lCompactGE
230134
231135-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
232136-- maximum specified size in bytes.
@@ -235,7 +139,7 @@ lpackArraysChunksOf n fld =
235139{-# INLINE compact #-}
236140compact :: (MonadIO m, Unbox a)
237141 => Int -> Stream m (Array a) -> Stream m (Array a)
238−compact = packArraysChunksOf
142+compact = Array.compactLE
239143
240144-- | Given a stream of arrays, splice them all together to generate a single
241145-- array. The stream must be /finite/.
@@ -243,71 +147,12 @@ compact = packArraysChunksOf
243147-- @since 0.7.0
244148{-# INLINE toArray #-}
245149toArray :: (MonadIO m, Unbox a) => Stream m (Array a) -> m (Array a)
246−toArray s =
247− fmap A.unsafeFreeze $ MA.fromChunksRealloced (fmap A.unsafeThaw s)
150+toArray = Array.fromChunks
248151
249152-------------------------------------------------------------------------------
250153-- Split
251154-------------------------------------------------------------------------------
252155
253−data SplitState s arr
254− = Initial s
255− | Buffering s arr
256− | Splitting s arr
257− | Yielding arr (SplitState s arr)
258− | Finishing
259−
260−-- | Split a stream of arrays on a given separator byte, dropping the separator
261−-- and coalescing all the arrays between two separators into a single array.
262−--
263−-- @since 0.7.0
264−{-# INLINE_NORMAL _splitOn #-}
265−_splitOn
266− :: MonadIO m
267− => Word8
268− -> D.Stream m (Array Word8)
269− -> D.Stream m (Array Word8)
270−_splitOn byte (D.Stream step state) = D.Stream step' (Initial state)
271−
272− where
273−
274− {-# INLINE_LATE step' #-}
275− step' gst (Initial st) = do
276− r <- step gst st
277− case r of
278− D.Yield arr s -> do
279− (arr1, marr2) <- A.breakOn byte arr
280− return $ case marr2 of
281− Nothing -> D.Skip (Buffering s arr1)
282− Just arr2 -> D.Skip (Yielding arr1 (Splitting s arr2))
283− D.Skip s -> return $ D.Skip (Initial s)
284− D.Stop -> return D.Stop
285−
286− step' gst (Buffering st buf) = do
287− r <- step gst st
288− case r of
289− D.Yield arr s -> do
290− (arr1, marr2) <- A.breakOn byte arr
291− -- XXX Should use MutArray.splice
292− buf' <- A.splice buf arr1
293− return $ case marr2 of
294− Nothing -> D.Skip (Buffering s buf')
295− Just x -> D.Skip (Yielding buf' (Splitting s x))
296− D.Skip s -> return $ D.Skip (Buffering s buf)
297− D.Stop -> return $
298− if A.byteLength buf == 0
299− then D.Stop
300− else D.Skip (Yielding buf Finishing)
301−
302− step' _ (Splitting st buf) = do
303− (arr1, marr2) <- A.breakOn byte buf
304− return $ case marr2 of
305− Nothing -> D.Skip $ Buffering st arr1
306− Just arr2 -> D.Skip $ Yielding arr1 (Splitting st arr2)
307−
308− step' _ (Yielding arr next) = return $ D.Yield arr next
309− step' _ Finishing = return D.Stop
310−
311156-- XXX Remove MonadIO constraint.
312157-- | Split a stream of arrays on a given separator byte, dropping the separator
313158-- and coalescing all the arrays between two separators into a single array.
@@ -319,11 +164,7 @@ splitOn
319164 => Word8
320165 -> Stream m (Array Word8)
321166 -> Stream m (Array Word8)
322−splitOn byte =
323− fmap A.unsafeFreeze
324− -- XXX use spliceExp and rightSize?
325− . D.splitInnerBy (MA.breakOn byte) MA.splice
326− . fmap A.unsafeThaw
167+splitOn = Array.compactOnByte
327168
328169{-# INLINE splitOnSuffix #-}
329170splitOnSuffix
@@ -331,13 +172,7 @@ splitOnSuffix
331172 => Word8
332173 -> Stream m (Array Word8)
333174 -> Stream m (Array Word8)
334−-- splitOn byte s = fromStreamD $ A.splitOn byte $ toStreamD s
335−splitOnSuffix byte =
336− fmap A.unsafeFreeze
337− -- XXX use spliceExp and rightSize?
338− . D.splitInnerBySuffix
339− (\arr -> MA.byteLength arr == 0) (MA.breakOn byte) MA.splice
340− . fmap A.unsafeThaw
175+splitOnSuffix = Array.compactOnByteSuffix
341176
342177-------------------------------------------------------------------------------
343178-- Elimination - Running folds