89ddf14 · streamly · git.iko.soy

git.iko.soy/streamly

implement getChunks and putChunks

commit89ddf14ed3da32cfdbe644fdb0b9cd4ea70e8fe7 authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 14 Nov 2019, 00:21 (7 years ago) parent56eb882

2 files changed+33−11

compare with master

unifiedsplit

src/Streamly/Internal/FileSystem/File.hs+9−9 src/Streamly/Internal/FileSystem/Handle.hs+24−2

src/Streamly/Internal/FileSystem/File.hs+9−9

@@ -77,7 +77,7 @@ module Streamly.Internal.FileSystem.File

7777

7878 -- -- * Array Write

7979 , writeArray

80− , writeChunks

80+ , fromChunks

8181

8282 -- ** Append To File

8383 , append

@@ -242,19 +242,19 @@ readTailForever = undefined

242242-- Writing

243243-------------------------------------------------------------------------------

244244

245−{-# INLINE writeChunksMode #-}

246−writeChunksMode :: (MonadAsync m, MonadCatch m, Storable a)

245+{-# INLINE fromChunksMode #-}

246+fromChunksMode :: (MonadAsync m, MonadCatch m, Storable a)

247247 => IOMode -> FilePath -> SerialT m (Array a) -> m ()

248−writeChunksMode mode file xs = S.drain $

248+fromChunksMode mode file xs = S.drain $

249249 withFile file mode (\h -> S.mapM (liftIO . FH.writeArray h) xs)

250250

251251-- | Write a stream of arrays to a file. Overwrites the file if it exists.

252252--

253253-- @since 0.7.0

254−{-# INLINE writeChunks #-}

255−writeChunks :: (MonadAsync m, MonadCatch m, Storable a)

254+{-# INLINE fromChunks #-}

255+fromChunks :: (MonadAsync m, MonadCatch m, Storable a)

256256 => FilePath -> SerialT m (Array a) -> m ()

257−writeChunks = writeChunksMode WriteMode

257+fromChunks = fromChunksMode WriteMode

258258

259259-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

260260--

@@ -272,7 +272,7 @@ writeChunks = writeChunksMode WriteMode

272272{-# INLINE writeWithBufferOf #-}

273273writeWithBufferOf :: (MonadAsync m, MonadCatch m)

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

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

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

276276

277277-- > write = 'writeWithBufferOf' defaultChunkSize

278278--

@@ -298,7 +298,7 @@ write = toHandleWith A.defaultChunkSize

298298{-# INLINE appendChunks #-}

299299appendChunks :: (MonadAsync m, MonadCatch m, Storable a)

300300 => FilePath -> SerialT m (Array a) -> m ()

301−appendChunks = writeChunksMode AppendMode

301+appendChunks = fromChunksMode AppendMode

302302

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

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

src/Streamly/Internal/FileSystem/Handle.hs+24−2

@@ -36,6 +36,7 @@ module Streamly.Internal.FileSystem.Handle

3636

3737 , toChunksWithBufferOf

3838 , toChunks

39+ , getChunks

3940

4041 -- ** Write to Handle

4142 -- Byte stream write (Folds)

@@ -55,8 +56,9 @@ module Streamly.Internal.FileSystem.Handle

5556 , writeChunksWithBufferOf

5657

5758 -- -- * Array stream Write

58− , fromChunks

5959 , fromChunksWithBufferOf

60+ , fromChunks

61+ , putChunks

6062

6163 -- -- * Random Access (Seek)

6264 -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or

@@ -105,7 +107,7 @@ import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

105107import Foreign.Ptr (minusPtr, plusPtr)

106108import Foreign.Storable (Storable(..))

107109import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

108−import System.IO (Handle, hGetBufSome, hPutBuf)

110+import System.IO (Handle, hGetBufSome, hPutBuf, stdin, stdout)

109111import Prelude hiding (read)

110112

111113import Streamly.Data.Fold (Fold)

@@ -229,6 +231,18 @@ readChunksWithBufferOf = Unfold step return

229231toChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)

230232toChunks = toChunksWithBufferOf defaultChunkSize

231233

234+-- | Read a stream of chunks from standard input. The maximum size of a single

235+-- chunk is limited to @defaultChunkSize@. The actual size read may be less

236+-- than @defaultChunkSize@.

237+--

238+-- > getChunks = toChunks stdin

239+--

240+-- /Internal/

241+--

242+{-# INLINE getChunks #-}

243+getChunks :: (IsStream t, MonadIO m) => t m (Array Word8)

244+getChunks = toChunks stdin

245+

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

233247-- device are performed using a buffer of size

234248-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'. The

@@ -320,6 +334,14 @@ fromChunks :: (MonadIO m, Storable a)

320334 => Handle -> SerialT m (Array a) -> m ()

321335fromChunks h m = S.mapM_ (liftIO . writeArray h) m

322336

337+-- | Write a stream of chunks to standard output.

338+--

339+-- /Internal/

340+--

341+{-# INLINE putChunks #-}

342+putChunks :: (MonadIO m, Storable a) => SerialT m (Array a) -> m ()

343+putChunks = fromChunks stdout

344+

323345-- | @fromChunksWithBufferOf bufsize handle stream@ writes a stream of arrays

324346-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.

325347-- The chunk size is only a maximum and the actual writes could be smaller as