Author: chirino
Date: Mon Apr 30 07:29:20 2007
New Revision: 533758
URL: http://svn.apache.org/viewvc?view=rev&rev=533758
Log:
- Renamed Component.resolveEndpoint to Component.createEndpoint
- DefaultCamelContext only registeres as singtons endpoints that say they are
singletons.
- Added some more transaction scenarios for JMS
- JmsConsumer was starting up multiple listeners. Removed the call to
listenerContainer.initialize() since listenerContainer.afterPropertiesSet()
calls it.
Modified:
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/Component.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/component/direct/DirectComponent.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultComponent.java
activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/component/queue/QueueRouteTest.java
activemq/camel/trunk/camel-jbi/src/main/java/org/apache/camel/component/jbi/CamelJbiComponent.java
activemq/camel/trunk/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConsumer.java
activemq/camel/trunk/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
activemq/camel/trunk/camel-jms/src/test/java/org/apache/camel/component/jms/TransactedJmsRouteTest.java
Modified:
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/Component.java
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/Component.java?view=diff&rev=533758&r1=533757&r2=533758
==============================================================================
---
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/Component.java
(original)
+++
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/Component.java
Mon Apr 30 07:29:20 2007
@@ -43,5 +43,5 @@
* @return a newly created endpoint or null if this component cannot
create instances of the given
* uri
*/
- Endpoint<E> resolveEndpoint(String uri) throws Exception;
+ Endpoint<E> createEndpoint(String uri) throws Exception;
}
Modified:
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/component/direct/DirectComponent.java
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/component/direct/DirectComponent.java?view=diff&rev=533758&r1=533757&r2=533758
==============================================================================
---
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/component/direct/DirectComponent.java
(original)
+++
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/component/direct/DirectComponent.java
Mon Apr 30 07:29:20 2007
@@ -47,7 +47,7 @@
return null;
}
- public Endpoint<E> resolveEndpoint(String uri) throws Exception {
+ public Endpoint<E> createEndpoint(String uri) throws Exception {
ObjectHelper.notNull(getCamelContext(), "camelContext");
URI u = new URI(uri);
Modified:
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java?view=diff&rev=533758&r1=533757&r2=533758
==============================================================================
---
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
(original)
+++
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
Mon Apr 30 07:29:20 2007
@@ -190,15 +190,19 @@
// Ask the component to resolve the endpoint.
if (component != null) {
- answer = component.resolveEndpoint(uri);
+
+ // Have the component create the endpoint if it can.
+ answer = component.createEndpoint(uri);
+
+ // If it's a singleton then auto register it.
+ if( answer!=null && answer.isSingleton() ) {
+ if (answer != null) {
+ startServices(answer);
+ endpoints.put(uri, answer);
+ }
+ }
}
- // HC: What's the idea behind starting an endpoint?
- // I don't think we have any endpoints that are services
do we?
- if (answer != null) {
- startServices(answer);
- endpoints.put(uri, answer);
- }
}
catch (Exception e) {
throw new ResolveEndpointFailedException(uri, e);
Modified:
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultComponent.java
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultComponent.java?view=diff&rev=533758&r1=533757&r2=533758
==============================================================================
---
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultComponent.java
(original)
+++
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultComponent.java
Mon Apr 30 07:29:20 2007
@@ -48,7 +48,7 @@
}
- public Endpoint<E> resolveEndpoint(String uri) throws Exception {
+ public Endpoint<E> createEndpoint(String uri) throws Exception {
ObjectHelper.notNull(getCamelContext(), "camelContext");
URI u = new URI(uri);
String path = u.getHost();
Modified:
activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/component/queue/QueueRouteTest.java
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/component/queue/QueueRouteTest.java?view=diff&rev=533758&r1=533757&r2=533758
==============================================================================
---
activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/component/queue/QueueRouteTest.java
(original)
+++
activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/component/queue/QueueRouteTest.java
Mon Apr 30 07:29:20 2007
@@ -95,7 +95,7 @@
container.start();
// now lets fire in a message
- Endpoint<Exchange> endpoint =
container.getComponent("queue").resolveEndpoint("queue:test.a");
+ Endpoint<Exchange> endpoint =
container.getComponent("queue").createEndpoint("queue:test.a");
Exchange exchange = endpoint.createExchange();
exchange.getIn().setHeader("cheese", 123);
Modified:
activemq/camel/trunk/camel-jbi/src/main/java/org/apache/camel/component/jbi/CamelJbiComponent.java
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/camel-jbi/src/main/java/org/apache/camel/component/jbi/CamelJbiComponent.java?view=diff&rev=533758&r1=533757&r2=533758
==============================================================================
---
activemq/camel/trunk/camel-jbi/src/main/java/org/apache/camel/component/jbi/CamelJbiComponent.java
(original)
+++
activemq/camel/trunk/camel-jbi/src/main/java/org/apache/camel/component/jbi/CamelJbiComponent.java
Mon Apr 30 07:29:20 2007
@@ -120,7 +120,7 @@
// Resolve Camel Endpoints
//-------------------------------------------------------------------------
- public Endpoint<Exchange> resolveEndpoint(String uri) {
+ public Endpoint<Exchange> createEndpoint(String uri) {
if (uri.startsWith("jbi:")) {
uri = uri.substring("jbi:".length());
Modified:
activemq/camel/trunk/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConsumer.java
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConsumer.java?view=diff&rev=533758&r1=533757&r2=533758
==============================================================================
---
activemq/camel/trunk/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConsumer.java
(original)
+++
activemq/camel/trunk/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConsumer.java
Mon Apr 30 07:29:20 2007
@@ -50,7 +50,6 @@
protected void doStart() throws Exception {
super.doStart();
listenerContainer.afterPropertiesSet();
- listenerContainer.initialize();
listenerContainer.start();
}
Modified:
activemq/camel/trunk/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java?view=diff&rev=533758&r1=533757&r2=533758
==============================================================================
---
activemq/camel/trunk/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
(original)
+++
activemq/camel/trunk/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
Mon Apr 30 07:29:20 2007
@@ -16,20 +16,15 @@
*/
package org.apache.camel.component.jms;
-import org.apache.camel.CamelContext;
+import javax.jms.Message;
+
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
-import org.apache.camel.util.URISupport;
-import org.apache.camel.util.IntrospectionSupport;
import org.apache.camel.impl.DefaultEndpoint;
import org.springframework.jms.core.JmsOperations;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
-
-import javax.jms.Message;
-import java.net.URI;
-import java.util.Map;
/**
* @version $Revision:520964 $
Modified:
activemq/camel/trunk/camel-jms/src/test/java/org/apache/camel/component/jms/TransactedJmsRouteTest.java
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/camel-jms/src/test/java/org/apache/camel/component/jms/TransactedJmsRouteTest.java?view=diff&rev=533758&r1=533757&r2=533758
==============================================================================
---
activemq/camel/trunk/camel-jms/src/test/java/org/apache/camel/component/jms/TransactedJmsRouteTest.java
(original)
+++
activemq/camel/trunk/camel-jms/src/test/java/org/apache/camel/component/jms/TransactedJmsRouteTest.java
Mon Apr 30 07:29:20 2007
@@ -28,7 +28,6 @@
import org.apache.camel.CamelContext;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Processor;
-import org.apache.camel.Route;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.processor.DelegateProcessor;
@@ -102,6 +101,9 @@
inheritErrorHandler(false);
// Used to validate messages are sent to the target.
from("activemq:queue:mock.a").trace().to("mock:a");
+
from("activemq:queue:mock.b").trace().to("mock:b");
+
from("activemq:queue:mock.c").trace().to("mock:c");
+
from("activemq:queue:mock.d").trace().to("mock:d");
// Receive from a and send to target in 1 tx.
from("activemq:queue:a").to("activemq:queue:mock.a");
@@ -134,15 +136,22 @@
//
// Sets up 2 consumers on single topic, one being
transacted the other not. Used to verify
// That each consumer can have independently configured
transaction settings.
- // Do a rollback, should cause the transacted consumer
to re-deliver but not the un-trasacted one.
+ // Do a rollback, should cause the transacted consumer
to re-deliver (since we are using a durable subscription) but not the
un-transacted one.
+ // TODO: find out why re-delivery is not working with a
non durable transacted topic.
JmsEndpoint endpoint1 = (JmsEndpoint)
endpoint("activemq:topic:f");
endpoint1.getConfiguration().setTransacted(true);
+ endpoint1.getConfiguration().setSubscriptionDurable(true);
+ endpoint1.getConfiguration().setClientId("client2");
+ endpoint1.getConfiguration().setDurableSubscriptionName("sub");
from(endpoint1).policy(requried).policy(rollback).to("activemq:queue:mock.a",
"mock:b");
JmsEndpoint endpoint2 = (JmsEndpoint)
endpoint("activemq:topic:f");
- endpoint1.getConfiguration().setTransacted(false);
+ endpoint2.getConfiguration().setTransacted(false);
+
endpoint2.getConfiguration().setAcknowledgementMode(Session.AUTO_ACKNOWLEDGE);
+ endpoint2.getConfiguration().setSubscriptionDurable(true);
+ endpoint2.getConfiguration().setClientId("client1");
+ endpoint2.getConfiguration().setDurableSubscriptionName("sub");
from(endpoint2).policy(requried).policy(rollback).to("activemq:queue:mock.c",
"mock:d");
-
}
};
}
@@ -162,9 +171,9 @@
protected void setUp() throws Exception {
super.setUp();
- for (Route route : this.context.getRoutes()) {
- System.out.println(route);
- }
+// for (Route route : this.context.getRoutes()) {
+// System.out.println(route);
+// }
mockEndpointA = (MockEndpoint) resolveMandatoryEndpoint("mock:a");
mockEndpointB = (MockEndpoint) resolveMandatoryEndpoint("mock:b");
@@ -178,6 +187,27 @@
spring.destroy();
}
+ /**
+ * This test seems to be fail every other run.
+ * @throws Exception
+ */
+ public void disabledtestSenarioF() throws Exception {
+ String expected = getName()+": "+System.currentTimeMillis();
+ mockEndpointA.expectedMessageCount(0);
+ mockEndpointB.expectedMinimumMessageCount(2);
+ mockEndpointC.expectedMessageCount(0);
+ mockEndpointD.expectedMessageCount(1);
+ send("activemq:topic:f", expected);
+
+ // Wait till the endpoints get their messages.
+ assertWait(10, TimeUnit.SECONDS,
mockEndpointA,mockEndpointB,mockEndpointC,mockEndpointD);
+
+ // Wait a little more to make sure extra messages are not received.
+ Thread.sleep(1000);
+
+
assertIsSatisfied(mockEndpointA,mockEndpointB,mockEndpointC,mockEndpointD);
+ }
+
public void testSenarioA() throws Exception {
String expected = getName()+": "+System.currentTimeMillis();
mockEndpointA.expectedBodiesReceived(expected);
@@ -237,21 +267,5 @@
assertIsSatisfied(mockEndpointA, mockEndpointB);
}
- public void disabletestSenarioF() throws Exception {
- String expected = getName()+": "+System.currentTimeMillis();
- mockEndpointA.expectedMessageCount(0);
- mockEndpointB.expectedMinimumMessageCount(2);
- mockEndpointC.expectedMessageCount(0);
- mockEndpointD.expectedMessageCount(1);
- send("activemq:queue:e", expected);
-
- // Wait till the endpoints get their messages.
- assertWait(5, TimeUnit.SECONDS, mockEndpointA,mockEndpointB);
-
- // Wait a little more to make sure extra messages are not received.
- Thread.sleep(1000);
-
- assertIsSatisfied(mockEndpointA, mockEndpointB);
- }
}
