Streamly.Internal.FileSystem.Handle

The fundamental singleton IO APIs are getChunk and putChunk and the fundamental stream IO APIs built on top of those are readChunksWith and writeChunks. Rest of this module is just combinatorial programming using these.

We can achieve line buffering by folding lines in the input stream into a stream of arrays using Stream.splitOn or Fold.takeEndBy_ and similar operations. One can wrap the input stream in Maybe type and then use writeMaybesWith to achieve user controlled buffering.

Setup

To execute the code examples provided in this module in ghci, please run the following commands first.

>>> :m
>>> import qualified Streamly.Data.Array as Array
>>> import qualified Streamly.FileSystem.Handle as Handle hiding (readChunks)
>>> import qualified Streamly.Data.Fold as Fold
>>> import qualified Streamly.Data.Stream as Stream
>>> import qualified Streamly.Data.Unfold as Unfold

For APIs that have not been released yet.

>>> import qualified Streamly.Internal.Data.Array as Array (writeNUnsafe)
>>> import qualified Streamly.Internal.Data.Unfold as Unfold (first)
>>> import qualified Streamly.Internal.FileSystem.Handle as Handle
>>> import qualified Streamly.Internal.System.IO as IO (defaultChunkSize)

Singleton APIs

getChunk :: MonadIO m => Int -> Handle -> m (Array Word8)

Read a ByteArray consisting of one or more bytes from a file handle. If no data is available on the handle it blocks until at least one byte becomes available. If any data is available then it immediately returns that data without blocking. As a result of this behavior, it may read less than or equal to the size requested.

getChunkOf :: Int -> Handle -> IO (Array Word8)

Read a ByteArray consisting of exactly the specified number of bytes from a file handle.

Unimplemented

putChunk :: MonadIO m => Handle -> Array a -> m ()

Write an Array to a file handle.

Streams

read :: MonadIO m => Handle -> Stream m Word8

Generate a byte stream from a file Handle.

>>> read h = Stream.unfoldMany Array.reader $ Handle.readChunks h

Pre-release

readWith :: MonadIO m => Int -> Handle -> Stream m Word8

readWith bufsize handle reads a byte stream from a file handle, reads are performed in chunks of up to bufsize.

>>> readWith size h = Stream.unfoldMany Array.reader $ Handle.readChunksWith size h

Pre-release

readChunksWith :: MonadIO m => Int -> Handle -> Stream m (Array Word8)

readChunksWith size handle reads a stream of arrays from the file handle handle. The maximum size of a single array is limited to size. The actual size read may be less than or equal to size.

>>> readChunksWith size h = Stream.unfold Handle.chunkReaderWith (size, h)

readChunks :: MonadIO m => Handle -> Stream m (Array Word8)

getChunks handle reads a stream of arrays from the specified file handle. The maximum size of a single array is limited to defaultChunkSize. The actual size read may be less than or equal to defaultChunkSize.

>>> readChunks = Handle.readChunksWith IO.defaultChunkSize

Pre-release

Unfolds

reader :: MonadIO m => Unfold m Handle Word8

Unfolds a file handle into a byte stream. IO requests to the device are performed in sizes of defaultChunkSize.

>>> reader = Unfold.many Array.reader Handle.chunkReader

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

Unfolds the tuple (bufsize, handle) into a byte stream, read requests to the IO device are performed using buffers of bufsize.

>>> readerWith = Unfold.many Array.reader Handle.chunkReaderWith

chunkReader :: MonadIO m => Unfold m Handle (Array Word8)

Unfolds a handle into a stream of Word8 arrays. Requests to the IO device are performed using a buffer of size defaultChunkSize. The size of arrays in the resulting stream are therefore less than or equal to defaultChunkSize.

>>> chunkReader = Unfold.first IO.defaultChunkSize Handle.chunkReaderWith

chunkReaderWith :: MonadIO m => Unfold m (Int, Handle) (Array Word8)

