You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2017/10/21 19:29:49 UTC
[camel] branch master updated (04160a1 -> 67fde94)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from 04160a1 CAMEL-11869: Removed the unused dependency to mockito from camel-jcache
new b61b8cc CAMEL-11931: camel-jms - Add better support for Stream JMS message type
new 3143225 CAMEL-11931: camel-jms - Add better support for Stream JMS message type
new d4ed964 CAMEL-11931: adding artemis example
new 38b78e9 Regen
new d2899e7 CAMEL-11931: adding artemis example and optimise camel-jms for Artemis in streaming mode for large messages.
new 67fde94 CAMEL-11931: Rename example
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../camel-amqp/src/main/docs/amqp-component.adoc | 6 +-
.../camel-jms/src/main/docs/jms-component.adoc | 6 +-
.../org/apache/camel/component/jms/JmsBinding.java | 115 +++++++++++++++++++--
.../apache/camel/component/jms/JmsComponent.java | 20 ++++
.../camel/component/jms/JmsConfiguration.java | 21 +++-
.../camel/component/jms/JmsMessageHelper.java | 33 ++++++
.../component/jms/StreamMessageInputStream.java | 90 ++++++++++++++++
...dersTest.java => JmsStreamMessageTypeTest.java} | 53 +++++-----
examples/README.adoc | 4 +-
.../camel-example-artemis-large-messages/README.md | 77 ++++++++++++++
.../pom.xml | 41 +++-----
.../src/main/resources/META-INF/LICENSE.txt | 0
.../src/main/resources/META-INF/NOTICE.txt | 0
.../resources/META-INF/spring/camel-context.xml | 41 ++++----
.../src/main/resources/log4j2.properties | 9 +-
examples/pom.xml | 1 +
.../springboot/AMQPComponentConfiguration.java | 18 ++++
.../jms/springboot/JmsComponentConfiguration.java | 36 +++++++
18 files changed, 481 insertions(+), 90 deletions(-)
create mode 100644 components/camel-jms/src/main/java/org/apache/camel/component/jms/StreamMessageInputStream.java
copy components/camel-jms/src/test/java/org/apache/camel/component/jms/{JmsAllowAdditionalHeadersTest.java => JmsStreamMessageTypeTest.java} (64%)
create mode 100644 examples/camel-example-artemis-large-messages/README.md
copy examples/{camel-example-jmx => camel-example-artemis-large-messages}/pom.xml (77%)
copy {tooling/maven/guice-maven-plugin => examples/camel-example-artemis-large-messages}/src/main/resources/META-INF/LICENSE.txt (100%)
copy {tooling/maven/guice-maven-plugin => examples/camel-example-artemis-large-messages}/src/main/resources/META-INF/NOTICE.txt (100%)
copy components/camel-apns/src/test/resources/org/apache/camel/component/apns/spring/SpringApnsConsumerTest-context.xml => examples/camel-example-artemis-large-messages/src/main/resources/META-INF/spring/camel-context.xml (52%)
copy {archetypes/camel-archetype-activemq/src/main/resources/archetype-resources => examples/camel-example-artemis-large-messages}/src/main/resources/log4j2.properties (90%)
--
To stop receiving notification emails like this one, please contact
['"commits@camel.apache.org" <co...@camel.apache.org>'].
[camel] 05/06: CAMEL-11931: adding artemis example and optimise
camel-jms for Artemis in streaming mode for large messages.
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit d2899e770734c7734244ac696973c3784a77c7eb
Author: Claus Ibsen <da...@apache.org>
AuthorDate: Sat Oct 21 21:02:09 2017 +0200
CAMEL-11931: adding artemis example and optimise camel-jms for Artemis in streaming mode for large messages.
---
.../org/apache/camel/component/jms/JmsBinding.java | 66 +++++++++++++++++++---
.../camel/component/jms/JmsMessageHelper.java | 33 +++++++++++
examples/camel-example-artemis-stream/README.md | 51 ++++++++++++++---
.../resources/META-INF/spring/camel-context.xml | 13 +++--
.../src/main/resources/log4j2.properties | 5 ++
5 files changed, 146 insertions(+), 22 deletions(-)
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
index 2684ea8..9ebe704 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
@@ -50,17 +50,21 @@ import org.apache.camel.NoTypeConversionAvailableException;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.StreamCache;
import org.apache.camel.WrappedFile;
+import org.apache.camel.converter.stream.CachedOutputStream;
import org.apache.camel.impl.DefaultExchangeHolder;
import org.apache.camel.spi.HeaderFilterStrategy;
import org.apache.camel.util.CamelContextHelper;
import org.apache.camel.util.EndpointHelper;
import org.apache.camel.util.ExchangeHelper;
import org.apache.camel.util.FileUtil;
+import org.apache.camel.util.IOHelper;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.apache.camel.component.jms.JmsConstants.JMS_X_GROUP_ID;
+import static org.apache.camel.component.jms.JmsMessageHelper.getSafeLongProperty;
+import static org.apache.camel.component.jms.JmsMessageHelper.isVendor;
import static org.apache.camel.component.jms.JmsMessageHelper.normalizeDestinationName;
import static org.apache.camel.component.jms.JmsMessageType.Bytes;
import static org.apache.camel.component.jms.JmsMessageType.Map;
@@ -154,11 +158,11 @@ public class JmsBinding {
return createMapFromMapMessage((MapMessage)message);
} else if (message instanceof BytesMessage) {
LOG.trace("Extracting body as a BytesMessage from JMS message: {}", message);
- return createByteArrayFromBytesMessage((BytesMessage)message);
+ return createByteArrayFromBytesMessage(exchange, (BytesMessage)message);
} else if (message instanceof StreamMessage) {
LOG.trace("Extracting body as a StreamMessage from JMS message: {}", message);
StreamMessage streamMessage = (StreamMessage)message;
- return createInputStreamFromStreamMessage(streamMessage);
+ return createInputStreamFromStreamMessage(exchange, streamMessage);
} else {
return null;
}
@@ -219,7 +223,11 @@ public class JmsBinding {
return map;
}
-
+
+ /**
+ * @deprecated not in use
+ */
+ @Deprecated
public Object getObjectProperty(Message jmsMessage, String name) throws JMSException {
// try a direct lookup first
Object answer = jmsMessage.getObjectProperty(name);
@@ -231,7 +239,26 @@ public class JmsBinding {
return answer;
}
- protected byte[] createByteArrayFromBytesMessage(BytesMessage message) throws JMSException {
+ protected Object createByteArrayFromBytesMessage(Exchange exchange, BytesMessage message) throws JMSException {
+ // ActiveMQ has special optimised mode for bytes message, so we should use streaming if possible
+ Long size = getSafeLongProperty(message, "_AMQ_LARGE_SIZE");
+ if (size != null && size > 0) {
+ LOG.trace("Optimised for Artemis: Reading from BytesMessage in streaming mode directly into CachedOutputStream payload");
+ CachedOutputStream cos = new CachedOutputStream(exchange, true);
+ // this will save the stream and wait until the entire message is written before continuing.
+ message.setObjectProperty("JMS_AMQ_SaveStream", cos);
+ try {
+ // and then lets get the input stream of this so we can read it
+ return cos.getInputStream();
+ } catch (IOException e) {
+ JMSException cause = new MessageFormatException(e.getMessage());
+ cause.initCause(e);
+ throw cause;
+ } finally {
+ IOHelper.close(cos);
+ }
+ }
+
if (message.getBodyLength() > Integer.MAX_VALUE) {
LOG.warn("Length of BytesMessage is too long: {}", message.getBodyLength());
return null;
@@ -241,7 +268,7 @@ public class JmsBinding {
return result;
}
- protected InputStream createInputStreamFromStreamMessage(StreamMessage message) {
+ protected Object createInputStreamFromStreamMessage(Exchange exchange, StreamMessage message) throws JMSException {
return new StreamMessageInputStream(message);
}
@@ -574,6 +601,15 @@ public class JmsBinding {
|| exchange.getContext().getTypeConverter().tryConvertTo(InputStream.class, body) != null) {
type = streamingEnabled ? Stream : Bytes;
}
+
+ if (type == Stream) {
+ boolean artemis = isVendor(session, "Artemis");
+ if (artemis) {
+ // if running ActiveMQ Artemis then it has optimised streaming mode using byte messages so enforce as bytes
+ type = Bytes;
+ }
+ }
+
return type;
}
@@ -596,8 +632,22 @@ public class JmsBinding {
case Bytes: {
BytesMessage message = session.createBytesMessage();
if (body != null) {
- byte[] payload = context.getTypeConverter().convertTo(byte[].class, exchange, body);
- message.writeBytes(payload);
+ try {
+ if (isVendor(session, "Artemis")) {
+ LOG.trace("Optimised for Artemis: Streaming payload in BytesMessage");
+ InputStream is = context.getTypeConverter().mandatoryConvertTo(InputStream.class, exchange, body);
+ message.setObjectProperty("JMS_AMQ_InputStream", is);
+ LOG.trace("Optimised for Artemis: Finished streaming payload in BytesMessage");
+ } else {
+ byte[] payload = context.getTypeConverter().mandatoryConvertTo(byte[].class, exchange, body);
+ message.writeBytes(payload);
+ }
+ } catch (NoTypeConversionAvailableException e) {
+ // cannot convert to inputstream then thrown an exception to avoid sending a null message
+ JMSException cause = new MessageFormatException(e.getMessage());
+ cause.initCause(e);
+ throw cause;
+ }
}
return message;
}
@@ -629,8 +679,8 @@ public class JmsBinding {
if (body != null) {
long size = 0;
try {
- LOG.trace("Writing payload in StreamMessage");
InputStream is = context.getTypeConverter().mandatoryConvertTo(InputStream.class, exchange, body);
+ LOG.trace("Writing payload in StreamMessage");
// assume streaming is bigger payload so use same buffer size as the file component
byte[] buffer = new byte[FileUtil.BUFFER_SIZE];
int len = 0;
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessageHelper.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessageHelper.java
index fe9ae21..f352e9e 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessageHelper.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsMessageHelper.java
@@ -24,6 +24,7 @@ import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
+import javax.jms.Session;
import org.apache.camel.Exchange;
import org.apache.camel.util.ExchangeHelper;
@@ -121,6 +122,38 @@ public final class JmsMessageHelper {
}
/**
+ * Gets a JMS property in a safe way
+ *
+ * @param jmsMessage the JMS message
+ * @param name name of the property to get
+ * @return the property value, or <tt>null</tt> if does not exists or failure to get the value
+ */
+ public static Long getSafeLongProperty(Message jmsMessage, String name) {
+ try {
+ return jmsMessage.getLongProperty(name);
+ } catch (Exception e) {
+ // ignore
+ }
+ return null;
+ }
+
+ /**
+ * Is the JMS session from a given vendor
+ *
+ * @param session the JMS session
+ * @param vendor the vendor, such as <tt>ActiveMQ</tt>, or <tt>Artemis</tt>
+ * @return <tt>true</tt> if from the vendor, <tt>false</tt> if not or not possible to determine
+ */
+ public static boolean isVendor(Session session, String vendor) {
+ if ("Artemis".equals(vendor)) {
+ return session.getClass().getName().startsWith("org.apache.activemq.artemis");
+ } else if ("ActiveMQ".equals(vendor)) {
+ return !isVendor(session, "Artemis") && session.getClass().getName().startsWith("org.apache.activemq");
+ }
+ return false;
+ }
+
+ /**
* Sets the property on the given JMS message.
*
* @param jmsMessage the JMS message
diff --git a/examples/camel-example-artemis-stream/README.md b/examples/camel-example-artemis-stream/README.md
index 608e931..1be77ea 100644
--- a/examples/camel-example-artemis-stream/README.md
+++ b/examples/camel-example-artemis-stream/README.md
@@ -2,8 +2,14 @@
### Introduction
-This example shows how to use Camel Aggregator EIP which offers (since Camel 2.3)
-database persistence.
+This example shows how to send large messages between Apache Camel and ActiveMQ Artemis.
+When we say large messages we refer to messages with sizes of GB.
+
+You should be able to run Camel and Artemis in JVMs with lower memory such as 256/512mb etc, and
+still be able to send messages in GB of sizes between them.
+
+This works by spool big messages to disk. Artemis spool large messages to its `data/large-messages`
+directory, and Camel uses stream caching to spool to a temporary directory during routing.
### Build
@@ -11,24 +17,53 @@ The example is run using Maven.
First compile the example by entering:
- mvn compile
+ mvn compile
### Install ActiveMQ Artemis
-TODO: How to install Artemis
+You download and unzip Apache ActiveMQ Artemis from: http://activemq.apache.org/artemis/download.html
+
+After unzipping the download, you can then create a new broker with the name `mybroker`:
+
+ $ cd apache-artemis-2.3.0
+ $ bin/artemis create mybroker
+
+### Run ActiveMQ Artemis
+
+You start ActiveMQ in a shell by running:
+
+ $ cd mybroker
+ $ bin/artemis run
-### Run
+Which startup Artemis in the foreground and keeps it running until you hit <kbd>ctrl</kbd>+<kbd>c</kbd>
+to shutdown Artemis.
-TODO: How to start Artemis
+### Run Camel
-To run the example type:
+Before running this example, then ensure the JVM has limited memory by executing
- mvn camel:run
+ export MAVEN_OPTS="-Xmx256m"
+
+And then start the Camel application:
+
+ mvn camel:run
+
+You can then copy files to `target/inbox` folder which is send to Artemis, and then
+back again to Camel and written to the `target/outbox` folder.
+
+This should work for small and big files such as files with sizes of GB.
+The JVM should not run out of memory.
To stop the example hit <kbd>ctrl</kbd>+<kbd>c</kbd>. If you restart it and resume
entering numbers you should see that it remembered previously entered values, as it
uses a persistent store.
+### ActiveMQ Artemis web console
+
+You can browse the Artemis web console: <http://localhost:8161/console>
+to see activity such as number of consumers and producers.
+You can also delete all messages from queues which is a handy operation.
+
### Forum, Help, etc
diff --git a/examples/camel-example-artemis-stream/src/main/resources/META-INF/spring/camel-context.xml b/examples/camel-example-artemis-stream/src/main/resources/META-INF/spring/camel-context.xml
index abc80eb..aa906da 100644
--- a/examples/camel-example-artemis-stream/src/main/resources/META-INF/spring/camel-context.xml
+++ b/examples/camel-example-artemis-stream/src/main/resources/META-INF/spring/camel-context.xml
@@ -17,23 +17,20 @@
limitations under the License.
-->
-<!-- START SNIPPET: e1 -->
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
-
+ <!-- setup ActiveMQ Artemis connection factory -->
<bean id="artemisConnectionFactory" class="org.apache.activemq.artemis.jms.client.ActiveMQJMSConnectionFactory">
<constructor-arg index="0" value="tcp://localhost:61616"/>
</bean>
<bean id="jms" class="org.apache.camel.component.jms.JmsComponent">
- <!-- specify connection factory -->
+ <!-- specify connection factory on JMS component -->
<property name="connectionFactory" ref="artemisConnectionFactory"/>
- <!-- turn on streaming message support -->
- <property name="streamMessageTypeEnabled" value="true"/>
</bean>
<camelContext id="myCamel" xmlns="http://camel.apache.org/schema/spring">
@@ -42,12 +39,16 @@
<from uri="file:target/inbox"/>
<log message="Sending file ${file:name} to Artemis"/>
<to uri="jms:queue:data"/>
+ <log message="Finish sending file to Artemis"/>
</route>
- <route>
+ <!-- turn on stream caching so we can stream big messages from Artemis
+ to spool disk to avoid reading into memory -->
+ <route streamCache="true">
<from uri="jms:queue:data"/>
<log message="Received data from Artemis"/>
<to uri="file:target/outbox"/>
+ <log message="Finish saving data from Artemis as file"/>
</route>
</camelContext>
diff --git a/examples/camel-example-artemis-stream/src/main/resources/log4j2.properties b/examples/camel-example-artemis-stream/src/main/resources/log4j2.properties
index d9f0508..56ee6f3 100644
--- a/examples/camel-example-artemis-stream/src/main/resources/log4j2.properties
+++ b/examples/camel-example-artemis-stream/src/main/resources/log4j2.properties
@@ -21,3 +21,8 @@ appender.out.layout.type = PatternLayout
appender.out.layout.pattern = [%30.30t] %-30.30c{1} %-5p %m%n
rootLogger.level = INFO
rootLogger.appenderRef.out.ref = out
+
+# turn on logging on camel-jms to see more activity
+# logger.jms.name = org.apache.camel.component.jms
+# logger.jms.level = TRACE
+
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.
[camel] 02/06: CAMEL-11931: camel-jms - Add better support for
Stream JMS message type
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 314322590adcf181348dcdb8ab8842fc289fbdf7
Author: Claus Ibsen <da...@apache.org>
AuthorDate: Fri Oct 20 16:48:29 2017 +0200
CAMEL-11931: camel-jms - Add better support for Stream JMS message type
---
.../camel-jms/src/main/docs/jms-component.adoc | 6 ++--
.../org/apache/camel/component/jms/JmsBinding.java | 16 ++++++----
.../apache/camel/component/jms/JmsComponent.java | 20 ++++++++++++
.../camel/component/jms/JmsConfiguration.java | 21 ++++++++++++-
.../component/jms/JmsStreamMessageTypeTest.java | 9 +++---
.../jms/springboot/JmsComponentConfiguration.java | 36 ++++++++++++++++++++++
6 files changed, 94 insertions(+), 14 deletions(-)
diff --git a/components/camel-jms/src/main/docs/jms-component.adoc b/components/camel-jms/src/main/docs/jms-component.adoc
index cf43483..351fdd0 100644
--- a/components/camel-jms/src/main/docs/jms-component.adoc
+++ b/components/camel-jms/src/main/docs/jms-component.adoc
@@ -199,7 +199,7 @@ about these properties by consulting the relevant Spring documentation.
// component options: START
-The JMS component supports 78 options which are listed below.
+The JMS component supports 79 options which are listed below.
@@ -282,6 +282,7 @@ The JMS component supports 78 options which are listed below.
| *subscriptionDurable* (consumer) | Set whether to make the subscription durable. The durable subscription name to be used can be specified through the subscriptionName property. Default is false. Set this to true to register a durable subscription typically in combination with a subscriptionName value (unless your message listener class name is good enough as subscription name). Only makes sense when listening to a topic (pub-sub domain) therefore this method switches the pubSubDomain [...]
| *subscriptionShared* (consumer) | Set whether to make the subscription shared. The shared subscription name to be used can be specified through the subscriptionName property. Default is false. Set this to true to register a shared subscription typically in combination with a subscriptionName value (unless your message listener class name is good enough as subscription name). Note that shared subscriptions may also be durable so this flag can (and often will) be combined with subscripti [...]
| *subscriptionName* (consumer) | Set the name of a subscription to create. To be applied in case of a topic (pub-sub domain) with a shared or durable subscription. The subscription name needs to be unique within this client's JMS client id. Default is the class name of the specified message listener. Note: Only 1 concurrent consumer (which is the default of this message listener container) is allowed for each subscription except for a shared subscription (which requires JMS 2.0). | | String
+| *streamMessageType Enabled* (producer) | Sets whether StreamMessage type is enabled or not. Message payloads of streaming kind such as files InputStream etc will either by sent as BytesMessage or StreamMessage. This option controls which kind will be used. By default BytesMessage is used which enforces the entire message payload to be read into memory. By enabling this option the message payload is read into memory in chunks and each chunk is then written to the StreamMessage until no [...]
| *headerFilterStrategy* (filter) | To use a custom org.apache.camel.spi.HeaderFilterStrategy to filter header to and from Camel message. | | HeaderFilterStrategy
| *resolveProperty Placeholders* (advanced) | Whether the component should resolve property placeholders on itself when starting. Only properties which are of String type can use property placeholders. | true | boolean
|===
@@ -322,7 +323,7 @@ with the following path and query parameters:
| *destinationName* | *Required* Name of the queue or topic to use as destination | | String
|===
-==== Query Parameters (89 parameters):
+==== Query Parameters (90 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|===
@@ -378,6 +379,7 @@ with the following path and query parameters:
| *includeSentJMSMessageID* (producer) | Only applicable when sending to JMS destination using InOnly (eg fire and forget). Enabling this option will enrich the Camel Exchange with the actual JMSMessageID that was used by the JMS client when the message was sent to the JMS destination. | false | boolean
| *replyToCacheLevelName* (producer) | Sets the cache level by name for the reply consumer when doing request/reply over JMS. This option only applies when using fixed reply queues (not temporary). Camel will by default use: CACHE_CONSUMER for exclusive or shared w/ replyToSelectorName. And CACHE_SESSION for shared without replyToSelectorName. Some JMS brokers such as IBM WebSphere may require to set the replyToCacheLevelName=CACHE_NONE to work. Note: If using temporary queues then CACHE [...]
| *replyToDestinationSelector Name* (producer) | Sets the JMS Selector using the fixed name to be used so you can filter out your own replies from the others when using a shared queue (that is if you are not using a temporary reply queue). | | String
+| *streamMessageTypeEnabled* (producer) | Sets whether StreamMessage type is enabled or not. Message payloads of streaming kind such as files InputStream etc will either by sent as BytesMessage or StreamMessage. This option controls which kind will be used. By default BytesMessage is used which enforces the entire message payload to be read into memory. By enabling this option the message payload is read into memory in chunks and each chunk is then written to the StreamMessage until no m [...]
| *allowSerializedHeaders* (advanced) | Controls whether or not to include serialized headers. Applies only when link isTransferExchange() is true. This requires that the objects are serializable. Camel will exclude any non-serializable objects and log it at WARN level. | false | boolean
| *asyncStartListener* (advanced) | Whether to startup the JmsConsumer message listener asynchronously when starting a route. For example if a JmsConsumer cannot get a connection to a remote JMS broker then it may block while retrying and/or failover. This will cause Camel to block while starting routes. By setting this option to true you will let routes startup while the JmsConsumer connects to the JMS broker using a dedicated thread in asynchronous mode. If this option is used then bew [...]
| *asyncStopListener* (advanced) | Whether to stop the JmsConsumer message listener asynchronously when stopping a route. | false | boolean
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
index 3a8439f..2684ea8 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
@@ -42,8 +42,6 @@ import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;
-import org.apache.camel.util.FileUtil;
-import org.apache.camel.util.IOHelper;
import org.w3c.dom.Node;
import org.apache.camel.CamelContext;
@@ -57,6 +55,7 @@ import org.apache.camel.spi.HeaderFilterStrategy;
import org.apache.camel.util.CamelContextHelper;
import org.apache.camel.util.EndpointHelper;
import org.apache.camel.util.ExchangeHelper;
+import org.apache.camel.util.FileUtil;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -66,6 +65,7 @@ import static org.apache.camel.component.jms.JmsMessageHelper.normalizeDestinati
import static org.apache.camel.component.jms.JmsMessageType.Bytes;
import static org.apache.camel.component.jms.JmsMessageType.Map;
import static org.apache.camel.component.jms.JmsMessageType.Object;
+import static org.apache.camel.component.jms.JmsMessageType.Stream;
import static org.apache.camel.component.jms.JmsMessageType.Text;
/**
@@ -293,7 +293,7 @@ public class JmsBinding {
answer = answer instanceof MapMessage ? answer : null;
} else if (type == JmsMessageType.Object) {
answer = answer instanceof ObjectMessage ? answer : null;
- } else if (type == JmsMessageType.Stream) {
+ } else if (type == Stream) {
answer = answer instanceof StreamMessage ? answer : null;
}
}
@@ -555,20 +555,24 @@ public class JmsBinding {
* @return type or null if no mapping was possible
*/
protected JmsMessageType getJMSMessageTypeForBody(Exchange exchange, Object body, Map<String, Object> headers, Session session, CamelContext context) {
+ boolean streamingEnabled = endpoint.getConfiguration().isStreamMessageTypeEnabled();
+
JmsMessageType type = null;
// let body determine the type
if (body instanceof Node || body instanceof String) {
type = Text;
- } else if (body instanceof byte[] || body instanceof WrappedFile || body instanceof File || body instanceof Reader
- || body instanceof InputStream || body instanceof ByteBuffer || body instanceof StreamCache) {
+ } else if (body instanceof byte[] || body instanceof ByteBuffer) {
type = Bytes;
+ } else if (body instanceof WrappedFile || body instanceof File || body instanceof Reader
+ || body instanceof InputStream || body instanceof StreamCache) {
+ type = streamingEnabled ? Stream : Bytes;
} else if (body instanceof Map) {
type = Map;
} else if (body instanceof Serializable) {
type = Object;
} else if (exchange.getContext().getTypeConverter().tryConvertTo(File.class, body) != null
|| exchange.getContext().getTypeConverter().tryConvertTo(InputStream.class, body) != null) {
- type = Bytes;
+ type = streamingEnabled ? Stream : Bytes;
}
return type;
}
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java
index 3e03966..134f8fa 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java
@@ -1213,6 +1213,26 @@ public class JmsComponent extends HeaderFilterStrategyComponent implements Appli
getConfiguration().setSubscriptionName(subscriptionName);
}
+
+ public boolean isStreamMessageTypeEnabled() {
+ return getConfiguration().isStreamMessageTypeEnabled();
+ }
+
+ /**
+ * Sets whether StreamMessage type is enabled or not.
+ * Message payloads of streaming kind such as files, InputStream, etc will either by sent as BytesMessage or StreamMessage.
+ * This option controls which kind will be used. By default BytesMessage is used which enforces the entire message payload to be read into memory.
+ * By enabling this option the message payload is read into memory in chunks and each chunk is then written to the StreamMessage until no more data.
+ */
+ @Metadata(label = "producer,advanced", description = "Sets whether StreamMessage type is enabled or not."
+ + " Message payloads of streaming kind such as files, InputStream, etc will either by sent as BytesMessage or StreamMessage."
+ + " This option controls which kind will be used. By default BytesMessage is used which enforces the entire message payload to be read into memory."
+ + " By enabling this option the message payload is read into memory in chunks and each chunk is then written to the StreamMessage until no more data.")
+ public void setStreamMessageTypeEnabled(boolean streamMessageTypeEnabled) {
+ getConfiguration().setStreamMessageTypeEnabled(streamMessageTypeEnabled);
+ }
+
+
// Implementation methods
// -------------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
index 27ac6b8..eb054ce 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
@@ -485,6 +485,12 @@ public class JmsConfiguration implements Cloneable {
+ " Requires a JMS 2.0 compatible message broker.")
private boolean subscriptionShared;
+ @UriParam(label = "producer,advanced", description = "Sets whether StreamMessage type is enabled or not."
+ + " Message payloads of streaming kind such as files, InputStream, etc will either by sent as BytesMessage or StreamMessage."
+ + " This option controls which kind will be used. By default BytesMessage is used which enforces the entire message payload to be read into memory."
+ + " By enabling this option the message payload is read into memory in chunks and each chunk is then written to the StreamMessage until no more data.")
+ private boolean streamMessageTypeEnabled;
+
public JmsConfiguration() {
}
@@ -2193,5 +2199,18 @@ public class JmsConfiguration implements Cloneable {
public void setSubscriptionName(String subscriptionName) {
this.subscriptionName = subscriptionName;
}
-
+
+ public boolean isStreamMessageTypeEnabled() {
+ return streamMessageTypeEnabled;
+ }
+
+ /**
+ * Sets whether StreamMessage type is enabled or not.
+ * Message payloads of streaming kind such as files, InputStream, etc will either by sent as BytesMessage or StreamMessage.
+ * This option controls which kind will be used. By default BytesMessage is used which enforces the entire message payload to be read into memory.
+ * By enabling this option the message payload is read into memory in chunks and each chunk is then written to the StreamMessage until no more data.
+ */
+ public void setStreamMessageTypeEnabled(boolean streamMessageTypeEnabled) {
+ this.streamMessageTypeEnabled = streamMessageTypeEnabled;
+ }
}
diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsStreamMessageTypeTest.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsStreamMessageTypeTest.java
index d9ee858..72d3d64 100644
--- a/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsStreamMessageTypeTest.java
+++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsStreamMessageTypeTest.java
@@ -28,9 +28,6 @@ import org.junit.Test;
import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge;
-/**
- * @version
- */
public class JmsStreamMessageTypeTest extends CamelTestSupport {
@Override
@@ -43,7 +40,9 @@ public class JmsStreamMessageTypeTest extends CamelTestSupport {
CamelContext camelContext = super.createCamelContext();
ConnectionFactory connectionFactory = CamelJmsTestHelper.createConnectionFactory();
- camelContext.addComponent("jms", jmsComponentAutoAcknowledge(connectionFactory));
+ JmsComponent jms = jmsComponentAutoAcknowledge(connectionFactory);
+ jms.setStreamMessageTypeEnabled(true); // turn on streaming
+ camelContext.addComponent("jms", jms);
return camelContext;
}
@@ -68,7 +67,7 @@ public class JmsStreamMessageTypeTest extends CamelTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("file:target/stream/in").to("jms:queue:foo?jmsMessageType=Stream");
+ from("file:target/stream/in").to("jms:queue:foo");
from("jms:queue:foo").to("file:target/stream/out").to("mock:result");
}
diff --git a/platforms/spring-boot/components-starter/camel-jms-starter/src/main/java/org/apache/camel/component/jms/springboot/JmsComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-jms-starter/src/main/java/org/apache/camel/component/jms/springboot/JmsComponentConfiguration.java
index 66f737b..4e76189 100644
--- a/platforms/spring-boot/components-starter/camel-jms-starter/src/main/java/org/apache/camel/component/jms/springboot/JmsComponentConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-jms-starter/src/main/java/org/apache/camel/component/jms/springboot/JmsComponentConfiguration.java
@@ -599,6 +599,16 @@ public class JmsComponentConfiguration
*/
private String subscriptionName;
/**
+ * Sets whether StreamMessage type is enabled or not. Message payloads of
+ * streaming kind such as files InputStream etc will either by sent as
+ * BytesMessage or StreamMessage. This option controls which kind will be
+ * used. By default BytesMessage is used which enforces the entire message
+ * payload to be read into memory. By enabling this option the message
+ * payload is read into memory in chunks and each chunk is then written to
+ * the StreamMessage until no more data.
+ */
+ private Boolean streamMessageTypeEnabled = false;
+ /**
* To use a custom org.apache.camel.spi.HeaderFilterStrategy to filter
* header to and from Camel message.
*/
@@ -1233,6 +1243,14 @@ public class JmsComponentConfiguration
this.subscriptionName = subscriptionName;
}
+ public Boolean getStreamMessageTypeEnabled() {
+ return streamMessageTypeEnabled;
+ }
+
+ public void setStreamMessageTypeEnabled(Boolean streamMessageTypeEnabled) {
+ this.streamMessageTypeEnabled = streamMessageTypeEnabled;
+ }
+
public HeaderFilterStrategy getHeaderFilterStrategy() {
return headerFilterStrategy;
}
@@ -1918,6 +1936,16 @@ public class JmsComponentConfiguration
* for a shared subscription (which requires JMS 2.0).
*/
private String subscriptionName;
+ /**
+ * Sets whether StreamMessage type is enabled or not. Message payloads
+ * of streaming kind such as files, InputStream, etc will either by sent
+ * as BytesMessage or StreamMessage. This option controls which kind
+ * will be used. By default BytesMessage is used which enforces the
+ * entire message payload to be read into memory. By enabling this
+ * option the message payload is read into memory in chunks and each
+ * chunk is then written to the StreamMessage until no more data.
+ */
+ private Boolean streamMessageTypeEnabled = false;
public ConsumerType getConsumerType() {
return consumerType;
@@ -2664,5 +2692,13 @@ public class JmsComponentConfiguration
public void setSubscriptionName(String subscriptionName) {
this.subscriptionName = subscriptionName;
}
+
+ public Boolean getStreamMessageTypeEnabled() {
+ return streamMessageTypeEnabled;
+ }
+
+ public void setStreamMessageTypeEnabled(Boolean streamMessageTypeEnabled) {
+ this.streamMessageTypeEnabled = streamMessageTypeEnabled;
+ }
}
}
\ No newline at end of file
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.
[camel] 04/06: Regen
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 38b78e96e21f05aff79e190a85426bd3f2122c42
Author: Claus Ibsen <da...@apache.org>
AuthorDate: Sat Oct 21 09:43:22 2017 +0200
Regen
---
.../camel-amqp/src/main/docs/amqp-component.adoc | 6 ++++--
.../amqp/springboot/AMQPComponentConfiguration.java | 18 ++++++++++++++++++
2 files changed, 22 insertions(+), 2 deletions(-)
diff --git a/components/camel-amqp/src/main/docs/amqp-component.adoc b/components/camel-amqp/src/main/docs/amqp-component.adoc
index be9bc8b..ae5566d 100644
--- a/components/camel-amqp/src/main/docs/amqp-component.adoc
+++ b/components/camel-amqp/src/main/docs/amqp-component.adoc
@@ -38,7 +38,7 @@ link:../../../../camel-jms/src/main/docs/readme.html[JMS] component after the de
// component options: START
-The AMQP component supports 78 options which are listed below.
+The AMQP component supports 79 options which are listed below.
@@ -121,6 +121,7 @@ The AMQP component supports 78 options which are listed below.
| *subscriptionDurable* (consumer) | Set whether to make the subscription durable. The durable subscription name to be used can be specified through the subscriptionName property. Default is false. Set this to true to register a durable subscription typically in combination with a subscriptionName value (unless your message listener class name is good enough as subscription name). Only makes sense when listening to a topic (pub-sub domain) therefore this method switches the pubSubDomain [...]
| *subscriptionShared* (consumer) | Set whether to make the subscription shared. The shared subscription name to be used can be specified through the subscriptionName property. Default is false. Set this to true to register a shared subscription typically in combination with a subscriptionName value (unless your message listener class name is good enough as subscription name). Note that shared subscriptions may also be durable so this flag can (and often will) be combined with subscripti [...]
| *subscriptionName* (consumer) | Set the name of a subscription to create. To be applied in case of a topic (pub-sub domain) with a shared or durable subscription. The subscription name needs to be unique within this client's JMS client id. Default is the class name of the specified message listener. Note: Only 1 concurrent consumer (which is the default of this message listener container) is allowed for each subscription except for a shared subscription (which requires JMS 2.0). | | String
+| *streamMessageType Enabled* (producer) | Sets whether StreamMessage type is enabled or not. Message payloads of streaming kind such as files InputStream etc will either by sent as BytesMessage or StreamMessage. This option controls which kind will be used. By default BytesMessage is used which enforces the entire message payload to be read into memory. By enabling this option the message payload is read into memory in chunks and each chunk is then written to the StreamMessage until no [...]
| *headerFilterStrategy* (filter) | To use a custom org.apache.camel.spi.HeaderFilterStrategy to filter header to and from Camel message. | | HeaderFilterStrategy
| *resolveProperty Placeholders* (advanced) | Whether the component should resolve property placeholders on itself when starting. Only properties which are of String type can use property placeholders. | true | boolean
|===
@@ -150,7 +151,7 @@ with the following path and query parameters:
| *destinationName* | *Required* Name of the queue or topic to use as destination | | String
|===
-==== Query Parameters (89 parameters):
+==== Query Parameters (90 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|===
@@ -206,6 +207,7 @@ with the following path and query parameters:
| *includeSentJMSMessageID* (producer) | Only applicable when sending to JMS destination using InOnly (eg fire and forget). Enabling this option will enrich the Camel Exchange with the actual JMSMessageID that was used by the JMS client when the message was sent to the JMS destination. | false | boolean
| *replyToCacheLevelName* (producer) | Sets the cache level by name for the reply consumer when doing request/reply over JMS. This option only applies when using fixed reply queues (not temporary). Camel will by default use: CACHE_CONSUMER for exclusive or shared w/ replyToSelectorName. And CACHE_SESSION for shared without replyToSelectorName. Some JMS brokers such as IBM WebSphere may require to set the replyToCacheLevelName=CACHE_NONE to work. Note: If using temporary queues then CACHE [...]
| *replyToDestinationSelector Name* (producer) | Sets the JMS Selector using the fixed name to be used so you can filter out your own replies from the others when using a shared queue (that is if you are not using a temporary reply queue). | | String
+| *streamMessageTypeEnabled* (producer) | Sets whether StreamMessage type is enabled or not. Message payloads of streaming kind such as files InputStream etc will either by sent as BytesMessage or StreamMessage. This option controls which kind will be used. By default BytesMessage is used which enforces the entire message payload to be read into memory. By enabling this option the message payload is read into memory in chunks and each chunk is then written to the StreamMessage until no m [...]
| *allowSerializedHeaders* (advanced) | Controls whether or not to include serialized headers. Applies only when transferExchange is true. This requires that the objects are serializable. Camel will exclude any non-serializable objects and log it at WARN level. | false | boolean
| *asyncStartListener* (advanced) | Whether to startup the JmsConsumer message listener asynchronously when starting a route. For example if a JmsConsumer cannot get a connection to a remote JMS broker then it may block while retrying and/or failover. This will cause Camel to block while starting routes. By setting this option to true you will let routes startup while the JmsConsumer connects to the JMS broker using a dedicated thread in asynchronous mode. If this option is used then bew [...]
| *asyncStopListener* (advanced) | Whether to stop the JmsConsumer message listener asynchronously when stopping a route. | false | boolean
diff --git a/platforms/spring-boot/components-starter/camel-amqp-starter/src/main/java/org/apache/camel/component/amqp/springboot/AMQPComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-amqp-starter/src/main/java/org/apache/camel/component/amqp/springboot/AMQPComponentConfiguration.java
index 5c85fc1..179b37f 100644
--- a/platforms/spring-boot/components-starter/camel-amqp-starter/src/main/java/org/apache/camel/component/amqp/springboot/AMQPComponentConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-amqp-starter/src/main/java/org/apache/camel/component/amqp/springboot/AMQPComponentConfiguration.java
@@ -597,6 +597,16 @@ public class AMQPComponentConfiguration
*/
private String subscriptionName;
/**
+ * Sets whether StreamMessage type is enabled or not. Message payloads of
+ * streaming kind such as files InputStream etc will either by sent as
+ * BytesMessage or StreamMessage. This option controls which kind will be
+ * used. By default BytesMessage is used which enforces the entire message
+ * payload to be read into memory. By enabling this option the message
+ * payload is read into memory in chunks and each chunk is then written to
+ * the StreamMessage until no more data.
+ */
+ private Boolean streamMessageTypeEnabled = false;
+ /**
* To use a custom org.apache.camel.spi.HeaderFilterStrategy to filter
* header to and from Camel message.
*/
@@ -1230,6 +1240,14 @@ public class AMQPComponentConfiguration
this.subscriptionName = subscriptionName;
}
+ public Boolean getStreamMessageTypeEnabled() {
+ return streamMessageTypeEnabled;
+ }
+
+ public void setStreamMessageTypeEnabled(Boolean streamMessageTypeEnabled) {
+ this.streamMessageTypeEnabled = streamMessageTypeEnabled;
+ }
+
public HeaderFilterStrategy getHeaderFilterStrategy() {
return headerFilterStrategy;
}
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.
[camel] 06/06: CAMEL-11931: Rename example
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 67fde946ab3d126f7db99f61eef007471587f7ad
Author: Claus Ibsen <da...@apache.org>
AuthorDate: Sat Oct 21 21:07:00 2017 +0200
CAMEL-11931: Rename example
---
examples/README.adoc | 4 +++-
.../README.md | 2 +-
.../pom.xml | 8 ++++----
.../src/main/resources/META-INF/LICENSE.txt | 0
.../src/main/resources/META-INF/NOTICE.txt | 0
.../src/main/resources/META-INF/spring/camel-context.xml | 0
.../src/main/resources/log4j2.properties | 0
examples/pom.xml | 2 +-
8 files changed, 9 insertions(+), 7 deletions(-)
diff --git a/examples/README.adoc b/examples/README.adoc
index fa1c537..6d281ee 100644
--- a/examples/README.adoc
+++ b/examples/README.adoc
@@ -11,7 +11,7 @@ View the individual example READMEs for details.
### Examples
// examples: START
-Number of Examples: 98 (8 deprecated)
+Number of Examples: 99 (8 deprecated)
[width="100%",cols="4,2,4",options="header"]
|===
@@ -139,6 +139,8 @@ Number of Examples: 98 (8 deprecated)
| link:camel-example-activemq-tomcat/README.md[ActiveMQ Tomcat] (camel-example-activemq-tomcat) | Messaging | An example using ActiveMQ Broker and Camel with Apache Tomcat
+| link:camel-example-artemis-large-messages/README.md[Artemis Large Messages] (camel-example-artemis-large-messages) | Messaging | Demonstrates sending large messages (handles GBs in size) between Apache Camel and ActiveMQ Artemis in streaming mode
+
| link:camel-example-google-pubsub/README.adoc[Google Pubsub] (camel-example-google-pubsub) | Messaging | An example for Google Pubsub
| link:camel-example-guice-jms/README.md[Guice JMS] (camel-example-guice-jms) | Messaging | *deprecated* An example showing how to work with Camel, Guice and JMS
diff --git a/examples/camel-example-artemis-stream/README.md b/examples/camel-example-artemis-large-messages/README.md
similarity index 98%
rename from examples/camel-example-artemis-stream/README.md
rename to examples/camel-example-artemis-large-messages/README.md
index 1be77ea..87a68d2 100644
--- a/examples/camel-example-artemis-stream/README.md
+++ b/examples/camel-example-artemis-large-messages/README.md
@@ -1,4 +1,4 @@
-# Camel Artemis Stream
+# Camel Artemis Large Messages
### Introduction
diff --git a/examples/camel-example-artemis-stream/pom.xml b/examples/camel-example-artemis-large-messages/pom.xml
similarity index 89%
rename from examples/camel-example-artemis-stream/pom.xml
rename to examples/camel-example-artemis-large-messages/pom.xml
index cb4de8c..3a95288 100644
--- a/examples/camel-example-artemis-stream/pom.xml
+++ b/examples/camel-example-artemis-large-messages/pom.xml
@@ -27,10 +27,10 @@
<version>2.21.0-SNAPSHOT</version>
</parent>
- <artifactId>camel-example-artemis-stream</artifactId>
+ <artifactId>camel-example-artemis-large-messages</artifactId>
<packaging>jar</packaging>
- <name>Camel :: Example :: ActiveMQ Artemis Stream messaging</name>
- <description>Demonstrates sending Stream messages to ActiveMQ Artemis</description>
+ <name>Camel :: Example :: ActiveMQ Artemis Large Messages</name>
+ <description>Demonstrates sending large messages (handles GBs in size) between Apache Camel and ActiveMQ Artemis in streaming mode</description>
<properties>
<category>Messaging</category>
@@ -48,7 +48,7 @@
<artifactId>camel-jms</artifactId>
</dependency>
- <!-- artemis -->
+ <!-- activemq artemis -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-spring-integration</artifactId>
diff --git a/examples/camel-example-artemis-stream/src/main/resources/META-INF/LICENSE.txt b/examples/camel-example-artemis-large-messages/src/main/resources/META-INF/LICENSE.txt
similarity index 100%
rename from examples/camel-example-artemis-stream/src/main/resources/META-INF/LICENSE.txt
rename to examples/camel-example-artemis-large-messages/src/main/resources/META-INF/LICENSE.txt
diff --git a/examples/camel-example-artemis-stream/src/main/resources/META-INF/NOTICE.txt b/examples/camel-example-artemis-large-messages/src/main/resources/META-INF/NOTICE.txt
similarity index 100%
rename from examples/camel-example-artemis-stream/src/main/resources/META-INF/NOTICE.txt
rename to examples/camel-example-artemis-large-messages/src/main/resources/META-INF/NOTICE.txt
diff --git a/examples/camel-example-artemis-stream/src/main/resources/META-INF/spring/camel-context.xml b/examples/camel-example-artemis-large-messages/src/main/resources/META-INF/spring/camel-context.xml
similarity index 100%
rename from examples/camel-example-artemis-stream/src/main/resources/META-INF/spring/camel-context.xml
rename to examples/camel-example-artemis-large-messages/src/main/resources/META-INF/spring/camel-context.xml
diff --git a/examples/camel-example-artemis-stream/src/main/resources/log4j2.properties b/examples/camel-example-artemis-large-messages/src/main/resources/log4j2.properties
similarity index 100%
rename from examples/camel-example-artemis-stream/src/main/resources/log4j2.properties
rename to examples/camel-example-artemis-large-messages/src/main/resources/log4j2.properties
diff --git a/examples/pom.xml b/examples/pom.xml
index a2441d8..3b406c4 100644
--- a/examples/pom.xml
+++ b/examples/pom.xml
@@ -37,7 +37,7 @@
<modules>
<module>camel-example-activemq-tomcat</module>
<module>camel-example-aggregate</module>
- <module>camel-example-artemis-stream</module>
+ <module>camel-example-artemis-large-messages</module>
<module>camel-example-bam</module>
<module>camel-example-cafe</module>
<module>camel-example-cassandra-kubernetes</module>
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.
[camel] 03/06: CAMEL-11931: adding artemis example
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit d4ed96467983b676a5c152bc4f95561ee802f07e
Author: Claus Ibsen <da...@apache.org>
AuthorDate: Sat Oct 21 09:05:30 2017 +0200
CAMEL-11931: adding artemis example
---
examples/camel-example-artemis-stream/README.md | 42 +++++
examples/camel-example-artemis-stream/pom.xml | 90 +++++++++
.../src/main/resources/META-INF/LICENSE.txt | 203 +++++++++++++++++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 ++
.../resources/META-INF/spring/camel-context.xml | 55 ++++++
.../src/main/resources/log4j2.properties | 23 +++
examples/pom.xml | 1 +
7 files changed, 425 insertions(+)
diff --git a/examples/camel-example-artemis-stream/README.md b/examples/camel-example-artemis-stream/README.md
new file mode 100644
index 0000000..608e931
--- /dev/null
+++ b/examples/camel-example-artemis-stream/README.md
@@ -0,0 +1,42 @@
+# Camel Artemis Stream
+
+### Introduction
+
+This example shows how to use Camel Aggregator EIP which offers (since Camel 2.3)
+database persistence.
+
+### Build
+
+The example is run using Maven.
+
+First compile the example by entering:
+
+ mvn compile
+
+### Install ActiveMQ Artemis
+
+TODO: How to install Artemis
+
+### Run
+
+TODO: How to start Artemis
+
+To run the example type:
+
+ mvn camel:run
+
+To stop the example hit <kbd>ctrl</kbd>+<kbd>c</kbd>. If you restart it and resume
+entering numbers you should see that it remembered previously entered values, as it
+uses a persistent store.
+
+
+### Forum, Help, etc
+
+If you hit an problems please let us know on the Camel Forums
+ <http://camel.apache.org/discussion-forums.html>
+
+Please help us make Apache Camel better - we appreciate any feedback you may
+have. Enjoy!
+
+
+The Camel riders!
diff --git a/examples/camel-example-artemis-stream/pom.xml b/examples/camel-example-artemis-stream/pom.xml
new file mode 100644
index 0000000..cb4de8c
--- /dev/null
+++ b/examples/camel-example-artemis-stream/pom.xml
@@ -0,0 +1,90 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.camel.example</groupId>
+ <artifactId>examples</artifactId>
+ <version>2.21.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-example-artemis-stream</artifactId>
+ <packaging>jar</packaging>
+ <name>Camel :: Example :: ActiveMQ Artemis Stream messaging</name>
+ <description>Demonstrates sending Stream messages to ActiveMQ Artemis</description>
+
+ <properties>
+ <category>Messaging</category>
+ </properties>
+
+ <dependencies>
+
+ <!-- camel -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-jms</artifactId>
+ </dependency>
+
+ <!-- artemis -->
+ <dependency>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>artemis-spring-integration</artifactId>
+ <version>${activemq-artemis-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.geronimo.specs</groupId>
+ <artifactId>geronimo-jms_2.0_spec</artifactId>
+ </dependency>
+
+ <!-- logging -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ </dependency>
+
+ </dependencies>
+
+
+ <build>
+ <plugins>
+ <!-- Allows the routes to be run via 'mvn camel:run' -->
+ <plugin>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-maven-plugin</artifactId>
+ <version>${project.version}</version>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
diff --git a/examples/camel-example-artemis-stream/src/main/resources/META-INF/LICENSE.txt b/examples/camel-example-artemis-stream/src/main/resources/META-INF/LICENSE.txt
new file mode 100644
index 0000000..6b0b127
--- /dev/null
+++ b/examples/camel-example-artemis-stream/src/main/resources/META-INF/LICENSE.txt
@@ -0,0 +1,203 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed 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.
+
diff --git a/examples/camel-example-artemis-stream/src/main/resources/META-INF/NOTICE.txt b/examples/camel-example-artemis-stream/src/main/resources/META-INF/NOTICE.txt
new file mode 100644
index 0000000..2e215bf
--- /dev/null
+++ b/examples/camel-example-artemis-stream/src/main/resources/META-INF/NOTICE.txt
@@ -0,0 +1,11 @@
+ =========================================================================
+ == NOTICE file corresponding to the section 4 d of ==
+ == the Apache License, Version 2.0, ==
+ == in this case for the Apache Camel distribution. ==
+ =========================================================================
+
+ This product includes software developed by
+ The Apache Software Foundation (http://www.apache.org/).
+
+ Please read the different LICENSE files present in the licenses directory of
+ this distribution.
diff --git a/examples/camel-example-artemis-stream/src/main/resources/META-INF/spring/camel-context.xml b/examples/camel-example-artemis-stream/src/main/resources/META-INF/spring/camel-context.xml
new file mode 100644
index 0000000..abc80eb
--- /dev/null
+++ b/examples/camel-example-artemis-stream/src/main/resources/META-INF/spring/camel-context.xml
@@ -0,0 +1,55 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<!-- START SNIPPET: e1 -->
+<beans xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="
+ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+ http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
+
+
+ <bean id="artemisConnectionFactory" class="org.apache.activemq.artemis.jms.client.ActiveMQJMSConnectionFactory">
+ <constructor-arg index="0" value="tcp://localhost:61616"/>
+ </bean>
+
+ <bean id="jms" class="org.apache.camel.component.jms.JmsComponent">
+ <!-- specify connection factory -->
+ <property name="connectionFactory" ref="artemisConnectionFactory"/>
+ <!-- turn on streaming message support -->
+ <property name="streamMessageTypeEnabled" value="true"/>
+ </bean>
+
+ <camelContext id="myCamel" xmlns="http://camel.apache.org/schema/spring">
+
+ <route>
+ <from uri="file:target/inbox"/>
+ <log message="Sending file ${file:name} to Artemis"/>
+ <to uri="jms:queue:data"/>
+ </route>
+
+ <route>
+ <from uri="jms:queue:data"/>
+ <log message="Received data from Artemis"/>
+ <to uri="file:target/outbox"/>
+ </route>
+
+ </camelContext>
+
+</beans>
diff --git a/examples/camel-example-artemis-stream/src/main/resources/log4j2.properties b/examples/camel-example-artemis-stream/src/main/resources/log4j2.properties
new file mode 100644
index 0000000..d9f0508
--- /dev/null
+++ b/examples/camel-example-artemis-stream/src/main/resources/log4j2.properties
@@ -0,0 +1,23 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+
+appender.out.type = Console
+appender.out.name = out
+appender.out.layout.type = PatternLayout
+appender.out.layout.pattern = [%30.30t] %-30.30c{1} %-5p %m%n
+rootLogger.level = INFO
+rootLogger.appenderRef.out.ref = out
diff --git a/examples/pom.xml b/examples/pom.xml
index 717f593..a2441d8 100644
--- a/examples/pom.xml
+++ b/examples/pom.xml
@@ -37,6 +37,7 @@
<modules>
<module>camel-example-activemq-tomcat</module>
<module>camel-example-aggregate</module>
+ <module>camel-example-artemis-stream</module>
<module>camel-example-bam</module>
<module>camel-example-cafe</module>
<module>camel-example-cassandra-kubernetes</module>
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.
[camel] 01/06: CAMEL-11931: camel-jms - Add better support for
Stream JMS message type
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit b61b8cc56718d4774eedb57b8cc23363c386f57f
Author: Claus Ibsen <da...@apache.org>
AuthorDate: Fri Oct 20 14:15:18 2017 +0200
CAMEL-11931: camel-jms - Add better support for Stream JMS message type
---
.../org/apache/camel/component/jms/JmsBinding.java | 43 ++++++++++-
.../component/jms/StreamMessageInputStream.java | 90 ++++++++++++++++++++++
.../component/jms/JmsStreamMessageTypeTest.java | 78 +++++++++++++++++++
3 files changed, 209 insertions(+), 2 deletions(-)
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
index 6b2a9a2..3a8439f 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.jms;
import java.io.File;
+import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
import java.io.Serializable;
@@ -41,6 +42,8 @@ import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;
+import org.apache.camel.util.FileUtil;
+import org.apache.camel.util.IOHelper;
import org.w3c.dom.Node;
import org.apache.camel.CamelContext;
@@ -154,7 +157,8 @@ public class JmsBinding {
return createByteArrayFromBytesMessage((BytesMessage)message);
} else if (message instanceof StreamMessage) {
LOG.trace("Extracting body as a StreamMessage from JMS message: {}", message);
- return message;
+ StreamMessage streamMessage = (StreamMessage)message;
+ return createInputStreamFromStreamMessage(streamMessage);
} else {
return null;
}
@@ -237,6 +241,10 @@ public class JmsBinding {
return result;
}
+ protected InputStream createInputStreamFromStreamMessage(StreamMessage message) {
+ return new StreamMessageInputStream(message);
+ }
+
/**
* Creates a JMS message from the Camel exchange and message
*
@@ -597,7 +605,7 @@ public class JmsBinding {
}
return message;
}
- case Object:
+ case Object: {
ObjectMessage message = session.createObjectMessage();
if (body != null) {
try {
@@ -611,6 +619,37 @@ public class JmsBinding {
}
}
return message;
+ }
+ case Stream: {
+ StreamMessage message = session.createStreamMessage();
+ if (body != null) {
+ long size = 0;
+ try {
+ LOG.trace("Writing payload in StreamMessage");
+ InputStream is = context.getTypeConverter().mandatoryConvertTo(InputStream.class, exchange, body);
+ // assume streaming is bigger payload so use same buffer size as the file component
+ byte[] buffer = new byte[FileUtil.BUFFER_SIZE];
+ int len = 0;
+ int count = 0;
+ while (len >= 0) {
+ count++;
+ len = is.read(buffer);
+ if (len >= 0) {
+ size += len;
+ LOG.trace("Writing payload chunk {} as bytes in StreamMessage", count);
+ message.writeBytes(buffer, 0, len);
+ }
+ }
+ LOG.trace("Finished writing payload (size {}) as bytes in StreamMessage", size);
+ } catch (NoTypeConversionAvailableException | IOException e) {
+ // cannot convert to inputstream then thrown an exception to avoid sending a null message
+ JMSException cause = new MessageFormatException(e.getMessage());
+ cause.initCause(e);
+ throw cause;
+ }
+ }
+ return message;
+ }
default:
break;
}
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
new file mode 100644
index 0000000..0d75ec7
--- /dev/null
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/StreamMessageInputStream.java
@@ -0,0 +1,90 @@
+/**
+ * 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.IOException;
+import java.io.InputStream;
+import javax.jms.JMSException;
+import javax.jms.MessageEOFException;
+import javax.jms.StreamMessage;
+
+public class StreamMessageInputStream extends InputStream {
+
+ private final StreamMessage message;
+ private volatile boolean eof;
+
+ public StreamMessageInputStream(StreamMessage message) {
+ this.message = message;
+ }
+
+ @Override
+ public int read() throws IOException {
+ byte[] array = new byte[1];
+ try {
+ return message.readBytes(array);
+ } catch (MessageEOFException e) {
+ eof = true;
+ return -1;
+ } catch (JMSException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public int read(byte[] array) throws IOException {
+ try {
+ int num = message.readBytes(array);
+ eof = num < 0;
+ return num;
+ } catch (MessageEOFException e) {
+ eof = true;
+ return -1;
+ } catch (JMSException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public int read(byte[] array, int off, int len) throws IOException {
+ // we cannot honor off and len, but assuming off is always 0
+ try {
+ int num = message.readBytes(array);
+ eof = num < 0;
+ return num;
+ } catch (MessageEOFException e) {
+ eof = true;
+ return -1;
+ } catch (JMSException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public synchronized void reset() throws IOException {
+ try {
+ message.reset();
+ } catch (JMSException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public int available() throws IOException {
+ // if we are end of file then there is no more data, otherwise assume there is at least one more byte
+ return eof ? 0 : 1;
+ }
+}
diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsStreamMessageTypeTest.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsStreamMessageTypeTest.java
new file mode 100644
index 0000000..d9ee858
--- /dev/null
+++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsStreamMessageTypeTest.java
@@ -0,0 +1,78 @@
+/**
+ * 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.File;
+import java.io.InputStream;
+import javax.jms.ConnectionFactory;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.apache.camel.util.FileUtil;
+import org.junit.Test;
+
+import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge;
+
+/**
+ * @version
+ */
+public class JmsStreamMessageTypeTest extends CamelTestSupport {
+
+ @Override
+ public void setUp() throws Exception {
+ deleteDirectory("target/stream");
+ super.setUp();
+ }
+
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+
+ ConnectionFactory connectionFactory = CamelJmsTestHelper.createConnectionFactory();
+ camelContext.addComponent("jms", jmsComponentAutoAcknowledge(connectionFactory));
+ return camelContext;
+ }
+
+ @Test
+ public void testStreamType() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+
+ // copy the file
+ FileUtil.copyFile(new File("src/test/data/message1.xml"), new File("target/stream/in/message1.xml"));
+
+ assertMockEndpointsSatisfied();
+
+ InputStream is = getMockEndpoint("mock:result").getReceivedExchanges().get(0).getIn().getBody(InputStream.class);
+ assertNotNull(is);
+ String xml = context.getTypeConverter().convertTo(String.class, is);
+
+ System.out.println(xml);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("file:target/stream/in").to("jms:queue:foo?jmsMessageType=Stream");
+
+ from("jms:queue:foo").to("file:target/stream/out").to("mock:result");
+ }
+ };
+ }
+
+}
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.