5c4bbc5 · streamly · git.iko.soy

git.iko.soy/streamly

Add write folds, fromBytes/toBytes for File module

commit5c4bbc5885c65ff8ecf01bd29f527b00367c8c53 authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 14 Nov 2019, 03:14 (7 years ago) parent89ddf14

7 files changed+205−102

compare with master

unifiedsplit

examples/CamelCase.hs+2−2 examples/FileIOExamples.hs+6−6 src/Streamly/Benchmark/FileIO/Stream.hs+3−3 src/Streamly/Internal/FileSystem/File.hs+110−16 src/Streamly/Internal/FileSystem/Handle.hs+19−19 src/Streamly/Internal/Network/Inet/TCP.hs+48−43 src/Streamly/Internal/Network/Socket.hs+17−13

examples/CamelCase.hs+2−2

@@ -10,12 +10,12 @@ import qualified Streamly.Internal.FileSystem.Handle as FH

1010

1111camelCase :: Handle -> Handle -> IO ()

1212camelCase src dst =

13− FH.fromStream dst

13+ FH.fromBytes dst

1414 $ S.map fromJust

1515 $ S.filter isJust

1616 $ S.map snd

1717 $ S.scanl' step (True, Nothing)

18− $ FH.toStream src

18+ $ FH.toBytes src

1919

2020 where

2121

examples/FileIOExamples.hs+6−6

@@ -9,12 +9,12 @@ import System.Environment (getArgs)

1010cat :: FilePath -> IO ()

1111cat src =

12− File.writeChunks "/dev/stdout"

12+ File.fromChunks "/dev/stdout"

1313 $ File.toChunksWithBufferOf (256*1024) src

1414

1515cp :: FilePath -> FilePath -> IO ()

1616cp src dst =

17− File.writeChunks dst

17+ File.fromChunks dst

1818 $ File.toChunksWithBufferOf (256*1024) src

1919

2020append :: FilePath -> FilePath -> IO ()

@@ -28,17 +28,17 @@ ord' = (fromIntegral . ord)

2828wcl :: FilePath -> IO ()

2929wcl src = print =<< (S.length

3030 $ S.splitOnSuffix (== ord' '\n') FL.drain

31− $ File.read src)

31+ $ File.toBytes src)

3232

3333grepc :: String -> FilePath -> IO ()