Unfold the tuple (bufsize, handle) into a stream of Word8 arrays. Read requests to the IO device are performed using a buffer of size bufsize. The size of an array in the resulting stream is always less than or equal to bufsize.

Folds

write :: MonadIO m => Handle -> Fold m Word8 ()

Write a byte stream to a file handle. Accumulates the input in chunks of up to defaultChunkSize before writing to the IO device.

>>> write = Handle.writeWith IO.defaultChunkSize

writeWith :: MonadIO m => Int -> Handle -> Fold m Word8 ()

writeWith reqSize handle writes the input stream to handle. Bytes in the input stream are collected into a buffer until we have a chunk of reqSize and then written to the IO device.

>>> writeWith n h = Fold.groupsOf n (Array.writeNUnsafe n) (Handle.writeChunks h)

writeChunks :: MonadIO m => Handle -> Fold m (Array a) ()

Write a stream of arrays to a handle. Each array in the stream is written to the device as a separate IO request.

writeChunks h = Fold.drainBy (Handle.putChunk h)

writeChunksWith :: (MonadIO m, Unbox a) => Int -> Handle -> Fold m (Array a) ()

writeChunksWith bufsize handle writes a stream of arrays to handle after coalescing the adjacent arrays in chunks of bufsize. We never split an array, if a single array is bigger than the specified size it emitted as it is. Multiple arrays are coalesed as long as the total size remains below the specified size.

writeMaybesWith :: MonadIO m => Int -> Handle -> Fold m (Maybe Word8) ()

Write a stream of Maybe values. Keep buffering the just values in an array until a Nothing is encountered or the buffer size exceeds the specified limit, at that point flush the buffer to the handle.

Pre-release

Refolds

writer :: MonadIO m => Refold m Handle Word8 ()

Like write but uses the experimental Refold API.

Internal

writerWith :: MonadIO m => Int -> Refold m Handle Word8 ()

Like writeWith but uses the experimental Refold API.

Internal

chunkWriter :: MonadIO m => Refold m Handle (Array a) ()

Like writeChunks but uses the experimental Refold API.

Internal

Stream writes

putBytes :: MonadIO m => Handle -> Stream m Word8 -> m ()

Write a byte stream to a file handle. Accumulates the input in chunks of up to defaultChunkSize before writing.

NOTE: This may perform better than the write fold, you can try this if you need some extra perf boost.

>>> putBytes = Handle.putBytesWith IO.defaultChunkSize

putBytesWith :: MonadIO m => Int -> Handle -> Stream m Word8 -> m ()

putBytesWith bufsize handle stream writes stream to handle in chunks of bufsize. A write is performed to the IO device as soon as we collect the required input size.

putChunksWith :: (MonadIO m, Unbox a) => Int -> Handle -> Stream m (Array a) -> m ()

putChunksWith bufsize handle stream writes a stream of arrays to handle after coalescing the adjacent arrays in chunks of bufsize. The chunk size is only a maximum and the actual writes could be smaller as we do not split the arrays to fit exactly to the specified size.

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

Write a stream of arrays to a handle.

>>> putChunks h = Stream.fold (Fold.drainBy (Handle.putChunk h))

Random Access (Seek)

Unlike the streaming APIs listed above, these APIs apply to devices or files that have random access or seek capability. This type of devices include disks, files, memory devices and exclude terminals, pipes, sockets and fifos.

chunkReaderFromToWith :: MonadIO m => Unfold m (Int, Int, Int, Handle) (Array Word8)

The input to the unfold is (from, to, bufferSize, handle). It starts reading from the offset from in the file and reads up to the offset to.

Deprecated

readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)

Deprecated: Please use chunkReaderWith instead.

Same as chunkReaderWith

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

Deprecated: Please use readerWith instead.

Same as readerWith

writeChunksWithBufferOf :: (MonadIO m, Unbox a) => Int -> Handle -> Fold m (Array a) ()

Deprecated: Please use writeChunksWith instead.

Same as writeChunksWith

writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()

Deprecated: Please use writeWith instead.

Same as writeWith