aef0a2e · streamly · git.iko.soy

git.iko.soy/streamly

Move Array.Stream.Foreign to streamly-core (#1863)

aef0a2eRanjeet Ranjan4 years ago

src/Streamly/Internal/Data/Array/Stream/Foreign.hs → core/src/Streamly/Internal/Data/Array/Stream/Foreign.hsrenamed+39−40

compare with master

unifiedsplit

@@ -69,9 +69,9 @@ import Streamly.Data.Stream (Stream)

69import Streamly.Internal.Data.Array.Unboxed.Type (Array(..))

70import Streamly.Internal.Data.Array.Stream.Fold.Foreign (ArrayFold(..))

71import Streamly.Internal.Data.Parser (ParseError(..))

72−import Streamly.Internal.Data.Stream.IsStream.Type

72+import Streamly.Internal.Data.Stream

73− (IsStream, fromStreamD, toStreamD)

73+ (fromStreamD, fromStreamK, toStreamD, toStreamK)

74−import Streamly.Internal.Data.SVar (adaptState, defState)

74+import Streamly.Internal.Data.SVar.Type (adaptState, defState)

75import Streamly.Internal.Data.Array.Unboxed.Mut.Type

76 (allocBytesToElemCount)

77import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

@@ -79,27 +79,22 @@ import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

79import qualified Streamly.Data.Fold as FL

80import qualified Streamly.Internal.Data.Array.Unboxed as A

81import qualified Streamly.Internal.Data.Array.Unboxed as Array

82−-- import qualified Streamly.Internal.Data.Array.Stream.Fold.Foreign

83−-- as ArrayFold

84import qualified Streamly.Internal.Data.Array.Unboxed.Type as A

82import qualified Streamly.Internal.Data.Array.Unboxed.Type as A

85import qualified Streamly.Internal.Data.Array.Unboxed.Mut.Type as MA

83import qualified Streamly.Internal.Data.Array.Unboxed.Mut.Type as MA

86import qualified Streamly.Internal.Data.Array.Stream.Mut.Foreign as AS

84import qualified Streamly.Internal.Data.Array.Stream.Mut.Foreign as AS

87−import qualified Streamly.Internal.Data.Fold.Type as FL

85+import qualified Streamly.Internal.Data.Fold.Type as FL (Fold(..), Step(..))

88− (Fold(..), Step(..))

89import qualified Streamly.Internal.Data.Parser as PR

86import qualified Streamly.Internal.Data.Parser as PR

90import qualified Streamly.Internal.Data.Parser.ParserD as PRD

87import qualified Streamly.Internal.Data.Parser.ParserD as PRD

91 (Parser(..), Initial(..), fromParserK)

88 (Parser(..), Initial(..), fromParserK)

92−import qualified Streamly.Internal.Data.Stream.IsStream as S

89+import qualified Streamly.Internal.Data.Stream as S

93import qualified Streamly.Internal.Data.Stream.StreamD as D

90import qualified Streamly.Internal.Data.Stream.StreamD as D

94 ( fromList, nil, cons, map

91 ( fromList, nil, cons, map

95− , unfoldMany, append, splitInnerBy, splitInnerBySuffix

92+ , unfoldMany, append, splitInnerBy, splitInnerBySuffix, foldlM'

96 )

93 )

97import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

94import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

98 (Step(Yield, Stop, Skip), Stream(Stream))

95 (Step(Yield, Stop, Skip), Stream(Stream))

99import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

96import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

100 (Stream, cons, nil, fromPure, foldStream)

97 (Stream, cons, nil, fromPure, foldStream)

101−import qualified Streamly.Internal.Data.Stream as Stream

102− (fromStreamK, toStreamK)

103

104-- XXX Since these are immutable arrays MonadIO constraint can be removed from

99-- XXX Since these are immutable arrays MonadIO constraint can be removed from

105-- most places.

100-- most places.

@@ -115,8 +110,8 @@ import qualified Streamly.Internal.Data.Stream as Stream

115--

110--

116-- /Pre-release/

111-- /Pre-release/

117{-# INLINE arraysOf #-}

112{-# INLINE arraysOf #-}

118−arraysOf :: (IsStream t, MonadIO m, Unboxed a)

113+arraysOf :: (MonadIO m, Unboxed a)

119− => Int -> t m a -> t m (Array a)

114+ => Int -> Stream m a -> Stream m (Array a)

120arraysOf n str = fromStreamD $ A.arraysOf n (toStreamD str)

115arraysOf n str = fromStreamD $ A.arraysOf n (toStreamD str)

121

116

122-------------------------------------------------------------------------------

117-------------------------------------------------------------------------------

@@ -135,7 +130,7 @@ arraysOf n str = fromStreamD $ A.arraysOf n (toStreamD str)

135--

130--

136-- @since 0.7.0

131-- @since 0.7.0

137{-# INLINE concat #-}

132{-# 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

139-- concat m = fromStreamD $ A.flattenArrays (toStreamD m)

134-- concat m = fromStreamD $ A.flattenArrays (toStreamD m)

140-- concat m = fromStreamD $ D.concatMap A.toStreamD (toStreamD m)

135-- concat m = fromStreamD $ D.concatMap A.toStreamD (toStreamD m)

141concat m = fromStreamD $ D.unfoldMany A.read (toStreamD m)

136concat m = fromStreamD $ D.unfoldMany A.read (toStreamD m)

@@ -147,7 +142,7 @@ concat m = fromStreamD $ D.unfoldMany A.read (toStreamD m)

147--

142--

148-- @since 0.7.0

143-- @since 0.7.0

149{-# INLINE concatRev #-}

144{-# 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

151-- concatRev m = fromStreamD $ A.flattenArraysRev (toStreamD m)

146-- concatRev m = fromStreamD $ A.flattenArraysRev (toStreamD m)

152concatRev m = fromStreamD $ D.unfoldMany A.readRev (toStreamD m)

147concatRev m = fromStreamD $ D.unfoldMany A.readRev (toStreamD m)

153

148

@@ -160,12 +155,12 @@ concatRev m = fromStreamD $ D.unfoldMany A.readRev (toStreamD m)

160--

155--

161-- /Pre-release/

156-- /Pre-release/

162{-# INLINE interpose #-}

157{-# 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

164interpose x = S.interpose x A.read

159interpose x = S.interpose x A.read

165

160

166{-# INLINE intercalateSuffix #-}

161{-# INLINE intercalateSuffix #-}

167−intercalateSuffix :: (Monad m, IsStream t, Unboxed a)

162+intercalateSuffix :: (Monad m, Unboxed a)

168− => Array a -> t m (Array a) -> t m a

163+ => Array a -> Stream m (Array a) -> Stream m a

169intercalateSuffix = S.intercalateSuffix A.read

164intercalateSuffix = S.intercalateSuffix A.read

170

165

171-- | Flatten a stream of arrays appending the given element after each

166-- | Flatten a stream of arrays appending the given element after each

@@ -173,8 +168,8 @@ intercalateSuffix = S.intercalateSuffix A.read

173--

168--

174-- @since 0.7.0

169-- @since 0.7.0

175{-# INLINE interposeSuffix #-}

170{-# INLINE interposeSuffix #-}

176−interposeSuffix :: (Monad m, IsStream t, Unboxed a)

171+interposeSuffix :: (Monad m, Unboxed a)

177− => a -> t m (Array a) -> t m a

172+ => a -> Stream m (Array a) -> Stream m a

178-- interposeSuffix x = fromStreamD . A.unlines x . toStreamD

173-- interposeSuffix x = fromStreamD . A.unlines x . toStreamD

179interposeSuffix x = S.interposeSuffix x A.read

174interposeSuffix x = S.interposeSuffix x A.read

180

175

@@ -302,19 +297,19 @@ _splitOn byte (D.Stream step state) = D.Stream step' (Initial state)

302-- @since 0.7.0

297-- @since 0.7.0

303{-# INLINE splitOn #-}

298{-# INLINE splitOn #-}

304splitOn

299splitOn

305− :: (IsStream t, MonadIO m)

300+ :: (MonadIO m)

306 => Word8

301 => Word8

307− -> t m (Array Word8)

302+ -> Stream m (Array Word8)

308− -> t m (Array Word8)

303+ -> Stream m (Array Word8)

309splitOn byte s =

304splitOn byte s =

310 fromStreamD $ D.splitInnerBy (A.breakOn byte) A.splice $ toStreamD s

305 fromStreamD $ D.splitInnerBy (A.breakOn byte) A.splice $ toStreamD s

311

306

312{-# INLINE splitOnSuffix #-}

307{-# INLINE splitOnSuffix #-}

313splitOnSuffix

308splitOnSuffix

314− :: (IsStream t, MonadIO m)

309+ :: (MonadIO m)

315 => Word8

310 => Word8

316− -> t m (Array Word8)

311+ -> Stream m (Array Word8)

317− -> t m (Array Word8)

312+ -> Stream m (Array Word8)

318-- splitOn byte s = fromStreamD $ A.splitOn byte $ toStreamD s

313-- splitOn byte s = fromStreamD $ A.splitOn byte $ toStreamD s

319splitOnSuffix byte s =

314splitOnSuffix byte s =

320 fromStreamD $ D.splitInnerBySuffix (A.breakOn byte) A.splice $ toStreamD s

315 fromStreamD $ D.splitInnerBySuffix (A.breakOn byte) A.splice $ toStreamD s

@@ -410,9 +405,9 @@ foldBreak ::

410 -> m (b, Stream m (A.Array a))

405 -> m (b, Stream m (A.Array a))

411-- foldBreak f s = fmap fromStreamD <$> foldBreakD f (toStreamD s)

406-- foldBreak f s = fmap fromStreamD <$> foldBreakD f (toStreamD s)

412foldBreak f =

407foldBreak f =

413− fmap (fmap Stream.fromStreamK)

408+ fmap (fmap fromStreamK)

414 . foldBreakK f

409 . foldBreakK f

415− . Stream.toStreamK

410+ . toStreamK

416-- If foldBreak performs better than runArrayFoldBreak we can use a rewrite

411-- If foldBreak performs better than runArrayFoldBreak we can use a rewrite

417-- rule to rewrite runArrayFoldBreak to fold.

412-- rule to rewrite runArrayFoldBreak to fold.

418-- foldBreak f = runArrayFoldBreak (ArrayFold.fromFold f)

413-- foldBreak f = runArrayFoldBreak (ArrayFold.fromFold f)

@@ -468,6 +463,10 @@ splitAtArrayListRev n ls

468-- Fold to a single Array

463-- Fold to a single Array

469-------------------------------------------------------------------------------

464-------------------------------------------------------------------------------

470

465

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+

471-- XXX Both of these implementations of splicing seem to perform equally well.

470-- XXX Both of these implementations of splicing seem to perform equally well.

472-- We need to perform benchmarks over a range of sizes though.

471-- We need to perform benchmarks over a range of sizes though.

473

472

@@ -478,16 +477,16 @@ spliceArraysLenUnsafe :: (MonadIO m, Unboxed a)

478 => Int -> Stream m (MA.Array a) -> m (MA.Array a)

477 => Int -> Stream m (MA.Array a) -> m (MA.Array a)

479spliceArraysLenUnsafe len buffered = do

478spliceArraysLenUnsafe len buffered = do

480 arr <- liftIO $ MA.newArray len

479 arr <- liftIO $ MA.newArray len

481− S.foldlM' MA.spliceUnsafe (return arr) buffered

480+ foldlM' MA.spliceUnsafe (return arr) buffered

482

481

483{-# INLINE _spliceArrays #-}

482{-# INLINE _spliceArrays #-}

484_spliceArrays :: (MonadIO m, Unboxed a)

483_spliceArrays :: (MonadIO m, Unboxed a)

485 => Stream m (Array a) -> m (Array a)

484 => Stream m (Array a) -> m (Array a)

486_spliceArrays s = do

485_spliceArrays s = do

487 buffered <- S.foldr S.cons S.nil s

486 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)

489 arr <- liftIO $ MA.newArray len

488 arr <- liftIO $ MA.newArray len

490− final <- S.foldlM' writeArr (return arr) s

489+ final <- foldlM' writeArr (return arr) s

491 return $ A.unsafeFreeze final

490 return $ A.unsafeFreeze final

492

491

493 where

492 where

@@ -499,8 +498,8 @@ _spliceArraysBuffered :: (MonadIO m, Unboxed a)

499 => Stream m (Array a) -> m (Array a)

498 => Stream m (Array a) -> m (Array a)

500_spliceArraysBuffered s = do

499_spliceArraysBuffered s = do

501 buffered <- S.foldr S.cons S.nil s

500 buffered <- S.foldr S.cons S.nil s

502− len <- S.sum (S.map Array.length buffered)

501+ len <- S.fold FL.sum (fmap Array.length buffered)

503− A.unsafeFreeze <$> spliceArraysLenUnsafe len (S.map A.unsafeThaw s)

502+ A.unsafeFreeze <$> spliceArraysLenUnsafe len (fmap A.unsafeThaw s)

504

503

505{-# INLINE spliceArraysRealloced #-}

504{-# INLINE spliceArraysRealloced #-}

506spliceArraysRealloced :: forall m a. (MonadIO m, Unboxed a)

505spliceArraysRealloced :: forall m a. (MonadIO m, Unboxed a)

@@ -509,7 +508,7 @@ spliceArraysRealloced s = do

509 let n = allocBytesToElemCount (undefined :: a) (4 * 1024)

508 let n = allocBytesToElemCount (undefined :: a) (4 * 1024)

510 idst = liftIO $ MA.newArray n

509 idst = liftIO $ MA.newArray n

511

510

512− arr <- S.foldlM' MA.spliceExp idst (S.map A.unsafeThaw s)

511+ arr <- foldlM' MA.spliceExp idst (fmap A.unsafeThaw s)

513 liftIO $ A.unsafeFreeze <$> MA.rightSize arr

512 liftIO $ A.unsafeFreeze <$> MA.rightSize arr

514

513

515-- XXX This should just be "fold A.write"

514-- XXX This should just be "fold A.write"

@@ -530,7 +529,7 @@ toArray = spliceArraysRealloced

530--

529--

531{-

530{-

532{-# INLINE toArraysInRange #-}

531{-# INLINE toArraysInRange #-}

533−toArraysInRange :: (IsStream t, MonadIO m, Unboxed a)

532+toArraysInRange :: (MonadIO m, Unboxed a)

534 => Int -> Int -> Fold m (Array a) b -> Fold m a b

533 => Int -> Int -> Fold m (Array a) b -> Fold m a b

535toArraysInRange low high (Fold step initial extract) =

534toArraysInRange low high (Fold step initial extract) =

536-}

535-}

@@ -785,9 +784,9 @@ parseBreak p s =

785 fmap fromStreamD <$> parseBreakD (PRD.fromParserK p) (toStreamD s)

784 fmap fromStreamD <$> parseBreakD (PRD.fromParserK p) (toStreamD s)

786-}

785-}

787parseBreak p =

786parseBreak p =

788− fmap (fmap Stream.fromStreamK)

787+ fmap (fmap fromStreamK)

789 . parseBreakK (PRD.fromParserK p)

788 . parseBreakK (PRD.fromParserK p)

790− . Stream.toStreamK

789+ . toStreamK

791

790

792-------------------------------------------------------------------------------

791-------------------------------------------------------------------------------

793-- Elimination - Running Array Folds and parsers

792-- Elimination - Running Array Folds and parsers

@@ -1155,8 +1154,8 @@ runArrayFoldManyD

1155-- /Pre-release/

1154-- /Pre-release/

1156{-# INLINE runArrayFoldMany #-}

1155{-# INLINE runArrayFoldMany #-}

1157runArrayFoldMany

1156runArrayFoldMany

1158− :: (IsStream t, MonadThrow m, Unboxed a)

1157+ :: (MonadThrow m, Unboxed a)

1159 => ArrayFold m a b

1158 => ArrayFold m a b

1160− -> t m (Array a)

1159+ -> Stream m (Array a)

1161− -> t m b

1160+ -> Stream m b

1162runArrayFoldMany p m = fromStreamD $ runArrayFoldManyD p (toStreamD m)

1161runArrayFoldMany p m = fromStreamD $ runArrayFoldManyD p (toStreamD m)

core/streamly-core.cabal+2

@@ -325,6 +325,8 @@ library

325 , Streamly.Internal.Data.Array.Stream.Mut.Foreign

326 , Streamly.Internal.Data.Array.Stream.Fold.Foreign

327

328+ , Streamly.Internal.Data.Array.Stream.Foreign

329+

328 -- Ring Arrays

330 -- Ring Arrays

329 , Streamly.Internal.Data.Ring.Foreign

331 , Streamly.Internal.Data.Ring.Foreign

330 , Streamly.Internal.Data.Ring

332 , Streamly.Internal.Data.Ring

src/Streamly/Internal/FileSystem/FD.hs+33−32

@@ -132,8 +132,6 @@ import Streamly.Internal.Data.Array.Unboxed.Type

132 (Array(..), byteLength, unsafeFreeze, asPtrUnsafe)

133import Streamly.Internal.System.IO (defaultChunkSize)

134import Streamly.Internal.Data.Stream.Type (Stream)

135−import Streamly.Internal.Data.Stream.IsStream.Type

136− (IsStream, mkStream, fromStreamD)

137#if !defined(mingw32_HOST_OS)

135#if !defined(mingw32_HOST_OS)

138{-

136{-

139import Streamly.Internal.Data.Stream.IsStream.Type (toStreamD)

137import Streamly.Internal.Data.Stream.IsStream.Type (toStreamD)

@@ -146,12 +144,15 @@ import qualified Streamly.Internal.System.IOVec.Type as RawIO

146-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

144-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

147

145

148import qualified Streamly.Data.Array.Unboxed as A

146import qualified Streamly.Data.Array.Unboxed as A

147+import qualified Streamly.Data.Fold as FL

149import qualified Streamly.Internal.Data.Array.Unboxed.Mut as MArray

148import qualified Streamly.Internal.Data.Array.Unboxed.Mut as MArray

150 (Array(..), newPinnedArrayBytes, asPtrUnsafe)

149 (Array(..), newPinnedArrayBytes, asPtrUnsafe)

151import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS

150import 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

153import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

152import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

154 (Stream(..), Step(..))

153 (Stream(..), Step(..))

154+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K (mkStream)

155+

155

156

156-------------------------------------------------------------------------------

157-------------------------------------------------------------------------------

157-- References

158-- References

@@ -281,21 +282,21 @@ writeIOVec (Handle fd) iov =

281-- The maximum size of a single array is specified by @size@. The actual size

282-- The maximum size of a single array is specified by @size@. The actual size

282-- read may be less than or equal to @size@.

283-- read may be less than or equal to @size@.

283{-# INLINE _readArraysOfUpto #-}

284{-# INLINE _readArraysOfUpto #-}

284−_readArraysOfUpto :: (IsStream t, MonadIO m)

285+_readArraysOfUpto :: (MonadIO m)

285− => Int -> Handle -> t m (Array Word8)

286+ => Int -> Handle -> Stream m (Array Word8)

286−_readArraysOfUpto size h = go

287+_readArraysOfUpto size h = S.fromStreamK go

287 where

288 where

288 -- XXX use cons/nil instead

289 -- XXX use cons/nil instead

289− go = mkStream $ \_ yld _ stp -> do

290+ go = K.mkStream $ \_ yld _ stp -> do

290 arr <- liftIO $ readArrayUpto size h

291 arr <- liftIO $ readArrayUpto size h

291 if A.length arr == 0

292 if A.length arr == 0

292 then stp

293 then stp

293 else yld arr go

294 else yld arr go

294

295

295{-# INLINE_NORMAL readArraysOfUpto #-}

296{-# INLINE_NORMAL readArraysOfUpto #-}

296−readArraysOfUpto :: (IsStream t, MonadIO m)

297+readArraysOfUpto :: (MonadIO m)

297− => Int -> Handle -> t m (Array Word8)

298+ => Int -> Handle -> Stream m (Array Word8)

298−readArraysOfUpto size h = fromStreamD (D.Stream step ())

299+readArraysOfUpto size h = S.fromStreamD (D.Stream step ())

299 where

300 where

300 {-# INLINE_LATE step #-}

301 {-# INLINE_LATE step #-}

301 step _ _ = do

302 step _ _ = do

@@ -316,7 +317,7 @@ readArraysOfUpto size h = fromStreamD (D.Stream step ())

316--

317--

317-- @since 0.7.0

318-- @since 0.7.0

318{-# INLINE readArrays #-}

319{-# INLINE readArrays #-}

319−readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)

320+readArrays :: (MonadIO m) => Handle -> Stream m (Array Word8)

320readArrays = readArraysOfUpto defaultChunkSize

321readArrays = readArraysOfUpto defaultChunkSize

321

322

322-------------------------------------------------------------------------------

323-------------------------------------------------------------------------------

@@ -332,11 +333,11 @@ readArrays = readArraysOfUpto defaultChunkSize

332-- as EOF is encountered.

333-- as EOF is encountered.

333--

334--

334{-# INLINE readInChunksOf #-}

335{-# INLINE readInChunksOf #-}

335−readInChunksOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

336+readInChunksOf :: (MonadIO m) => Int -> Handle -> Stream m Word8

336readInChunksOf chunkSize h = AS.concat $ readArraysOfUpto chunkSize h

337readInChunksOf chunkSize h = AS.concat $ readArraysOfUpto chunkSize h

337

338

338-- TODO

339-- TODO

339−-- read :: (IsStream t, MonadIO m, Unboxed a) => Handle -> t m a

340+-- read :: (MonadIO m, Unboxed a) => Handle -> Stream m a

340--

341--

341-- > read = 'readByChunks' A.defaultChunkSize

342-- > read = 'readByChunks' A.defaultChunkSize

342-- | Generate a stream of elements of the given type from a file 'Handle'. The

343-- | 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

344--

345--

345-- @since 0.7.0

346-- @since 0.7.0

346{-# INLINE read #-}

347{-# INLINE read #-}

347−read :: (IsStream t, MonadIO m) => Handle -> t m Word8

348+read :: (MonadIO m) => Handle -> Stream m Word8

348read = AS.concat . readArrays

349read = AS.concat . readArrays

349

350

350-------------------------------------------------------------------------------

351-------------------------------------------------------------------------------

@@ -356,7 +357,7 @@ read = AS.concat . readArrays

356-- @since 0.7.0

357-- @since 0.7.0

357{-# INLINE writeArrays #-}

358{-# INLINE writeArrays #-}

358writeArrays :: (MonadIO m, Unboxed a) => Handle -> Stream m (Array a) -> m ()

359writeArrays :: (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))

360

361

361-- | Write a stream of arrays to a handle after coalescing them in chunks of

362-- | Write a stream of arrays to a handle after coalescing them in chunks of

362-- specified size. The chunk size is only a maximum and the actual writes could

363-- specified size. The chunk size is only a maximum and the actual writes could

@@ -440,7 +441,7 @@ write = toHandleWith A.defaultChunkSize

440--

441--

441-- @since 0.7.0

442-- @since 0.7.0

442{-# INLINE readUtf8 #-}

443{-# INLINE readUtf8 #-}

443−readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char

444+readUtf8 :: (MonadIO m) => Handle -> Stream m Char

444readUtf8 = decodeUtf8 . read

445readUtf8 = decodeUtf8 . read

445

446

446-- |

447-- |

@@ -462,7 +463,7 @@ writeUtf8 h s = write h $ encodeUtf8 s

462--

463--

463-- @since 0.7.0

464-- @since 0.7.0

464{-# INLINE writeUtf8ByLines #-}

465{-# INLINE writeUtf8ByLines #-}

465−writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()

466+writeUtf8ByLines :: (MonadIO m) => Handle -> Stream m Char -> m ()

466writeUtf8ByLines = undefined

467writeUtf8ByLines = undefined

467

468

468-- | Read UTF-8 lines from a file handle and apply the specified fold to each

469-- | Read UTF-8 lines from a file handle and apply the specified fold to each

@@ -470,7 +471,7 @@ writeUtf8ByLines = undefined

470--

471--

471-- @since 0.7.0

472-- @since 0.7.0

472{-# INLINE readLines #-}

473{-# 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

474readLines h f = foldLines (readUtf8 h) f

475readLines h f = foldLines (readUtf8 h) f

475

476

476-------------------------------------------------------------------------------

477-------------------------------------------------------------------------------

@@ -483,8 +484,8 @@ readLines h f = foldLines (readUtf8 h) f

483--

484--

484-- @since 0.7.0

485-- @since 0.7.0

485{-# INLINE readFrames #-}

486{-# INLINE readFrames #-}

486−readFrames :: (IsStream t, MonadIO m, Unboxed a)

487+readFrames :: (MonadIO m, Unboxed a)

487− => Array a -> Handle -> Fold m a b -> t m b

488+ => Array a -> Handle -> Fold m a b -> Stream m b

488readFrames = undefined -- foldFrames . read

489readFrames = undefined -- foldFrames . read

489

490

490-- | Write a stream to the given file handle buffering up to frames separated

491-- | Write a stream to the given file handle buffering up to frames separated

@@ -492,8 +493,8 @@ readFrames = undefined -- foldFrames . read

492--

493--

493-- @since 0.7.0

494-- @since 0.7.0

494{-# INLINE writeByFrames #-}

495{-# INLINE writeByFrames #-}

495−writeByFrames :: (IsStream t, MonadIO m, Unboxed a)

496+writeByFrames :: (MonadIO m, Unboxed a)

496− => Array a -> Handle -> t m a -> m ()

497+ => Array a -> Handle -> Stream m a -> m ()

497writeByFrames = undefined

498writeByFrames = undefined

498

499

499-------------------------------------------------------------------------------

500-------------------------------------------------------------------------------

@@ -530,8 +531,8 @@ readIndex arr i = undefined

530-- chunk is aligned with @chunkSize@ from second chunk onwards.

531-- chunk is aligned with @chunkSize@ from second chunk onwards.

531--

532--

532{-# INLINE readSliceWith #-}

533{-# INLINE readSliceWith #-}

533−readSliceWith :: (IsStream t, MonadIO m, Unboxed a)

534+readSliceWith :: (MonadIO m, Unboxed a)

534− => Int -> Handle -> Int -> Int -> t m a

535+ => Int -> Handle -> Int -> Int -> Stream m a

535readSliceWith chunkSize h pos len = undefined

536readSliceWith chunkSize h pos len = undefined

536

537

537-- | @readSlice h i count@ streams a slice from the file handle @h@ starting

538-- | @readSlice h i count@ streams a slice from the file handle @h@ starting

@@ -540,8 +541,8 @@ readSliceWith chunkSize h pos len = undefined

540--

541--

541-- @since 0.7.0

542-- @since 0.7.0

542{-# INLINE readSlice #-}

543{-# INLINE readSlice #-}

543−readSlice :: (IsStream t, MonadIO m, Unboxed a)

544+readSlice :: (MonadIO m, Unboxed a)

544− => Handle -> Int -> Int -> t m a

545+ => Handle -> Int -> Int -> Stream m a

545readSlice = readSliceWith defaultChunkSize

546readSlice = readSliceWith defaultChunkSize

546

547

547-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting

548-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting

@@ -550,8 +551,8 @@ readSlice = readSliceWith defaultChunkSize

550--

551--

551-- @since 0.7.0

552-- @since 0.7.0

552{-# INLINE readSliceRev #-}

553{-# INLINE readSliceRev #-}

553−readSliceRev :: (IsStream t, MonadIO m, Unboxed a)

554+readSliceRev :: (MonadIO m, Unboxed a)

554− => Handle -> Int -> Int -> t m a

555+ => Handle -> Int -> Int -> Stream m a

555readSliceRev h i count = undefined

556readSliceRev h i count = undefined

556

557

557-- | Write the given element at the given index in the file.

558-- | Write the given element at the given index in the file.

@@ -567,8 +568,8 @@ writeIndex h i a = undefined

567--

568--

568-- @since 0.7.0

569-- @since 0.7.0

569{-# INLINE writeSlice #-}

570{-# INLINE writeSlice #-}

570−writeSlice :: (IsStream t, Monad m, Unboxed a)

571+writeSlice :: (Monad m, Unboxed a)

571− => Handle -> Int -> Int -> t m a -> m ()

572+ => Handle -> Int -> Int -> Stream m a -> m ()

572writeSlice h i len s = undefined

573writeSlice h i len s = undefined

573

574

574-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@

575-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@

@@ -577,7 +578,7 @@ writeSlice h i len s = undefined

577--

578--

578-- @since 0.7.0

579-- @since 0.7.0

579{-# INLINE writeSliceRev #-}

580{-# INLINE writeSliceRev #-}

580−writeSliceRev :: (IsStream t, Monad m, Unboxed a)

581+writeSliceRev :: (Monad m, Unboxed a)

581− => Handle -> Int -> Int -> t m a -> m ()

582+ => Handle -> Int -> Int -> Stream m a -> m ()

582writeSliceRev arr i len s = undefined

583writeSliceRev arr i len s = undefined

583-}

584-}

src/Streamly/Internal/FileSystem/Handle.hs+31−32

@@ -124,8 +124,6 @@ import Streamly.Internal.Data.Unfold.Type (Unfold(..))

124import Streamly.Internal.Data.Array.Unboxed.Type

125 (Array(..), writeNUnsafe, unsafeFreezeWithShrink, byteLength)

126import Streamly.Internal.Data.Stream.Type (Stream)

127−import Streamly.Internal.Data.Stream.IsStream.Type

128− (IsStream, mkStream, fromStreamD)

129import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)

127import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)

130-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

128-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

131import Streamly.Internal.System.IO (defaultChunkSize)

129import Streamly.Internal.System.IO (defaultChunkSize)

@@ -139,10 +137,11 @@ import qualified Streamly.Internal.Data.Array.Unboxed.Mut as MArray

139import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS

137import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS

140import qualified Streamly.Internal.Data.Refold.Type as Refold

138import qualified Streamly.Internal.Data.Refold.Type as Refold

141import qualified Streamly.Internal.Data.Fold.Type as FL(refoldMany)

139import 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

143import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

141import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

144 (Stream(..), Step(..))

142 (Stream(..), Step(..))

145import qualified Streamly.Internal.Data.Unfold as UF

143import qualified Streamly.Internal.Data.Unfold as UF

144+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K (mkStream)

146

145

147-- $setup

146-- $setup

148-- >>> import qualified Streamly.Data.Array.Unboxed as Array

147-- >>> import qualified Streamly.Data.Array.Unboxed as Array

@@ -213,12 +212,12 @@ getChunkOf = undefined

213-- read may be less than or equal to @size@.

212-- read may be less than or equal to @size@.

214-- @since 0.9.0

213-- @since 0.9.0

215{-# INLINE _getChunksWith #-}

214{-# INLINE _getChunksWith #-}

216−_getChunksWith :: (IsStream t, MonadIO m)

215+_getChunksWith :: (MonadIO m)

217− => Int -> Handle -> t m (Array Word8)

216+ => Int -> Handle -> Stream m (Array Word8)

218−_getChunksWith size h = go

217+_getChunksWith size h = S.fromStreamK go

219 where

218 where

220 -- XXX use cons/nil instead

219 -- XXX use cons/nil instead

221− go = mkStream $ \_ yld _ stp -> do

220+ go = K.mkStream $ \_ yld _ stp -> do

222 arr <- getChunk size h

221 arr <- getChunk size h

223 if byteLength arr == 0

222 if byteLength arr == 0

224 then stp

223 then stp

@@ -232,9 +231,9 @@ _getChunksWith size h = go

232--

231--

233-- @since 0.9.0

232-- @since 0.9.0

234{-# INLINE_NORMAL getChunksWith #-}

233{-# INLINE_NORMAL getChunksWith #-}

235−getChunksWith :: (IsStream t, MonadIO m) =>

234+getChunksWith :: (MonadIO m) =>

236− Int -> Handle -> t m (Array Word8)

235+ Int -> Handle -> Stream m (Array Word8)

237−getChunksWith size h = fromStreamD (D.Stream step ())

236+getChunksWith size h = S.fromStreamD (D.Stream step ())

238 where

237 where

239 {-# INLINE_LATE step #-}

238 {-# INLINE_LATE step #-}

240 step _ _ = do

239 step _ _ = do

@@ -314,7 +313,7 @@ readChunksFromToWith = Unfold step inject

314--

313--

315-- @since 0.9.0

314-- @since 0.9.0

316{-# INLINE getChunks #-}

315{-# INLINE getChunks #-}

317−getChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)

316+getChunks :: (MonadIO m) => Handle -> Stream m (Array Word8)

318getChunks = getChunksWith defaultChunkSize

317getChunks = getChunksWith defaultChunkSize

319

318

320-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO

319-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO

@@ -363,12 +362,12 @@ readWithBufferOf = readWith

363--

362--

364-- /Pre-release/

363-- /Pre-release/

365{-# INLINE getBytesWith #-}

364{-# INLINE getBytesWith #-}

366−getBytesWith :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

365+getBytesWith :: (MonadIO m) => Int -> Handle -> Stream m Word8

367getBytesWith size h = AS.concat $ getChunksWith size h

366getBytesWith size h = AS.concat $ getChunksWith size h

368

367

369-- TODO

368-- TODO

370-- Generate a stream of elements of the given type from a file 'Handle'.

369-- 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

372--

371--

373-- | Unfolds a file handle into a byte stream. IO requests to the device are

372-- | Unfolds a file handle into a byte stream. IO requests to the device are

374-- performed in sizes of

373-- performed in sizes of

@@ -387,7 +386,7 @@ read = UF.many A.read readChunks

387--

386--

388-- /Pre-release/

387-- /Pre-release/

389{-# INLINE getBytes #-}

388{-# INLINE getBytes #-}

390−getBytes :: (IsStream t, MonadIO m) => Handle -> t m Word8

389+getBytes :: (MonadIO m) => Handle -> Stream m Word8

391getBytes = AS.concat . getChunks

390getBytes = AS.concat . getChunks

392

391

393-------------------------------------------------------------------------------

392-------------------------------------------------------------------------------

@@ -428,7 +427,7 @@ putChunk h arr = A.asPtrUnsafe arr $ \ptr ->

428-- @since 0.7.0

427-- @since 0.7.0

429{-# INLINE putChunks #-}

428{-# INLINE putChunks #-}

430putChunks :: MonadIO m => Handle -> Stream m (Array a) -> m ()

429putChunks :: MonadIO m => Handle -> Stream m (Array a) -> m ()

431−putChunks h = S.mapM_ (putChunk h)

430+putChunks h = S.fold (FL.drainBy (putChunk h))

432

431

433-- XXX AS.compact can be written idiomatically in terms of foldMany, just like

432-- XXX AS.compact can be written idiomatically in terms of foldMany, just like

434-- AS.concat is written in terms of foldMany. Once that is done we can write

433-- AS.concat is written in terms of foldMany. Once that is done we can write

@@ -601,7 +600,7 @@ write = toHandleWith A.defaultChunkSize

601--

600--

602-- @since 0.7.0

601-- @since 0.7.0

603{-# INLINE readUtf8 #-}

602{-# INLINE readUtf8 #-}

604−readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char

603+readUtf8 :: (MonadIO m) => Handle -> Stream m Char

605readUtf8 = decodeUtf8 . read

604readUtf8 = decodeUtf8 . read

606

605

607-- |

606-- |

@@ -623,7 +622,7 @@ writeUtf8 h s = write h $ encodeUtf8 s

623--

622--

624-- @since 0.7.0

623-- @since 0.7.0

625{-# INLINE writeUtf8ByLines #-}

624{-# INLINE writeUtf8ByLines #-}

626−writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()

625+writeUtf8ByLines :: (MonadIO m) => Handle -> Stream m Char -> m ()

627writeUtf8ByLines = undefined

626writeUtf8ByLines = undefined

628

627

629-- | Read UTF-8 lines from a file handle and apply the specified fold to each

628-- | Read UTF-8 lines from a file handle and apply the specified fold to each

@@ -631,7 +630,7 @@ writeUtf8ByLines = undefined

631--

630--

632-- @since 0.7.0

631-- @since 0.7.0

633{-# INLINE readLines #-}

632{-# 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

635readLines h f = foldLines (readUtf8 h) f

634readLines h f = foldLines (readUtf8 h) f

636

635

637-------------------------------------------------------------------------------

636-------------------------------------------------------------------------------

@@ -644,8 +643,8 @@ readLines h f = foldLines (readUtf8 h) f

644--

643--

645-- @since 0.7.0

644-- @since 0.7.0

646{-# INLINE readFrames #-}

645{-# INLINE readFrames #-}

647−readFrames :: (IsStream t, MonadIO m, Unboxed a)

646+readFrames :: (MonadIO m, Unboxed a)

648− => Array a -> Handle -> Fold m a b -> t m b

647+ => Array a -> Handle -> Fold m a b -> Stream m b

649readFrames = undefined -- foldFrames . read

648readFrames = undefined -- foldFrames . read

650

649

651-- | Write a stream to the given file handle buffering up to frames separated

650-- | Write a stream to the given file handle buffering up to frames separated

@@ -653,8 +652,8 @@ readFrames = undefined -- foldFrames . read

653--

652--

654-- @since 0.7.0

653-- @since 0.7.0

655{-# INLINE writeByFrames #-}

654{-# INLINE writeByFrames #-}

656−writeByFrames :: (IsStream t, MonadIO m, Unboxed a)

655+writeByFrames :: (MonadIO m, Unboxed a)

657− => Array a -> Handle -> t m a -> m ()

656+ => Array a -> Handle -> Stream m a -> m ()

658writeByFrames = undefined

657writeByFrames = undefined

659

658

660-------------------------------------------------------------------------------

659-------------------------------------------------------------------------------

@@ -708,8 +707,8 @@ readIndex arr i = undefined

708-- chunk is aligned with @chunkSize@ from second chunk onwards.

707-- chunk is aligned with @chunkSize@ from second chunk onwards.

709--

708--

710{-# INLINE readSliceWith #-}

709{-# INLINE readSliceWith #-}

711−readSliceWith :: (IsStream t, MonadIO m, Unboxed a)

710+readSliceWith :: (MonadIO m, Unboxed a)

712− => Int -> Handle -> Int -> Int -> t m a

711+ => Int -> Handle -> Int -> Int -> Stream m a

713readSliceWith chunkSize h pos len = undefined

712readSliceWith chunkSize h pos len = undefined

714

713

715-- | @readSlice h i count@ streams a slice from the file handle @h@ starting

714-- | @readSlice h i count@ streams a slice from the file handle @h@ starting

@@ -718,8 +717,8 @@ readSliceWith chunkSize h pos len = undefined

718--

717--

719-- @since 0.7.0

718-- @since 0.7.0

720{-# INLINE readSlice #-}

719{-# INLINE readSlice #-}

721−readSlice :: (IsStream t, MonadIO m, Unboxed a)

720+readSlice :: (MonadIO m, Unboxed a)

722− => Handle -> Int -> Int -> t m a

721+ => Handle -> Int -> Int -> Stream m a

723readSlice = readSliceWith A.defaultChunkSize

722readSlice = readSliceWith A.defaultChunkSize

724

723

725-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting

724-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting

@@ -728,8 +727,8 @@ readSlice = readSliceWith A.defaultChunkSize

728--

727--

729-- @since 0.7.0

728-- @since 0.7.0

730{-# INLINE readSliceRev #-}

729{-# INLINE readSliceRev #-}

731−readSliceRev :: (IsStream t, MonadIO m, Unboxed a)

730+readSliceRev :: (MonadIO m, Unboxed a)

732− => Handle -> Int -> Int -> t m a

731+ => Handle -> Int -> Int -> Stream m a

733readSliceRev h i count = undefined

732readSliceRev h i count = undefined

734

733

735-- | Write the given element at the given index in the file.

734-- | Write the given element at the given index in the file.

@@ -745,8 +744,8 @@ writeIndex h i a = undefined

745--

744--

746-- @since 0.7.0

745-- @since 0.7.0

747{-# INLINE writeSlice #-}

746{-# INLINE writeSlice #-}

748−writeSlice :: (IsStream t, Monad m, Unboxed a)

747+writeSlice :: (Monad m, Unboxed a)

749− => Handle -> Int -> Int -> t m a -> m ()

748+ => Handle -> Int -> Int -> Stream m a -> m ()

750writeSlice h i len s = undefined

749writeSlice h i len s = undefined

751

750

752-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@

751-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@

@@ -755,7 +754,7 @@ writeSlice h i len s = undefined

755--

754--

756-- @since 0.7.0

755-- @since 0.7.0

757{-# INLINE writeSliceRev #-}

756{-# INLINE writeSliceRev #-}

758−writeSliceRev :: (IsStream t, Monad m, Unboxed a)

757+writeSliceRev :: (Monad m, Unboxed a)

759− => Handle -> Int -> Int -> t m a -> m ()

758+ => Handle -> Int -> Int -> Stream m a -> m ()

760writeSliceRev arr i len s = undefined

759writeSliceRev arr i len s = undefined

761-}

760-}

src/Streamly/Internal/Network/Inet/TCP.hs+9−10

@@ -107,7 +107,6 @@ import Streamly.Internal.Control.Concurrent (MonadAsync)

107import Streamly.Internal.Control.ForkLifted (fork)

108import Streamly.Internal.Data.Array.Unboxed.Type (Array(..), writeNUnsafe)

109import Streamly.Internal.Data.Fold.Type (Fold(..))

110−import Streamly.Internal.Data.Stream.IsStream.Type (IsStream)

111import Streamly.Internal.Data.Stream (Stream)

110import Streamly.Internal.Data.Stream (Stream)

112import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

111import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

113import Streamly.Internal.Data.Unfold.Type (Unfold(..))

112import Streamly.Internal.Data.Unfold.Type (Unfold(..))

@@ -302,8 +301,8 @@ usingConnection =

302--

301--

303-- /Pre-release/

302-- /Pre-release/

304{-# INLINE withConnection #-}

303{-# INLINE withConnection #-}

305−withConnection :: (IsStream t, MonadCatch m, MonadAsync m)

304+withConnection :: (MonadCatch m, MonadAsync m)

306− => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> t m a) -> t m a

305+ => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> Stream m a) -> Stream m a

307withConnection addr port =

306withConnection addr port =

308 S.bracket (liftIO $ connect addr port) (liftIO . Net.close)

307 S.bracket (liftIO $ connect addr port) (liftIO . Net.close)

309

308

@@ -323,8 +322,8 @@ read = UF.many A.read (usingConnection ISK.readChunks)

323--

322--

324-- @since 0.7.0

323-- @since 0.7.0

325{-# INLINE toBytes #-}

324{-# INLINE toBytes #-}

326−toBytes :: (IsStream t, MonadCatch m, MonadAsync m)

325+toBytes :: (MonadCatch m, MonadAsync m)

327− => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8

326+ => (Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Word8

328toBytes addr port = AS.concat $ withConnection addr port ISK.toChunks

327toBytes addr port = AS.concat $ withConnection addr port ISK.toChunks

329

328

330-------------------------------------------------------------------------------

329-------------------------------------------------------------------------------

@@ -424,12 +423,12 @@ write = writeWithBufferOf defaultChunkSize

424

423

425{-# INLINE withInputConnect #-}

424{-# INLINE withInputConnect #-}

426withInputConnect

425withInputConnect

427− :: (IsStream t, MonadCatch m, MonadAsync m)

426+ :: (MonadCatch m, MonadAsync m)

428 => (Word8, Word8, Word8, Word8)

427 => (Word8, Word8, Word8, Word8)

429 -> PortNumber

428 -> PortNumber

430 -> Stream m Word8

429 -> Stream m Word8

431− -> (Socket -> t m a)

430+ -> (Socket -> Stream m a)

432− -> t m a

431+ -> Stream m a

433withInputConnect addr port input f = S.bracket pre post handler

432withInputConnect addr port input f = S.bracket pre post handler

434

433

435 where

434 where

@@ -452,9 +451,9 @@ withInputConnect addr port input f = S.bracket pre post handler

452--

451--

453{-# INLINE processBytes #-}

452{-# INLINE processBytes #-}

454processBytes

453processBytes

455− :: (IsStream t, MonadAsync m, MonadCatch m)

454+ :: (MonadAsync m, MonadCatch m)

456 => (Word8, Word8, Word8, Word8)

455 => (Word8, Word8, Word8, Word8)

457 -> PortNumber

456 -> PortNumber

458 -> Stream m Word8

457 -> Stream m Word8

459− -> t m Word8

458+ -> Stream m Word8

460processBytes addr port input = withInputConnect addr port input ISK.toBytes

459processBytes 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)

94import Streamly.Internal.Data.Array.Unboxed.Type (Array(..))

95import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)

96import Streamly.Internal.Data.Fold (Fold)

97−import Streamly.Internal.Data.Stream.IsStream.Type

98− (IsStream, mkStream, fromStreamD)

99import Streamly.Internal.Data.Stream.Type (Stream)

97import Streamly.Internal.Data.Stream.Type (Stream)

100import Streamly.Internal.Data.Unfold.Type (Unfold(..))

98import Streamly.Internal.Data.Unfold.Type (Unfold(..))

101-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

99-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

@@ -108,11 +106,12 @@ import qualified Streamly.Internal.Data.Array.Unboxed.Type as A

108import qualified Streamly.Internal.Data.Array.Unboxed.Mut as MArray

106import qualified Streamly.Internal.Data.Array.Unboxed.Mut as MArray

109 (Array(..), newPinnedArrayBytes, asPtrUnsafe)

107 (Array(..), newPinnedArrayBytes, asPtrUnsafe)

110import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS

108import 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

112import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

110import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

113 (Stream(..), Step(..))

111 (Stream(..), Step(..))

114import qualified Streamly.Data.Unfold as UF

112import qualified Streamly.Data.Unfold as UF

115import qualified Streamly.Internal.Data.Unfold as UF (first, map)

113import qualified Streamly.Internal.Data.Unfold as UF (first, map)

114+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K (mkStream)

116

115

117-- | @'forSocketM' action socket@ runs the monadic computation @action@ passing

116-- | @'forSocketM' action socket@ runs the monadic computation @action@ passing

118-- the socket handle to it. The handle will be closed on exit from

117-- 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))

132--

131--

133-- /Internal/

132-- /Internal/

134{-# INLINE withSocket #-}

133{-# INLINE withSocket #-}

135−withSocket :: (IsStream t, MonadAsync m, MonadCatch m)

134+withSocket :: (MonadAsync m, MonadCatch m)

136− => Socket -> (Socket -> t m a) -> t m a

135+ => Socket -> (Socket -> Stream m a) -> Stream m a

137withSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)

136withSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)

138

137

139-------------------------------------------------------------------------------

138-------------------------------------------------------------------------------

@@ -331,13 +330,13 @@ writeChunk = writeArrayWith sendAll

331-------------------------------------------------------------------------------

330-------------------------------------------------------------------------------

332

331

333{-# INLINE _readChunksUptoWith #-}

332{-# INLINE _readChunksUptoWith #-}

334−_readChunksUptoWith :: (IsStream t, MonadIO m)

333+_readChunksUptoWith :: (MonadIO m)

335 => (Int -> h -> IO (Array Word8))

334 => (Int -> h -> IO (Array Word8))

336− -> Int -> h -> t m (Array Word8)

335+ -> Int -> h -> Stream m (Array Word8)

337−_readChunksUptoWith f size h = go

336+_readChunksUptoWith f size h = S.fromStreamK go

338 where

337 where

339 -- XXX use cons/nil instead

338 -- XXX use cons/nil instead

340− go = mkStream $ \_ yld _ stp -> do

339+ go = K.mkStream $ \_ yld _ stp -> do

341 arr <- liftIO $ f size h

340 arr <- liftIO $ f size h

342 if A.length arr == 0

341 if A.length arr == 0

343 then stp

342 then stp

@@ -348,10 +347,10 @@ _readChunksUptoWith f size h = go

348-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'

347-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'

349-- on the 'Handle'.

348-- on the 'Handle'.

350{-# INLINE_NORMAL toChunksWith #-}

349{-# INLINE_NORMAL toChunksWith #-}

351−toChunksWith :: (IsStream t, MonadIO m)

350+toChunksWith :: (MonadIO m)

352− => Int -> Socket -> t m (Array Word8)

351+ => Int -> Socket -> Stream m (Array Word8)

353-- toChunksWith = _readChunksUptoWith readChunk

352-- toChunksWith = _readChunksUptoWith readChunk

354−toChunksWith size h = fromStreamD (D.Stream step ())

353+toChunksWith size h = S.fromStreamD (D.Stream step ())

355 where

354 where

356 {-# INLINE_LATE step #-}

355 {-# INLINE_LATE step #-}

357 step _ _ = do

356 step _ _ = do

@@ -368,7 +367,7 @@ toChunksWith size h = fromStreamD (D.Stream step ())

368--

367--

369-- @since 0.7.0

368-- @since 0.7.0

370{-# INLINE toChunks #-}

369{-# INLINE toChunks #-}

371−toChunks :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)

370+toChunks :: (MonadIO m) => Socket -> Stream m (Array Word8)

372toChunks = toChunksWith defaultChunkSize

371toChunks = toChunksWith defaultChunkSize

373

372

374-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.

373-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.

@@ -422,12 +421,12 @@ readChunks = UF.first defaultChunkSize readChunksWith

422-- as soon as EOF is encountered.

421-- as soon as EOF is encountered.

423--

422--

424{-# INLINE readWith #-}

423{-# INLINE readWith #-}

425−readWith :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

424+readWith :: (MonadIO m) => Int -> Handle -> Stream m Word8

426readWith chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h

425readWith chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h

427-}

426-}

428

427

429-- TODO

428-- TODO

430−-- read :: (IsStream t, MonadIO m, Unboxed a) => Handle -> t m a

429+-- read :: (MonadIO m, Unboxed a) => Handle -> Stream m a

431--

430--

432-- > read = 'readByChunks' defaultChunkSize

431-- > read = 'readByChunks' defaultChunkSize

433-- | Generate a stream of elements of the given type from a socket. The

432-- | 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

435--

434--

436-- @since 0.7.0

435-- @since 0.7.0

437{-# INLINE toBytes #-}

436{-# INLINE toBytes #-}

438−toBytes :: (IsStream t, MonadIO m) => Socket -> t m Word8

437+toBytes :: (MonadIO m) => Socket -> Stream m Word8

439toBytes = AS.concat . toChunks

438toBytes = AS.concat . toChunks

440

439

441-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests

440-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests

@@ -473,7 +472,7 @@ read = UF.first defaultChunkSize readWith

473{-# INLINE putChunks #-}

472{-# INLINE putChunks #-}

474putChunks :: (MonadIO m, Unboxed a)

473putChunks :: (MonadIO m, Unboxed a)

475 => Socket -> Stream m (Array a) -> m ()

474 => Socket -> Stream m (Array a) -> m ()

476−putChunks h = S.mapM_ (liftIO . writeChunk h)

475+putChunks h = S.fold (FL.drainBy (liftIO . writeChunk h))

477

476

478-- | Write a stream of arrays to a socket. Each array in the stream is written

477-- | Write a stream of arrays to a socket. Each array in the stream is written

479-- to the socket as a separate IO request.

478-- to the socket as a separate IO request.

@@ -591,7 +590,7 @@ write = toHandleWith defaultChunkSize

591--

590--

592-- @since 0.7.0

591-- @since 0.7.0

593{-# INLINE readUtf8 #-}

592{-# INLINE readUtf8 #-}

594−readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char

593+readUtf8 :: (MonadIO m) => Handle -> Stream m Char

595readUtf8 = decodeUtf8 . read

594readUtf8 = decodeUtf8 . read

596

595

597-- |

596-- |

@@ -613,7 +612,7 @@ writeUtf8 h s = write h $ encodeUtf8 s

613--

612--

614-- @since 0.7.0

613-- @since 0.7.0

615{-# INLINE writeUtf8ByLines #-}

614{-# INLINE writeUtf8ByLines #-}

616−writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()

615+writeUtf8ByLines :: (MonadIO m) => Handle -> Stream m Char -> m ()

617writeUtf8ByLines = undefined

616writeUtf8ByLines = undefined

618

617

619-- | Read UTF-8 lines from a file handle and apply the specified fold to each

618-- | Read UTF-8 lines from a file handle and apply the specified fold to each

@@ -621,7 +620,7 @@ writeUtf8ByLines = undefined

621--

620--

622-- @since 0.7.0

621-- @since 0.7.0

623{-# INLINE readLines #-}

622{-# 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

625readLines h f = foldLines (readUtf8 h) f

624readLines h f = foldLines (readUtf8 h) f

626

625

627-------------------------------------------------------------------------------

626-------------------------------------------------------------------------------

@@ -634,8 +633,8 @@ readLines h f = foldLines (readUtf8 h) f

634--

633--

635-- @since 0.7.0

634-- @since 0.7.0

636{-# INLINE readFrames #-}

635{-# INLINE readFrames #-}

637−readFrames :: (IsStream t, MonadIO m, Unboxed a)

636+readFrames :: (MonadIO m, Unboxed a)

638− => Array a -> Handle -> Fold m a b -> t m b

637+ => Array a -> Handle -> Fold m a b -> Stream m b

639readFrames = undefined -- foldFrames . read

638readFrames = undefined -- foldFrames . read

640

639

641-- | Write a stream to the given file handle buffering up to frames separated

640-- | Write a stream to the given file handle buffering up to frames separated

@@ -643,7 +642,7 @@ readFrames = undefined -- foldFrames . read

643--

642--

644-- @since 0.7.0

643-- @since 0.7.0

645{-# INLINE writeByFrames #-}

644{-# INLINE writeByFrames #-}

646−writeByFrames :: (IsStream t, MonadIO m, Unboxed a)

645+writeByFrames :: (MonadIO m, Unboxed a)

647− => Array a -> Handle -> t m a -> m ()

646+ => Array a -> Handle -> Stream m a -> m ()

648writeByFrames = undefined

647writeByFrames = undefined

649-}

648-}

streamly.cabal+1−2

@@ -205,6 +205,7 @@ extra-source-files:

205 core/src/Streamly/Internal/Data/List.hs

206 core/src/Streamly/Internal/Data/Array.hs

207 core/src/Streamly/Internal/Data/Array/Unboxed.hs

208+ core/src/Streamly/Internal/Data/Array/Stream/Foreign.hs

208 core/src/Streamly/Internal/Data/Array/Stream/Mut/Foreign.hs

209 core/src/Streamly/Internal/Data/Array/Stream/Mut/Foreign.hs

209 core/src/Streamly/Internal/Data/Array/Stream/Fold/Foreign.hs

210 core/src/Streamly/Internal/Data/Array/Stream/Fold/Foreign.hs

210 core/src/Streamly/Internal/Data/Ring/Foreign.hs

211 core/src/Streamly/Internal/Data/Ring/Foreign.hs

@@ -475,8 +476,6 @@ library

475 , Streamly.Internal.Data.Stream.IsStream.Async

476 , Streamly.Internal.Data.Stream.IsStream.Async

476 , Streamly.Internal.Data.Stream.IsStream

477 , Streamly.Internal.Data.Stream.IsStream

477

478

478− , Streamly.Internal.Data.Array.Stream.Foreign

479−

480 -- streamly-serde

479 -- streamly-serde

481 , Streamly.Internal.Data.Binary.Decode

480 , Streamly.Internal.Data.Binary.Decode

482

481