You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by le...@apache.org on 2013/01/17 22:24:28 UTC
svn commit: r1434922 - in /incubator/streams/trunk: streams-eip-routes/
streams-eip-routes/src/main/java/org/apache/streams/messaging/processors/
streams-eip-routes/src/main/java/org/apache/streams/messaging/routers/impl/
streams-eip-routes/src/main/re...
Author: letourneau
Date: Thu Jan 17 21:24:28 2013
New Revision: 1434922
URL: http://svn.apache.org/viewvc?rev=1434922&view=rev
Log:
fixed the null body message route...publish to splitting message and placing on Q infrastructure is in place with dynamic routing
Modified:
incubator/streams/trunk/streams-eip-routes/pom.xml
incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/processors/ActivityPublisherProcessor.java
incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/routers/impl/ActivityConsumerRouter.java
incubator/streams/trunk/streams-eip-routes/src/main/resources/META-INF/spring/camelContext.xml
incubator/streams/trunk/streams-eip-routes/src/main/resources/META-INF/spring/osgi-component-import.xml
incubator/streams/trunk/streams-osgi-components/activity-consumer/pom.xml
incubator/streams/trunk/streams-osgi-components/activity-consumer/src/main/java/org/apache/streams/osgi/components/activityconsumer/ActivityConsumer.java
incubator/streams/trunk/streams-osgi-components/activity-consumer/src/main/java/org/apache/streams/osgi/components/activityconsumer/impl/PushActivityConsumer.java
Modified: incubator/streams/trunk/streams-eip-routes/pom.xml
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-eip-routes/pom.xml?rev=1434922&r1=1434921&r2=1434922&view=diff
==============================================================================
--- incubator/streams/trunk/streams-eip-routes/pom.xml (original)
+++ incubator/streams/trunk/streams-eip-routes/pom.xml Thu Jan 17 21:24:28 2013
@@ -69,9 +69,9 @@
<instructions>
<Bundle-SymbolicName>${bundle.symbolicName}</Bundle-SymbolicName>
<Bundle-Version>${pom.version}</Bundle-Version>
- <Export-Package>${bundle.namespace};version="${pom.version}",org.apache.streams.messaging.routers,org.apache.streams.messaging.rules,org.apache.streams.messaging.processors</Export-Package>
+ <Export-Package>${bundle.namespace};version="${pom.version}",org.apache.streams.messaging.routers,org.apache.streams.messaging.rules,org.apache.streams.messaging.processors,org.apache.activemq</Export-Package>
<Private-Package>${bundle.namespace}.messaging.routers.impl.*,${bundle.namespace}.messaging.rules.impl.*</Private-Package>
- <Import-Package>org.apache.camel.*;version="2.8.5",org.springframework.*;version="3.0.6.RELEASE",org.apache.commons.logging,org.apache.streams.*,org.apache.streams.osgi.components,org.apache.streams.osgi.components.activityconsumer,org.apache.streams.messaging.processors</Import-Package>
+ <Import-Package>org.apache.camel.*;version="2.8.5",org.apache.activemq.camel.component,org.apache.activemq,org.apache.activemq.pool,org.apache.camel.component.jms,org.springframework.*;version="3.0.6.RELEASE",org.apache.commons.logging,org.apache.streams.*,org.apache.streams.osgi.components,org.apache.streams.osgi.components.activityconsumer,org.apache.streams.messaging.processors,javax.jms, javax.net.ssl, javax.transaction.xa, org.apache.activemq.advisory, org.apache.activemq.blob, org.apache.activemq.broker, org.apache.activemq.broker.region, org.apache.activemq.command, org.apache.activemq.filter, org.apache.activemq.jndi, org.apache.activemq.management, org.apache.activemq.selector, org.apache.activemq.state, org.apache.activemq.thread, org.apache.activemq.transaction, org.apache.activemq.transport, org.apache.activemq.transport.failover, org.apache.activemq.transport.tcp, org.apache.activemq.usage, org.apache.activemq.util, org.slf4j</Import-Pack
age>
<Include-Resource>src/main/resources</Include-Resource>
</instructions>
</configuration>
@@ -87,6 +87,18 @@
</dependency>
<dependency>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>activemq-camel</artifactId>
+ <version>5.5.1</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>activemq-pool</artifactId>
+ <version>5.5.1</version>
+ </dependency>
+
+ <dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<version>3.0.6.RELEASE</version>
Modified: incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/processors/ActivityPublisherProcessor.java
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/processors/ActivityPublisherProcessor.java?rev=1434922&r1=1434921&r2=1434922&view=diff
==============================================================================
--- incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/processors/ActivityPublisherProcessor.java (original)
+++ incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/processors/ActivityPublisherProcessor.java Thu Jan 17 21:24:28 2013
@@ -26,8 +26,11 @@ public class ActivityPublisherProcessor
String body = exchange.getIn().getBody(String.class);
try{
URI publisherUrl = new URI(body);
+ //todo: this will probably come from the activity set itself...maybe?
exchange.getOut().setHeader("SRC",body);
+ //todo: set this to the activity set json
exchange.getOut().setBody("got some message from a registered publisher.");
+
}catch(URISyntaxException e){
exchange.getOut().setFault(true);
exchange.getOut().setHeader(Exchange.HTTP_RESPONSE_CODE,400);
Modified: incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/routers/impl/ActivityConsumerRouter.java
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/routers/impl/ActivityConsumerRouter.java?rev=1434922&r1=1434921&r2=1434922&view=diff
==============================================================================
--- incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/routers/impl/ActivityConsumerRouter.java (original)
+++ incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/routers/impl/ActivityConsumerRouter.java Thu Jan 17 21:24:28 2013
@@ -16,6 +16,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.CamelContext;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.camel.ExchangePattern;
@@ -57,13 +58,17 @@ public class ActivityConsumerRouter exte
public String slip(Exchange exchange, String body, @Header("SRC") String src){
//if not sent from a SRC, kill the routing chain
if (src==null){
+ LOG.info("end of route reached, the body is: " + body);
+ exchange.getOut().setBody(null);
return src;
- }
+ }
+ LOG.info("body of message at router: " + body);
//lookup the route to the activityconsumer registered to receive messages from SRC
String outRoute = activityConsumerWarehouse.findConsumerBySrc(src).getInRoute();
//for now, just one route out...set SRC to null
exchange.getOut().setHeader("SRC", null);
+ exchange.getOut().setBody(body);
return outRoute;
}
@@ -88,7 +93,11 @@ public class ActivityConsumerRouter exte
@Override
public void configure() throws Exception {
- from(from).bean(activityConsumer, "receive");
+
+ //todo: this message to the bean is always NULL!!!
+ from(from).bean(activityConsumer, "receive").split().method(activityConsumer, "split").to("direct:activityQ");
+
+
}
}
Modified: incubator/streams/trunk/streams-eip-routes/src/main/resources/META-INF/spring/camelContext.xml
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-eip-routes/src/main/resources/META-INF/spring/camelContext.xml?rev=1434922&r1=1434921&r2=1434922&view=diff
==============================================================================
--- incubator/streams/trunk/streams-eip-routes/src/main/resources/META-INF/spring/camelContext.xml (original)
+++ incubator/streams/trunk/streams-eip-routes/src/main/resources/META-INF/spring/camelContext.xml Thu Jan 17 21:24:28 2013
@@ -49,6 +49,11 @@
</dynamicRouter>
</route>
+ <route>
+ <from uri="direct:activityQ"/>
+ <inOnly uri="activemq:queue:activities"/>
+ </route>
+
</camelContext>
Modified: incubator/streams/trunk/streams-eip-routes/src/main/resources/META-INF/spring/osgi-component-import.xml
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-eip-routes/src/main/resources/META-INF/spring/osgi-component-import.xml?rev=1434922&r1=1434921&r2=1434922&view=diff
==============================================================================
--- incubator/streams/trunk/streams-eip-routes/src/main/resources/META-INF/spring/osgi-component-import.xml (original)
+++ incubator/streams/trunk/streams-eip-routes/src/main/resources/META-INF/spring/osgi-component-import.xml Thu Jan 17 21:24:28 2013
@@ -22,7 +22,27 @@
<property name="camelContext" ref="context"/>
</bean>
+ <bean id="jmsConnectionFactory"
+ class="org.apache.activemq.ActiveMQConnectionFactory">
+ <property name="brokerURL" value="tcp://localhost:61616" />
+ </bean>
+
+ <bean id="pooledConnectionFactory"
+ class="org.apache.activemq.pool.PooledConnectionFactory" init-method="start" destroy-method="stop">
+ <property name="maxConnections" value="8" />
+ <property name="connectionFactory" ref="jmsConnectionFactory" />
+ </bean>
+ <bean id="jmsConfig"
+ class="org.apache.camel.component.jms.JmsConfiguration">
+ <property name="connectionFactory" ref="pooledConnectionFactory"/>
+ <property name="concurrentConsumers" value="10"/>
+ </bean>
+
+ <bean id="activemq"
+ class="org.apache.activemq.camel.component.ActiveMQComponent">
+ <property name="configuration" ref="jmsConfig"/>
+ </bean>
</beans>
\ No newline at end of file
Modified: incubator/streams/trunk/streams-osgi-components/activity-consumer/pom.xml
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-osgi-components/activity-consumer/pom.xml?rev=1434922&r1=1434921&r2=1434922&view=diff
==============================================================================
--- incubator/streams/trunk/streams-osgi-components/activity-consumer/pom.xml (original)
+++ incubator/streams/trunk/streams-osgi-components/activity-consumer/pom.xml Thu Jan 17 21:24:28 2013
@@ -86,6 +86,7 @@
<scope>provided</scope>
<optional>true</optional>
</dependency>
+
<dependency>
<groupId>org.osgi</groupId>
<artifactId>osgi_R4_compendium</artifactId>
Modified: incubator/streams/trunk/streams-osgi-components/activity-consumer/src/main/java/org/apache/streams/osgi/components/activityconsumer/ActivityConsumer.java
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-osgi-components/activity-consumer/src/main/java/org/apache/streams/osgi/components/activityconsumer/ActivityConsumer.java?rev=1434922&r1=1434921&r2=1434922&view=diff
==============================================================================
--- incubator/streams/trunk/streams-osgi-components/activity-consumer/src/main/java/org/apache/streams/osgi/components/activityconsumer/ActivityConsumer.java (original)
+++ incubator/streams/trunk/streams-osgi-components/activity-consumer/src/main/java/org/apache/streams/osgi/components/activityconsumer/ActivityConsumer.java Thu Jan 17 21:24:28 2013
@@ -2,7 +2,7 @@ package org.apache.streams.osgi.componen
public interface ActivityConsumer {
- public void receive(Object activity);
+ public void receive(String activity);
public void init();
public String getSrc();
public void setInRoute(String route);
Modified: incubator/streams/trunk/streams-osgi-components/activity-consumer/src/main/java/org/apache/streams/osgi/components/activityconsumer/impl/PushActivityConsumer.java
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-osgi-components/activity-consumer/src/main/java/org/apache/streams/osgi/components/activityconsumer/impl/PushActivityConsumer.java?rev=1434922&r1=1434921&r2=1434922&view=diff
==============================================================================
--- incubator/streams/trunk/streams-osgi-components/activity-consumer/src/main/java/org/apache/streams/osgi/components/activityconsumer/impl/PushActivityConsumer.java (original)
+++ incubator/streams/trunk/streams-osgi-components/activity-consumer/src/main/java/org/apache/streams/osgi/components/activityconsumer/impl/PushActivityConsumer.java Thu Jan 17 21:24:28 2013
@@ -3,6 +3,8 @@ package org.apache.streams.osgi.componen
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.streams.osgi.components.activityconsumer.ActivityConsumer;
+import java.util.List;
+import java.util.ArrayList;
public class PushActivityConsumer implements ActivityConsumer {
@@ -25,10 +27,24 @@ public class PushActivityConsumer implem
this.src = src;
}
- public void receive (Object activity){
- //receive activity...do anything that is necessary
+
+ public void receive (String activity){
+ //receive activities...do anything that is necessary
LOG.info("got a message i subscribed to: " + activity);
- //pass off to activity splitter ?
+
+ //pass off to activity splitter
+
+ }
+
+ public List<String> split(String activities){
+ LOG.info("I am going to split this message: " + activities);
+ ArrayList<String> activitiesList = new ArrayList<String>();
+ activitiesList.add("this");
+ activitiesList.add("is");
+ activitiesList.add("my");
+ activitiesList.add("split");
+ activitiesList.add("message");
+ return activitiesList;
}
public void init(){