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/02/04 14:43:04 UTC
svn commit: r1442116 - in /incubator/streams/trunk: streams-eip-routes/
streams-eip-routes/src/main/java/org/apache/streams/messaging/configuration/
streams-eip-routes/src/main/java/org/apache/streams/messaging/processors/
streams-osgi-components/activ...
Author: letourneau
Date: Mon Feb 4 13:43:03 2013
New Revision: 1442116
URL: http://svn.apache.org/viewvc?rev=1442116&view=rev
Log:
setting up subscriptions using JSON post configuration
Modified:
incubator/streams/trunk/streams-eip-routes/pom.xml
incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/configuration/EipConfigurator.java
incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/processors/ActivityStreamsSubscriberRegistrationProcessor.java
incubator/streams/trunk/streams-osgi-components/activity-subscriber/pom.xml
incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/ActivityStreamsSubscriber.java
incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/impl/ActivityStreamsSubscriberDelegate.java
incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/impl/ActivityStreamsSubscriptionImpl.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=1442116&r1=1442115&r2=1442116&view=diff
==============================================================================
--- incubator/streams/trunk/streams-eip-routes/pom.xml (original)
+++ incubator/streams/trunk/streams-eip-routes/pom.xml Mon Feb 4 13:43:03 2013
@@ -70,9 +70,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.configuration,org.apache.streams.messaging.routers,org.apache.streams.messaging.rules,org.apache.streams.messaging.processors,org.apache.activemq</Export-Package>
+ <Export-Package>${bundle.namespace};version="${pom.version}",org.apache.streams.messaging.configuration,org.apache.streams.messaging.routers,org.apache.streams.messaging.rules,org.apache.streams.messaging.processors,org.apache.activemq,org.codehaus.jackson.*;version="${jackson.version}"</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.apache.streams.messaging.configuration,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.osgi.components.activitysubscriber,org.apache.streams.osgi.components.activitysubscriber.impl,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.trans
port, org.apache.activemq.transport.failover, org.apache.activemq.transport.tcp, org.apache.activemq.usage, org.apache.activemq.util, org.slf4j</Import-Package>
+ <Import-Package>org.apache.camel.*;version="2.8.5",org.apache.streams.messaging.configuration,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.osgi.components.activitysubscriber,org.apache.streams.osgi.components.activitysubscriber.impl,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.trans
port, org.apache.activemq.transport.failover, org.apache.activemq.transport.tcp, org.apache.activemq.usage, org.apache.activemq.util, org.slf4j,org.codehaus.jackson;version="${jackson.version}",javax.xml.datatype, javax.xml.namespace, javax.xml.parsers, org.joda.time, org.joda.time.format, org.w3c.dom, org.w3c.dom.bootstrap, org.w3c.dom.ls, org.xml.sax</Import-Package>
<Include-Resource>src/main/resources</Include-Resource>
</instructions>
</configuration>
Modified: incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/configuration/EipConfigurator.java
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/configuration/EipConfigurator.java?rev=1442116&r1=1442115&r2=1442116&view=diff
==============================================================================
--- incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/configuration/EipConfigurator.java (original)
+++ incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/configuration/EipConfigurator.java Mon Feb 4 13:43:03 2013
@@ -30,7 +30,7 @@ public class EipConfigurator {
public static String SUBSCRIBER_URL_RESOURCE = "/streams/subscriber";
public static String SUBSCRIBER_RECEIVE_METHOD = "receive";
- public static String SUBSCRIBER_POST_METHOD = "addSrc";
+ public static String SUBSCRIBER_POST_METHOD = "updateActivityStreamsSubscriberConfiguration";
public static String SUBSCRIBER_GET_METHOD = "getStream";
public String getConsumerInRouteHost() {
Modified: incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/processors/ActivityStreamsSubscriberRegistrationProcessor.java
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/processors/ActivityStreamsSubscriberRegistrationProcessor.java?rev=1442116&r1=1442115&r2=1442116&view=diff
==============================================================================
--- incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/processors/ActivityStreamsSubscriberRegistrationProcessor.java (original)
+++ incubator/streams/trunk/streams-eip-routes/src/main/java/org/apache/streams/messaging/processors/ActivityStreamsSubscriberRegistrationProcessor.java Mon Feb 4 13:43:03 2013
@@ -2,12 +2,16 @@ package org.apache.streams.messaging.pro
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.streams.osgi.components.activitysubscriber.ActivityStreamsSubscription;
+import org.apache.streams.osgi.components.activitysubscriber.impl.ActivityStreamsSubscriptionImpl;
+import org.codehaus.jackson.map.DeserializationConfig;
import org.codehaus.jackson.map.ObjectMapper;
public class ActivityStreamsSubscriberRegistrationProcessor implements Processor{
-
+ private static final transient Log LOG = LogFactory.getLog(ActivityStreamsSubscriberRegistrationProcessor.class);
public void process(Exchange exchange){
//add the necessary headers to the message so that the activity registration component
//can do a lookup to either make a new processor and endpoint, or pass the message to the right one
@@ -29,13 +33,15 @@ public class ActivityStreamsSubscriberRe
//maybe its a list of URLs to subscribe to subscriptions=1,2,3,4&auth_token=XXXX
ObjectMapper mapper = new ObjectMapper();
+ mapper.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES,false);
try {
// read from file, convert it to user class
- ActivityStreamsSubscription configuration = mapper.readValue(body, ActivityStreamsSubscription.class);
+ ActivityStreamsSubscription configuration = mapper.readValue(body, ActivityStreamsSubscriptionImpl.class);
exchange.getOut().setBody(configuration);
} catch (Exception e) {
+ LOG.info("exception" + e);
exchange.getOut().setFault(true);
exchange.getOut().setHeader(Exchange.HTTP_RESPONSE_CODE,400);
exchange.getOut().setBody("POST should contain a valid Subscription configuration object.");
Modified: incubator/streams/trunk/streams-osgi-components/activity-subscriber/pom.xml
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-osgi-components/activity-subscriber/pom.xml?rev=1442116&r1=1442115&r2=1442116&view=diff
==============================================================================
--- incubator/streams/trunk/streams-osgi-components/activity-subscriber/pom.xml (original)
+++ incubator/streams/trunk/streams-osgi-components/activity-subscriber/pom.xml Mon Feb 4 13:43:03 2013
@@ -7,6 +7,7 @@
<bundle.symbolicName>activity-subscriber-bundle</bundle.symbolicName>
<bundle.namespace>org.apache.streams.osgi.components.activitysubscriber</bundle.namespace>
<commons.log>1.1</commons.log>
+ <jackson.version>1.9.11</jackson.version>
</properties>
<modelVersion>4.0.0</modelVersion>
@@ -70,7 +71,7 @@
<Bundle-Version>${pom.version}</Bundle-Version>
<Export-Package>${bundle.namespace};version="${pom.version}",org.apache.streams.osgi.components.activitysubscriber.impl</Export-Package>
<Private-Package>${bundle.namespace}.impl.*</Private-Package>
- <Import-Package>org.apache.streams.osgi.components.activitysubscriber,org.apache.commons.logging</Import-Package>
+ <Import-Package>org.apache.streams.osgi.components.activitysubscriber,org.apache.commons.logging,org.codehaus.jackson.*;version="${jackson.version}",javax.xml.datatype, javax.xml.namespace, javax.xml.parsers, org.joda.time, org.joda.time.format, org.w3c.dom, org.w3c.dom.bootstrap, org.w3c.dom.ls, org.xml.sax</Import-Package>
<Include-Resource>src/main/resources</Include-Resource>
</instructions>
</configuration>
@@ -88,6 +89,18 @@
</dependency>
<dependency>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-mrbean</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-mapper-asl</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
+
+ <dependency>
<groupId>org.osgi</groupId>
<artifactId>osgi_R4_compendium</artifactId>
<version>1.0</version>
Modified: incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/ActivityStreamsSubscriber.java
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/ActivityStreamsSubscriber.java?rev=1442116&r1=1442115&r2=1442116&view=diff
==============================================================================
--- incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/ActivityStreamsSubscriber.java (original)
+++ incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/ActivityStreamsSubscriber.java Mon Feb 4 13:43:03 2013
@@ -8,6 +8,6 @@ public interface ActivityStreamsSubscrib
public void setInRoute(String route);
public String getInRoute();
public void setActivityStreamsSubscriberConfiguration(ActivityStreamsSubscription config);
- public void updateActivityStreamsSubscriberConfiguration(ActivityStreamsSubscription config);
+ public void updateActivityStreamsSubscriberConfiguration(String config);
}
Modified: incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/impl/ActivityStreamsSubscriberDelegate.java
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/impl/ActivityStreamsSubscriberDelegate.java?rev=1442116&r1=1442115&r2=1442116&view=diff
==============================================================================
--- incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/impl/ActivityStreamsSubscriberDelegate.java (original)
+++ incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/impl/ActivityStreamsSubscriberDelegate.java Mon Feb 4 13:43:03 2013
@@ -4,6 +4,8 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.streams.osgi.components.activitysubscriber.ActivityStreamsSubscriber;
import org.apache.streams.osgi.components.activitysubscriber.ActivityStreamsSubscription;
+import org.codehaus.jackson.map.DeserializationConfig;
+import org.codehaus.jackson.map.ObjectMapper;
import java.util.ArrayList;
@@ -34,8 +36,20 @@ public class ActivityStreamsSubscriberDe
this.activityStreamsSubscriberConfiguration = activityStreamsSubscriberConfiguration;
}
- public void updateActivityStreamsSubscriberConfiguration(ActivityStreamsSubscription activityStreamsSubscriberConfiguration) {
- this.activityStreamsSubscriberConfiguration = activityStreamsSubscriberConfiguration;
+ public void updateActivityStreamsSubscriberConfiguration(String activityStreamsSubscriberConfiguration) {
+ ObjectMapper mapper = new ObjectMapper();
+ mapper.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES,false);
+
+ try {
+ // read from file, convert it to user class
+ ActivityStreamsSubscription configuration = mapper.readValue(activityStreamsSubscriberConfiguration, ActivityStreamsSubscriptionImpl.class);
+ this.activityStreamsSubscriberConfiguration = configuration;
+
+ } catch (Exception e) {
+ LOG.info("exception" + e);
+
+ }
+
}
public String getInRoute() {
Modified: incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/impl/ActivityStreamsSubscriptionImpl.java
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/impl/ActivityStreamsSubscriptionImpl.java?rev=1442116&r1=1442115&r2=1442116&view=diff
==============================================================================
--- incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/impl/ActivityStreamsSubscriptionImpl.java (original)
+++ incubator/streams/trunk/streams-osgi-components/activity-subscriber/src/main/java/org/apache/streams/osgi/components/activitysubscriber/impl/ActivityStreamsSubscriptionImpl.java Mon Feb 4 13:43:03 2013
@@ -12,6 +12,7 @@ public class ActivityStreamsSubscription
private ArrayList<ActivityStreamsSubscriptionFilter> activityStreamsSubscriptionFilters;
private ArrayList<ActivityStreamsSubscriptionOutput> activityStreamsSubscriptionOutputs;
+
@Override
public ArrayList<ActivityStreamsSubscriptionFilter> getActivityStreamsSubscriptionFilters() {
return activityStreamsSubscriptionFilters;