diff --git a/handwritten/storage/src/file.ts b/handwritten/storage/src/file.ts index 9cbd6ee7578..ad622534b65 100644 --- a/handwritten/storage/src/file.ts +++ b/handwritten/storage/src/file.ts @@ -337,6 +337,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; @@ -547,6 +552,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; } @@ -1612,6 +1622,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 @@ -1682,12 +1695,13 @@ 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; let validateStream: HashStreamValidator | undefined = undefined; - const throughStream = new PassThroughShim(); + const throughStream = new PassThroughShim({highWaterMark}); let crc32c = true; let md5 = false; @@ -2063,6 +2077,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. @@ -2218,6 +2235,7 @@ class File extends ServiceObject { * fs.createWriteStream({uri, resumeCRC32C}); */ createWriteStream(options: CreateWriteStreamOptions = {}): Writable { + const {highWaterMark} = options; options.metadata ??= {}; if (options.contentType) { @@ -2261,6 +2279,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 @@ -2283,13 +2302,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 @@ -4873,7 +4892,9 @@ class File extends ServiceObject { options.preconditionOpts, ); - 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 8f9e7d170fc..33bb8faad8f 100644 --- a/handwritten/storage/test/file.ts +++ b/handwritten/storage/test/file.ts @@ -17,6 +17,8 @@ import assert from 'assert'; import { Bucket, CRC32C, + CreateReadStreamOptions, + CreateWriteStreamOptions, File, GaxiosError, GaxiosOptionsPrepared, @@ -1179,6 +1181,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', () => { file.storageTransport.makeRequest = sandbox.stub().callsFake(opts => { @@ -2259,6 +2267,48 @@ 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, + }; + + file.startResumableUpload_ = sandbox.stub().callsFake((dup, opts) => { + assert.strictEqual(dup.writableHighWaterMark, highWaterMark); + assert.strictEqual(opts.highWaterMark, highWaterMark); + done(); + }); + + 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, + }; + + file.startSimpleUpload_ = sandbox.stub().callsFake((dup, opts) => { + assert.strictEqual(dup.writableHighWaterMark, highWaterMark); + assert.strictEqual(opts.highWaterMark, highWaterMark); + 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}); @@ -5748,6 +5798,7 @@ describe('File', () => { uri: 'http://resumable-uri', userProject: 'user-project-id', chunkSize: 262144, // 256 KiB + highWaterMark: 12345, }; const resumableUpload = { @@ -5765,6 +5816,7 @@ describe('File', () => { assert.strictEqual(opts.uri, options.uri); assert.strictEqual(opts.userProject, options.userProject); assert.strictEqual(opts.chunkSize, options.chunkSize); + assert.strictEqual(opts.highWaterMark, options.highWaterMark); setImmediate(done); return new PassThrough(); @@ -5775,6 +5827,18 @@ describe('File', () => { file.startResumableUpload_(duplexify(), options); }); + it('should pass highWaterMark to the upload stream', done => { + const dup = duplexify(); + const highWaterMark = 12345; + + dup.setWritable = (stream: Duplex) => { + assert.strictEqual(stream.writableHighWaterMark, highWaterMark); + done(); + }; + + file.startResumableUpload_(dup, {highWaterMark}); + }); + it('should emit the response', done => { const resp = {}; const uploadStream = new PassThrough(); @@ -6035,6 +6099,34 @@ describe('File', () => { await file.startSimpleUpload_(duplexify(), options); }); + it('should pass highWaterMark if set', async () => { + const options: CreateWriteStreamOptions = { + highWaterMark: 12345, + }; + const dup = duplexify(); + let setWritableStream: Duplex | undefined; + dup.setWritable = sandbox.stub().callsFake((stream: Duplex) => { + setWritableStream = stream; + }); + file.storageTransport.makeRequest = sandbox + .stub() + .callsFake((options_: StorageRequestOptions) => { + const mediaPart = options_.multipart?.[1]; + assert.strictEqual(mediaPart?.content, setWritableStream); + assert.strictEqual( + setWritableStream?.writableHighWaterMark, + options.highWaterMark, + ); + assert.strictEqual( + setWritableStream?.readableHighWaterMark, + options.highWaterMark, + ); + }) + .resolves({}); + + await file.startSimpleUpload_(dup, options); + }); + describe('request', () => { describe('error', () => { const error = new Error('Error.');