Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

refactor(deps): Remove get-stream in favor of async iterators #1925

Merged
merged 4 commits into from
May 16, 2022
Merged

refactor(deps): Remove get-stream in favor of async iterators #1925

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4ab85e0
refactor(deps): Remove get-stream in favor of async iterators
ddelgrosso1 May 12, 2022
5bafb5c
explicitly handle stream error for node < 16
ddelgrosso1 May 12, 2022
f2afb79
clean up implementation, fix unit tests
ddelgrosso1 May 12, 2022
a009034
update testbench docker image to 0.20.0
ddelgrosso1 May 12, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,6 @@
"ent": "^2.2.0",
"extend": "^3.0.2",
"gaxios": "^4.0.0",
"get-stream": "^6.0.0",
"google-auth-library": "^7.14.1",
"hash-stream-validation": "^0.2.2",
"mime": "^3.0.0",
Expand Down
22 changes: 16 additions & 6 deletions src/file.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@ import {
import {promisifyAll} from '@google-cloud/promisify';

import compressible = require('compressible');
import getStream = require('get-stream');
import * as crypto from 'crypto';
import * as dateFormat from 'date-and-time';
import * as extend from 'extend';
Expand Down Expand Up @@ -1394,8 +1393,8 @@ class File extends ServiceObject<File> {
) => {
if (err) {
// Get error message from the body.
getStream(rawResponseStream).then(body => {
err.message = body;
this.getBufferFromReadable(rawResponseStream).then(body => {
err.message = body.toString('utf-8');
throughStream.destroy(err);
});

Expand Down Expand Up @@ -2172,10 +2171,9 @@ class File extends ServiceObject<File> {
fileStream.pipe(writable).on('error', callback).on('finish', callback);
});
} else {
getStream
.buffer(fileStream)
this.getBufferFromReadable(fileStream)
.then(contents => callback?.(null, contents))
.catch(callback as (error: RequestError) => void);
.catch(callback as (err: RequestError) => void);
}
}

Expand Down Expand Up @@ -4104,6 +4102,18 @@ class File extends ServiceObject<File> {
this.storage.retryOptions.autoRetry = false;
}
}

private async getBufferFromReadable(readable: Readable): Promise<Buffer> {
readable.once('error', e => {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There seems to be some difference in how node < 16 handles errors with streams / async iterators. Test cases were failing with Node 12 / 14 until I explicitly handled a stream error. However, node 16 handled it with just the iterator.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Broke other things... investigating a better way.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yea, this block handled the error a bit differently - the throw wouldn't populate in getBufferFromReadable, but would have a separate context as it is in a callback.

I actually don't think the catch is necessary - for await (const chunk of readable) { should populate the throw. E.g. https://nodejs.org/api/stream.html#readablesymbolasynciterator

throw e;
});
let buf = Buffer.alloc(0);
for await (const chunk of readable) {
buf = Buffer.concat([buf, chunk]);
}

return buf;
}
}

/*! Developer Documentation
Expand Down