From 9aa5990d7e9320fbb8d0f725ccd83799d88a0f6a Mon Sep 17 00:00:00 2001 From: Thiyagu K Date: Fri, 23 Oct 2026 05:38:15 +0000 Subject: [PATCH] feat: add support for highWaterMark option in createReadStream and createWriteStream --- handwritten/storage/src/file.ts | 26 ++++++- handwritten/storage/src/nodejs-common/util.ts | 9 ++- handwritten/storage/test/file.ts | 68 +++++++++++++++++++ .../storage/test/nodejs-common/util.ts | 15 ++++ 4 files changed, 114 insertions(+), 4 deletions(-) diff --git a/handwritten/storage/src/file.ts b/handwritten/storage/src/file.ts index 977e3df4a3ca..f68e27f25585 100644 --- a/handwritten/storage/src/file.ts +++ b/handwritten/storage/src/file.ts @@ -332,6 +332,11 @@ export interface CreateResumableUploadCallback { export interface CreateWriteStreamOptions extends CreateResumableUploadOptions { contentType?: string; gzip?: string | boolean; + /** + * Stream `highWaterMark` in bytes. Configures the maximum number of bytes + * buffered in the underlying writable stream before backpressure is applied. + */ + highWaterMark?: number; resumable?: boolean; timeout?: number; validation?: string | boolean; @@ -530,6 +535,11 @@ export interface CreateReadStreamOptions { start?: number; end?: number; decompress?: boolean; + /** + * Stream `highWaterMark` in bytes. Configures the maximum number of bytes + * buffered in the underlying readable stream. + */ + highWaterMark?: number; [GCCL_GCS_CMD_KEY]?: string; } @@ -1577,6 +1587,9 @@ class File extends ServiceObject { * received data. By default this option is set to `true`. * Applicable in cases where the data was uploaded with * `gzip: true` option. See {@link File#createWriteStream}. + * @property {number} [highWaterMark] Stream `highWaterMark` in bytes. + * Configures the maximum number of bytes buffered in the underlying + * readable stream. */ /** * Create a readable stream to read the contents of the remote file. It can be @@ -1647,6 +1660,7 @@ class File extends ServiceObject { */ createReadStream(options: CreateReadStreamOptions = {}): Readable { options = Object.assign({decompress: true}, options); + const {highWaterMark} = options; const rangeRequest = typeof options.start === 'number' || typeof options.end === 'number'; const tailRequest = options.end! < 0; @@ -1654,7 +1668,7 @@ class File extends ServiceObject { let validateStream: HashStreamValidator | undefined = undefined; let request: TeenyRequest | undefined = undefined; - const throughStream = new PassThroughShim(); + const throughStream = new PassThroughShim({highWaterMark}); let crc32c = true; let md5 = false; @@ -1996,6 +2010,9 @@ class File extends ServiceObject { * If set to `auto`, the contentType is used to determine if the file * should be gzipped. This will set `options.metadata.contentEncoding` to * `gzip` if necessary. + * @property {number} [highWaterMark] Stream `highWaterMark` in bytes. + * Configures the maximum number of bytes buffered in the underlying + * writable stream before backpressure is applied. * @property {object} [metadata] See the examples below or * {@link https://cloud.google.com/storage/docs/json_api/v1/objects/insert#request_properties_JSON| Objects: insert request body} * for more details. @@ -2152,6 +2169,7 @@ class File extends ServiceObject { */ // eslint-disable-next-line @typescript-eslint/no-explicit-any createWriteStream(options: CreateWriteStreamOptions = {}): Writable { + const {highWaterMark} = options; options.metadata ??= {}; if (options.contentType) { @@ -2195,6 +2213,7 @@ class File extends ServiceObject { // A stream for consumer to write to const writeStream = new Writable({ + highWaterMark, final(cb) { // Set the pipeline callback to this callback so the pipeline's results // can be populated to the consumer @@ -2217,13 +2236,13 @@ class File extends ServiceObject { emitStream.destroy(); }); - const emitStream = new PassThroughShim(); + const emitStream = new PassThroughShim({highWaterMark}); // If `writeStream` is destroyed before the `writing` event, `emitStream` will not have any listeners. This prevents an unhandled error. const noop = () => {}; emitStream.on('error', noop); - const fileWriteStream = duplexify(); + const fileWriteStream = duplexify(undefined, undefined, {highWaterMark}); let fileWriteStreamMetadataReceived = false; // Handing off emitted events to users @@ -4766,6 +4785,7 @@ class File extends ServiceObject { }, metadata: options.metadata, request: reqOpts, + highWaterMark: options.highWaterMark, }); } diff --git a/handwritten/storage/src/nodejs-common/util.ts b/handwritten/storage/src/nodejs-common/util.ts index d79c5ca5ce10..1b83f49eef8a 100644 --- a/handwritten/storage/src/nodejs-common/util.ts +++ b/handwritten/storage/src/nodejs-common/util.ts @@ -230,6 +230,11 @@ export interface MakeWritableStreamOptions { */ connection?: {}; + /** + * Stream `highWaterMark` in bytes for the underlying writable stream. + */ + highWaterMark?: number; + /** * Metadata to send at the head of the request. */ @@ -525,7 +530,9 @@ export class Util { ) { onComplete = onComplete || util.noop; - const writeStream = new ProgressStream(); + const writeStream = new ProgressStream({ + highWaterMark: options.highWaterMark, + }); writeStream.on('progress', evt => dup.emit('progress', evt)); dup.setWritable(writeStream); diff --git a/handwritten/storage/test/file.ts b/handwritten/storage/test/file.ts index 938e8e317bba..7411c3d90cd0 100644 --- a/handwritten/storage/test/file.ts +++ b/handwritten/storage/test/file.ts @@ -52,6 +52,8 @@ import { GetSignedUrlConfig, GenerateSignedPostPolicyV2Options, CRC32C, + CreateReadStreamOptions, + CreateWriteStreamOptions, } from '../src/index.js'; import { SignedPostPolicyV4Output, @@ -1320,6 +1322,12 @@ describe('File', () => { .resume(); }); + it('should pass highWaterMark to the returned stream', () => { + const options: CreateReadStreamOptions = {highWaterMark: 12345}; + const stream = file.createReadStream(options); + assert.strictEqual(stream.readableHighWaterMark, 12345); + }); + describe('authenticating', () => { it('should create an authenticated request', done => { file.requestStream = (opts: DecorateRequestOptions) => { @@ -2355,6 +2363,51 @@ describe('File', () => { writable.write('data'); }); + it('should set highWaterMark on the returned stream', () => { + const options: CreateWriteStreamOptions = {highWaterMark: 12345}; + const writable = file.createWriteStream(options); + assert.strictEqual(writable.writableHighWaterMark, 12345); + }); + + it('should propagate highWaterMark to resumable upload', done => { + const highWaterMark = 12345; + const options: CreateWriteStreamOptions = { + highWaterMark, + resumable: true, + }; + + resumableUploadOverride = { + upload(opts: resumableUpload.UploadConfig) { + assert.strictEqual(opts.highWaterMark, highWaterMark); + setImmediate(done); + return new PassThrough({highWaterMark: opts.highWaterMark}); + }, + }; + + const writable = file.createWriteStream(options); + assert.strictEqual(writable.writableHighWaterMark, highWaterMark); + writable.write('data'); + }); + + it('should propagate highWaterMark to simple upload', done => { + const highWaterMark = 12345; + const options: CreateWriteStreamOptions = { + highWaterMark, + resumable: false, + }; + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + makeWritableStreamOverride = (dup: duplexify.Duplexify, opts: any) => { + assert.strictEqual(dup.writableHighWaterMark, highWaterMark); + assert.strictEqual(opts.highWaterMark, highWaterMark); + setImmediate(done); + }; + + const writable = file.createWriteStream(options); + assert.strictEqual(writable.writableHighWaterMark, highWaterMark); + writable.write('data'); + }); + it('should alias contentType to metadata object', done => { const contentType = 'text/html'; const writable = file.createWriteStream({contentType}); @@ -5954,6 +6007,7 @@ describe('File', () => { uri: 'http://resumable-uri', userProject: 'user-project-id', chunkSize: 262144, // 256 KiB + highWaterMark: 12345, }; file.generation = 3; @@ -6009,6 +6063,7 @@ describe('File', () => { }); assert.strictEqual(opts.params, storage.preconditionOpts); assert.strictEqual(opts.chunkSize, options.chunkSize); + assert.strictEqual(opts.highWaterMark, options.highWaterMark); setImmediate(done); return new PassThrough(); @@ -6237,6 +6292,19 @@ describe('File', () => { file.startSimpleUpload_(duplexify(), options); }); + it('should pass highWaterMark if set', done => { + const options: CreateWriteStreamOptions = { + highWaterMark: 12345, + }; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + makeWritableStreamOverride = (stream: {}, options_: any) => { + assert.strictEqual(options_.highWaterMark, options.highWaterMark); + done(); + }; + + file.startSimpleUpload_(duplexify(), options); + }); + describe('request', () => { describe('error', () => { const error = new Error('Error.'); diff --git a/handwritten/storage/test/nodejs-common/util.ts b/handwritten/storage/test/nodejs-common/util.ts index d7ebc515c3fb..2d3889afb245 100644 --- a/handwritten/storage/test/nodejs-common/util.ts +++ b/handwritten/storage/test/nodejs-common/util.ts @@ -606,6 +606,21 @@ describe('common/util', () => { util.makeWritableStream(dup, {makeAuthenticatedRequest() {}}); }); + it('should pass highWaterMark to the writable stream', done => { + const dup = duplexify(); + const highWaterMark = 12345; + + dup.setWritable = (writable: stream.Writable) => { + assert.strictEqual(writable.writableHighWaterMark, highWaterMark); + done(); + }; + + util.makeWritableStream(dup, { + highWaterMark, + makeAuthenticatedRequest() {}, + }); + }); + it('dup should emit a progress event with the bytes written', done => { let happened = false;