grpc / grpc/grpc-node

ClientReadableStream: override/extend .destroy(err?) method

Open
#1,238 16 comments 1 reaction 0 assignees View on GitHub
feature request package: @grpc/grpc-js package: grpc
Dominant language
TypeScript
Stars
4.8k
Forks
716
Avg merge
2d 3h
Merged PRs (30d)
10

Description

### Is your feature request related to a problem? Please describe.
Yes, and no.

The documentation for the `ClientReadableStream` object states that it extends `Readable`, and therefore, I'd expect to be able to invoke base methods, like `.destroy(err?)`.

I've spent too many hours now trying to determine why I'm able to process data from the server until suddenly, no data flows back from the server. Long story short, my process creates 10 ClientReadableStreams and then each stream is eventually destroyed (using `.destroy(err?)`).. After a short delay, the process then tries to continue processing and creates a new ClientReadableStream so that I always process 10 streams at a time. Eventually, this happens enough times for the data flow to just grind to a halt.

I then noticed the `.cancel()` function and decided to see what results I get out of that, and sure enough, I had no further issues. So clearly, calling `.destroy(err?)` doesn't inform the server of a cancelled call and the server then reaches its max cap of streams its prepared to handle on this channel, and then no data flows between the parties.

### Describe the solution you'd like
I'd like to see the `.destroy(err?)` method implemented to work. I believe the base function can be wrapped with code to ensure the workflow doesn't end abruptly.

The overriding function could match something like:

```ts
class ClientReadableStreams extends Readable {
destroy(err?: Error): void {
// Call the server with a CANCELLED status
// invoke the base .destroy(err?) function ... Do not throw the CANCELLED error currently thrown by .cancel()
}
}
```

Consider this example:
```ts
return new Promise((resolve, reject) => {
const stream: ClientReadableStream = await getStreamSomehow();

const complete = (err?: Error) => {
if (err) {
return reject(err);
}

resolve();
}

let dataPacketsHandled = 0;
stream.on('data', async data => {
stream.pause();

try {
await determineIfDataIsValid(data); // this could throw an InvalidDataError
await storeData(data); // this could a variety of errors
} catch (err) {
return stream.destroy(err)
}

if (++dataPacketsHandled >= 50) {
// time to go handle something else
return stream.destroy();
}

stream.resume()
});
stream.on('error', err => complete(err));
stream.on('close', () => complete());
});
```

With this promise, I can get exact information on what error occurred (if an error occurred), rather than just being aware that the stream was cancelled.

To me, this is intuitive and conforms to how I would handle a non-grpc stream.

### Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

### Additional context
If this feature is not approved, please add information to the documentation telling us not to use `.destroy(err?)`

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.