Author: davsclaus
Date: Mon Aug 31 11:42:10 2009
New Revision: 809546
URL: http://svn.apache.org/viewvc?rev=809546&view=rev
Log:
CAMEL-1940: Fixed pollEnrich from a JMS queue. Also fixed some other failing
unit test due class renames.
Added:
camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/jms/JmsPollEnrichTest.java
(with props)
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/SortDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/PollEnricher.java
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsPollingConsumerTest.java
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/management/jmxInstrumentationWithConnector.xml
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/SortDefinition.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/SortDefinition.java?rev=809546&r1=809545&r2=809546&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/model/SortDefinition.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/model/SortDefinition.java
Mon Aug 31 11:42:10 2009
@@ -90,6 +90,10 @@
}
Expression exp = expression.getExpression();
+ // fallback to the type when its been initialized from JAXB using
camel-spring
+ if (exp == null) {
+ exp = expression.getExpressionType();
+ }
return new SortProcessor(exp, getComparator());
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/PollEnricher.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/PollEnricher.java?rev=809546&r1=809545&r2=809546&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/PollEnricher.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/PollEnricher.java
Mon Aug 31 11:42:10 2009
@@ -24,9 +24,9 @@
import org.apache.camel.impl.ServiceSupport;
import org.apache.camel.processor.aggregate.AggregationStrategy;
import org.apache.camel.util.ExchangeHelper;
+import static org.apache.camel.util.ExchangeHelper.copyResultsPreservePattern;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import static org.apache.camel.util.ExchangeHelper.copyResultsPreservePattern;
/**
* A content enricher that enriches input data by first obtaining additional
@@ -131,6 +131,14 @@
resourceExchange = consumer.receive(timeout);
}
+ if (LOG.isDebugEnabled()) {
+ if (resourceExchange == null) {
+ LOG.debug("Consumer received no exchange");
+ } else {
+ LOG.debug("Consumer received: " + resourceExchange);
+ }
+ }
+
if (resourceExchange != null && resourceExchange.isFailed()) {
// copy resource exchange onto original exchange (preserving
pattern)
copyResultsPreservePattern(exchange, resourceExchange);
@@ -147,8 +155,10 @@
// prepare the exchanges for aggregation
ExchangeHelper.prepareAggregation(exchange, resourceExchange);
Exchange aggregatedExchange =
aggregationStrategy.aggregate(exchange, resourceExchange);
- // copy aggregation result onto original exchange (preserving
pattern)
- copyResultsPreservePattern(exchange, aggregatedExchange);
+ if (aggregatedExchange != null) {
+ // copy aggregation result onto original exchange
(preserving pattern)
+ copyResultsPreservePattern(exchange, aggregatedExchange);
+ }
} else {
if (LOG.isTraceEnabled()) {
LOG.trace("Cannot aggregate exchange as its filtered: " +
resourceExchange);
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java?rev=809546&r1=809545&r2=809546&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
Mon Aug 31 11:42:10 2009
@@ -258,7 +258,7 @@
// no need to copy
return;
}
-
+
// copy in message
result.getIn().copyFrom(source.getIn());
Modified:
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java?rev=809546&r1=809545&r2=809546&view=diff
==============================================================================
---
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
(original)
+++
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
Mon Aug 31 11:42:10 2009
@@ -42,20 +42,21 @@
}
public Exchange receiveNoWait() {
- // spring have changed the sematic of the receive timeout mode
- // so we need to deterime if running spring 2.0.x or 2.5.x or newer
+ // spring have changed the semantic of the receive timeout mode
+ // so we need to determine if running spring 2.0.x or 2.5.x or newer
if (spring20x) {
// spring 2.0.x
return receive(0L);
} else {
// spring 2.5.x
- return receive(-1L);
+ // no wait using -1L does not work properly so wait at most 1
millis to simulate no wait
+ return receive(1);
}
}
public Exchange receive() {
- // spring have changed the sematic of the receive timeout mode
- // so we need to deterime if running spring 2.0.x or 2.5.x or newer
+ // spring have changed the semantic of the receive timeout mode
+ // so we need to determine if running spring 2.0.x or 2.5.x or newer
if (spring20x) {
// spring 2.0.x
return receive(-1L);
Modified:
camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsPollingConsumerTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsPollingConsumerTest.java?rev=809546&r1=809545&r2=809546&view=diff
==============================================================================
---
camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsPollingConsumerTest.java
(original)
+++
camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsPollingConsumerTest.java
Mon Aug 31 11:42:10 2009
@@ -33,12 +33,12 @@
public class JmsPollingConsumerTest extends CamelTestSupport {
@Test
- public void testJmsPollingConsumer() throws Exception {
+ public void testJmsPollingConsumerWait() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result");
mock.expectedBodiesReceived("Hello Claus");
// use another thread for polling consumer to demonstrate that we can
wait before
- // the message is sent to the quueue
+ // the message is sent to the queue
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
String body = consumer.receiveBody("activemq:queue.start",
String.class);
@@ -47,7 +47,77 @@
});
// wait a little to demonstrate we can start poll before we have a msg
on the queue
- Thread.sleep(50);
+ Thread.sleep(500);
+
+ template.sendBody("direct:start", "Hello");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testJmsPollingConsumerNoWait() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("Hello Claus");
+
+ // use another thread for polling consumer to demonstrate that we can
wait before
+ // the message is sent to the queue
+ Executors.newSingleThreadExecutor().execute(new Runnable() {
+ public void run() {
+ String body =
consumer.receiveBodyNoWait("activemq:queue.start", String.class);
+ assertNull("Should be null", body);
+
+ template.sendBody("activemq:queue.foo", "Hello Claus");
+ }
+ });
+
+ // wait a little to demonstrate we can start poll before we have a msg
on the queue
+ Thread.sleep(500);
+
+ template.sendBody("direct:start", "Hello");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testJmsPollingConsumerLowTimeout() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("Hello Claus");
+
+ // use another thread for polling consumer to demonstrate that we can
wait before
+ // the message is sent to the queue
+ Executors.newSingleThreadExecutor().execute(new Runnable() {
+ public void run() {
+ String body = consumer.receiveBody("activemq:queue.start",
100, String.class);
+ assertNull("Should be null", body);
+
+ template.sendBody("activemq:queue.foo", "Hello Claus");
+ }
+ });
+
+ // wait a little to demonstrate we can start poll before we have a msg
on the queue
+ Thread.sleep(500);
+
+ template.sendBody("direct:start", "Hello");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testJmsPollingConsumerHighTimeout() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("Hello Claus");
+
+ // use another thread for polling consumer to demonstrate that we can
wait before
+ // the message is sent to the queue
+ Executors.newSingleThreadExecutor().execute(new Runnable() {
+ public void run() {
+ String body = consumer.receiveBody("activemq:queue.start",
3000, String.class);
+ template.sendBody("activemq:queue.foo", body + " Claus");
+ }
+ });
+
+ // wait a little to demonstrate we can start poll before we have a msg
on the queue
+ Thread.sleep(500);
template.sendBody("direct:start", "Hello");
Modified:
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/management/jmxInstrumentationWithConnector.xml
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/management/jmxInstrumentationWithConnector.xml?rev=809546&r1=809545&r2=809546&view=diff
==============================================================================
---
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/management/jmxInstrumentationWithConnector.xml
(original)
+++
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/management/jmxInstrumentationWithConnector.xml
Mon Aug 31 11:42:10 2009
@@ -23,7 +23,7 @@
">
<!-- START SNIPPET: example -->
- <bean id="agent"
class="org.apache.camel.management.DefaultInstrumentationAgent">
+ <bean id="agent" class="org.apache.camel.management.DefaultManagementAgent">
<property name="server" ref="mbeanServer"/>
</bean>
@@ -31,7 +31,7 @@
<property name="managementAgent" ref="agent"/>
</bean>
- <bean id="lifeCycleStrategy"
class="org.apache.camel.management.DefaultManagedLifecycleStrategy">
+ <bean id="lifeCycleStrategy"
class="org.apache.camel.management.DefaultManagementLifecycleStrategy">
<constructor-arg index="0" ref="camel"/>
</bean>
Added:
camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/jms/JmsPollEnrichTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/jms/JmsPollEnrichTest.java?rev=809546&view=auto
==============================================================================
---
camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/jms/JmsPollEnrichTest.java
(added)
+++
camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/jms/JmsPollEnrichTest.java
Mon Aug 31 11:42:10 2009
@@ -0,0 +1,65 @@
+/**
+ * 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.itest.jms;
+
+import javax.naming.Context;
+
+import org.apache.activemq.camel.component.ActiveMQComponent;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.apache.camel.util.jndi.JndiContext;
+import org.junit.Test;
+
+/**
+ * @version $Revision$
+ */
+public class JmsPollEnrichTest extends CamelTestSupport {
+
+ @Test
+ public void testPollEnrichJms() throws Exception {
+ template.sendBody("jms:queue:foo", "Bye World");
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("Bye World");
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() {
+
from("direct:start").pollEnrich("jms:queue:foo").to("mock:result");
+ }
+ };
+ }
+
+ @Override
+ protected Context createJndiContext() throws Exception {
+ JndiContext answer = new JndiContext();
+
+ // add ActiveMQ with embedded broker
+ ActiveMQComponent amq =
ActiveMQComponent.activeMQComponent("vm://localhost?broker.persistent=false");
+ amq.setCamelContext(context);
+ answer.bind("jms", amq);
+ return answer;
+ }
+
+}
Propchange:
camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/jms/JmsPollEnrichTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/jms/JmsPollEnrichTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
