This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 3809ed64d72d CAMEL-25546: camel-whatsapp - Complete the exchange when
the HTTP request fails (#27694)
3809ed64d72d is described below
commit 3809ed64d72dba18c1d60caf81df35ad1cc5aaa1
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 12 23:56:54 2026 +0530
CAMEL-25546: camel-whatsapp - Complete the exchange when the HTTP request
fails (#27694)
The producer handled the result of HttpClient.sendAsync with thenAccept,
which
only runs for a response. A request that failed without one (connection
refused or reset, DNS failure, connect timeout) left the callback uncalled,
so
the exchange never completed: the caller blocked forever and graceful
shutdown
waited for its timeout.
The result is now handled with whenComplete, which calls the callback once
in
all cases. A failed request fails the exchange with a RuntimeCamelException
"Could not send message to <uri>" with the HTTP client exception as cause;
the
URI is sanitized (the access token is sent in the Authorization header, not
the
URI). The exception for an error status also names the sanitized URI, and
any
exception from the response handling is now set on the exchange instead of
being lost.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../service/WhatsAppServiceRestAPIAdapter.java | 27 +++--
.../whatsapp/WhatsAppProducerFailureTest.java | 132 +++++++++++++++++++++
2 files changed, 152 insertions(+), 7 deletions(-)
diff --git
a/components/camel-whatsapp/src/main/java/org/apache/camel/component/whatsapp/service/WhatsAppServiceRestAPIAdapter.java
b/components/camel-whatsapp/src/main/java/org/apache/camel/component/whatsapp/service/WhatsAppServiceRestAPIAdapter.java
index 895bb0bfb249..53ea9c2c2796 100644
---
a/components/camel-whatsapp/src/main/java/org/apache/camel/component/whatsapp/service/WhatsAppServiceRestAPIAdapter.java
+++
b/components/camel-whatsapp/src/main/java/org/apache/camel/component/whatsapp/service/WhatsAppServiceRestAPIAdapter.java
@@ -30,6 +30,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.function.Supplier;
import com.fasterxml.jackson.annotation.JsonInclude;
@@ -51,6 +52,7 @@ import
org.apache.camel.component.whatsapp.model.TextMessageRequest;
import org.apache.camel.component.whatsapp.model.UploadMediaRequest;
import org.apache.camel.component.whatsapp.util.FileUploadStreamSupplier;
import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.URISupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -197,8 +199,20 @@ public class WhatsAppServiceRestAPIAdapter implements
WhatsAppService {
CompletableFuture<HttpResponse<String>> asyncResponse
= httpClient.sendAsync(httpRequestBuilder.build(),
HttpResponse.BodyHandlers.ofString());
- asyncResponse.thenAccept(response -> {
+ // whenComplete (and not thenAccept), so that the exchange is also
completed when there is no response,
+ // for example when the connection failed or the connect timeout
expired. The request itself has no
+ // timeout, so a server that accepts the connection and never
answers is not covered here.
+ asyncResponse.whenComplete((response, failure) -> {
try {
+ if (failure != null) {
+ Throwable cause = failure instanceof
CompletionException && failure.getCause() != null
+ ? failure.getCause() : failure;
+ // the exception of the HTTP client often has no
message (a refused connection), so say
+ // where the request went
+ exchange.setException(new RuntimeCamelException(
+ "Could not send message to " +
URISupport.sanitizeUri(uri), cause));
+ return;
+ }
int statusCode = response.statusCode();
LOG.info("Response received with status {} and body {}",
statusCode, response.body());
final boolean success = statusCode >= 200 && statusCode <
300;
@@ -208,14 +222,13 @@ public class WhatsAppServiceRestAPIAdapter implements
WhatsAppService {
exchange.getMessage().setBody(result);
} else {
LOG.debug("Error response Headers {}",
response.headers());
- RuntimeException exception = new RuntimeCamelException(
- uri + " responded: " + statusCode + " and
body: " + response.body());
- exchange.setException(exception);
- throw exception;
+ exchange.setException(new RuntimeCamelException(
+ URISupport.sanitizeUri(uri) + " responded: " +
statusCode + " and body: "
+ +
response.body()));
}
- } catch (JsonProcessingException e) {
+ } catch (Exception e) {
+ // any failure while handling the response (not only a
JSON one) must fail the exchange
exchange.setException(e);
- throw RuntimeCamelException.wrapRuntimeCamelException(e);
} finally {
callback.done(false);
}
diff --git
a/components/camel-whatsapp/src/test/java/org/apache/camel/component/whatsapp/WhatsAppProducerFailureTest.java
b/components/camel-whatsapp/src/test/java/org/apache/camel/component/whatsapp/WhatsAppProducerFailureTest.java
new file mode 100644
index 000000000000..4198a831b137
--- /dev/null
+++
b/components/camel-whatsapp/src/test/java/org/apache/camel/component/whatsapp/WhatsAppProducerFailureTest.java
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.whatsapp;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.TimeUnit;
+
+import com.sun.net.httpserver.HttpServer;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.whatsapp.model.MessageResponse;
+import org.apache.camel.component.whatsapp.model.TextMessage;
+import org.apache.camel.component.whatsapp.model.TextMessageRequest;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The producer must complete the exchange for every outcome of the HTTP
request: a response, an error response, and no
+ * response at all.
+ */
+class WhatsAppProducerFailureTest extends CamelTestSupport {
+
+ private HttpServer server;
+ private volatile int status = 200;
+ private volatile String responseBody = "{}";
+ // nothing listens on this port
+ private final int closedPort = AvailablePortFinder.getNextAvailable();
+
+ @Override
+ protected void setupResources() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
+ server.createContext("/", exchange -> {
+ byte[] body = responseBody.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(status, body.length);
+ exchange.getResponseBody().write(body);
+ exchange.close();
+ });
+ server.start();
+ }
+
+ @Override
+ protected void cleanupResources() {
+ server.stop(0);
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ WhatsAppComponent component = new WhatsAppComponent();
+ component.setAuthorizationToken("token");
+ context.addComponent("whatsapp", component);
+ return context;
+ }
+
+ @Test
+ void failsTheExchangeWhenTheConnectionFails() throws Exception {
+ // an asynchronous send, so that an exchange that is never completed
fails the test instead of blocking it
+ Exchange exchange = template.asyncSend("direct:unreachable", e ->
e.getIn().setBody(textMessage()))
+ .get(30, TimeUnit.SECONDS);
+
+ // the exception of the HTTP client (a ConnectException without a
message) is the cause, and the message says
+ // where the request went
+
assertThat(exchange.getException()).isInstanceOf(RuntimeCamelException.class)
+ .hasMessageContaining("Could not send message to
http://localhost:"; + closedPort)
+ .hasCauseInstanceOf(IOException.class);
+ }
+
+ @Test
+ void failsTheExchangeForAnErrorResponse() throws Exception {
+ status = 400;
+ responseBody = "{\"error\":{\"message\":\"Unsupported post
request\",\"code\":100}}";
+
+ Exchange exchange = template.asyncSend("direct:server", e ->
e.getIn().setBody(textMessage()))
+ .get(30, TimeUnit.SECONDS);
+
+
assertThat(exchange.getException()).isInstanceOf(RuntimeCamelException.class)
+ .hasMessageContaining("http://localhost:"; +
server.getAddress().getPort())
+ .hasMessageContaining("responded:
400").hasMessageContaining("Unsupported post request");
+ }
+
+ @Test
+ void returnsTheResponse() throws Exception {
+ responseBody =
"{\"messaging_product\":\"whatsapp\",\"messages\":[{\"id\":\"wamid.1\"}]}";
+
+ Exchange exchange = template.asyncSend("direct:server", e ->
e.getIn().setBody(textMessage()))
+ .get(30, TimeUnit.SECONDS);
+
+ assertThat(exchange.getException()).isNull();
+
assertThat(exchange.getMessage().getBody()).isInstanceOf(MessageResponse.class);
+ }
+
+ private static TextMessageRequest textMessage() {
+ TextMessageRequest request = new TextMessageRequest();
+ request.setTo("15550001111");
+ request.setText(new TextMessage());
+ request.getText().setBody("Hello");
+ return request;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:unreachable").to("whatsapp:123?baseUri=http://localhost:"; +
closedPort);
+
from("direct:server").to("whatsapp:123?baseUri=http://localhost:"; +
server.getAddress().getPort());
+ }
+ };
+ }
+}
