davsclaus commented on code in PR #27761:
URL: https://github.com/apache/camel/pull/27761#discussion_r4255126488
##########
components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramMessageHandler.java:
##########
@@ -63,18 +64,26 @@ public void sendMessage(Exchange exchange, AsyncCallback
callback, T message) {
builder.setHeader("Accept", "application/json");
builder.POST(bodyPublisher.newPublisher());
- try {
- final TelegramAsyncHandler telegramAsyncHandler
- = new TelegramAsyncHandler(uri, resultClass, mapper,
exchange, callback);
+ final TelegramAsyncHandler telegramAsyncHandler = new
TelegramAsyncHandler(uri, resultClass, mapper, exchange);
+ final CompletableFuture<Object> response
+ = apiClient.client().sendAsync(builder.build(),
HttpResponse.BodyHandlers.ofInputStream())
+
.thenApply(telegramAsyncHandler::handleCompressedResponse);
- apiClient.client().sendAsync(builder.build(),
HttpResponse.BodyHandlers.ofInputStream())
-
.thenApply(telegramAsyncHandler::handleCompressedResponse).get();
+ try {
+ // the calling thread waits for the answer, so a caller sends one
message after the other
+ response.get();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- throw new RuntimeException(e);
+ // nobody waits for the answer any more: keep the stage from
writing to the completed exchange later
+ response.cancel(true);
Review Comment:
`response` is the `thenApply` stage, and cancelling a dependent stage does
not cancel the source future. So the HTTP request keeps going. When its answer
arrives the handler is skipped, and the `ofInputStream()` body is never read or
closed, so the connection is not released.
Since Java 16 the `HttpClient` aborts the exchange when the future returned
by `sendAsync` is cancelled. Keeping that future and cancelling it would abort
the request, and the dependent stage gets cancelled with it:
```java
CompletableFuture<HttpResponse<InputStream>> sent =
apiClient.client().sendAsync(...);
CompletableFuture<Object> response =
sent.thenApply(telegramAsyncHandler::handleCompressedResponse);
...
sent.cancel(true);
```
This only matters on interrupt (for example a forced shutdown), so it is not
blocking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