3434grepc pat src = print . (subtract 1) =<< (S.length

3535 $ IP.splitOnSeq (A.fromList (map ord' pat)) FL.drain

36− $ File.read src)

36+ $ File.toBytes src)

3737

3838avgll :: FilePath -> IO ()

3939avgll src = print =<< (S.fold avg

4040 $ S.splitOnSuffix (== ord' '\n') FL.length

41− $ File.read src)

41+ $ File.toBytes src)

4242 where avg = (/) <$> toDouble FL.sum <*> toDouble FL.length

4343 toDouble = fmap (fromIntegral :: Int -> Double)

4444

@@ -46,7 +46,7 @@ llhisto :: FilePath -> IO ()

4646llhisto src = print =<< (S.fold (FL.classify FL.length)

4747 $ S.map bucket

4848 $ S.splitOnSuffix (== ord' '\n') FL.length

49− $ File.read src)

49+ $ File.toBytes src)

5050 where

5151 bucket n = let i = n `div` 10 in if i > 9 then (9,n) else (i,n)

5252

src/Streamly/Benchmark/FileIO/Stream.hs+3−3

@@ -188,7 +188,7 @@ inspect $ 'cat `hasNoType` ''D.ConcatMapUState

188188-- | Send the file contents to /dev/null

189189{-# INLINE catStreamWrite #-}

190190catStreamWrite :: Handle -> Handle -> IO ()

191−catStreamWrite devNull inh = IFH.fromStream devNull $ S.unfold FH.read inh

191+catStreamWrite devNull inh = IFH.fromBytes devNull $ S.unfold FH.read inh

192192

193193#ifdef INSPECTION

194194inspect $ hasNoTypeClasses 'catStreamWrite

@@ -216,8 +216,8 @@ inspect $ hasNoTypeClasses 'catBracket

216216catBracketStream :: Handle -> Handle -> IO ()

217217catBracketStream devNull inh =

218218 let readEx = S.bracket (return ()) (\_ -> hClose inh)

219− (\_ -> IFH.toStream inh)

220− in IFH.fromStream devNull $ readEx

219+ (\_ -> IFH.toBytes inh)

220+ in IFH.fromBytes devNull $ readEx

221221

222222#ifdef INSPECTION

223223inspect $ hasNoTypeClasses 'catBracketStream

src/Streamly/Internal/FileSystem/File.hs+110−16

@@ -61,11 +61,12 @@ module Streamly.Internal.FileSystem.File

6161 -- , readFrames

6262 -- , readChunks

6363

64+ , toBytes

65+

6466 -- -- * Array Read

6567 -- , readArrayOf

6668

6769 , toChunksWithBufferOf

68− -- , toChunksWithBufferOf

6970 , toChunks

7071

7172 -- ** Write To File

@@ -75,8 +76,12 @@ module Streamly.Internal.FileSystem.File

7576 -- , writeByFrames

7677 , writeWithBufferOf

7778

79+ , fromBytes

80+ , fromBytesWithBufferOf

81+

7882 -- -- * Array Write

7983 , writeArray

84+ , writeChunks

8085 , fromChunks

8186

8287 -- ** Append To File

@@ -95,17 +100,24 @@ import Foreign.Storable (Storable(..))

95100import System.IO (Handle, openFile, IOMode(..), hClose)

96101import Prelude hiding (read)

97102

103+import qualified Control.Monad.Catch as MC

98104import qualified System.IO as SIO

99105

100−import Streamly.Internal.Memory.Array.Types (Array(..), defaultChunkSize)

106+import Streamly.Internal.Data.Fold.Types (Fold(..))

107+import Streamly.Internal.Data.Unfold.Types (Unfold(..))

108+import Streamly.Internal.Memory.Array.Types

109+ (Array(..), defaultChunkSize, writeNUnsafe)

101110import Streamly.Streams.Serial (SerialT)

102111import Streamly.Streams.StreamK.Type (IsStream)

103112import Streamly.Internal.Data.SVar (MonadAsync)

104113-- import Streamly.Data.Fold (Fold)

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

106115

116+import qualified Streamly.Internal.Data.Fold.Types as FL

117+import qualified Streamly.Internal.Data.Unfold as UF

107118import qualified Streamly.Internal.FileSystem.Handle as FH

108119import qualified Streamly.Internal.Memory.ArrayStream as AS

120+import qualified Streamly.Memory.Array as A

109121import qualified Streamly.Prelude as S

110122

111123-------------------------------------------------------------------------------

@@ -131,11 +143,29 @@ import qualified Streamly.Prelude as S

131143-- raising an exception. If closing the handle raises an exception, then

132144-- this exception will be raised by 'withFile' rather than any exception

133145-- raised by 'act'.

146+--

147+-- /Internal/

148+--

134149{-# INLINE withFile #-}

135150withFile :: (IsStream t, MonadCatch m, MonadIO m)

136151 => FilePath -> IOMode -> (Handle -> t m a) -> t m a

137152withFile file mode = S.bracket (liftIO $ openFile file mode) (liftIO . hClose)

138153

154+-- | Transform an 'Unfold' from a 'Handle' to an unfold from a 'FilePath'. The

155+-- resulting unfold opens a handle in 'ReadMode', uses it using the supplied

156+-- unfold and then makes sure that the handle is closed on normal termination

157+-- or in case of an exception. If closing the handle raises an exception, then

158+-- this exception will be raised by 'usingFile'.

159+--

160+-- /Internal/

161+--

162+{-# INLINABLE usingFile #-}

163+usingFile :: (MonadCatch m, MonadIO m)

164+ => Unfold m Handle a -> Unfold m FilePath a

165+usingFile =

166+ UF.bracket (\file -> liftIO $ openFile file ReadMode)

167+ (liftIO . hClose)

168+

139169-------------------------------------------------------------------------------

140170-- Array IO (Input)

141171-------------------------------------------------------------------------------

@@ -195,6 +225,25 @@ toChunks = toChunksWithBufferOf defaultChunkSize

195225-- read requests at the same time. For serial case we can use async IO. We can

196226-- also control the read throughput in mbps or IOPS.

197227

228+{-

229+-- | Unfolds the tuple @(bufsize, filepath)@ into a byte stream, read requests

230+-- to the IO device are performed using buffers of @bufsize@.

231+--

232+-- @since 0.7.0

233+{-# INLINE readWithBufferOf #-}

234+readWithBufferOf :: MonadIO m => Unfold m (Int, FilePath) Word8

235+readWithBufferOf = UF.concat (usingFilexxx FH.readChunksWithBufferOf) A.read

236+-}

237+

238+-- | Unfolds a file path into a byte stream. IO requests to the device are

239+-- performed in sizes of

240+-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.

241+--

242+-- @since 0.7.0

243+{-# INLINE read #-}

244+read :: (MonadCatch m, MonadIO m) => Unfold m FilePath Word8

245+read = UF.concat (usingFile FH.readChunks) A.read

246+

198247{-

199248-- | @readInChunksOf chunkSize handle@ reads a byte stream from a file

200249-- handle, reads are performed in chunks of up to @chunkSize@. The stream ends

@@ -209,14 +258,15 @@ readInChunksOf chunkSize h = A.flattenArrays $ toChunksWithBufferOf chunkSize h

209258-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a

210259--

211260-- > read = 'readByChunks' defaultChunkSize

212−-- | Generate a stream of elements of the given type from a file specified by

213−-- path. The stream ends when EOF is encountered. File is locked using multiple

214−-- reader and single writer locking mode.

261+-- | Generate a stream of bytes from a file specified by path. The stream ends

262+-- when EOF is encountered. File is locked using multiple reader and single

263+-- writer locking mode.

215264--

216−-- @since 0.7.0

217−{-# INLINE read #-}

218−read :: (IsStream t, MonadCatch m, MonadIO m) => FilePath -> t m Word8

219−read file = AS.concat $ withFile file ReadMode FH.toChunks

265+-- /Internal/

266+--

267+{-# INLINE toBytes #-}

268+toBytes :: (IsStream t, MonadCatch m, MonadIO m) => FilePath -> t m Word8

269+toBytes file = AS.concat $ withFile file ReadMode FH.toChunks

220270

221271{-

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

@@ -269,10 +319,10 @@ fromChunks = fromChunksMode WriteMode

269319-- input elements.

270320--

271321-- @since 0.7.0

272−{-# INLINE writeWithBufferOf #-}

273−writeWithBufferOf :: (MonadAsync m, MonadCatch m)

322+{-# INLINE fromBytesWithBufferOf #-}

323+fromBytesWithBufferOf :: (MonadAsync m, MonadCatch m)

274324 => Int -> FilePath -> SerialT m Word8 -> m ()

275−writeWithBufferOf n file xs = fromChunks file $ AS.arraysOf n xs

325+fromBytesWithBufferOf n file xs = fromChunks file $ AS.arraysOf n xs

276326

277327-- > write = 'writeWithBufferOf' defaultChunkSize

278328--

@@ -281,10 +331,10 @@ writeWithBufferOf n file xs = fromChunks file $ AS.arraysOf n xs

281331-- truncated to zero size before writing. If the file does not exist it is

282332-- created. File is locked using single writer locking mode.

283333--

284−-- @since 0.7.0

285−{-# INLINE write #-}

286−write :: (MonadAsync m, MonadCatch m) => FilePath -> SerialT m Word8 -> m ()

287−write = writeWithBufferOf defaultChunkSize

334+-- /Internal/

335+{-# INLINE fromBytes #-}

336+fromBytes :: (MonadAsync m, MonadCatch m) => FilePath -> SerialT m Word8 -> m ()

337+fromBytes = fromBytesWithBufferOf defaultChunkSize

288338

289339{-

290340{-# INLINE write #-}

@@ -292,6 +342,50 @@ write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()

292342write = toHandleWith A.defaultChunkSize

293343-}

294344

345+-- | Write a stream of chunks to a handle. Each chunk in the stream is written

346+-- to the device as a separate IO request.

347+--

348+-- /Internal/

349+{-# INLINE writeChunks #-}

350+writeChunks :: (MonadIO m, MonadCatch m, Storable a)

351+ => FilePath -> Fold m (Array a) ()

352+writeChunks path = Fold step initial extract

353+ where

354+ initial = do

355+ h <- liftIO (openFile path WriteMode)

356+ fld <- FL.initialize (FH.writeChunks h)

357+ `MC.onException` (liftIO $ hClose h)

358+ return (fld, h)

359+ step (fld, h) x = do

360+ r <- FL.runStep fld x `MC.onException` (liftIO $ hClose h)

361+ return (r, h)

362+ extract ((Fold _ initial1 extract1), h) = do

363+ liftIO $ hClose h

364+ initial1 >>= extract1

365+

366+-- | @writeWithBufferOf chunkSize handle@ writes the input stream to @handle@.

367+-- Bytes in the input stream are collected into a buffer until we have a chunk

368+-- of size @chunkSize@ and then written to the IO device.

369+--

370+-- /Internal/

371+{-# INLINE writeWithBufferOf #-}

372+writeWithBufferOf :: (MonadIO m, MonadCatch m)

373+ => Int -> FilePath -> Fold m Word8 ()

374+writeWithBufferOf n path =

375+ FL.lchunksOf n (writeNUnsafe n) (writeChunks path)

376+

377+-- > write = 'writeWithBufferOf' A.defaultChunkSize

378+--

379+-- | Write a byte stream to a file. Accumulates the input in chunks of up to

380+-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing to

381+-- the IO device.

382+--

383+-- /Internal/

384+--

385+{-# INLINE write #-}

386+write :: (MonadIO m, MonadCatch m) => FilePath -> Fold m Word8 ()

387+write = writeWithBufferOf defaultChunkSize

388+

295389-- | Append a stream of arrays to a file.

296390--

297391-- @since 0.7.0

src/Streamly/Internal/FileSystem/Handle.hs+19−19

@@ -25,8 +25,8 @@ module Streamly.Internal.FileSystem.Handle

2525 -- , readFrames

2626 , readWithBufferOf

2727

28− , toStream

29− , toStreamWithBufferOf

28+ , toBytes

29+ , toBytesWithBufferOf

3030

3131 -- -- * Array Read

3232 -- , readArrayUpto

@@ -47,8 +47,8 @@ module Streamly.Internal.FileSystem.Handle

4747 , writeWithBufferOf

4848

4949 -- Byte stream write (Streams)

50− , fromStream

51− , fromStreamWithBufferOf

50+ , fromBytes

51+ , fromBytesWithBufferOf

5252

5353 -- -- * Array Write

5454 , writeArray

@@ -270,13 +270,13 @@ readChunks = UF.supplyFirst readChunksWithBufferOf defaultChunkSize

270270readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8

271271readWithBufferOf = UF.concat readChunksWithBufferOf A.read

272272

273−-- | @toStreamWithBufferOf bufsize handle@ reads a byte stream from a file

273+-- | @toBytesWithBufferOf bufsize handle@ reads a byte stream from a file

274274-- handle, reads are performed in chunks of up to @bufsize@.

275275--

276276-- /Internal/

277−{-# INLINE toStreamWithBufferOf #-}

278−toStreamWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

279−toStreamWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h

277+{-# INLINE toBytesWithBufferOf #-}

278+toBytesWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

279+toBytesWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h

280280

281281-- TODO

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

@@ -294,9 +294,9 @@ read = UF.supplyFirst readWithBufferOf defaultChunkSize

294294-- | Generate a byte stream from a file 'Handle'.

295295--

296296-- /Internal/

297−{-# INLINE toStream #-}

298−toStream :: (IsStream t, MonadIO m) => Handle -> t m Word8

299−toStream = AS.concat . toChunks

297+{-# INLINE toBytes #-}

298+toBytes :: (IsStream t, MonadIO m) => Handle -> t m Word8

299+toBytes = AS.concat . toChunks

300300

301301-------------------------------------------------------------------------------

302302-- Writing

@@ -353,15 +353,15 @@ fromChunksWithBufferOf :: (MonadIO m, Storable a)

353353 => Int -> Handle -> SerialT m (Array a) -> m ()

354354fromChunksWithBufferOf n h xs = fromChunks h $ AS.compact n xs

355355

356−-- | @fromStreamWithBufferOf bufsize handle stream@ writes @stream@ to @handle@

356+-- | @fromBytesWithBufferOf bufsize handle stream@ writes @stream@ to @handle@

357357-- in chunks of @bufsize@. A write is performed to the IO device as soon as we

358358-- collect the required input size.

359359--

360360-- @since 0.7.0

361−{-# INLINE fromStreamWithBufferOf #-}

362−fromStreamWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()

363−fromStreamWithBufferOf n h m = fromChunks h $ S.arraysOf n m

364−-- fromStreamWithBufferOf n h m = fromChunks h $ AS.arraysOf n m

361+{-# INLINE fromBytesWithBufferOf #-}

362+fromBytesWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()

363+fromBytesWithBufferOf n h m = fromChunks h $ S.arraysOf n m

364+-- fromBytesWithBufferOf n h m = fromChunks h $ AS.arraysOf n m

365365

366366-- > write = 'writeWithBufferOf' A.defaultChunkSize

367367--

@@ -372,9 +372,9 @@ fromStreamWithBufferOf n h m = fromChunks h $ S.arraysOf n m

372372-- need some extra perf boost.

373373--

374374-- @since 0.7.0

375−{-# INLINE fromStream #-}

376−fromStream :: MonadIO m => Handle -> SerialT m Word8 -> m ()

377−fromStream = fromStreamWithBufferOf defaultChunkSize

375+{-# INLINE fromBytes #-}

376+fromBytes :: MonadIO m => Handle -> SerialT m Word8 -> m ()

377+fromBytes = fromBytesWithBufferOf defaultChunkSize

378378

379379-- | Write a stream of arrays to a handle. Each array in the stream is written

380380-- to the device as a separate IO request.

src/Streamly/Internal/Network/Inet/TCP.hs+48−43

@@ -35,16 +35,16 @@ module Streamly.Internal.Network.Inet.TCP

3535 -- * TCP clients

3636 -- | IP Address based operations.

3737 , connect

38− , withServerAddrM

38+ , withConnectionM

3939

4040 -- ** Unfolds

41− , withServerAddr

41+ , usingConnection

4242 , read

4343

4444 -- ** Streams

4545 , withConnection

4646 -- *** Source

47− , toStream

47+ , toBytes

4848 -- , readUtf8

4949 -- , readLines

5050 -- , readFrames

@@ -63,11 +63,14 @@ module Streamly.Internal.Network.Inet.TCP

6363 -- , writeUtf8

6464 -- , writeUtf8ByLines

6565 -- , writeByFrames

66− -- , writeRequestsOf

66+ , writeWithBufferOf

67+ , fromBytes

68+ , fromBytesWithBufferOf

6769

6870 -- -- * Array Write

6971 -- , writeArray

7072 , writeChunks

73+ , fromChunks

7174 {-

7275 -- ** Sink Servers

7376

@@ -109,7 +112,9 @@ import Streamly.Streams.Serial (SerialT)

109112import Streamly.Internal.Memory.Array.Types (Array(..), defaultChunkSize, writeNUnsafe)

110113import Streamly.Streams.StreamK.Type (IsStream)

111114

115+import qualified Control.Monad.Catch as MC

112116import qualified Network.Socket as Net

117+

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

114119import qualified Streamly.Internal.Memory.Array as A

115120import qualified Streamly.Internal.Memory.ArrayStream as AS

@@ -225,16 +230,16 @@ connect addr port = do

225230 return sock

226231

227232-- | Connect to a remote host using IP address and port and run the supplied

228−-- action on the resulting socket. 'withServerAddrM' makes sure that the

233+-- action on the resulting socket. 'withConnectionM' makes sure that the

229234-- socket is closed on normal termination or in case of an exception. If

230235-- closing the socket raises an exception, then this exception will be raised

231−-- by 'withServerAddrM'.

236+-- by 'withConnectionM'.

232237--

233238-- /Internal/

234−{-# INLINABLE withServerAddrM #-}

235−withServerAddrM :: (MonadMask m, MonadIO m)

239+{-# INLINABLE withConnectionM #-}

240+withConnectionM :: (MonadMask m, MonadIO m)

236241 => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> m ()) -> m ()

237−withServerAddrM addr port =

242+withConnectionM addr port =

238243 bracket (liftIO $ connect addr port) (liftIO . Net.close)

239244

240245-------------------------------------------------------------------------------

@@ -245,14 +250,14 @@ withServerAddrM addr port =

245250-- address and port. The resulting unfold opens a socket, uses it using the

246251-- supplied unfold and then makes sure that the socket is closed on normal

247252-- termination or in case of an exception. If closing the socket raises an

248−-- exception, then this exception will be raised by 'withServerAddr'.

253+-- exception, then this exception will be raised by 'usingConnection'.

249254--

250255-- /Internal/

251−{-# INLINABLE withServerAddr #-}

252−withServerAddr :: (MonadCatch m, MonadIO m)

256+{-# INLINABLE usingConnection #-}

257+usingConnection :: (MonadCatch m, MonadIO m)

253258 => Unfold m Socket a

254259 -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) a

255−withServerAddr =

260+usingConnection =

256261 UF.bracket (\(addr, port) -> liftIO $ connect addr port)

257262 (liftIO . Net.close)

258263

@@ -284,35 +289,33 @@ withConnection addr port =

284289{-# INLINE read #-}

285290read :: (MonadCatch m, MonadIO m)

286291 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Word8

287−read = UF.concat (withServerAddr ISK.readChunks) A.read

292+read = UF.concat (usingConnection ISK.readChunks) A.read

288293

289294-- | Read a stream from the supplied IPv4 host address and port number.

290295--

291296-- @since 0.7.0

292−{-# INLINE toStream #-}

293−toStream :: (IsStream t, MonadCatch m, MonadIO m)

297+{-# INLINE toBytes #-}

298+toBytes :: (IsStream t, MonadCatch m, MonadIO m)

294299 => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8

295−toStream addr port = AS.concat $ withConnection addr port ISK.toChunks

300+toBytes addr port = AS.concat $ withConnection addr port ISK.toChunks

296301

297302-------------------------------------------------------------------------------

298303-- Writing

299304-------------------------------------------------------------------------------

300305

301−{-

302306-- | Write a stream of arrays to the supplied IPv4 host address and port

303307-- number.

304308--

305309-- @since 0.7.0

306−{-# INLINE writeChunks #-}

307−writeChunks

310+{-# INLINE fromChunks #-}

311+fromChunks

308312 :: (MonadCatch m, MonadAsync m)

309313 => (Word8, Word8, Word8, Word8)

310314 -> PortNumber

311315 -> SerialT m (Array Word8)

312316 -> m ()

313−writeChunks addr port xs =

314− S.drain $ withConnection addr port (\sk -> S.yieldM $ SK.writeChunks sk xs)

315−-}

317+fromChunks addr port xs =

318+ S.drain $ withConnection addr port (\sk -> S.yieldM $ ISK.fromChunks sk xs)

316319

317320-- | Write a stream of arrays to the supplied IPv4 host address and port

318321-- number.

@@ -320,7 +323,7 @@ writeChunks addr port xs =

320323-- @since 0.7.0

321324{-# INLINE writeChunks #-}

322325writeChunks

323− :: (MonadAsync m)

326+ :: (MonadAsync m, MonadCatch m)

324327 => (Word8, Word8, Word8, Word8)

325328 -> PortNumber

326329 -> Fold m (Array Word8) ()

@@ -328,56 +331,58 @@ writeChunks addr port = Fold step initial extract

328331 where

329332 initial = do

330333 skt <- liftIO (connect addr port)

331− FL.initialize (SK.writeChunks skt)

332− step = FL.runStep

333− extract (Fold _ initial1 extract1) = initial1 >>= extract1

334+ fld <- FL.initialize (SK.writeChunks skt)

335+ `MC.onException` (liftIO $ Net.close skt)

336+ return (fld, skt)

337+ step (fld, skt) x = do

338+ r <- FL.runStep fld x `MC.onException` (liftIO $ Net.close skt)

339+ return (r, skt)

340+ extract ((Fold _ initial1 extract1), skt) = do

341+ liftIO $ Net.close skt

342+ initial1 >>= extract1

334343

335−{-

336344-- | Like 'write' but provides control over the write buffer. Output will

337345-- be written to the IO device as soon as we collect the specified number of

338346-- input elements.

339347--

340348-- @since 0.7.0

341−{-# INLINE writeRequestsOf #-}

342−writeRequestsOf

349+{-# INLINE fromBytesWithBufferOf #-}

350+fromBytesWithBufferOf

343351 :: (MonadCatch m, MonadAsync m)

344352 => Int

345353 -> (Word8, Word8, Word8, Word8)

346354 -> PortNumber

347355 -> SerialT m Word8

348356 -> m ()

349−writeRequestsOf n addr port m = writeChunks addr port $ AS.arraysOf n m

350−-}

357+fromBytesWithBufferOf n addr port m = fromChunks addr port $ AS.arraysOf n m

351358

352359-- | Like 'write' but provides control over the write buffer. Output will

353360-- be written to the IO device as soon as we collect the specified number of

354361-- input elements.

355362--

356363-- @since 0.7.0

357−{-# INLINE writeRequestsOf #-}

358−writeRequestsOf

359− :: (MonadAsync m)

364+{-# INLINE writeWithBufferOf #-}

365+writeWithBufferOf

366+ :: (MonadAsync m, MonadCatch m)

360367 => Int

361368 -> (Word8, Word8, Word8, Word8)

362369 -> PortNumber

363370 -> Fold m Word8 ()

364−writeRequestsOf n addr port =

371+writeWithBufferOf n addr port =

365372 FL.lchunksOf n (writeNUnsafe n) (writeChunks addr port)

366373

367−{-

368374-- | Write a stream to the supplied IPv4 host address and port number.

369375--

370376-- @since 0.7.0

371−{-# INLINE write #-}

372−write :: (MonadCatch m, MonadAsync m)

377+{-# INLINE fromBytes #-}

378+fromBytes :: (MonadCatch m, MonadAsync m)

373379 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()

374−write = writeRequestsOf defaultChunkSize

375−-}

380+fromBytes = fromBytesWithBufferOf defaultChunkSize

376381

377382-- | Write a stream to the supplied IPv4 host address and port number.

378383--

379384-- @since 0.7.0

380385{-# INLINE write #-}

381−write :: MonadAsync m

386+write :: (MonadAsync m, MonadCatch m)

382387 => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()

383−write = writeRequestsOf defaultChunkSize

388+write = writeWithBufferOf defaultChunkSize

src/Streamly/Internal/Network/Socket.hs+17−13

@@ -35,7 +35,6 @@ module Streamly.Internal.Network.Socket

3535 -- , readLines

3636 -- , readFrames

3737 -- , readByChunks

38− , toStream

3938

4039 -- -- * Array Read

4140 -- , readArrayUpto

@@ -47,6 +46,7 @@ module Streamly.Internal.Network.Socket

4746

4847 , toChunksWithBufferOf

4948 , toChunks

49+ , toBytes

5050

5151 -- * Write to connection

5252 , write

@@ -55,6 +55,10 @@ module Streamly.Internal.Network.Socket

5555 -- , writeByFrames

5656 , writeWithBufferOf

5757

58+ , fromChunks

59+ , fromBytesWithBufferOf

60+ , fromBytes

61+

5862 -- -- * Array Write

5963 , writeArray

6064 , writeChunks

@@ -370,9 +374,9 @@ readWithBufferOf chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h

370374-- stream ends when EOF is encountered.

371375--

372376-- @since 0.7.0

373−{-# INLINE toStream #-}

374−toStream :: (IsStream t, MonadIO m) => Socket -> t m Word8

375−toStream = AS.concat . toChunks

377+{-# INLINE toBytes #-}

378+toBytes :: (IsStream t, MonadIO m) => Socket -> t m Word8

379+toBytes = AS.concat . toChunks

376380

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

378382-- to the socket are performed using buffers of @bufsize@.

@@ -398,10 +402,10 @@ read = UF.supplyFirst readWithBufferOf A.defaultChunkSize

398402-- | Write a stream of arrays to a handle.

399403--

400404-- @since 0.7.0

401−{-# INLINE writeChunksS #-}

402−writeChunksS :: (MonadIO m, Storable a)

405+{-# INLINE fromChunks #-}

406+fromChunks :: (MonadIO m, Storable a)

403407 => Socket -> SerialT m (Array a) -> m ()

404−writeChunksS h m = S.mapM_ (liftIO . writeArray h) m

408+fromChunks h m = S.mapM_ (liftIO . writeArray h) m

405409

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

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

@@ -424,9 +428,9 @@ writeChunks h = FL.drainBy (liftIO . writeArray h)

424428-- input elements.

425429--

426430-- @since 0.7.0

427−{-# INLINE writeWithBufferOfS #-}

428−writeWithBufferOfS :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()

429−writeWithBufferOfS n h m = writeChunksS h $ AS.arraysOf n m

431+{-# INLINE fromBytesWithBufferOf #-}

432+fromBytesWithBufferOf :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()

433+fromBytesWithBufferOf n h m = fromChunks h $ AS.arraysOf n m

430434

431435-- | Write a byte stream to a socket. Accumulates the input in chunks of

432436-- specified number of bytes before writing.

@@ -443,9 +447,9 @@ writeWithBufferOf n h = FL.lchunksOf n (A.writeNUnsafe n) (writeChunks h)

443447-- depends on the 'IOMode' and the current seek position of the handle.

444448--

445449-- @since 0.7.0

446−{-# INLINE _writeS #-}

447−_writeS :: MonadIO m => Socket -> SerialT m Word8 -> m ()

448−_writeS = writeWithBufferOfS A.defaultChunkSize

450+{-# INLINE fromBytes #-}

451+fromBytes :: MonadIO m => Socket -> SerialT m Word8 -> m ()

452+fromBytes = fromBytesWithBufferOf A.defaultChunkSize

449453

450454-- | Write a byte stream to a socket. Accumulates the input in chunks of

451455-- up to 'A.defaultChunkSize' bytes before writing.