feat: implement createReadStream and createWriteStream on FileHandle (#1076)

* feat: implement `createReadStream` and `createWriteStream` on `FileHandle`

Closes #1063

* test for FileHandle#create{Read,Write}Stream

Author: graue

src/__tests__/promises.test.ts

@@ -1,3 +1,4 @@
+import { promisify } from 'util';
 import { Volume } from '../volume';
 import { Readable } from 'stream';
 
@@ -82,6 +83,39 @@ describe('Promises API', () => {
       });
     });
     // close(): covered by all other tests
+    it('supports createReadStream()', done => {
+      const vol = Volume.fromJSON({
+        '/test.txt': 'Hello',
+      });
+      vol.promises
+        .open('/test.txt', 'r')
+        .then(fh => {
+          const readStream = fh.createReadStream({});
+          readStream.setEncoding('utf8');
+          let readData = '';
+          readStream.on('readable', () => {
+            const chunk = readStream.read();
+            if (chunk != null) readData += chunk;
+          });
+          readStream.on('end', () => {
+            expect(readData).toEqual('Hello');
+            done();
+          });
+        })
+        .catch(err => {
+          expect(err).toBeNull();
+        });
+    });
+    it('supports createWriteStream()', async () => {
+      const vol = new Volume();
+      const fh = await vol.promises.open('/test.txt', 'wx', 0o600);
+      const writeStream = fh.createWriteStream({});
+      await promisify(writeStream.write.bind(writeStream))(Buffer.from('Hello'));
+      await promisify(writeStream.close.bind(writeStream))();
+      expect(vol.toJSON()).toEqual({
+        '/test.txt': 'Hello',
+      });
+    });
     describe('datasync()', () => {
       const vol = new Volume();
       const { promises } = vol;

src/node/FileHandle.ts

@@ -1,6 +1,6 @@
 import { promisify } from './util';
 import type * as opts from './types/options';
-import type { IFileHandle, IStats, TData, TDataOut, TMode, TTime } from './types/misc';
+import type { IFileHandle, IReadStream, IWriteStream, IStats, TData, TDataOut, TMode, TTime } from './types/misc';
 import type { FsCallbackApi } from './types';
 
 export class FileHandle implements IFileHandle {
@@ -33,6 +33,14 @@ export class FileHandle implements IFileHandle {
     return promisify(this.fs, 'fdatasync')(this.fd);
   }
 
+  createReadStream(options: opts.IFileHandleReadStreamOptions): IReadStream {
+    return this.fs.createReadStream('', { ...options, fd: this });
+  }
+
+  createWriteStream(options: opts.IFileHandleWriteStreamOptions): IWriteStream {
+    return this.fs.createWriteStream('', { ...options, fd: this });
+  }
+
   readableWebStream(options?: opts.IReadableWebStreamOptions): ReadableStream {
     return new ReadableStream({
       pull: async controller => {

src/node/types/misc.ts

@@ -4,6 +4,8 @@ import type { EventEmitter } from 'events';
 import type { TSetTimeout } from '../../setTimeoutUnref';
 import type {
   IAppendFileOptions,
+  IFileHandleReadStreamOptions,
+  IFileHandleWriteStreamOptions,
   IReadableWebStreamOptions,
   IReadFileOptions,
   IStatOptions,
@@ -138,6 +140,8 @@ export interface IFileHandle {
   chmod(mode: TMode): Promise<void>;
   chown(uid: number, gid: number): Promise<void>;
   close(): Promise<void>;
+  createReadStream(options: IFileHandleReadStreamOptions): IReadStream;
+  createWriteStream(options: IFileHandleWriteStreamOptions): IWriteStream;
   datasync(): Promise<void>;
   readableWebStream(options?: IReadableWebStreamOptions): ReadableStream;
   read(buffer: Buffer | Uint8Array, offset: number, length: number, position: number): Promise<TFileHandleReadResult>;

src/node/types/options.ts

@@ -36,6 +36,26 @@ export interface IReadableWebStreamOptions {
   type?: 'bytes' | undefined;
 }
 
+export interface IFileHandleReadStreamOptions {
+  encoding?: BufferEncoding;
+  autoClose?: boolean;
+  emitClose?: boolean;
+  start?: number | undefined;
+  end?: number;
+  highWaterMark?: number;
+  flush?: boolean;
+  signal?: AbortSignal | undefined;
+}
+
+export interface IFileHandleWriteStreamOptions {
+  encoding?: BufferEncoding;
+  autoClose?: boolean;
+  emitClose?: boolean;
+  start?: number;
+  highWaterMark?: number;
+  flush?: boolean;
+}
+
 export interface IReaddirOptions extends IOptions {
   recursive?: boolean;
   withFileTypes?: boolean;