googleapis / googleapis/google-cloud-java

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

Abierto
#12,410 0 comentarios 0 reacciones 1 asignado Reclamado por @blakeli0 Ver en GitHub
priority: p3 type: feature request
Lenguaje dominante
Java
Estrellas
2.1k
Forks
1.2k
Merge medio
1 d 23 h
PR fusionados (30 d)
154

Descripción

**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.

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.