This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26506-jms-transferexception-sjms-stream in repository https://gitbox.apache.org/repos/asf/camel.git
commit 0cb14be370720d3d2c44c6255c85241fe7620bf0 Author: Claus Ibsen <[email protected]> AuthorDate: Thu Nov 12 17:37:23 2026 +0100 CAMEL-26506: camel-jms, camel-sjms - an exception sent back with transferException is not rolled back, and sjms maps a StreamMessage to an InputStream With transferException=true, the consumer sent the exception as the reply, and then looked the failure up again with getException(RuntimeCamelException.class), which walks the causes. When the exception was or wrapped a RuntimeCamelException it was rethrown, so a transacted session rolled back the request and the reply with it, and the request was redelivered. The consumer now rethrows only the failure its callback decided on, in camel-jms and camel-sjms (and so camel-sjms2). The camel-sjms consumer gave a received StreamMessage as the body itself. It is now an InputStream over its bytes, as in camel-jms. StreamMessageInputStream (camel-jms, and the new copy in camel-sjms) honours the offset and length of read(byte[], int, int), which readAllBytes and InputStreamReader use, and read() no longer calls readByte, which a byte[] field does not allow. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../org/apache/camel/catalog/components/sjms.json | 2 +- .../org/apache/camel/catalog/components/sjms2.json | 2 +- .../apache/camel/catalog/docs/sjms-component.adoc | 3 +- .../apache/camel/catalog/docs/sjms2-component.adoc | 3 +- .../component/jms/EndpointMessageListener.java | 18 ++- .../component/jms/StreamMessageInputStream.java | 53 +++++--- .../jms/JmsTransferExceptionTransactedTest.java | 120 +++++++++++++++++ .../jms/StreamMessageInputStreamTest.java | 148 +++++++++++++++++++++ .../org/apache/camel/component/sjms/sjms.json | 2 +- .../camel-sjms/src/main/docs/sjms-component.adoc | 3 +- .../apache/camel/component/sjms/SjmsEndpoint.java | 12 +- .../sjms/consumer/EndpointMessageListener.java | 19 ++- .../camel/component/sjms/jms/JmsBinding.java | 8 +- .../sjms}/jms/StreamMessageInputStream.java | 55 +++++--- .../sjms/consumer/StreamMessageConsumerTest.java | 73 ++++++++++ .../consumer/TransferExceptionTransactedTest.java | 86 ++++++++++++ .../sjms/jms/StreamMessageInputStreamTest.java | 148 +++++++++++++++++++++ .../org/apache/camel/component/sjms2/sjms2.json | 2 +- .../camel-sjms2/src/main/docs/sjms2-component.adoc | 3 +- .../camel/component/sjms2/Sjms2Endpoint.java | 10 +- .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 14 ++ 21 files changed, 720 insertions(+), 64 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms.json index eca5f4897ba7..ce7324138151 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms.json @@ -48,7 +48,7 @@ "CamelSjmsBatchSize": { "index": 4, "kind": "header", "displayName": "", "group": "batch", "label": "consumer,batch", "required": false, "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "description": "The size of the batch when using the batching consumer option.", "constantName": "org.apache.camel.component.sjms.SjmsConstants#SJMS_BATCH_SIZE_HEADER" } }, "bodyTypes": { - "consumerOut": { "carriers": [ "text", "bytes", "map", "pojo", "null" ], "byOption": [ { "option": "mapJmsMessage", "value": "false", "carriers": [ "pojo" ], "javaType": "jakarta.jms.Message" }, { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "map", "value", "list", "pojo", "null" ] } ], "description": "By the JMS message received: a String for a TextMessage, a byte array for a BytesMessage, a Map for a MapMessage, the jakarta.jms.StreamMessage its [...] + "consumerOut": { "carriers": [ "text", "bytes", "map", "stream", "null" ], "byOption": [ { "option": "mapJmsMessage", "value": "false", "carriers": [ "pojo" ], "javaType": "jakarta.jms.Message" }, { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "map", "stream", "value", "list", "pojo", "null" ] } ], "description": "By the JMS message received: a String for a TextMessage, a byte array for a BytesMessage, a Map for a MapMessage, an InputStream for a [...] "consumerReply": { "carriers": [ "text", "bytes", "stream", "file", "map", "null" ], "dependsOn": [ "jmsMessageType" ], "byOption": [ { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description": "The reply, sent as the producer sends a message. A reply that cannot be sent, such as a number or a List without objectMessageEnabled, fails the incoming message (rolled back when transacted [...] "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map", "null" ], "dependsOn": [ "jmsMessageType", "allowNullBody" ], "byOption": [ { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description": "Text is sent as a TextMessage; bytes, a stream or a file as a BytesMessage; a Map of simple values as a MapMessage; null as an empty Message. A number, a List or another Serial [...] "producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option": "exchangePattern", "value": "InOnly", "carriers": [ "unchanged" ] }, { "option": "disableReplyTo", "value": "true", "carriers": [ "unchanged" ] } ], "description": "Unchanged for a message that expects no reply (InOnly). A message that expects one (InOut) gets the body of the reply, as the consumer gives it" } diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms2.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms2.json index 33fd35b4d562..9c8b91beff0c 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms2.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms2.json @@ -48,7 +48,7 @@ "CamelSjmsBatchSize": { "index": 4, "kind": "header", "displayName": "", "group": "batch", "label": "consumer,batch", "required": false, "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "description": "The size of the batch when using the batching consumer option.", "constantName": "org.apache.camel.component.sjms.SjmsConstants#SJMS_BATCH_SIZE_HEADER" } }, "bodyTypes": { - "consumerOut": { "carriers": [ "text", "bytes", "map", "pojo", "null" ], "byOption": [ { "option": "mapJmsMessage", "value": "false", "carriers": [ "pojo" ], "javaType": "jakarta.jms.Message" }, { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "map", "value", "list", "pojo", "null" ] } ], "description": "By the JMS message received: a String for a TextMessage, a byte array for a BytesMessage, a Map for a MapMessage, the jakarta.jms.StreamMessage its [...] + "consumerOut": { "carriers": [ "text", "bytes", "map", "stream", "null" ], "byOption": [ { "option": "mapJmsMessage", "value": "false", "carriers": [ "pojo" ], "javaType": "jakarta.jms.Message" }, { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "map", "stream", "value", "list", "pojo", "null" ] } ], "description": "By the JMS message received: a String for a TextMessage, a byte array for a BytesMessage, a Map for a MapMessage, an InputStream for a [...] "consumerReply": { "carriers": [ "text", "bytes", "stream", "file", "map", "null" ], "dependsOn": [ "jmsMessageType" ], "byOption": [ { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description": "The reply, sent as the producer sends a message. A reply that cannot be sent, such as a number or a List without objectMessageEnabled, fails the incoming message (rolled back when transacted [...] "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map", "null" ], "dependsOn": [ "jmsMessageType", "allowNullBody" ], "byOption": [ { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description": "Text is sent as a TextMessage; bytes, a stream or a file as a BytesMessage; a Map of simple values as a MapMessage; null as an empty Message. A number, a List or another Serial [...] "producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option": "exchangePattern", "value": "InOnly", "carriers": [ "unchanged" ] }, { "option": "disableReplyTo", "value": "true", "carriers": [ "unchanged" ] } ], "description": "Unchanged for a message that expects no reply (InOnly). A message that expects one (InOut) gets the body of the reply, as the consumer gives it" } diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms-component.adoc index d12792d5ff25..318255a79d83 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms-component.adoc @@ -302,6 +302,7 @@ The body is mapped to and from a JMS message as follows (unless the option `jmsM | `String`, other text (a `CharSequence` such as a `StringBuilder`) or XML `Node` | `TextMessage` | `String` | `byte[]`, `InputStream`, `Reader`, `ByteBuffer`, `File`, `Path` or `StreamCache` | `BytesMessage` | `byte[]` | `Map` (of simple values) | `MapMessage` | `Map` +| Any body that converts to an `InputStream`, with `jmsMessageType=Stream` | `StreamMessage` | `InputStream` | `null` | `Message` | `null` | Other `Serializable` object (such as a number or a `List`) | `ObjectMessage` (only with `objectMessageEnabled=true`, fails otherwise) | the object | Any other object that converts to an `InputStream` (or `File`) | `BytesMessage` | `byte[]` @@ -309,7 +310,7 @@ The body is mapped to and from a JMS message as follows (unless the option `jmsM |=== With `jmsMessageType=Stream` the body is converted to an `InputStream`, and its bytes are written to a `StreamMessage`. -A received `StreamMessage` is not mapped: the body is the `jakarta.jms.StreamMessage` itself. +A received `StreamMessage` is mapped to an `InputStream` over its bytes, as in the JMS component. With `mapJmsMessage=false` the body is always the received `jakarta.jms.Message`. === Clustering diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms2-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms2-component.adoc index 9fa42cf85235..7bc5b9e0f2c8 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms2-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms2-component.adoc @@ -306,6 +306,7 @@ The body is mapped to and from a JMS message as follows (unless the option `jmsM | `String`, other text (a `CharSequence` such as a `StringBuilder`) or XML `Node` | `TextMessage` | `String` | `byte[]`, `InputStream`, `Reader`, `ByteBuffer`, `File`, `Path` or `StreamCache` | `BytesMessage` | `byte[]` | `Map` (of simple values) | `MapMessage` | `Map` +| Any body that converts to an `InputStream`, with `jmsMessageType=Stream` | `StreamMessage` | `InputStream` | `null` | `Message` | `null` | Other `Serializable` object (such as a number or a `List`) | `ObjectMessage` (only with `objectMessageEnabled=true`, fails otherwise) | the object | Any other object that converts to an `InputStream` (or `File`) | `BytesMessage` | `byte[]` @@ -313,7 +314,7 @@ The body is mapped to and from a JMS message as follows (unless the option `jmsM |=== With `jmsMessageType=Stream` the body is converted to an `InputStream`, and its bytes are written to a `StreamMessage`. -A received `StreamMessage` is not mapped: the body is the `jakarta.jms.StreamMessage` itself. +A received `StreamMessage` is mapped to an `InputStream` over its bytes, as in the JMS component. With `mapJmsMessage=false` the body is always the received `jakarta.jms.Message`. === Clustering diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java index 29be0c3b4838..b34926d0a741 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java @@ -118,7 +118,7 @@ public class EndpointMessageListener implements SessionAwareMessageListener { // process the exchange either asynchronously or synchronous LOG.trace("onMessage.process START"); - AsyncCallback callback + EndpointMessageListenerAsyncCallback callback = new EndpointMessageListenerAsyncCallback(message, exchange, endpoint, sendReply, replyDestination); // async is by default false, which mean we by default will process the exchange synchronously @@ -150,8 +150,10 @@ public class EndpointMessageListener implements SessionAwareMessageListener { return; } } - // if we failed processed the exchange from the async callback task, then grab the exception - rce = exchange.getException(RuntimeCamelException.class); + // rethrow only the failure the callback decided on: an exception that was sent back as the reply + // (transferException) is not rethrown, as a transacted session would then roll back the request + // and the reply with it, and the request would be redelivered + rce = callback.getFailure(); // release back when synchronous mode consumer.releaseExchange(exchange, false); @@ -182,6 +184,7 @@ public class EndpointMessageListener implements SessionAwareMessageListener { private final JmsEndpoint endpoint; private final boolean sendReply; private final Object replyDestination; + private RuntimeCamelException failure; private EndpointMessageListenerAsyncCallback(Message message, Exchange exchange, JmsEndpoint endpoint, boolean sendReply, Object replyDestination) { @@ -198,6 +201,7 @@ public class EndpointMessageListener implements SessionAwareMessageListener { // Evaluate the processing of the exchange and determine if it was a success or failure RuntimeCamelException rce = evaluateExchangeResult(); + failure = rce; // Handle reply and exception handleReplyIfNeeded(rce); @@ -209,6 +213,14 @@ public class EndpointMessageListener implements SessionAwareMessageListener { } } + /** + * The failure to rethrow from onMessage when the exchange was done synchronously, or null when the message was + * processed, including when the exception was sent back as the reply (transferException). + */ + RuntimeCamelException getFailure() { + return failure; + } + private RuntimeCamelException evaluateExchangeResult() { if (exchange.isRollbackOnly()) { return wrapRuntimeCamelException(new RollbackExchangeException(exchange)); diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/StreamMessageInputStream.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/StreamMessageInputStream.java index 400e96dd2db5..e8260f8fb340 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/StreamMessageInputStream.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/StreamMessageInputStream.java @@ -18,6 +18,7 @@ package org.apache.camel.component.jms; import java.io.IOException; import java.io.InputStream; +import java.util.Objects; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -25,6 +26,10 @@ import jakarta.jms.JMSException; import jakarta.jms.MessageEOFException; import jakarta.jms.StreamMessage; +/** + * An {@link InputStream} over the bytes of a received {@link StreamMessage}, as written by the producer: one or more + * {@code byte[]} fields, which are read one after the other. + */ public class StreamMessageInputStream extends InputStream { private final Lock lock = new ReentrantLock(); @@ -37,27 +42,44 @@ public class StreamMessageInputStream extends InputStream { @Override public int read() throws IOException { - try { - return message.readByte(); - } catch (MessageEOFException e) { - eof = true; - return -1; - } catch (JMSException e) { - throw new IOException(e); - } + // the message holds byte[] fields, which cannot be read with readByte, so read one byte through the array + byte[] single = new byte[1]; + int num = read(single, 0, 1); + return num < 0 ? -1 : single[0] & 0xFF; } @Override public int read(byte[] array) throws IOException { - return doRead(array); + return read(array, 0, array.length); + } + + @Override + public int read(byte[] array, int off, int len) throws IOException { + Objects.checkFromIndexSize(off, len, array.length); + if (len == 0) { + return 0; + } + if (off == 0 && len == array.length) { + return doRead(array); + } + // readBytes always fills from the start of the array, so read into a buffer and copy it to the offset + byte[] buffer = new byte[len]; + int num = doRead(buffer); + if (num > 0) { + System.arraycopy(buffer, 0, array, off, num); + } + return num; } private int doRead(byte[] array) throws IOException { + if (eof) { + return -1; + } try { int num = message.readBytes(array); - if (num < 0) { - //the first 128K(FileUtil.BUFFER_SIZE/128K is used when sending JMS StreamMessage) - //buffer reached, give a chance to see if there is the next 128K buffer + if (num <= 0) { + // the end of the current byte[] field (the producer writes the payload in chunks) + // so continue with the next field, if there is one num = message.readBytes(array); } eof = num < 0; @@ -70,17 +92,12 @@ public class StreamMessageInputStream extends InputStream { } } - @Override - public int read(byte[] array, int off, int len) throws IOException { - // we cannot honor off and len, but assuming off is always 0 - return doRead(array); - } - @Override public void reset() throws IOException { lock.lock(); try { message.reset(); + eof = false; } catch (JMSException e) { throw new IOException(e); } finally { diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsTransferExceptionTransactedTest.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsTransferExceptionTransactedTest.java new file mode 100644 index 000000000000..72a93ec7b2e4 --- /dev/null +++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsTransferExceptionTransactedTest.java @@ -0,0 +1,120 @@ +/* + * 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.jms; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import jakarta.jms.ConnectionFactory; + +import org.apache.camel.CamelContext; +import org.apache.camel.CamelExecutionException; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.RuntimeCamelException; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.test.infra.core.CamelContextExtension; +import org.apache.camel.test.infra.core.TransientCamelContextExtension; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Order; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * With transferException, an exception that is sent back as the reply commits the transacted session, whatever its + * type: the reply reaches the caller and the request is not redelivered. + */ +public class JmsTransferExceptionTransactedTest extends AbstractJMSTest { + + @Order(2) + @RegisterExtension + public static CamelContextExtension camelContextExtension = new TransientCamelContextExtension(); + + private static final String REQUEST_OPTIONS = "?transferException=true&requestTimeout=5000"; + private static final String CONSUMER_OPTIONS = "?transacted=true&transferException=true"; + + private final AtomicInteger counter = new AtomicInteger(); + private final AtomicInteger wrappedCounter = new AtomicInteger(); + protected CamelContext context; + protected ProducerTemplate template; + + @Test + public void testRuntimeCamelExceptionIsSentBack() { + assertSentBackOnce("activemq:queue:JmsTransferExceptionTransactedTest", counter); + } + + @Test + public void testExceptionCausedByRuntimeCamelExceptionIsSentBack() { + assertSentBackOnce("activemq:queue:JmsTransferExceptionTransactedTest.wrapped", wrappedCounter); + } + + private void assertSentBackOnce(String uri, AtomicInteger processed) { + CamelExecutionException e = assertThrows(CamelExecutionException.class, + () -> template.requestBody(uri + REQUEST_OPTIONS, "Hello World")); + assertEquals("Kaboom", e.getCause().getMessage()); + + // the request was committed together with the reply, so it is not redelivered + await().during(1, TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS).until(() -> processed.get() == 1); + } + + @Override + protected String getComponentName() { + return "activemq"; + } + + @Override + protected JmsComponent setupComponent( + CamelContext camelContext, ConnectionFactory connectionFactory, String componentName) { + JmsComponent component = super.setupComponent(camelContext, connectionFactory, componentName); + component.setObjectMessageEnabled(true); + return component; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("activemq:queue:JmsTransferExceptionTransactedTest" + CONSUMER_OPTIONS) + .process(exchange -> { + counter.incrementAndGet(); + throw new RuntimeCamelException("Kaboom"); + }); + + from("activemq:queue:JmsTransferExceptionTransactedTest.wrapped" + CONSUMER_OPTIONS) + .process(exchange -> { + wrappedCounter.incrementAndGet(); + throw new IllegalStateException("Kaboom", new RuntimeCamelException("Inner")); + }); + } + }; + } + + @Override + public CamelContextExtension getCamelContextExtension() { + return camelContextExtension; + } + + @BeforeEach + void setUpRequirements() { + context = camelContextExtension.getContext(); + template = camelContextExtension.getProducerTemplate(); + } +} diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/StreamMessageInputStreamTest.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/StreamMessageInputStreamTest.java new file mode 100644 index 000000000000..b711690fa89a --- /dev/null +++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/StreamMessageInputStreamTest.java @@ -0,0 +1,148 @@ +/* + * 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.jms; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.Reader; +import java.io.StringWriter; +import java.lang.reflect.Proxy; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; + +import jakarta.jms.MessageEOFException; +import jakarta.jms.MessageFormatException; +import jakarta.jms.StreamMessage; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class StreamMessageInputStreamTest { + + private static final byte[] PAYLOAD = "Hello World from a StreamMessage".getBytes(StandardCharsets.UTF_8); + + @Test + public void testReadAllBytesOverSeveralFields() throws IOException { + // readAllBytes reads into the array at an offset after the first read + InputStream is = new StreamMessageInputStream(streamMessage(PAYLOAD, 5)); + + assertArrayEquals(PAYLOAD, is.readAllBytes()); + assertEquals(-1, is.read()); + } + + @Test + public void testReadOneByteAtATime() throws IOException { + // the fields are byte[] fields, which cannot be read with readByte + InputStream is = new StreamMessageInputStream(streamMessage(PAYLOAD, 7)); + + ByteArrayOutputStream out = new ByteArrayOutputStream(); + int b; + while ((b = is.read()) != -1) { + out.write(b); + } + assertArrayEquals(PAYLOAD, out.toByteArray()); + } + + @Test + public void testReadWithBufferSmallerThanField() throws IOException { + InputStream is = new StreamMessageInputStream(streamMessage(PAYLOAD, 20)); + + ByteArrayOutputStream out = new ByteArrayOutputStream(); + byte[] buffer = new byte[3]; + int num; + while ((num = is.read(buffer)) != -1) { + out.write(buffer, 0, num); + } + assertArrayEquals(PAYLOAD, out.toByteArray()); + } + + @Test + public void testReadTextWithMultiByteCharactersAcrossFields() throws IOException { + String text = "Grüße aus Århus, 東京からこんにちは ".repeat(500); + InputStream is = new StreamMessageInputStream(streamMessage(text.getBytes(StandardCharsets.UTF_8), 1001)); + + StringWriter out = new StringWriter(); + try (Reader reader = new InputStreamReader(is, StandardCharsets.UTF_8)) { + reader.transferTo(out); + } + assertEquals(text, out.toString()); + } + + @Test + public void testResetReadsAgain() throws IOException { + InputStream is = new StreamMessageInputStream(streamMessage(PAYLOAD, 9)); + + assertArrayEquals(PAYLOAD, is.readAllBytes()); + is.reset(); + assertArrayEquals(PAYLOAD, is.readAllBytes()); + } + + /** + * A StreamMessage holding the payload as byte[] fields of the given size, as the producer writes it, which follows + * the readBytes contract of the JMS API. + */ + private static StreamMessage streamMessage(byte[] payload, int fieldSize) { + int count = (payload.length + fieldSize - 1) / fieldSize; + byte[][] fields = new byte[count][]; + for (int i = 0; i < count; i++) { + int from = i * fieldSize; + fields[i] = Arrays.copyOfRange(payload, from, Math.min(payload.length, from + fieldSize)); + } + // the current field, and the position in it (-1 when its reading has not started) + int[] state = { 0, -1 }; + return (StreamMessage) Proxy.newProxyInstance(StreamMessageInputStreamTest.class.getClassLoader(), + new Class<?>[] { StreamMessage.class }, (proxy, method, args) -> { + switch (method.getName()) { + case "readBytes": { + byte[] target = (byte[]) args[0]; + if (state[0] >= fields.length) { + throw new MessageEOFException("No more fields"); + } + byte[] field = fields[state[0]]; + if (state[1] == field.length) { + // the field was read completely with a full buffer: -1, then the next field + state[0]++; + state[1] = -1; + return -1; + } + int pos = Math.max(state[1], 0); + int num = Math.min(field.length - pos, target.length); + System.arraycopy(field, pos, target, 0, num); + state[1] = pos + num; + if (num < target.length) { + // fewer bytes than the buffer: the field is complete + state[0]++; + state[1] = -1; + } + return num; + } + case "readByte": + throw new MessageFormatException("A byte[] field cannot be read as a byte"); + case "reset": + state[0] = 0; + state[1] = -1; + return null; + default: + throw new UnsupportedOperationException(method.getName()); + } + }); + } +} diff --git a/components/camel-sjms/src/generated/resources/META-INF/org/apache/camel/component/sjms/sjms.json b/components/camel-sjms/src/generated/resources/META-INF/org/apache/camel/component/sjms/sjms.json index eca5f4897ba7..ce7324138151 100644 --- a/components/camel-sjms/src/generated/resources/META-INF/org/apache/camel/component/sjms/sjms.json +++ b/components/camel-sjms/src/generated/resources/META-INF/org/apache/camel/component/sjms/sjms.json @@ -48,7 +48,7 @@ "CamelSjmsBatchSize": { "index": 4, "kind": "header", "displayName": "", "group": "batch", "label": "consumer,batch", "required": false, "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "description": "The size of the batch when using the batching consumer option.", "constantName": "org.apache.camel.component.sjms.SjmsConstants#SJMS_BATCH_SIZE_HEADER" } }, "bodyTypes": { - "consumerOut": { "carriers": [ "text", "bytes", "map", "pojo", "null" ], "byOption": [ { "option": "mapJmsMessage", "value": "false", "carriers": [ "pojo" ], "javaType": "jakarta.jms.Message" }, { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "map", "value", "list", "pojo", "null" ] } ], "description": "By the JMS message received: a String for a TextMessage, a byte array for a BytesMessage, a Map for a MapMessage, the jakarta.jms.StreamMessage its [...] + "consumerOut": { "carriers": [ "text", "bytes", "map", "stream", "null" ], "byOption": [ { "option": "mapJmsMessage", "value": "false", "carriers": [ "pojo" ], "javaType": "jakarta.jms.Message" }, { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "map", "stream", "value", "list", "pojo", "null" ] } ], "description": "By the JMS message received: a String for a TextMessage, a byte array for a BytesMessage, a Map for a MapMessage, an InputStream for a [...] "consumerReply": { "carriers": [ "text", "bytes", "stream", "file", "map", "null" ], "dependsOn": [ "jmsMessageType" ], "byOption": [ { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description": "The reply, sent as the producer sends a message. A reply that cannot be sent, such as a number or a List without objectMessageEnabled, fails the incoming message (rolled back when transacted [...] "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map", "null" ], "dependsOn": [ "jmsMessageType", "allowNullBody" ], "byOption": [ { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description": "Text is sent as a TextMessage; bytes, a stream or a file as a BytesMessage; a Map of simple values as a MapMessage; null as an empty Message. A number, a List or another Serial [...] "producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option": "exchangePattern", "value": "InOnly", "carriers": [ "unchanged" ] }, { "option": "disableReplyTo", "value": "true", "carriers": [ "unchanged" ] } ], "description": "Unchanged for a message that expects no reply (InOnly). A message that expects one (InOut) gets the body of the reply, as the consumer gives it" } diff --git a/components/camel-sjms/src/main/docs/sjms-component.adoc b/components/camel-sjms/src/main/docs/sjms-component.adoc index d12792d5ff25..318255a79d83 100644 --- a/components/camel-sjms/src/main/docs/sjms-component.adoc +++ b/components/camel-sjms/src/main/docs/sjms-component.adoc @@ -302,6 +302,7 @@ The body is mapped to and from a JMS message as follows (unless the option `jmsM | `String`, other text (a `CharSequence` such as a `StringBuilder`) or XML `Node` | `TextMessage` | `String` | `byte[]`, `InputStream`, `Reader`, `ByteBuffer`, `File`, `Path` or `StreamCache` | `BytesMessage` | `byte[]` | `Map` (of simple values) | `MapMessage` | `Map` +| Any body that converts to an `InputStream`, with `jmsMessageType=Stream` | `StreamMessage` | `InputStream` | `null` | `Message` | `null` | Other `Serializable` object (such as a number or a `List`) | `ObjectMessage` (only with `objectMessageEnabled=true`, fails otherwise) | the object | Any other object that converts to an `InputStream` (or `File`) | `BytesMessage` | `byte[]` @@ -309,7 +310,7 @@ The body is mapped to and from a JMS message as follows (unless the option `jmsM |=== With `jmsMessageType=Stream` the body is converted to an `InputStream`, and its bytes are written to a `StreamMessage`. -A received `StreamMessage` is not mapped: the body is the `jakarta.jms.StreamMessage` itself. +A received `StreamMessage` is mapped to an `InputStream` over its bytes, as in the JMS component. With `mapJmsMessage=false` the body is always the received `jakarta.jms.Message`. === Clustering diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java index 199db6762a0d..051ed7e4414b 100644 --- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java +++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java @@ -72,18 +72,18 @@ import org.slf4j.LoggerFactory; */ @UriEndpoint(firstVersion = "2.11.0", scheme = "sjms", title = "Simple JMS", syntax = "sjms:destinationType:destinationName", category = { Category.MESSAGING }, headersClass = SjmsConstants.class) -@BodyTypes(consumerOut = @BodyType(value = { Carrier.TEXT, Carrier.BYTES, Carrier.MAP, Carrier.POJO, Carrier.NULL }, +@BodyTypes(consumerOut = @BodyType(value = { Carrier.TEXT, Carrier.BYTES, Carrier.MAP, Carrier.STREAM, Carrier.NULL }, byOption = { @BodyType.When(option = "mapJmsMessage", value = "false", carriers = Carrier.POJO, javaType = "jakarta.jms.Message"), @BodyType.When(option = "objectMessageEnabled", value = "true", carriers = { - Carrier.TEXT, Carrier.BYTES, Carrier.MAP, Carrier.VALUE, - Carrier.LIST, Carrier.POJO, Carrier.NULL }) }, + Carrier.TEXT, Carrier.BYTES, Carrier.MAP, Carrier.STREAM, + Carrier.VALUE, Carrier.LIST, Carrier.POJO, Carrier.NULL }) }, description = "By the JMS message received: a String for a TextMessage, a byte array for a" - + " BytesMessage, a Map for a MapMessage, the jakarta.jms.StreamMessage itself" - + " for a StreamMessage, and null for a plain Message. An ObjectMessage fails" - + " unless objectMessageEnabled"), + + " BytesMessage, a Map for a MapMessage, an InputStream for a StreamMessage," + + " and null for a plain Message. An ObjectMessage fails unless" + + " objectMessageEnabled"), consumerReply = @BodyType(value = { Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.FILE, Carrier.MAP, Carrier.NULL }, diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java index fcc1da6368fb..3c9bb148ec50 100644 --- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java +++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java @@ -186,7 +186,7 @@ public class EndpointMessageListener implements SessionMessageListener { // process the exchange either asynchronously or synchronous LOG.trace("onMessage.process START"); - AsyncCallback callback + EndpointMessageListenerAsyncCallback callback = new EndpointMessageListenerAsyncCallback( session, message, exchange, endpoint, sendReply, replyDestination); @@ -219,8 +219,10 @@ public class EndpointMessageListener implements SessionMessageListener { return; } } - // if we failed processed the exchange from the async callback task, then grab the exception - rce = exchange.getException(RuntimeCamelException.class); + // rethrow only the failure the callback decided on: an exception that was sent back as the reply + // (transferException) is not rethrown, as a transacted session would then roll back the request + // and the reply with it, and the request would be redelivered + rce = callback.getFailure(); // release back when synchronous mode consumer.releaseExchange(exchange, false); @@ -391,6 +393,7 @@ public class EndpointMessageListener implements SessionMessageListener { private final SjmsEndpoint endpoint; private final boolean sendReply; private final Object replyDestination; + private RuntimeCamelException failure; private EndpointMessageListenerAsyncCallback(Session session, Message message, Exchange exchange, SjmsEndpoint endpoint, boolean sendReply, Object replyDestination) { @@ -402,6 +405,14 @@ public class EndpointMessageListener implements SessionMessageListener { this.replyDestination = replyDestination; } + /** + * The failure to rethrow from onMessage when the exchange was done synchronously, or null when the message was + * processed, including when the exception was sent back as the reply (transferException). + */ + RuntimeCamelException getFailure() { + return failure; + } + @Override public void done(boolean doneSync) { LOG.trace("onMessage.process END"); @@ -458,6 +469,8 @@ public class EndpointMessageListener implements SessionMessageListener { } } + failure = rce; + // if an exception occurred if (rce != null) { if (doneSync) { diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsBinding.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsBinding.java index bea673b2193d..82b273f08379 100644 --- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsBinding.java +++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsBinding.java @@ -211,9 +211,9 @@ public class JmsBinding { } else if (message instanceof BytesMessage bytesMessage) { LOG.trace("Extracting body as a BytesMessage from JMS message: {}", message); return createByteArrayFromBytesMessage(bytesMessage); - } else if (message instanceof StreamMessage) { + } else if (message instanceof StreamMessage streamMessage) { LOG.trace("Extracting body as a StreamMessage from JMS message: {}", message); - return message; + return createInputStreamFromStreamMessage(streamMessage); } else { return null; } @@ -286,6 +286,10 @@ public class JmsBinding { return answer; } + protected Object createInputStreamFromStreamMessage(StreamMessage message) { + return new StreamMessageInputStream(message); + } + protected byte[] createByteArrayFromBytesMessage(BytesMessage message) throws JMSException { if (message.getBodyLength() > Integer.MAX_VALUE) { LOG.warn("Length of BytesMessage is too long: {}", message.getBodyLength()); diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/StreamMessageInputStream.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/StreamMessageInputStream.java similarity index 66% copy from components/camel-jms/src/main/java/org/apache/camel/component/jms/StreamMessageInputStream.java copy to components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/StreamMessageInputStream.java index 400e96dd2db5..72b88f5f9e46 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/StreamMessageInputStream.java +++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/StreamMessageInputStream.java @@ -14,10 +14,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.camel.component.jms; +package org.apache.camel.component.sjms.jms; import java.io.IOException; import java.io.InputStream; +import java.util.Objects; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -25,6 +26,10 @@ import jakarta.jms.JMSException; import jakarta.jms.MessageEOFException; import jakarta.jms.StreamMessage; +/** + * An {@link InputStream} over the bytes of a received {@link StreamMessage}, as written by the producer: one or more + * {@code byte[]} fields, which are read one after the other. + */ public class StreamMessageInputStream extends InputStream { private final Lock lock = new ReentrantLock(); @@ -37,27 +42,44 @@ public class StreamMessageInputStream extends InputStream { @Override public int read() throws IOException { - try { - return message.readByte(); - } catch (MessageEOFException e) { - eof = true; - return -1; - } catch (JMSException e) { - throw new IOException(e); - } + // the message holds byte[] fields, which cannot be read with readByte, so read one byte through the array + byte[] single = new byte[1]; + int num = read(single, 0, 1); + return num < 0 ? -1 : single[0] & 0xFF; } @Override public int read(byte[] array) throws IOException { - return doRead(array); + return read(array, 0, array.length); + } + + @Override + public int read(byte[] array, int off, int len) throws IOException { + Objects.checkFromIndexSize(off, len, array.length); + if (len == 0) { + return 0; + } + if (off == 0 && len == array.length) { + return doRead(array); + } + // readBytes always fills from the start of the array, so read into a buffer and copy it to the offset + byte[] buffer = new byte[len]; + int num = doRead(buffer); + if (num > 0) { + System.arraycopy(buffer, 0, array, off, num); + } + return num; } private int doRead(byte[] array) throws IOException { + if (eof) { + return -1; + } try { int num = message.readBytes(array); - if (num < 0) { - //the first 128K(FileUtil.BUFFER_SIZE/128K is used when sending JMS StreamMessage) - //buffer reached, give a chance to see if there is the next 128K buffer + if (num <= 0) { + // the end of the current byte[] field (the producer writes the payload in chunks) + // so continue with the next field, if there is one num = message.readBytes(array); } eof = num < 0; @@ -70,17 +92,12 @@ public class StreamMessageInputStream extends InputStream { } } - @Override - public int read(byte[] array, int off, int len) throws IOException { - // we cannot honor off and len, but assuming off is always 0 - return doRead(array); - } - @Override public void reset() throws IOException { lock.lock(); try { message.reset(); + eof = false; } catch (JMSException e) { throw new IOException(e); } finally { diff --git a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/StreamMessageConsumerTest.java b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/StreamMessageConsumerTest.java new file mode 100644 index 000000000000..c8e72142bb5a --- /dev/null +++ b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/StreamMessageConsumerTest.java @@ -0,0 +1,73 @@ +/* + * 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.sjms.consumer; + +import java.io.InputStream; +import java.util.Random; + +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.sjms.support.JmsTestSupport; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; + +/** + * A received StreamMessage is mapped to an InputStream over its bytes, as in the jms component. + */ +public class StreamMessageConsumerTest extends JmsTestSupport { + + private volatile Object body; + private volatile byte[] received; + + @Override + protected boolean useJmx() { + return false; + } + + @Test + public void testStreamMessageIsInputStream() throws Exception { + // larger than one chunk of the producer, so the message holds several byte[] fields + byte[] payload = new byte[300 * 1024]; + new Random(42).nextBytes(payload); + + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedMessageCount(1); + + template.sendBody("sjms:queue:StreamMessageConsumerTest?jmsMessageType=Stream", payload); + + MockEndpoint.assertIsSatisfied(context); + assertInstanceOf(InputStream.class, body); + assertArrayEquals(payload, received); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + from("sjms:queue:StreamMessageConsumerTest") + .streamCache("false") + .process(exchange -> { + body = exchange.getIn().getBody(); + received = exchange.getIn().getBody(InputStream.class).readAllBytes(); + }) + .to("mock:result"); + } + }; + } +} diff --git a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/TransferExceptionTransactedTest.java b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/TransferExceptionTransactedTest.java new file mode 100644 index 000000000000..aba3bafc9044 --- /dev/null +++ b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/TransferExceptionTransactedTest.java @@ -0,0 +1,86 @@ +/* + * 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.sjms.consumer; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.CamelExecutionException; +import org.apache.camel.RuntimeCamelException; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.sjms.support.JmsTestSupport; +import org.junit.jupiter.api.Test; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * With transferException, an exception that is sent back as the reply commits the transacted session, whatever its + * type: the reply reaches the caller and the request is not redelivered. + */ +public class TransferExceptionTransactedTest extends JmsTestSupport { + + private static final String REQUEST_OPTIONS = "?transferException=true&objectMessageEnabled=true&requestTimeout=5000"; + private static final String CONSUMER_OPTIONS = "?transacted=true&transferException=true&objectMessageEnabled=true"; + + private final AtomicInteger counter = new AtomicInteger(); + private final AtomicInteger wrappedCounter = new AtomicInteger(); + + @Override + protected boolean useJmx() { + return false; + } + + @Test + public void testRuntimeCamelExceptionIsSentBack() { + assertSentBackOnce("sjms:queue:TransferExceptionTransactedTest", counter); + } + + @Test + public void testExceptionCausedByRuntimeCamelExceptionIsSentBack() { + assertSentBackOnce("sjms:queue:TransferExceptionTransactedTest.wrapped", wrappedCounter); + } + + private void assertSentBackOnce(String uri, AtomicInteger processed) { + CamelExecutionException e = assertThrows(CamelExecutionException.class, + () -> template.requestBody(uri + REQUEST_OPTIONS, "Hello World")); + assertEquals("Kaboom", e.getCause().getMessage()); + + // the request was committed together with the reply, so it is not redelivered + await().during(1, TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS).until(() -> processed.get() == 1); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + from("sjms:queue:TransferExceptionTransactedTest" + CONSUMER_OPTIONS) + .process(exchange -> { + counter.incrementAndGet(); + throw new RuntimeCamelException("Kaboom"); + }); + + from("sjms:queue:TransferExceptionTransactedTest.wrapped" + CONSUMER_OPTIONS) + .process(exchange -> { + wrappedCounter.incrementAndGet(); + throw new IllegalStateException("Kaboom", new RuntimeCamelException("Inner")); + }); + } + }; + } +} diff --git a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/jms/StreamMessageInputStreamTest.java b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/jms/StreamMessageInputStreamTest.java new file mode 100644 index 000000000000..993e939f0ef6 --- /dev/null +++ b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/jms/StreamMessageInputStreamTest.java @@ -0,0 +1,148 @@ +/* + * 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.sjms.jms; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.Reader; +import java.io.StringWriter; +import java.lang.reflect.Proxy; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; + +import jakarta.jms.MessageEOFException; +import jakarta.jms.MessageFormatException; +import jakarta.jms.StreamMessage; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class StreamMessageInputStreamTest { + + private static final byte[] PAYLOAD = "Hello World from a StreamMessage".getBytes(StandardCharsets.UTF_8); + + @Test + public void testReadAllBytesOverSeveralFields() throws IOException { + // readAllBytes reads into the array at an offset after the first read + InputStream is = new StreamMessageInputStream(streamMessage(PAYLOAD, 5)); + + assertArrayEquals(PAYLOAD, is.readAllBytes()); + assertEquals(-1, is.read()); + } + + @Test + public void testReadOneByteAtATime() throws IOException { + // the fields are byte[] fields, which cannot be read with readByte + InputStream is = new StreamMessageInputStream(streamMessage(PAYLOAD, 7)); + + ByteArrayOutputStream out = new ByteArrayOutputStream(); + int b; + while ((b = is.read()) != -1) { + out.write(b); + } + assertArrayEquals(PAYLOAD, out.toByteArray()); + } + + @Test + public void testReadWithBufferSmallerThanField() throws IOException { + InputStream is = new StreamMessageInputStream(streamMessage(PAYLOAD, 20)); + + ByteArrayOutputStream out = new ByteArrayOutputStream(); + byte[] buffer = new byte[3]; + int num; + while ((num = is.read(buffer)) != -1) { + out.write(buffer, 0, num); + } + assertArrayEquals(PAYLOAD, out.toByteArray()); + } + + @Test + public void testReadTextWithMultiByteCharactersAcrossFields() throws IOException { + String text = "Grüße aus Århus, 東京からこんにちは ".repeat(500); + InputStream is = new StreamMessageInputStream(streamMessage(text.getBytes(StandardCharsets.UTF_8), 1001)); + + StringWriter out = new StringWriter(); + try (Reader reader = new InputStreamReader(is, StandardCharsets.UTF_8)) { + reader.transferTo(out); + } + assertEquals(text, out.toString()); + } + + @Test + public void testResetReadsAgain() throws IOException { + InputStream is = new StreamMessageInputStream(streamMessage(PAYLOAD, 9)); + + assertArrayEquals(PAYLOAD, is.readAllBytes()); + is.reset(); + assertArrayEquals(PAYLOAD, is.readAllBytes()); + } + + /** + * A StreamMessage holding the payload as byte[] fields of the given size, as the producer writes it, which follows + * the readBytes contract of the JMS API. + */ + private static StreamMessage streamMessage(byte[] payload, int fieldSize) { + int count = (payload.length + fieldSize - 1) / fieldSize; + byte[][] fields = new byte[count][]; + for (int i = 0; i < count; i++) { + int from = i * fieldSize; + fields[i] = Arrays.copyOfRange(payload, from, Math.min(payload.length, from + fieldSize)); + } + // the current field, and the position in it (-1 when its reading has not started) + int[] state = { 0, -1 }; + return (StreamMessage) Proxy.newProxyInstance(StreamMessageInputStreamTest.class.getClassLoader(), + new Class<?>[] { StreamMessage.class }, (proxy, method, args) -> { + switch (method.getName()) { + case "readBytes": { + byte[] target = (byte[]) args[0]; + if (state[0] >= fields.length) { + throw new MessageEOFException("No more fields"); + } + byte[] field = fields[state[0]]; + if (state[1] == field.length) { + // the field was read completely with a full buffer: -1, then the next field + state[0]++; + state[1] = -1; + return -1; + } + int pos = Math.max(state[1], 0); + int num = Math.min(field.length - pos, target.length); + System.arraycopy(field, pos, target, 0, num); + state[1] = pos + num; + if (num < target.length) { + // fewer bytes than the buffer: the field is complete + state[0]++; + state[1] = -1; + } + return num; + } + case "readByte": + throw new MessageFormatException("A byte[] field cannot be read as a byte"); + case "reset": + state[0] = 0; + state[1] = -1; + return null; + default: + throw new UnsupportedOperationException(method.getName()); + } + }); + } +} diff --git a/components/camel-sjms2/src/generated/resources/META-INF/org/apache/camel/component/sjms2/sjms2.json b/components/camel-sjms2/src/generated/resources/META-INF/org/apache/camel/component/sjms2/sjms2.json index 33fd35b4d562..9c8b91beff0c 100644 --- a/components/camel-sjms2/src/generated/resources/META-INF/org/apache/camel/component/sjms2/sjms2.json +++ b/components/camel-sjms2/src/generated/resources/META-INF/org/apache/camel/component/sjms2/sjms2.json @@ -48,7 +48,7 @@ "CamelSjmsBatchSize": { "index": 4, "kind": "header", "displayName": "", "group": "batch", "label": "consumer,batch", "required": false, "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "description": "The size of the batch when using the batching consumer option.", "constantName": "org.apache.camel.component.sjms.SjmsConstants#SJMS_BATCH_SIZE_HEADER" } }, "bodyTypes": { - "consumerOut": { "carriers": [ "text", "bytes", "map", "pojo", "null" ], "byOption": [ { "option": "mapJmsMessage", "value": "false", "carriers": [ "pojo" ], "javaType": "jakarta.jms.Message" }, { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "map", "value", "list", "pojo", "null" ] } ], "description": "By the JMS message received: a String for a TextMessage, a byte array for a BytesMessage, a Map for a MapMessage, the jakarta.jms.StreamMessage its [...] + "consumerOut": { "carriers": [ "text", "bytes", "map", "stream", "null" ], "byOption": [ { "option": "mapJmsMessage", "value": "false", "carriers": [ "pojo" ], "javaType": "jakarta.jms.Message" }, { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "map", "stream", "value", "list", "pojo", "null" ] } ], "description": "By the JMS message received: a String for a TextMessage, a byte array for a BytesMessage, a Map for a MapMessage, an InputStream for a [...] "consumerReply": { "carriers": [ "text", "bytes", "stream", "file", "map", "null" ], "dependsOn": [ "jmsMessageType" ], "byOption": [ { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description": "The reply, sent as the producer sends a message. A reply that cannot be sent, such as a number or a List without objectMessageEnabled, fails the incoming message (rolled back when transacted [...] "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map", "null" ], "dependsOn": [ "jmsMessageType", "allowNullBody" ], "byOption": [ { "option": "objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description": "Text is sent as a TextMessage; bytes, a stream or a file as a BytesMessage; a Map of simple values as a MapMessage; null as an empty Message. A number, a List or another Serial [...] "producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option": "exchangePattern", "value": "InOnly", "carriers": [ "unchanged" ] }, { "option": "disableReplyTo", "value": "true", "carriers": [ "unchanged" ] } ], "description": "Unchanged for a message that expects no reply (InOnly). A message that expects one (InOut) gets the body of the reply, as the consumer gives it" } diff --git a/components/camel-sjms2/src/main/docs/sjms2-component.adoc b/components/camel-sjms2/src/main/docs/sjms2-component.adoc index 9fa42cf85235..7bc5b9e0f2c8 100644 --- a/components/camel-sjms2/src/main/docs/sjms2-component.adoc +++ b/components/camel-sjms2/src/main/docs/sjms2-component.adoc @@ -306,6 +306,7 @@ The body is mapped to and from a JMS message as follows (unless the option `jmsM | `String`, other text (a `CharSequence` such as a `StringBuilder`) or XML `Node` | `TextMessage` | `String` | `byte[]`, `InputStream`, `Reader`, `ByteBuffer`, `File`, `Path` or `StreamCache` | `BytesMessage` | `byte[]` | `Map` (of simple values) | `MapMessage` | `Map` +| Any body that converts to an `InputStream`, with `jmsMessageType=Stream` | `StreamMessage` | `InputStream` | `null` | `Message` | `null` | Other `Serializable` object (such as a number or a `List`) | `ObjectMessage` (only with `objectMessageEnabled=true`, fails otherwise) | the object | Any other object that converts to an `InputStream` (or `File`) | `BytesMessage` | `byte[]` @@ -313,7 +314,7 @@ The body is mapped to and from a JMS message as follows (unless the option `jmsM |=== With `jmsMessageType=Stream` the body is converted to an `InputStream`, and its bytes are written to a `StreamMessage`. -A received `StreamMessage` is not mapped: the body is the `jakarta.jms.StreamMessage` itself. +A received `StreamMessage` is mapped to an `InputStream` over its bytes, as in the JMS component. With `mapJmsMessage=false` the body is always the received `jakarta.jms.Message`. === Clustering diff --git a/components/camel-sjms2/src/main/java/org/apache/camel/component/sjms2/Sjms2Endpoint.java b/components/camel-sjms2/src/main/java/org/apache/camel/component/sjms2/Sjms2Endpoint.java index 4f2beb461ac4..7fa89581d607 100644 --- a/components/camel-sjms2/src/main/java/org/apache/camel/component/sjms2/Sjms2Endpoint.java +++ b/components/camel-sjms2/src/main/java/org/apache/camel/component/sjms2/Sjms2Endpoint.java @@ -36,17 +36,17 @@ import org.apache.camel.spi.UriParam; @UriEndpoint(firstVersion = "2.19.0", scheme = "sjms2", extendsScheme = "sjms", title = "Simple JMS2", syntax = "sjms2:destinationType:destinationName", category = { Category.MESSAGING }, headersClass = SjmsConstants.class) -@BodyTypes(consumerOut = @BodyType(value = { Carrier.TEXT, Carrier.BYTES, Carrier.MAP, Carrier.POJO, Carrier.NULL }, +@BodyTypes(consumerOut = @BodyType(value = { Carrier.TEXT, Carrier.BYTES, Carrier.MAP, Carrier.STREAM, Carrier.NULL }, byOption = { @BodyType.When(option = "mapJmsMessage", value = "false", carriers = Carrier.POJO, javaType = "jakarta.jms.Message"), @BodyType.When(option = "objectMessageEnabled", value = "true", carriers = { - Carrier.TEXT, Carrier.BYTES, Carrier.MAP, Carrier.VALUE, - Carrier.LIST, Carrier.POJO, Carrier.NULL }) }, + Carrier.TEXT, Carrier.BYTES, Carrier.MAP, Carrier.STREAM, + Carrier.VALUE, Carrier.LIST, Carrier.POJO, Carrier.NULL }) }, description = "By the JMS message received: a String for a TextMessage, a byte array for a" - + " BytesMessage, a Map for a MapMessage, the jakarta.jms.StreamMessage itself" - + " for a StreamMessage, and null for a plain Message. An ObjectMessage fails" + + " BytesMessage, a Map for a MapMessage, an InputStream for a StreamMessage," + + " and null for a plain Message. An ObjectMessage fails" + " unless objectMessageEnabled"), consumerReply = @BodyType(value = { Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.FILE, Carrier.MAP, diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 03577d5cf348..dbef3ee34b49 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -258,6 +258,20 @@ is skipped. Before, the source transaction was rolled back and the record replay the route of the send (for example with `direct`) behaves the same way, as it did in 4.23. To replay the record, throw the exception again in the `doCatch` block, or end it with `rollback()`. +=== camel-jms, camel-sjms, camel-sjms2 - an exception sent back with transferException is not rolled back + +With `transferException=true`, a consumer that sends the exception of a failed exchange back as the reply now always +treats the message as processed. Before, when the exception was a `RuntimeCamelException`, or had one as a cause, the +consumer also rethrew it: a transacted session was rolled back, so the request was redelivered and the reply was lost +with it. Now the outcome does not depend on the type of the exception. To have a failed request redelivered, do not +use `transferException` on that route. + +=== camel-sjms, camel-sjms2 - a received StreamMessage is an InputStream + +The camel-sjms and camel-sjms2 consumers now map a received `jakarta.jms.StreamMessage` to an `InputStream` over its +bytes, as the camel-jms consumer does, so a body sent with `jmsMessageType=Stream` arrives as the bytes that were sent. +Before, the body was the `jakarta.jms.StreamMessage` itself. To get the JMS message, use `mapJmsMessage=false`. + === Java *Breaking change:* Camel 4.24 requires *Java 21* as the minimum runtime version.
