You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by ds...@apache.org on 2013/09/12 15:34:21 UTC

svn commit: r1522571 - in /incubator/streams/branches/cassandra/streams-eip-routes/src/main/java/org/apache/streams/messaging: aggregation/ActivityAggregator.java routers/impl/ActivityStreamsSubscriberRouter.java

Author: dsullivan
Date: Thu Sep 12 13:34:20 2013
New Revision: 1522571

URL: http://svn.apache.org/r1522571
Log:
immediate update of freshly registered subscribers

Modified:
    incubator/streams/branches/cassandra/streams-eip-routes/src/main/java/org/apache/streams/messaging/aggregation/ActivityAggregator.java
    incubator/streams/branches/cassandra/streams-eip-routes/src/main/java/org/apache/streams/messaging/routers/impl/ActivityStreamsSubscriberRouter.java

Modified: incubator/streams/branches/cassandra/streams-eip-routes/src/main/java/org/apache/streams/messaging/aggregation/ActivityAggregator.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/cassandra/streams-eip-routes/src/main/java/org/apache/streams/messaging/aggregation/ActivityAggregator.java?rev=1522571&r1=1522570&r2=1522571&view=diff
==============================================================================
--- incubator/streams/branches/cassandra/streams-eip-routes/src/main/java/org/apache/streams/messaging/aggregation/ActivityAggregator.java (original)
+++ incubator/streams/branches/cassandra/streams-eip-routes/src/main/java/org/apache/streams/messaging/aggregation/ActivityAggregator.java Thu Sep 12 13:34:20 2013
@@ -27,11 +27,15 @@ public class ActivityAggregator {
     @Scheduled(fixedRate=30000)
     public void distributeToSubscribers() {
         for (ActivityStreamsSubscriber subscriber : activityStreamsSubscriberWarehouse.getAllSubscribers()) {
-            Set<String> activities = new HashSet<String>();
-            activities.addAll(activityService.getActivitiesForFilters(subscriber.getActivityStreamsSubscriberConfiguration().getFilters(), subscriber.getLastUpdated()));
-            //TODO: an activity posted in between the cql query and setting the lastUpdated field will be lost
-            subscriber.setLastUpdated(new Date());
-            subscriber.receive(new ArrayList<String>(activities));
+              updateSubscriber(subscriber);
         }
     }
+
+    public void updateSubscriber(ActivityStreamsSubscriber subscriber){
+        Set<String> activities = new HashSet<String>();
+        activities.addAll(activityService.getActivitiesForFilters(subscriber.getActivityStreamsSubscriberConfiguration().getFilters(), subscriber.getLastUpdated()));
+        //TODO: an activity posted in between the cql query and setting the lastUpdated field will be lost
+        subscriber.setLastUpdated(new Date());
+        subscriber.receive(new ArrayList<String>(activities));
+    }
 }

Modified: incubator/streams/branches/cassandra/streams-eip-routes/src/main/java/org/apache/streams/messaging/routers/impl/ActivityStreamsSubscriberRouter.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/cassandra/streams-eip-routes/src/main/java/org/apache/streams/messaging/routers/impl/ActivityStreamsSubscriberRouter.java?rev=1522571&r1=1522570&r2=1522571&view=diff
==============================================================================
--- incubator/streams/branches/cassandra/streams-eip-routes/src/main/java/org/apache/streams/messaging/routers/impl/ActivityStreamsSubscriberRouter.java (original)
+++ incubator/streams/branches/cassandra/streams-eip-routes/src/main/java/org/apache/streams/messaging/routers/impl/ActivityStreamsSubscriberRouter.java Thu Sep 12 13:34:20 2013
@@ -6,6 +6,7 @@ import org.apache.camel.Exchange;
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.streams.messaging.aggregation.ActivityAggregator;
 import org.apache.streams.messaging.configuration.EipConfigurator;
 import org.apache.streams.messaging.routers.ActivityStreamsSubscriberRouteBuilder;
 import org.apache.streams.osgi.components.activitysubscriber.ActivityStreamsSubscriber;
@@ -24,6 +25,9 @@ public class ActivityStreamsSubscriberRo
 
     protected CamelContext camelContext;
 
+    @Autowired
+    private ActivityAggregator activityAggregator;
+
     private ActivityStreamsSubscriberWarehouse activityStreamsSubscriberWarehouse;
 
     public void setCamelContext(CamelContext camelContext) {
@@ -58,6 +62,7 @@ public class ActivityStreamsSubscriberRo
                 //setup a message queue for this consumer.getInRoute()
                 camelContext.addRoutes(new DynamicSubscriberRouteBuilder(configuration,camelContext, configuration.getSubscriberEndpointProtocol() + activityStreamsSubscriber.getInRoute(), activityStreamsSubscriber));
 
+                activityAggregator.updateSubscriber(activityStreamsSubscriber);
                 activityStreamsSubscriberWarehouse.register(activityStreamsSubscriber);
             }catch (Exception e){
                 exchange.getOut().setHeader(Exchange.HTTP_RESPONSE_CODE,500);