googleapis / googleapis/google-cloud-java

[sdk-platform-java] `ServerStream.stream()` should release underlying resources when calling `Stream.close()` method

未关闭
#12,410 0 条评论 0 个 reaction 已指派 1 人 已被 @blakeli0 认领 在 GitHub 查看
priority: p3 type: feature request
主要语言
Java
星标
2.1k
派生
1.2k
平均合并
1 天 23 小时
30 天内合并 PR
154

描述

**Is your feature request related to a problem? Please describe.**
When I use `ServerStream.stream()` and call any Stream terminal operation that does not consume all elements of the stream like `findAny()`, `anyMatch(Predicate)`, all underlying resources of the `ServerStream` are not released.
```
ServerStream serverStream= ...;
try(Stream stream = serverStream.stream()) {
//consume partially the stream
}
//At this point serverStream underlying resources are not released
```

**Describe the solution you'd like**
The derived stream should close all underlying resources of the main `ServerStream` when it is closed.
To do that, the `stream()` method should register an `onClose(Runnable)` on the stream before returning it.
This `Runnable` should cancel the `ServerStream` when the stream is not fully consumed.
With this, the stream user should add the statement try-with-resources to ensure all underlying resources are released in all cases.

```
ServerStream serverStream= ...;
try(Stream stream = serverStream.stream()) {
//consume totally or partially the stream
}
//At this point serverStream is fully consumed or canceled
```

***Proposal implementation steps***

In `ServerStreamIterator`, a new method can be created to be able to know if the iterator is fully consumed.
```
boolean isFullyConsumed() {
return last == QueuingResponseObserver.EOF_MARKER;
}
```

In `ServerStream`, a new method can be created to cancel only when iterator is not fully consumed.
```
void cancelIfNecessary() {
if (!iterator.isFullyConsumed()){
cancel();
}
}
```

In `ServerStream`, the `stream()` method can be updated to register the `onClose(Runnable)` to call the above method
```
public Stream stream() {
return StreamSupport.stream(this.spliterator(), false).onClose(this::cancelIfNecessary);
}
```

**Describe alternatives you've considered**
I have tried two different workarounds.

***Register a `onClose(Runnable)` that cancel the `ServerStream`***
```
ServerStream serverStream= ...;
return serverStream.stream().onClose(serverStream::cancel)
```
But with this, `ServerStream` is canceled even if all its items have been successfully consumed. I don't think it's a good practice to do this.

***Not use `stream()` method, but create the stream from the `iterator()` and register the `onClose(Runnable)` on it that check result of `iterator.hasNext()` to cancel or not the `ServerStream`***
```
ServerStream serverStream= ...;
Iterator iterator = serverStream.iterator();
return StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, 0), false).onClose(() -> {
if (iterator.hasNext()) {
serverStream.cancel();
}
})
```
But `iterator.hasNext()` can force the `Thread` to wait until next element is received, which is a waste of time in that case.

贡献指南

打开贡献指南

评估

这个 Issue 还没有评估数据。

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。