git.iko.soy/streamly
on master help sign in
Move Array.Stream.Foreign to streamly-core (#1863)
commitaef0a2e793559abf4fbc1857fcbed39ad79d167d authorRanjeet Ranjan <ranjeetr@composewell.com> dateTue 27 Sep 2022, 13:38 (4 years ago) parenta519c85
7 files changed+138−140
compare with master
unifiedsplit
src/Streamly/Internal/Data/Array/Stream/Foreign.hs → core/src/Streamly/Internal/Data/Array/Stream/Foreign.hs+39−40 core/streamly-core.cabal+2 src/Streamly/Internal/FileSystem/FD.hs+33−32 src/Streamly/Internal/FileSystem/Handle.hs+31−32 src/Streamly/Internal/Network/Inet/TCP.hs+9−10 src/Streamly/Internal/Network/Socket.hs+23−24 streamly.cabal+1−2
src/Streamly/Internal/Data/Array/Stream/Foreign.hs → core/src/Streamly/Internal/Data/Array/Stream/Foreign.hsrenamed+39−40
@@ -69,9 +69,9 @@ import Streamly.Data.Stream (Stream)
6969import Streamly.Internal.Data.Array.Unboxed.Type (Array(..))
7070import Streamly.Internal.Data.Array.Stream.Fold.Foreign (ArrayFold(..))
7171import Streamly.Internal.Data.Parser (ParseError(..))
72−import Streamly.Internal.Data.Stream.IsStream.Type
73− (IsStream, fromStreamD, toStreamD)
74−import Streamly.Internal.Data.SVar (adaptState, defState)
72+import Streamly.Internal.Data.Stream
73+ (fromStreamD, fromStreamK, toStreamD, toStreamK)
74+import Streamly.Internal.Data.SVar.Type (adaptState, defState)
7575import Streamly.Internal.Data.Array.Unboxed.Mut.Type
7676 (allocBytesToElemCount)
7777import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
@@ -79,27 +79,22 @@ import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
7979import qualified Streamly.Data.Fold as FL
8080import qualified Streamly.Internal.Data.Array.Unboxed as A
8181import qualified Streamly.Internal.Data.Array.Unboxed as Array
82−-- import qualified Streamly.Internal.Data.Array.Stream.Fold.Foreign
83−-- as ArrayFold
8482import qualified Streamly.Internal.Data.Array.Unboxed.Type as A
8583import qualified Streamly.Internal.Data.Array.Unboxed.Mut.Type as MA
8684import qualified Streamly.Internal.Data.Array.Stream.Mut.Foreign as AS
87−import qualified Streamly.Internal.Data.Fold.Type as FL
88− (Fold(..), Step(..))
85+import qualified Streamly.Internal.Data.Fold.Type as FL (Fold(..), Step(..))
8986import qualified Streamly.Internal.Data.Parser as PR
9087import qualified Streamly.Internal.Data.Parser.ParserD as PRD
9188 (Parser(..), Initial(..), fromParserK)
92−import qualified Streamly.Internal.Data.Stream.IsStream as S
89+import qualified Streamly.Internal.Data.Stream as S
9390import qualified Streamly.Internal.Data.Stream.StreamD as D
9491 ( fromList, nil, cons, map
95− , unfoldMany, append, splitInnerBy, splitInnerBySuffix
92+ , unfoldMany, append, splitInnerBy, splitInnerBySuffix, foldlM'
9693 )
9794import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
9895 (Step(Yield, Stop, Skip), Stream(Stream))
9996import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
10097 (Stream, cons, nil, fromPure, foldStream)
101−import qualified Streamly.Internal.Data.Stream as Stream
102− (fromStreamK, toStreamK)
10398
10499-- XXX Since these are immutable arrays MonadIO constraint can be removed from
105100-- most places.
@@ -115,8 +110,8 @@ import qualified Streamly.Internal.Data.Stream as Stream
115110--
116111-- /Pre-release/
117112{-# INLINE arraysOf #-}
118−arraysOf :: (IsStream t, MonadIO m, Unboxed a)
119− => Int -> t m a -> t m (Array a)
113+arraysOf :: (MonadIO m, Unboxed a)
114+ => Int -> Stream m a -> Stream m (Array a)
120115arraysOf n str = fromStreamD $ A.arraysOf n (toStreamD str)
121116
122117-------------------------------------------------------------------------------
@@ -135,7 +130,7 @@ arraysOf n str = fromStreamD $ A.arraysOf n (toStreamD str)
135130--
136131-- @since 0.7.0
137132{-# INLINE concat #-}
138−concat :: (IsStream t, Monad m, Unboxed a) => t m (Array a) -> t m a
133+concat :: (Monad m, Unboxed a) => Stream m (Array a) -> Stream m a
139134-- concat m = fromStreamD $ A.flattenArrays (toStreamD m)
140135-- concat m = fromStreamD $ D.concatMap A.toStreamD (toStreamD m)
141136concat m = fromStreamD $ D.unfoldMany A.read (toStreamD m)
@@ -147,7 +142,7 @@ concat m = fromStreamD $ D.unfoldMany A.read (toStreamD m)
147142--
148143-- @since 0.7.0
149144{-# INLINE concatRev #-}
150−concatRev :: (IsStream t, Monad m, Unboxed a) => t m (Array a) -> t m a
145+concatRev :: (Monad m, Unboxed a) => Stream m (Array a) -> Stream m a
151146-- concatRev m = fromStreamD $ A.flattenArraysRev (toStreamD m)
152147concatRev m = fromStreamD $ D.unfoldMany A.readRev (toStreamD m)
153148
@@ -160,12 +155,12 @@ concatRev m = fromStreamD $ D.unfoldMany A.readRev (toStreamD m)
160155--
161156-- /Pre-release/
162157{-# INLINE interpose #-}
163−interpose :: (Monad m, IsStream t, Unboxed a) => a -> t m (Array a) -> t m a
158+interpose :: (Monad m, Unboxed a) => a -> Stream m (Array a) -> Stream m a
164159interpose x = S.interpose x A.read
165160
166161{-# INLINE intercalateSuffix #-}
167−intercalateSuffix :: (Monad m, IsStream t, Unboxed a)
168− => Array a -> t m (Array a) -> t m a
162+intercalateSuffix :: (Monad m, Unboxed a)
163+ => Array a -> Stream m (Array a) -> Stream m a
169164intercalateSuffix = S.intercalateSuffix A.read
170165
171166-- | Flatten a stream of arrays appending the given element after each
@@ -173,8 +168,8 @@ intercalateSuffix = S.intercalateSuffix A.read
173168--
174169-- @since 0.7.0
175170{-# INLINE interposeSuffix #-}
176−interposeSuffix :: (Monad m, IsStream t, Unboxed a)
177− => a -> t m (Array a) -> t m a
171+interposeSuffix :: (Monad m, Unboxed a)
172+ => a -> Stream m (Array a) -> Stream m a
178173-- interposeSuffix x = fromStreamD . A.unlines x . toStreamD
179174interposeSuffix x = S.interposeSuffix x A.read
180175
@@ -302,19 +297,19 @@ _splitOn byte (D.Stream step state) = D.Stream step' (Initial state)
302297-- @since 0.7.0
303298{-# INLINE splitOn #-}
304299splitOn
305− :: (IsStream t, MonadIO m)
300+ :: (MonadIO m)
306301 => Word8
307− -> t m (Array Word8)
308− -> t m (Array Word8)
302+ -> Stream m (Array Word8)
303+ -> Stream m (Array Word8)
309304splitOn byte s =
310305 fromStreamD $ D.splitInnerBy (A.breakOn byte) A.splice $ toStreamD s
311306
312307{-# INLINE splitOnSuffix #-}
313308splitOnSuffix
314− :: (IsStream t, MonadIO m)
309+ :: (MonadIO m)
315310 => Word8
316− -> t m (Array Word8)
317− -> t m (Array Word8)
311+ -> Stream m (Array Word8)
312+ -> Stream m (Array Word8)
318313-- splitOn byte s = fromStreamD $ A.splitOn byte $ toStreamD s
319314splitOnSuffix byte s =
320315 fromStreamD $ D.splitInnerBySuffix (A.breakOn byte) A.splice $ toStreamD s
@@ -410,9 +405,9 @@ foldBreak ::
410405 -> m (b, Stream m (A.Array a))
411406-- foldBreak f s = fmap fromStreamD <$> foldBreakD f (toStreamD s)
412407foldBreak f =
413− fmap (fmap Stream.fromStreamK)
408+ fmap (fmap fromStreamK)
414409 . foldBreakK f
415− . Stream.toStreamK
410+ . toStreamK
416411-- If foldBreak performs better than runArrayFoldBreak we can use a rewrite
417412-- rule to rewrite runArrayFoldBreak to fold.
418413-- foldBreak f = runArrayFoldBreak (ArrayFold.fromFold f)
@@ -468,6 +463,10 @@ splitAtArrayListRev n ls
468463-- Fold to a single Array
469464-------------------------------------------------------------------------------
470465
466+{-# INLINE foldlM' #-}
467+foldlM' :: Monad m => (b -> a -> m b) -> m b -> Stream m a -> m b
468+foldlM' step begin = D.foldlM' step begin . S.toStreamD
469+
471470-- XXX Both of these implementations of splicing seem to perform equally well.
472471-- We need to perform benchmarks over a range of sizes though.
473472
@@ -478,16 +477,16 @@ spliceArraysLenUnsafe :: (MonadIO m, Unboxed a)
478477 => Int -> Stream m (MA.Array a) -> m (MA.Array a)
479478spliceArraysLenUnsafe len buffered = do
480479 arr <- liftIO $ MA.newArray len
481− S.foldlM' MA.spliceUnsafe (return arr) buffered
480+ foldlM' MA.spliceUnsafe (return arr) buffered
482481
483482{-# INLINE _spliceArrays #-}
484483_spliceArrays :: (MonadIO m, Unboxed a)
485484 => Stream m (Array a) -> m (Array a)
486485_spliceArrays s = do
487486 buffered <- S.foldr S.cons S.nil s
488− len <- S.sum (S.map Array.length buffered)
487+ len <- S.fold FL.sum (fmap Array.length buffered)
489488 arr <- liftIO $ MA.newArray len
490− final <- S.foldlM' writeArr (return arr) s
489+ final <- foldlM' writeArr (return arr) s
491490 return $ A.unsafeFreeze final
492491
493492 where
@@ -499,8 +498,8 @@ _spliceArraysBuffered :: (MonadIO m, Unboxed a)
499498 => Stream m (Array a) -> m (Array a)
500499_spliceArraysBuffered s = do
501500 buffered <- S.foldr S.cons S.nil s
502− len <- S.sum (S.map Array.length buffered)
503− A.unsafeFreeze <$> spliceArraysLenUnsafe len (S.map A.unsafeThaw s)
501+ len <- S.fold FL.sum (fmap Array.length buffered)
502+ A.unsafeFreeze <$> spliceArraysLenUnsafe len (fmap A.unsafeThaw s)
504503
505504{-# INLINE spliceArraysRealloced #-}
506505spliceArraysRealloced :: forall m a. (MonadIO m, Unboxed a)
@@ -509,7 +508,7 @@ spliceArraysRealloced s = do
509508 let n = allocBytesToElemCount (undefined :: a) (4 * 1024)
510509 idst = liftIO $ MA.newArray n
511510
512− arr <- S.foldlM' MA.spliceExp idst (S.map A.unsafeThaw s)
511+ arr <- foldlM' MA.spliceExp idst (fmap A.unsafeThaw s)
513512 liftIO $ A.unsafeFreeze <$> MA.rightSize arr
514513
515514-- XXX This should just be "fold A.write"
@@ -530,7 +529,7 @@ toArray = spliceArraysRealloced
530529--
531530{-
532531{-# INLINE toArraysInRange #-}
533−toArraysInRange :: (IsStream t, MonadIO m, Unboxed a)
532+toArraysInRange :: (MonadIO m, Unboxed a)
534533 => Int -> Int -> Fold m (Array a) b -> Fold m a b
535534toArraysInRange low high (Fold step initial extract) =
536535-}
@@ -785,9 +784,9 @@ parseBreak p s =
785784 fmap fromStreamD <$> parseBreakD (PRD.fromParserK p) (toStreamD s)
786785-}
787786parseBreak p =
788− fmap (fmap Stream.fromStreamK)
787+ fmap (fmap fromStreamK)
789788 . parseBreakK (PRD.fromParserK p)
790− . Stream.toStreamK
789+ . toStreamK
791790
792791-------------------------------------------------------------------------------
793792-- Elimination - Running Array Folds and parsers
@@ -1155,8 +1154,8 @@ runArrayFoldManyD
11551154-- /Pre-release/
11561155{-# INLINE runArrayFoldMany #-}
11571156runArrayFoldMany
1158− :: (IsStream t, MonadThrow m, Unboxed a)
1157+ :: (MonadThrow m, Unboxed a)
11591158 => ArrayFold m a b
1160− -> t m (Array a)
1161− -> t m b
1159+ -> Stream m (Array a)
1160+ -> Stream m b
11621161runArrayFoldMany p m = fromStreamD $ runArrayFoldManyD p (toStreamD m)
core/streamly-core.cabal+2
@@ -325,6 +325,8 @@ library
325325 , Streamly.Internal.Data.Array.Stream.Mut.Foreign
326326 , Streamly.Internal.Data.Array.Stream.Fold.Foreign
327327
328+ , Streamly.Internal.Data.Array.Stream.Foreign
329+
328330 -- Ring Arrays
329331 , Streamly.Internal.Data.Ring.Foreign
330332 , Streamly.Internal.Data.Ring
src/Streamly/Internal/FileSystem/FD.hs+33−32
@@ -132,8 +132,6 @@ import Streamly.Internal.Data.Array.Unboxed.Type
132132 (Array(..), byteLength, unsafeFreeze, asPtrUnsafe)
133133import Streamly.Internal.System.IO (defaultChunkSize)
134134import Streamly.Internal.Data.Stream.Type (Stream)
135−import Streamly.Internal.Data.Stream.IsStream.Type
136− (IsStream, mkStream, fromStreamD)
137135#if !defined(mingw32_HOST_OS)
138136{-
139137import Streamly.Internal.Data.Stream.IsStream.Type (toStreamD)
@@ -146,12 +144,15 @@ import qualified Streamly.Internal.System.IOVec.Type as RawIO
146144-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
147145
148146import qualified Streamly.Data.Array.Unboxed as A
147+import qualified Streamly.Data.Fold as FL
149148import qualified Streamly.Internal.Data.Array.Unboxed.Mut as MArray
150149 (Array(..), newPinnedArrayBytes, asPtrUnsafe)
151150import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS
152−import qualified Streamly.Prelude as S
151+import qualified Streamly.Internal.Data.Stream as S
153152import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
154153 (Stream(..), Step(..))
154+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K (mkStream)
155+
155156
156157-------------------------------------------------------------------------------
157158-- References
@@ -281,21 +282,21 @@ writeIOVec (Handle fd) iov =
281282-- The maximum size of a single array is specified by @size@. The actual size
282283-- read may be less than or equal to @size@.
283284{-# INLINE _readArraysOfUpto #-}
284−_readArraysOfUpto :: (IsStream t, MonadIO m)
285− => Int -> Handle -> t m (Array Word8)
286−_readArraysOfUpto size h = go
285+_readArraysOfUpto :: (MonadIO m)
286+ => Int -> Handle -> Stream m (Array Word8)
287+_readArraysOfUpto size h = S.fromStreamK go
287288 where
288289 -- XXX use cons/nil instead
289− go = mkStream $ \_ yld _ stp -> do
290+ go = K.mkStream $ \_ yld _ stp -> do
290291 arr <- liftIO $ readArrayUpto size h
291292 if A.length arr == 0
292293 then stp
293294 else yld arr go
294295
295296{-# INLINE_NORMAL readArraysOfUpto #-}
296−readArraysOfUpto :: (IsStream t, MonadIO m)
297− => Int -> Handle -> t m (Array Word8)
298−readArraysOfUpto size h = fromStreamD (D.Stream step ())
297+readArraysOfUpto :: (MonadIO m)
298+ => Int -> Handle -> Stream m (Array Word8)
299+readArraysOfUpto size h = S.fromStreamD (D.Stream step ())
299300 where
300301 {-# INLINE_LATE step #-}
301302 step _ _ = do
@@ -316,7 +317,7 @@ readArraysOfUpto size h = fromStreamD (D.Stream step ())
316317--
317318-- @since 0.7.0
318319{-# INLINE readArrays #-}
319−readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
320+readArrays :: (MonadIO m) => Handle -> Stream m (Array Word8)
320321readArrays = readArraysOfUpto defaultChunkSize
321322
322323-------------------------------------------------------------------------------
@@ -332,11 +333,11 @@ readArrays = readArraysOfUpto defaultChunkSize
332333-- as EOF is encountered.
333334--
334335{-# INLINE readInChunksOf #-}
335−readInChunksOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
336+readInChunksOf :: (MonadIO m) => Int -> Handle -> Stream m Word8
336337readInChunksOf chunkSize h = AS.concat $ readArraysOfUpto chunkSize h
337338
338339-- TODO
339−-- read :: (IsStream t, MonadIO m, Unboxed a) => Handle -> t m a
340+-- read :: (MonadIO m, Unboxed a) => Handle -> Stream m a
340341--
341342-- > read = 'readByChunks' A.defaultChunkSize
342343-- | Generate a stream of elements of the given type from a file 'Handle'. The
@@ -344,7 +345,7 @@ readInChunksOf chunkSize h = AS.concat $ readArraysOfUpto chunkSize h
344345--
345346-- @since 0.7.0
346347{-# INLINE read #-}
347−read :: (IsStream t, MonadIO m) => Handle -> t m Word8
348+read :: (MonadIO m) => Handle -> Stream m Word8
348349read = AS.concat . readArrays
349350
350351-------------------------------------------------------------------------------
@@ -356,7 +357,7 @@ read = AS.concat . readArrays
356357-- @since 0.7.0
357358{-# INLINE writeArrays #-}
358359writeArrays :: (MonadIO m, Unboxed a) => Handle -> Stream m (Array a) -> m ()
359−writeArrays h = S.mapM_ (liftIO . writeArray h)
360+writeArrays h = S.fold (FL.drainBy (liftIO . writeArray h))
360361
361362-- | Write a stream of arrays to a handle after coalescing them in chunks of
362363-- specified size. The chunk size is only a maximum and the actual writes could
@@ -440,7 +441,7 @@ write = toHandleWith A.defaultChunkSize
440441--
441442-- @since 0.7.0
442443{-# INLINE readUtf8 #-}
443−readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
444+readUtf8 :: (MonadIO m) => Handle -> Stream m Char
444445readUtf8 = decodeUtf8 . read
445446
446447-- |
@@ -462,7 +463,7 @@ writeUtf8 h s = write h $ encodeUtf8 s
462463--
463464-- @since 0.7.0
464465{-# INLINE writeUtf8ByLines #-}
465−writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
466+writeUtf8ByLines :: (MonadIO m) => Handle -> Stream m Char -> m ()
466467writeUtf8ByLines = undefined
467468
468469-- | Read UTF-8 lines from a file handle and apply the specified fold to each
@@ -470,7 +471,7 @@ writeUtf8ByLines = undefined
470471--
471472-- @since 0.7.0
472473{-# INLINE readLines #-}
473−readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
474+readLines :: (MonadIO m) => Handle -> Fold m Char b -> Stream m b
474475readLines h f = foldLines (readUtf8 h) f
475476
476477-------------------------------------------------------------------------------
@@ -483,8 +484,8 @@ readLines h f = foldLines (readUtf8 h) f
483484--
484485-- @since 0.7.0
485486{-# INLINE readFrames #-}
486−readFrames :: (IsStream t, MonadIO m, Unboxed a)
487− => Array a -> Handle -> Fold m a b -> t m b
487+readFrames :: (MonadIO m, Unboxed a)
488+ => Array a -> Handle -> Fold m a b -> Stream m b
488489readFrames = undefined -- foldFrames . read
489490
490491-- | Write a stream to the given file handle buffering up to frames separated
@@ -492,8 +493,8 @@ readFrames = undefined -- foldFrames . read
492493--
493494-- @since 0.7.0
494495{-# INLINE writeByFrames #-}
495−writeByFrames :: (IsStream t, MonadIO m, Unboxed a)
496− => Array a -> Handle -> t m a -> m ()
496+writeByFrames :: (MonadIO m, Unboxed a)
497+ => Array a -> Handle -> Stream m a -> m ()
497498writeByFrames = undefined
498499
499500-------------------------------------------------------------------------------
@@ -530,8 +531,8 @@ readIndex arr i = undefined
530531-- chunk is aligned with @chunkSize@ from second chunk onwards.
531532--
532533{-# INLINE readSliceWith #-}
533−readSliceWith :: (IsStream t, MonadIO m, Unboxed a)
534− => Int -> Handle -> Int -> Int -> t m a
534+readSliceWith :: (MonadIO m, Unboxed a)
535+ => Int -> Handle -> Int -> Int -> Stream m a
535536readSliceWith chunkSize h pos len = undefined
536537
537538-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
@@ -540,8 +541,8 @@ readSliceWith chunkSize h pos len = undefined
540541--
541542-- @since 0.7.0
542543{-# INLINE readSlice #-}
543−readSlice :: (IsStream t, MonadIO m, Unboxed a)
544− => Handle -> Int -> Int -> t m a
544+readSlice :: (MonadIO m, Unboxed a)
545+ => Handle -> Int -> Int -> Stream m a
545546readSlice = readSliceWith defaultChunkSize
546547
547548-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
@@ -550,8 +551,8 @@ readSlice = readSliceWith defaultChunkSize
550551--
551552-- @since 0.7.0
552553{-# INLINE readSliceRev #-}
553−readSliceRev :: (IsStream t, MonadIO m, Unboxed a)
554− => Handle -> Int -> Int -> t m a
554+readSliceRev :: (MonadIO m, Unboxed a)
555+ => Handle -> Int -> Int -> Stream m a
555556readSliceRev h i count = undefined
556557
557558-- | Write the given element at the given index in the file.
@@ -567,8 +568,8 @@ writeIndex h i a = undefined
567568--
568569-- @since 0.7.0
569570{-# INLINE writeSlice #-}
570−writeSlice :: (IsStream t, Monad m, Unboxed a)
571− => Handle -> Int -> Int -> t m a -> m ()
571+writeSlice :: (Monad m, Unboxed a)
572+ => Handle -> Int -> Int -> Stream m a -> m ()
572573writeSlice h i len s = undefined
573574
574575-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
@@ -577,7 +578,7 @@ writeSlice h i len s = undefined
577578--
578579-- @since 0.7.0
579580{-# INLINE writeSliceRev #-}
580−writeSliceRev :: (IsStream t, Monad m, Unboxed a)
581− => Handle -> Int -> Int -> t m a -> m ()
581+writeSliceRev :: (Monad m, Unboxed a)
582+ => Handle -> Int -> Int -> Stream m a -> m ()
582583writeSliceRev arr i len s = undefined
583584-}
src/Streamly/Internal/FileSystem/Handle.hs+31−32
@@ -124,8 +124,6 @@ import Streamly.Internal.Data.Unfold.Type (Unfold(..))
124124import Streamly.Internal.Data.Array.Unboxed.Type
125125 (Array(..), writeNUnsafe, unsafeFreezeWithShrink, byteLength)
126126import Streamly.Internal.Data.Stream.Type (Stream)
127−import Streamly.Internal.Data.Stream.IsStream.Type
128− (IsStream, mkStream, fromStreamD)
129127import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)
130128-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
131129import Streamly.Internal.System.IO (defaultChunkSize)
@@ -139,10 +137,11 @@ import qualified Streamly.Internal.Data.Array.Unboxed.Mut as MArray
139137import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS
140138import qualified Streamly.Internal.Data.Refold.Type as Refold
141139import qualified Streamly.Internal.Data.Fold.Type as FL(refoldMany)
142−import qualified Streamly.Internal.Data.Stream.IsStream as S
140+import qualified Streamly.Internal.Data.Stream as S
143141import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
144142 (Stream(..), Step(..))
145143import qualified Streamly.Internal.Data.Unfold as UF
144+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K (mkStream)
146145
147146-- $setup
148147-- >>> import qualified Streamly.Data.Array.Unboxed as Array
@@ -213,12 +212,12 @@ getChunkOf = undefined
213212-- read may be less than or equal to @size@.
214213-- @since 0.9.0
215214{-# INLINE _getChunksWith #-}
216−_getChunksWith :: (IsStream t, MonadIO m)
217− => Int -> Handle -> t m (Array Word8)
218−_getChunksWith size h = go
215+_getChunksWith :: (MonadIO m)
216+ => Int -> Handle -> Stream m (Array Word8)
217+_getChunksWith size h = S.fromStreamK go
219218 where
220219 -- XXX use cons/nil instead
221− go = mkStream $ \_ yld _ stp -> do
220+ go = K.mkStream $ \_ yld _ stp -> do
222221 arr <- getChunk size h
223222 if byteLength arr == 0
224223 then stp
@@ -232,9 +231,9 @@ _getChunksWith size h = go
232231--
233232-- @since 0.9.0
234233{-# INLINE_NORMAL getChunksWith #-}
235−getChunksWith :: (IsStream t, MonadIO m) =>
236− Int -> Handle -> t m (Array Word8)
237−getChunksWith size h = fromStreamD (D.Stream step ())
234+getChunksWith :: (MonadIO m) =>
235+ Int -> Handle -> Stream m (Array Word8)
236+getChunksWith size h = S.fromStreamD (D.Stream step ())
238237 where
239238 {-# INLINE_LATE step #-}
240239 step _ _ = do
@@ -314,7 +313,7 @@ readChunksFromToWith = Unfold step inject
314313--
315314-- @since 0.9.0
316315{-# INLINE getChunks #-}
317−getChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
316+getChunks :: (MonadIO m) => Handle -> Stream m (Array Word8)
318317getChunks = getChunksWith defaultChunkSize
319318
320319-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO
@@ -363,12 +362,12 @@ readWithBufferOf = readWith
363362--
364363-- /Pre-release/
365364{-# INLINE getBytesWith #-}
366−getBytesWith :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
365+getBytesWith :: (MonadIO m) => Int -> Handle -> Stream m Word8
367366getBytesWith size h = AS.concat $ getChunksWith size h
368367
369368-- TODO
370369-- Generate a stream of elements of the given type from a file 'Handle'.
371−-- read :: (IsStream t, MonadIO m, Unboxed a) => Handle -> t m a
370+-- read :: (MonadIO m, Unboxed a) => Handle -> Stream m a
372371--
373372-- | Unfolds a file handle into a byte stream. IO requests to the device are
374373-- performed in sizes of
@@ -387,7 +386,7 @@ read = UF.many A.read readChunks
387386--
388387-- /Pre-release/
389388{-# INLINE getBytes #-}
390−getBytes :: (IsStream t, MonadIO m) => Handle -> t m Word8
389+getBytes :: (MonadIO m) => Handle -> Stream m Word8
391390getBytes = AS.concat . getChunks
392391
393392-------------------------------------------------------------------------------
@@ -428,7 +427,7 @@ putChunk h arr = A.asPtrUnsafe arr $ \ptr ->
428427-- @since 0.7.0
429428{-# INLINE putChunks #-}
430429putChunks :: MonadIO m => Handle -> Stream m (Array a) -> m ()
431−putChunks h = S.mapM_ (putChunk h)
430+putChunks h = S.fold (FL.drainBy (putChunk h))
432431
433432-- XXX AS.compact can be written idiomatically in terms of foldMany, just like
434433-- AS.concat is written in terms of foldMany. Once that is done we can write
@@ -601,7 +600,7 @@ write = toHandleWith A.defaultChunkSize
601600--
602601-- @since 0.7.0
603602{-# INLINE readUtf8 #-}
604−readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
603+readUtf8 :: (MonadIO m) => Handle -> Stream m Char
605604readUtf8 = decodeUtf8 . read
606605
607606-- |
@@ -623,7 +622,7 @@ writeUtf8 h s = write h $ encodeUtf8 s
623622--
624623-- @since 0.7.0
625624{-# INLINE writeUtf8ByLines #-}
626−writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
625+writeUtf8ByLines :: (MonadIO m) => Handle -> Stream m Char -> m ()
627626writeUtf8ByLines = undefined
628627
629628-- | Read UTF-8 lines from a file handle and apply the specified fold to each
@@ -631,7 +630,7 @@ writeUtf8ByLines = undefined
631630--
632631-- @since 0.7.0
633632{-# INLINE readLines #-}
634−readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
633+readLines :: (MonadIO m) => Handle -> Fold m Char b -> Stream m b
635634readLines h f = foldLines (readUtf8 h) f
636635
637636-------------------------------------------------------------------------------
@@ -644,8 +643,8 @@ readLines h f = foldLines (readUtf8 h) f
644643--
645644-- @since 0.7.0
646645{-# INLINE readFrames #-}
647−readFrames :: (IsStream t, MonadIO m, Unboxed a)
648− => Array a -> Handle -> Fold m a b -> t m b
646+readFrames :: (MonadIO m, Unboxed a)
647+ => Array a -> Handle -> Fold m a b -> Stream m b
649648readFrames = undefined -- foldFrames . read
650649
651650-- | Write a stream to the given file handle buffering up to frames separated
@@ -653,8 +652,8 @@ readFrames = undefined -- foldFrames . read
653652--
654653-- @since 0.7.0
655654{-# INLINE writeByFrames #-}
656−writeByFrames :: (IsStream t, MonadIO m, Unboxed a)
657− => Array a -> Handle -> t m a -> m ()
655+writeByFrames :: (MonadIO m, Unboxed a)
656+ => Array a -> Handle -> Stream m a -> m ()
658657writeByFrames = undefined
659658
660659-------------------------------------------------------------------------------
@@ -708,8 +707,8 @@ readIndex arr i = undefined
708707-- chunk is aligned with @chunkSize@ from second chunk onwards.
709708--
710709{-# INLINE readSliceWith #-}
711−readSliceWith :: (IsStream t, MonadIO m, Unboxed a)
712− => Int -> Handle -> Int -> Int -> t m a
710+readSliceWith :: (MonadIO m, Unboxed a)
711+ => Int -> Handle -> Int -> Int -> Stream m a
713712readSliceWith chunkSize h pos len = undefined
714713
715714-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
@@ -718,8 +717,8 @@ readSliceWith chunkSize h pos len = undefined
718717--
719718-- @since 0.7.0
720719{-# INLINE readSlice #-}
721−readSlice :: (IsStream t, MonadIO m, Unboxed a)
722− => Handle -> Int -> Int -> t m a
720+readSlice :: (MonadIO m, Unboxed a)
721+ => Handle -> Int -> Int -> Stream m a
723722readSlice = readSliceWith A.defaultChunkSize
724723
725724-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
@@ -728,8 +727,8 @@ readSlice = readSliceWith A.defaultChunkSize
728727--
729728-- @since 0.7.0
730729{-# INLINE readSliceRev #-}
731−readSliceRev :: (IsStream t, MonadIO m, Unboxed a)
732− => Handle -> Int -> Int -> t m a
730+readSliceRev :: (MonadIO m, Unboxed a)
731+ => Handle -> Int -> Int -> Stream m a
733732readSliceRev h i count = undefined
734733
735734-- | Write the given element at the given index in the file.
@@ -745,8 +744,8 @@ writeIndex h i a = undefined
745744--
746745-- @since 0.7.0
747746{-# INLINE writeSlice #-}
748−writeSlice :: (IsStream t, Monad m, Unboxed a)
749− => Handle -> Int -> Int -> t m a -> m ()
747+writeSlice :: (Monad m, Unboxed a)
748+ => Handle -> Int -> Int -> Stream m a -> m ()
750749writeSlice h i len s = undefined
751750
752751-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
@@ -755,7 +754,7 @@ writeSlice h i len s = undefined
755754--
756755-- @since 0.7.0
757756{-# INLINE writeSliceRev #-}
758−writeSliceRev :: (IsStream t, Monad m, Unboxed a)
759− => Handle -> Int -> Int -> t m a -> m ()
757+writeSliceRev :: (Monad m, Unboxed a)
758+ => Handle -> Int -> Int -> Stream m a -> m ()
760759writeSliceRev arr i len s = undefined
761760-}
src/Streamly/Internal/Network/Inet/TCP.hs+9−10
@@ -107,7 +107,6 @@ import Streamly.Internal.Control.Concurrent (MonadAsync)
107107import Streamly.Internal.Control.ForkLifted (fork)
108108import Streamly.Internal.Data.Array.Unboxed.Type (Array(..), writeNUnsafe)
109109import Streamly.Internal.Data.Fold.Type (Fold(..))
110−import Streamly.Internal.Data.Stream.IsStream.Type (IsStream)
111110import Streamly.Internal.Data.Stream (Stream)
112111import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
113112import Streamly.Internal.Data.Unfold.Type (Unfold(..))
@@ -302,8 +301,8 @@ usingConnection =
302301--
303302-- /Pre-release/
304303{-# INLINE withConnection #-}
305−withConnection :: (IsStream t, MonadCatch m, MonadAsync m)
306− => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> t m a) -> t m a
304+withConnection :: (MonadCatch m, MonadAsync m)
305+ => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> Stream m a) -> Stream m a
307306withConnection addr port =
308307 S.bracket (liftIO $ connect addr port) (liftIO . Net.close)
309308
@@ -323,8 +322,8 @@ read = UF.many A.read (usingConnection ISK.readChunks)
323322--
324323-- @since 0.7.0
325324{-# INLINE toBytes #-}
326−toBytes :: (IsStream t, MonadCatch m, MonadAsync m)
327− => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8
325+toBytes :: (MonadCatch m, MonadAsync m)
326+ => (Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Word8
328327toBytes addr port = AS.concat $ withConnection addr port ISK.toChunks
329328
330329-------------------------------------------------------------------------------
@@ -424,12 +423,12 @@ write = writeWithBufferOf defaultChunkSize
424423
425424{-# INLINE withInputConnect #-}
426425withInputConnect
427− :: (IsStream t, MonadCatch m, MonadAsync m)
426+ :: (MonadCatch m, MonadAsync m)
428427 => (Word8, Word8, Word8, Word8)
429428 -> PortNumber
430429 -> Stream m Word8
431− -> (Socket -> t m a)
432− -> t m a
430+ -> (Socket -> Stream m a)
431+ -> Stream m a
433432withInputConnect addr port input f = S.bracket pre post handler
434433
435434 where
@@ -452,9 +451,9 @@ withInputConnect addr port input f = S.bracket pre post handler
452451--
453452{-# INLINE processBytes #-}
454453processBytes
455− :: (IsStream t, MonadAsync m, MonadCatch m)
454+ :: (MonadAsync m, MonadCatch m)
456455 => (Word8, Word8, Word8, Word8)
457456 -> PortNumber
458457 -> Stream m Word8
459− -> t m Word8
458+ -> Stream m Word8
460459processBytes addr port input = withInputConnect addr port input ISK.toBytes
src/Streamly/Internal/Network/Socket.hs+23−24
@@ -94,8 +94,6 @@ import Streamly.Internal.Control.Concurrent (MonadAsync)
9494import Streamly.Internal.Data.Array.Unboxed.Type (Array(..))
9595import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)
9696import Streamly.Internal.Data.Fold (Fold)
97−import Streamly.Internal.Data.Stream.IsStream.Type
98− (IsStream, mkStream, fromStreamD)
9997import Streamly.Internal.Data.Stream.Type (Stream)
10098import Streamly.Internal.Data.Unfold.Type (Unfold(..))
10199-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
@@ -108,11 +106,12 @@ import qualified Streamly.Internal.Data.Array.Unboxed.Type as A
108106import qualified Streamly.Internal.Data.Array.Unboxed.Mut as MArray
109107 (Array(..), newPinnedArrayBytes, asPtrUnsafe)
110108import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS
111−import qualified Streamly.Internal.Data.Stream.IsStream as S
109+import qualified Streamly.Internal.Data.Stream as S
112110import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
113111 (Stream(..), Step(..))
114112import qualified Streamly.Data.Unfold as UF
115113import qualified Streamly.Internal.Data.Unfold as UF (first, map)
114+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K (mkStream)
116115
117116-- | @'forSocketM' action socket@ runs the monadic computation @action@ passing
118117-- the socket handle to it. The handle will be closed on exit from
@@ -132,8 +131,8 @@ forSocketM f sk = finally (f sk) (liftIO (Net.close sk))
132131--
133132-- /Internal/
134133{-# INLINE withSocket #-}
135−withSocket :: (IsStream t, MonadAsync m, MonadCatch m)
136− => Socket -> (Socket -> t m a) -> t m a
134+withSocket :: (MonadAsync m, MonadCatch m)
135+ => Socket -> (Socket -> Stream m a) -> Stream m a
137136withSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)
138137
139138-------------------------------------------------------------------------------
@@ -331,13 +330,13 @@ writeChunk = writeArrayWith sendAll
331330-------------------------------------------------------------------------------
332331
333332{-# INLINE _readChunksUptoWith #-}
334−_readChunksUptoWith :: (IsStream t, MonadIO m)
333+_readChunksUptoWith :: (MonadIO m)
335334 => (Int -> h -> IO (Array Word8))
336− -> Int -> h -> t m (Array Word8)
337−_readChunksUptoWith f size h = go
335+ -> Int -> h -> Stream m (Array Word8)
336+_readChunksUptoWith f size h = S.fromStreamK go
338337 where
339338 -- XXX use cons/nil instead
340− go = mkStream $ \_ yld _ stp -> do
339+ go = K.mkStream $ \_ yld _ stp -> do
341340 arr <- liftIO $ f size h
342341 if A.length arr == 0
343342 then stp
@@ -348,10 +347,10 @@ _readChunksUptoWith f size h = go
348347-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'
349348-- on the 'Handle'.
350349{-# INLINE_NORMAL toChunksWith #-}
351−toChunksWith :: (IsStream t, MonadIO m)
352− => Int -> Socket -> t m (Array Word8)
350+toChunksWith :: (MonadIO m)
351+ => Int -> Socket -> Stream m (Array Word8)
353352-- toChunksWith = _readChunksUptoWith readChunk
354−toChunksWith size h = fromStreamD (D.Stream step ())
353+toChunksWith size h = S.fromStreamD (D.Stream step ())
355354 where
356355 {-# INLINE_LATE step #-}
357356 step _ _ = do
@@ -368,7 +367,7 @@ toChunksWith size h = fromStreamD (D.Stream step ())
368367--
369368-- @since 0.7.0
370369{-# INLINE toChunks #-}
371−toChunks :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)
370+toChunks :: (MonadIO m) => Socket -> Stream m (Array Word8)
372371toChunks = toChunksWith defaultChunkSize
373372
374373-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.
@@ -422,12 +421,12 @@ readChunks = UF.first defaultChunkSize readChunksWith
422421-- as soon as EOF is encountered.
423422--
424423{-# INLINE readWith #-}
425−readWith :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
424+readWith :: (MonadIO m) => Int -> Handle -> Stream m Word8
426425readWith chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h
427426-}
428427
429428-- TODO
430−-- read :: (IsStream t, MonadIO m, Unboxed a) => Handle -> t m a
429+-- read :: (MonadIO m, Unboxed a) => Handle -> Stream m a
431430--
432431-- > read = 'readByChunks' defaultChunkSize
433432-- | Generate a stream of elements of the given type from a socket. The
@@ -435,7 +434,7 @@ readWith chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h
435434--
436435-- @since 0.7.0
437436{-# INLINE toBytes #-}
438−toBytes :: (IsStream t, MonadIO m) => Socket -> t m Word8
437+toBytes :: (MonadIO m) => Socket -> Stream m Word8
439438toBytes = AS.concat . toChunks
440439
441440-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests
@@ -473,7 +472,7 @@ read = UF.first defaultChunkSize readWith
473472{-# INLINE putChunks #-}
474473putChunks :: (MonadIO m, Unboxed a)
475474 => Socket -> Stream m (Array a) -> m ()
476−putChunks h = S.mapM_ (liftIO . writeChunk h)
475+putChunks h = S.fold (FL.drainBy (liftIO . writeChunk h))
477476
478477-- | Write a stream of arrays to a socket. Each array in the stream is written
479478-- to the socket as a separate IO request.
@@ -591,7 +590,7 @@ write = toHandleWith defaultChunkSize
591590--
592591-- @since 0.7.0
593592{-# INLINE readUtf8 #-}
594−readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
593+readUtf8 :: (MonadIO m) => Handle -> Stream m Char
595594readUtf8 = decodeUtf8 . read
596595
597596-- |
@@ -613,7 +612,7 @@ writeUtf8 h s = write h $ encodeUtf8 s
613612--
614613-- @since 0.7.0
615614{-# INLINE writeUtf8ByLines #-}
616−writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
615+writeUtf8ByLines :: (MonadIO m) => Handle -> Stream m Char -> m ()
617616writeUtf8ByLines = undefined
618617
619618-- | Read UTF-8 lines from a file handle and apply the specified fold to each
@@ -621,7 +620,7 @@ writeUtf8ByLines = undefined
621620--
622621-- @since 0.7.0
623622{-# INLINE readLines #-}
624−readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
623+readLines :: (MonadIO m) => Handle -> Fold m Char b -> Stream m b
625624readLines h f = foldLines (readUtf8 h) f
626625
627626-------------------------------------------------------------------------------
@@ -634,8 +633,8 @@ readLines h f = foldLines (readUtf8 h) f
634633--
635634-- @since 0.7.0
636635{-# INLINE readFrames #-}
637−readFrames :: (IsStream t, MonadIO m, Unboxed a)
638− => Array a -> Handle -> Fold m a b -> t m b
636+readFrames :: (MonadIO m, Unboxed a)
637+ => Array a -> Handle -> Fold m a b -> Stream m b
639638readFrames = undefined -- foldFrames . read
640639
641640-- | Write a stream to the given file handle buffering up to frames separated
@@ -643,7 +642,7 @@ readFrames = undefined -- foldFrames . read
643642--
644643-- @since 0.7.0
645644{-# INLINE writeByFrames #-}
646−writeByFrames :: (IsStream t, MonadIO m, Unboxed a)
647− => Array a -> Handle -> t m a -> m ()
645+writeByFrames :: (MonadIO m, Unboxed a)
646+ => Array a -> Handle -> Stream m a -> m ()
648647writeByFrames = undefined
649648-}
streamly.cabal+1−2
@@ -205,6 +205,7 @@ extra-source-files:
205205 core/src/Streamly/Internal/Data/List.hs
206206 core/src/Streamly/Internal/Data/Array.hs
207207 core/src/Streamly/Internal/Data/Array/Unboxed.hs
208+ core/src/Streamly/Internal/Data/Array/Stream/Foreign.hs
208209 core/src/Streamly/Internal/Data/Array/Stream/Mut/Foreign.hs
209210 core/src/Streamly/Internal/Data/Array/Stream/Fold/Foreign.hs
210211 core/src/Streamly/Internal/Data/Ring/Foreign.hs
@@ -475,8 +476,6 @@ library
475476 , Streamly.Internal.Data.Stream.IsStream.Async
476477 , Streamly.Internal.Data.Stream.IsStream
477478
478− , Streamly.Internal.Data.Array.Stream.Foreign
479−
480479 -- streamly-serde
481480 , Streamly.Internal.Data.Binary.Decode
482481