davsclaus commented on code in PR #29935:
URL: https://github.com/apache/camel/pull/29935#discussion_r4575288369
##########
core/camel-console/src/main/java/org/apache/camel/impl/console/SendDevConsole.java:
##########
@@ -270,20 +280,59 @@ private Exchange findToTarget(
if (poll) {
out = consumer.receive(target, timeout);
} else {
- final String mep = exchangePattern;
- out = producer.send(target, exchange -> {
- exchange.getMessage().setBody(inputBody);
- if (!inputHeaders.isEmpty()) {
- exchange.getMessage().setHeaders(inputHeaders);
- }
- exchange.setPattern(
- "InOut".equals(mep) ? ExchangePattern.InOut :
ExchangePattern.InOnly);
- });
+ Exchange exchange = target.createExchange(
+ "InOut".equals(exchangePattern) ?
ExchangePattern.InOut : ExchangePattern.InOnly);
+ exchange.getMessage().setBody(inputBody);
+ if (!inputHeaders.isEmpty()) {
+ exchange.getMessage().setHeaders(inputHeaders);
+ }
+ if (isDirectWithoutConsumer(target)) {
+ // direct waits 30 seconds for a consumer by default: no
route will start to consume it now
+ throw new CamelExchangeException("No consumers available
on endpoint: " + target, exchange);
+ }
+ CompletableFuture<Exchange> future =
producer.asyncSend(target, exchange);
+ try {
+ out = future.get(timeout, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ // the exchange goes on in the integration, but the caller
has its answer
+ throw new SendTimeoutException(exchange, timeout);
+ } catch (ExecutionException e) {
+ throw e.getCause() instanceof Exception ex ? ex : e;
+ }
Review Comment:
Right, fixed in 1b7c5447a4e2: an `InterruptedException` restores the
interrupt flag and fails the send with "The send was interrupted".
`testAnInterruptedSendKeepsTheInterrupt` checks both.
_Claude Code on behalf of davsclaus_
##########
core/camel-console/src/main/java/org/apache/camel/impl/console/SendDevConsole.java:
##########
@@ -270,20 +280,59 @@ private Exchange findToTarget(
if (poll) {
out = consumer.receive(target, timeout);
} else {
- final String mep = exchangePattern;
- out = producer.send(target, exchange -> {
- exchange.getMessage().setBody(inputBody);
- if (!inputHeaders.isEmpty()) {
- exchange.getMessage().setHeaders(inputHeaders);
- }
- exchange.setPattern(
- "InOut".equals(mep) ? ExchangePattern.InOut :
ExchangePattern.InOnly);
- });
+ Exchange exchange = target.createExchange(
+ "InOut".equals(exchangePattern) ?
ExchangePattern.InOut : ExchangePattern.InOnly);
+ exchange.getMessage().setBody(inputBody);
+ if (!inputHeaders.isEmpty()) {
+ exchange.getMessage().setHeaders(inputHeaders);
+ }
+ if (isDirectWithoutConsumer(target)) {
+ // direct waits 30 seconds for a consumer by default: no
route will start to consume it now
+ throw new CamelExchangeException("No consumers available
on endpoint: " + target, exchange);
+ }
+ CompletableFuture<Exchange> future =
producer.asyncSend(target, exchange);
+ try {
+ out = future.get(timeout, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ // the exchange goes on in the integration, but the caller
has its answer
+ throw new SendTimeoutException(exchange, timeout);
+ } catch (ExecutionException e) {
Review Comment:
Right, fixed in 1b7c5447a4e2: an `InterruptedException` restores the
interrupt flag and fails the send with "The send was interrupted".
`testAnInterruptedSendKeepsTheInterrupt` checks both.
_Claude Code on behalf of davsclaus_
--
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]
