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/11/12 17:01:42 UTC
svn commit: r1541121 - in /incubator/streams/branches/webservice:
streams-components/src/main/java/org/apache/streams/components/activitysubscriber/impl/
streams-components/src/main/java/org/apache/streams/components/service/
streams-components/src/mai...
Author: dsullivan
Date: Tue Nov 12 16:01:41 2013
New Revision: 1541121
URL: http://svn.apache.org/r1541121
Log:
changing tags to filters and the way to see activity
Added:
incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsFiltersService.java
incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/StreamsFiltersServiceImpl.java
incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsFiltersServiceTest.java
Modified:
incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/activitysubscriber/impl/ActivityStreamsSubscriberWarehouseImpl.java
incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsActivityRepositoryService.java
incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsSubscriptionRepositoryService.java
incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/CassandraActivityService.java
incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/CassandraSubscriptionService.java
incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsActivityRepositoryServiceTest.java
incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsSubscriptionRepositoryServiceTest.java
incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/model/ActivityStreamsSubscription.java
incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/model/cassandra/CassandraSubscription.java
incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/ActivityStreamsRepository.java
incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/SubscriptionRepository.java
incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/cassandra/CassandraActivityStreamsRepository.java
incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/cassandra/CassandraSubscriptionRepository.java
incubator/streams/branches/webservice/streams-persistence/src/main/resources/cassandra.properties
incubator/streams/branches/webservice/streams-persistence/src/test/java/org/apache/streams/persistence/repository/cassandra/CassandraActiivtyStreamsRepositoryTest.java
incubator/streams/branches/webservice/streams-persistence/src/test/java/org/apache/streams/persistence/repository/cassandra/CassandraSubscriptionRepositoryTest.java
incubator/streams/branches/webservice/streams-web/src/main/java/org/apache/streams/mvc/controller/StreamsWebController.java
incubator/streams/branches/webservice/streams-web/src/test/java/org/apache/streams/mvc/integration/integration/IntegrationTest.java
Modified: incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/activitysubscriber/impl/ActivityStreamsSubscriberWarehouseImpl.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/activitysubscriber/impl/ActivityStreamsSubscriberWarehouseImpl.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/activitysubscriber/impl/ActivityStreamsSubscriberWarehouseImpl.java (original)
+++ incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/activitysubscriber/impl/ActivityStreamsSubscriberWarehouseImpl.java Tue Nov 12 16:01:41 2013
@@ -53,7 +53,7 @@ public class ActivityStreamsSubscriberWa
if (subscriber != null) {
//TODO: an activity posted in between the cql query and setting the lastUpdated field will be lost
Set<String> activities = new TreeSet<String>();
- activities.addAll(activityService.getActivitiesForTags(subscription.getTags(), subscriber.getLastUpdated()));
+ activities.addAll(activityService.getActivitiesForProviders(subscription.getFilters(), subscriber.getLastUpdated()));
subscriber.setLastUpdated(new Date());
subscriber.receive(new ArrayList<String>(activities));
}
Modified: incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsActivityRepositoryService.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsActivityRepositoryService.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsActivityRepositoryService.java (original)
+++ incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsActivityRepositoryService.java Tue Nov 12 16:01:41 2013
@@ -11,5 +11,5 @@ public interface StreamsActivityReposito
void receiveActivity(ActivityStreamsPublisher publisher, String activityJSON) throws Exception;
- List<String> getActivitiesForTags(Set<String> tags, Date lastUpdated);
+ List<String> getActivitiesForProviders(Set<String> providers, Date lastUpdated);
}
Added: incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsFiltersService.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsFiltersService.java?rev=1541121&view=auto
==============================================================================
--- incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsFiltersService.java (added)
+++ incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsFiltersService.java Tue Nov 12 16:01:41 2013
@@ -0,0 +1,10 @@
+package org.apache.streams.components.service;
+
+import org.codehaus.jackson.JsonParseException;
+
+import java.io.IOException;
+
+public interface StreamsFiltersService {
+ String getFilters(String subscriberId) throws Exception;
+ String updateFilters(String subscriberId, String tagsJson) throws Exception;
+}
Modified: incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsSubscriptionRepositoryService.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsSubscriptionRepositoryService.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsSubscriptionRepositoryService.java (original)
+++ incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/StreamsSubscriptionRepositoryService.java Tue Nov 12 16:01:41 2013
@@ -8,7 +8,7 @@ import java.util.Set;
public interface StreamsSubscriptionRepositoryService{
void saveSubscription(ActivityStreamsSubscription subscription);
List<ActivityStreamsSubscription> getAllSubscriptions();
- void updateTags(String subscriberId, Set<String> add, Set<String> remove);
+ void updateFilters(String subscriberId, Set<String> add, Set<String> remove);
ActivityStreamsSubscription getSubscriptionByUsername(String username);
ActivityStreamsSubscription getSubscriptionByInRoute(String inRoute);
}
Modified: incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/CassandraActivityService.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/CassandraActivityService.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/CassandraActivityService.java (original)
+++ incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/CassandraActivityService.java Tue Nov 12 16:01:41 2013
@@ -41,8 +41,8 @@ public class CassandraActivityService im
}
@Override
- public List<String> getActivitiesForTags(Set<String> tags, Date lastUpdated){
- List<ActivityStreamsEntry> activityObjects = activityStreamsRepository.getActivitiesForTags(tags, lastUpdated);
+ public List<String> getActivitiesForProviders(Set<String> providers, Date lastUpdated){
+ List<ActivityStreamsEntry> activityObjects = activityStreamsRepository.getActivitiesForProviders(providers, lastUpdated);
Collections.sort(activityObjects, Collections.reverseOrder());
//TODO: make the number of streams returned configurable
return getJsonList(activityObjects.subList(0, Math.min(activityObjects.size(), 10)));
Modified: incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/CassandraSubscriptionService.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/CassandraSubscriptionService.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/CassandraSubscriptionService.java (original)
+++ incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/CassandraSubscriptionService.java Tue Nov 12 16:01:41 2013
@@ -30,8 +30,8 @@ public class CassandraSubscriptionServic
}
@Override
- public void updateTags(String subscriberId, Set<String> add, Set<String> remove){
- repository.updateTags(subscriberId, add, remove);
+ public void updateFilters(String subscriberId, Set<String> add, Set<String> remove){
+ repository.updateFilters(subscriberId, add, remove);
}
@Override
Added: incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/StreamsFiltersServiceImpl.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/StreamsFiltersServiceImpl.java?rev=1541121&view=auto
==============================================================================
--- incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/StreamsFiltersServiceImpl.java (added)
+++ incubator/streams/branches/webservice/streams-components/src/main/java/org/apache/streams/components/service/impl/StreamsFiltersServiceImpl.java Tue Nov 12 16:01:41 2013
@@ -0,0 +1,50 @@
+package org.apache.streams.components.service.impl;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.streams.components.activitysubscriber.ActivityStreamsSubscriberWarehouse;
+import org.apache.streams.components.service.StreamsFiltersService;
+import org.apache.streams.components.service.StreamsSubscriptionRepositoryService;
+import org.apache.streams.persistence.model.ActivityStreamsSubscription;
+import org.codehaus.jackson.map.DeserializationConfig;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+
+@Component
+public class StreamsFiltersServiceImpl implements StreamsFiltersService {
+ private static final Log LOG = LogFactory.getLog(StreamsFiltersServiceImpl.class);
+
+ private StreamsSubscriptionRepositoryService repositoryService;
+ private ActivityStreamsSubscriberWarehouse subscriberWarehouse;
+ private ObjectMapper mapper;
+
+ @Autowired
+ public StreamsFiltersServiceImpl(StreamsSubscriptionRepositoryService repositoryService, ActivityStreamsSubscriberWarehouse subscriberWarehouse) {
+ this.repositoryService = repositoryService;
+ this.subscriberWarehouse = subscriberWarehouse;
+ this.mapper = new ObjectMapper();
+ mapper.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ }
+
+ @Override
+ public String getFilters(String subscriberId) throws Exception{
+ ActivityStreamsSubscription subscription = repositoryService.getSubscriptionByInRoute(subscriberId);
+ return mapper.writeValueAsString(subscription.getFilters());
+ }
+
+ @Override
+ public String updateFilters(String subscriberId, String tagsJson) throws Exception {
+ LOG.info("updating filters for " + subscriberId);
+
+ Map<String, List> updateFilters = (Map<String, List>) mapper.readValue(tagsJson, Map.class);
+ repositoryService.updateFilters(subscriberId, new HashSet<String>(updateFilters.get("add")), new HashSet<String>(updateFilters.get("remove")));
+ subscriberWarehouse.updateSubscriber(repositoryService.getSubscriptionByInRoute(subscriberId));
+
+ return "Filters Updated Successfully!";
+ }
+}
Modified: incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsActivityRepositoryServiceTest.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsActivityRepositoryServiceTest.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsActivityRepositoryServiceTest.java (original)
+++ incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsActivityRepositoryServiceTest.java Tue Nov 12 16:01:41 2013
@@ -84,10 +84,10 @@ public class StreamsActivityRepositorySe
activities.add(entry1);
activities.add(entry2);
- expect(activityStreamsRepository.getActivitiesForTags(tags, begin)).andReturn(activities);
+ expect(activityStreamsRepository.getActivitiesForProviders(tags, begin)).andReturn(activities);
replay(activityStreamsRepository);
- List<String> returned = activityRepositoryService.getActivitiesForTags(tags,begin);
+ List<String> returned = activityRepositoryService.getActivitiesForProviders(tags,begin);
assertThat(returned.get(0),is(equalTo(entry2Json)));
assertThat(returned.get(1),is(equalTo(entry1Json)));
Added: incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsFiltersServiceTest.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsFiltersServiceTest.java?rev=1541121&view=auto
==============================================================================
--- incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsFiltersServiceTest.java (added)
+++ incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsFiltersServiceTest.java Tue Nov 12 16:01:41 2013
@@ -0,0 +1,62 @@
+package org.apache.streams.components.service;
+
+import org.apache.streams.components.activitysubscriber.ActivityStreamsSubscriberWarehouse;
+import org.apache.streams.components.service.impl.StreamsFiltersServiceImpl;
+import org.apache.streams.persistence.model.ActivityStreamsSubscription;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.easymock.EasyMock.*;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+public class StreamsFiltersServiceTest {
+ private StreamsFiltersService filtersService;
+ private StreamsSubscriptionRepositoryService repositoryService;
+ private ActivityStreamsSubscriberWarehouse subscriberWarehouse;
+
+ @Before
+ public void setup(){
+ repositoryService = createMock(StreamsSubscriptionRepositoryService.class);
+ subscriberWarehouse = createMock(ActivityStreamsSubscriberWarehouse.class);
+ filtersService = new StreamsFiltersServiceImpl(repositoryService, subscriberWarehouse);
+ }
+
+ @Test
+ public void updateFiltersTest() throws Exception {
+ String subscriberId = "subscriberId";
+ String tagsJson = "{\"add\":[\"this\"], \"remove\":[\"that\"]}";
+ ActivityStreamsSubscription subscription = createMock(ActivityStreamsSubscription.class);
+
+ repositoryService.updateFilters(eq(subscriberId), isA(Set.class), isA(Set.class));
+ expectLastCall();
+ expect(repositoryService.getSubscriptionByInRoute(subscriberId)).andReturn(subscription);
+ subscriberWarehouse.updateSubscriber(subscription);
+ expectLastCall();
+ replay(repositoryService,subscriberWarehouse);
+
+ String returned = filtersService.updateFilters(subscriberId,tagsJson);
+
+ assertThat(returned,is(equalTo("Filters Updated Successfully!")));
+ }
+
+ @Test
+ public void getTagsTest() throws Exception {
+ String subscriberId = "subscriberId";
+ ActivityStreamsSubscription subscription = createMock(ActivityStreamsSubscription.class);
+ Set<String> filters = new HashSet<String>(Arrays.asList("tags"));
+
+ expect(repositoryService.getSubscriptionByInRoute(subscriberId)).andReturn(subscription);
+ expect(subscription.getFilters()).andReturn(filters);
+ replay(subscription,repositoryService);
+
+ String returned = filtersService.getFilters(subscriberId);
+
+ assertThat(returned, is(equalTo("[\"tags\"]")));
+ }
+}
Modified: incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsSubscriptionRepositoryServiceTest.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsSubscriptionRepositoryServiceTest.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsSubscriptionRepositoryServiceTest.java (original)
+++ incubator/streams/branches/webservice/streams-components/src/test/java/org/apache/streams/components/service/StreamsSubscriptionRepositoryServiceTest.java Tue Nov 12 16:01:41 2013
@@ -49,16 +49,16 @@ public class StreamsSubscriptionReposito
}
@Test
- public void updateTagsTest(){
+ public void updateFiltersTest(){
String subscriberId = "subscriberId";
Set<String> add = new HashSet<String>(Arrays.asList("added"));
Set<String> remove = new HashSet<String>(Arrays.asList("removed"));
- repository.updateTags(subscriberId,add,remove);
+ repository.updateFilters(subscriberId,add,remove);
expectLastCall();
replay(repository);
- subscriptionRepositoryService.updateTags(subscriberId,add,remove);
+ subscriptionRepositoryService.updateFilters(subscriberId,add,remove);
verify(repository);
}
Modified: incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/model/ActivityStreamsSubscription.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/model/ActivityStreamsSubscription.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/model/ActivityStreamsSubscription.java (original)
+++ incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/model/ActivityStreamsSubscription.java Tue Nov 12 16:01:41 2013
@@ -4,8 +4,8 @@ import java.util.Set;
public interface ActivityStreamsSubscription {
- void setTags(Set<String> tags);
- Set<String> getTags();
+ void setFilters(Set<String> tags);
+ Set<String> getFilters();
String getInRoute();
void setInRoute(String inRoute);
Modified: incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/model/cassandra/CassandraSubscription.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/model/cassandra/CassandraSubscription.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/model/cassandra/CassandraSubscription.java (original)
+++ incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/model/cassandra/CassandraSubscription.java Tue Nov 12 16:01:41 2013
@@ -9,19 +9,19 @@ import java.util.Set;
public class CassandraSubscription implements ActivityStreamsSubscription {
@JsonDeserialize(as=HashSet.class)
- private Set<String> tags;
+ private Set<String> filters;
private String username;
private String inRoute;
- public void setTags(Set<String> tags) {
+ public void setFilters(Set<String> filters) {
//TODO: it's possible that this could be null
- this.tags = tags;
+ this.filters = filters;
}
@Override
- public Set<String> getTags(){
- return tags;
+ public Set<String> getFilters(){
+ return filters;
}
@Override
Modified: incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/ActivityStreamsRepository.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/ActivityStreamsRepository.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/ActivityStreamsRepository.java (original)
+++ incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/ActivityStreamsRepository.java Tue Nov 12 16:01:41 2013
@@ -9,7 +9,7 @@ import java.util.Set;
public interface ActivityStreamsRepository {
void save(ActivityStreamsEntry entry);
- List<ActivityStreamsEntry> getActivitiesForTags(Set<String> tags, Date lastUpdated);
+ List<ActivityStreamsEntry> getActivitiesForProviders(Set<String> providers, Date lastUpdated);
void dropTable(String table);
}
Modified: incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/SubscriptionRepository.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/SubscriptionRepository.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/SubscriptionRepository.java (original)
+++ incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/SubscriptionRepository.java Tue Nov 12 16:01:41 2013
@@ -10,5 +10,5 @@ public interface SubscriptionRepository
ActivityStreamsSubscription getSubscriptionByUsername(String username);
List<ActivityStreamsSubscription> getAllSubscriptions();
void save(ActivityStreamsSubscription subscription);
- void updateTags(String subscriberId, Set<String> add, Set<String> remove);
+ void updateFilters(String subscriberId, Set<String> add, Set<String> remove);
}
Modified: incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/cassandra/CassandraActivityStreamsRepository.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/cassandra/CassandraActivityStreamsRepository.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/cassandra/CassandraActivityStreamsRepository.java (original)
+++ incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/cassandra/CassandraActivityStreamsRepository.java Tue Nov 12 16:01:41 2013
@@ -33,11 +33,10 @@ public class CassandraActivityStreamsRep
this.keyspace = keyspace;
try {
- String createKeyspaceCql = "CREATE TABLE " + configuration.getActivitystreamsColumnFamilyName() + " (" +
+ String createTableCql = "CREATE TABLE " + configuration.getActivitystreamsColumnFamilyName() + " (" +
"id text, " +
"published timestamp, " +
"verb text, " +
- "tags text, " +
"actor_displayname text, " +
"actor_id text, " +
@@ -49,19 +48,16 @@ public class CassandraActivityStreamsRep
"target_url text, " +
"provider_url text, " +
+ "provider_displayname text, " +
"object_url text, " +
"object_displayname text, " +
"object_id text, " +
"object_objecttype text, " +
- "PRIMARY KEY (id));";
- String publishedIndexCql = "CREATE INDEX ON " + configuration.getActivitystreamsColumnFamilyName() + " (published)";
- String tagsIndexCql = "CREATE INDEX ON " + configuration.getActivitystreamsColumnFamilyName() + " (tags)";
-
- keyspace.getSession().execute(createKeyspaceCql);
- keyspace.getSession().execute(tagsIndexCql);
- keyspace.getSession().execute(publishedIndexCql);
+ "PRIMARY KEY (provider_displayname, published));";
+
+ keyspace.getSession().execute(createTableCql);
} catch (AlreadyExistsException ignored) {
}
}
@@ -72,7 +68,6 @@ public class CassandraActivityStreamsRep
.value("id", entry.getId())
.value("published", entry.getPublished().getTime())
.value("verb", entry.getVerb())
- .value("tags", entry.getTags())
.value("actor_displayname", entry.getActor().getDisplayName())
.value("actor_objecttype", entry.getActor().getObjectType())
@@ -84,6 +79,7 @@ public class CassandraActivityStreamsRep
.value("target_url", entry.getTarget().getUrl())
.value("provider_url", entry.getProvider().getUrl())
+ .value("provider_displayname", entry.getProvider().getDisplayName())
.value("object_displayname", entry.getObject().getDisplayName())
.value("object_objecttype", entry.getObject().getObjectType())
@@ -94,15 +90,14 @@ public class CassandraActivityStreamsRep
}
@Override
- public List<ActivityStreamsEntry> getActivitiesForTags(Set<String> tags, Date lastUpdated) {
+ public List<ActivityStreamsEntry> getActivitiesForProviders(Set<String> providers, Date lastUpdated) {
List<ActivityStreamsEntry> results = new ArrayList<ActivityStreamsEntry>();
- for (String tag : tags) {
+ for (String providerName : providers) {
Select query = QueryBuilder.select().from(configuration.getActivitystreamsColumnFamilyName())
- .where(QueryBuilder.eq("tags", tag))
+ .where(QueryBuilder.eq("provider_displayname", providerName))
.and(QueryBuilder.gt("published", lastUpdated))
- .limit(10)
- .allowFiltering();
+ .orderBy(QueryBuilder.desc("published")).limit(100).allowFiltering();
ResultSet set = keyspace.getSession().execute(query);
//iterate through the results and create a new ActivityStreamsEntry for every result returned
@@ -129,11 +124,11 @@ public class CassandraActivityStreamsRep
object.setId(row.getString("object_id"));
provider.setUrl(row.getString("provider_url"));
+ provider.setUrl(row.getString("provider_displayname"));
entry.setPublished(row.getDate("published"));
entry.setVerb(row.getString("verb"));
entry.setId(row.getString("id"));
- entry.setTags(row.getString("tags"));
entry.setActor(actor);
entry.setTarget(target);
entry.setObject(object);
Modified: incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/cassandra/CassandraSubscriptionRepository.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/cassandra/CassandraSubscriptionRepository.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/cassandra/CassandraSubscriptionRepository.java (original)
+++ incubator/streams/branches/webservice/streams-persistence/src/main/java/org/apache/streams/persistence/repository/cassandra/CassandraSubscriptionRepository.java Tue Nov 12 16:01:41 2013
@@ -40,7 +40,7 @@ public class CassandraSubscriptionReposi
String createTableCQL = "CREATE TABLE " + configuration.getSubscriptionColumnFamilyName() + " (" +
"inroute text, " +
"username text, " +
- "tags set<text>, " +
+ "filters set<text>, " +
"PRIMARY KEY (inroute));";
String usernameIndexCQL = "CREATE INDEX ON " + configuration.getSubscriptionColumnFamilyName() + "(username)";
@@ -65,7 +65,7 @@ public class CassandraSubscriptionReposi
subscription.setInRoute(row.getString("inroute"));
subscription.setUsername(row.getString("username"));
- subscription.setTags(row.getSet("tags", String.class));
+ subscription.setFilters(row.getSet("filters", String.class));
return subscription;
}
@@ -85,7 +85,7 @@ public class CassandraSubscriptionReposi
subscription.setInRoute(row.getString("inroute"));
subscription.setUsername(row.getString("username"));
- subscription.setTags(row.getSet("tags", String.class));
+ subscription.setFilters(row.getSet("filters", String.class));
return subscription;
}
@@ -106,7 +106,7 @@ public class CassandraSubscriptionReposi
subscription.setInRoute(row.getString("inroute"));
subscription.setUsername(row.getString("username"));
- subscription.setTags(row.getSet("tags", String.class));
+ subscription.setFilters(row.getSet("filters", String.class));
results.add(subscription);
}
@@ -122,17 +122,17 @@ public class CassandraSubscriptionReposi
Insert query = QueryBuilder.insertInto(configuration.getSubscriptionColumnFamilyName())
.value("inroute", subscription.getInRoute())
.value("username", subscription.getUsername())
- .value("tags", new HashSet<String>());
+ .value("filters", new HashSet<String>());
keyspace.getSession().execute(query);
}
@Override
- public void updateTags(String subscriberId, Set<String> add, Set<String> remove) {
- LOG.info("updating tags inRoute: " + subscriberId + " add: " + add + " remove: " + remove);
+ public void updateFilters(String subscriberId, Set<String> add, Set<String> remove) {
+ LOG.info("updating filters inRoute: " + subscriberId + " add: " + add + " remove: " + remove);
Update.Where query = QueryBuilder.update(configuration.getSubscriptionColumnFamilyName())
- .with(QueryBuilder.addAll("tags", add))
- .and(QueryBuilder.removeAll("tags", remove))
+ .with(QueryBuilder.addAll("filters", add))
+ .and(QueryBuilder.removeAll("filters", remove))
.where(QueryBuilder.eq("inroute", subscriberId));
keyspace.getSession().execute(query);
}
Modified: incubator/streams/branches/webservice/streams-persistence/src/main/resources/cassandra.properties
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-persistence/src/main/resources/cassandra.properties?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-persistence/src/main/resources/cassandra.properties (original)
+++ incubator/streams/branches/webservice/streams-persistence/src/main/resources/cassandra.properties Tue Nov 12 16:01:41 2013
@@ -1,5 +1,5 @@
cassandra.keyspaceName=keyspaceB
-cassandra.activitystreamsColumnFamilyName=activitystreamsD
-cassandra.subscriptionColumnFamilyName=subscriptionsD
+cassandra.activitystreamsColumnFamilyName=activitystreamsE
+cassandra.subscriptionColumnFamilyName=subscriptionsE
cassandra.publisherColumnFamilyName=publishersE
cassandra.cassandraPort=127.0.0.1
\ No newline at end of file
Modified: incubator/streams/branches/webservice/streams-persistence/src/test/java/org/apache/streams/persistence/repository/cassandra/CassandraActiivtyStreamsRepositoryTest.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-persistence/src/test/java/org/apache/streams/persistence/repository/cassandra/CassandraActiivtyStreamsRepositoryTest.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-persistence/src/test/java/org/apache/streams/persistence/repository/cassandra/CassandraActiivtyStreamsRepositoryTest.java (original)
+++ incubator/streams/branches/webservice/streams-persistence/src/test/java/org/apache/streams/persistence/repository/cassandra/CassandraActiivtyStreamsRepositoryTest.java Tue Nov 12 16:01:41 2013
@@ -12,6 +12,7 @@ import org.junit.Test;
import java.util.Arrays;
import java.util.Date;
import java.util.HashSet;
+import java.util.List;
public class CassandraActiivtyStreamsRepositoryTest {
private CassandraActivityStreamsRepository repository;
@@ -20,7 +21,7 @@ public class CassandraActiivtyStreamsRep
public void setup(){
CassandraConfiguration configuration = new CassandraConfiguration();
configuration.setCassandraPort("127.0.0.1");
- configuration.setActivitystreamsColumnFamilyName("acitivites_Test34");
+ configuration.setActivitystreamsColumnFamilyName("acitivites_Test7");
configuration.setKeyspaceName("keyspacetest");
CassandraKeyspace keyspace = new CassandraKeyspace(configuration);
@@ -30,13 +31,13 @@ public class CassandraActiivtyStreamsRep
@Ignore
@Test
public void testNullTags(){
- repository.getActivitiesForTags(null,new Date(0));
+ repository.getActivitiesForProviders(null,new Date(0));
}
@Ignore
@Test
public void getActivitiesForTagsTest(){
- repository.getActivitiesForTags(new HashSet<String>(Arrays.asList("tags")),new Date(0));
+ List<ActivityStreamsEntry> activities = repository.getActivitiesForProviders(new HashSet<String>(Arrays.asList("provider_displayname")),new Date(0));
}
@Ignore
@@ -63,10 +64,11 @@ public class CassandraActiivtyStreamsRep
object.setId("object_id");
provider.setUrl("provider_url");
+ provider.setDisplayName("provider_displayname");
entry.setPublished(new Date());
entry.setVerb("verb");
- entry.setId("id");
+ entry.setId("newid");
entry.setTags("tags");
entry.setActor(actor);
entry.setTarget(target);
Modified: incubator/streams/branches/webservice/streams-persistence/src/test/java/org/apache/streams/persistence/repository/cassandra/CassandraSubscriptionRepositoryTest.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-persistence/src/test/java/org/apache/streams/persistence/repository/cassandra/CassandraSubscriptionRepositoryTest.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-persistence/src/test/java/org/apache/streams/persistence/repository/cassandra/CassandraSubscriptionRepositoryTest.java (original)
+++ incubator/streams/branches/webservice/streams-persistence/src/test/java/org/apache/streams/persistence/repository/cassandra/CassandraSubscriptionRepositoryTest.java Tue Nov 12 16:01:41 2013
@@ -66,6 +66,6 @@ public class CassandraSubscriptionReposi
Set<String> remove = new HashSet<String>(Arrays.asList("one","three"));
- repository.updateTags("randomID2", add, remove);
+ repository.updateFilters("randomID2", add, remove);
}
}
Modified: incubator/streams/branches/webservice/streams-web/src/main/java/org/apache/streams/mvc/controller/StreamsWebController.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-web/src/main/java/org/apache/streams/mvc/controller/StreamsWebController.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-web/src/main/java/org/apache/streams/mvc/controller/StreamsWebController.java (original)
+++ incubator/streams/branches/webservice/streams-web/src/main/java/org/apache/streams/mvc/controller/StreamsWebController.java Tue Nov 12 16:01:41 2013
@@ -23,7 +23,7 @@ public class StreamsWebController {
private StreamsSubscriberRegistrationService subscriberRegistrationService;
private StreamsActivityPublishingService activityPublishingService;
private StreamsActivityReceivingService activityReceivingService;
- private StreamsTagsService tagsService;
+ private StreamsFiltersService filtersService;
@Autowired
public StreamsWebController(
@@ -31,12 +31,12 @@ public class StreamsWebController {
StreamsSubscriberRegistrationService subscriberRegistrationService,
StreamsActivityPublishingService activityPublishingService,
StreamsActivityReceivingService activityReceivingService,
- StreamsTagsService tagsService) {
+ StreamsFiltersService filtersService) {
this.publisherRegistrationService = publisherRegistrationService;
this.subscriberRegistrationService = subscriberRegistrationService;
this.activityPublishingService = activityPublishingService;
this.activityReceivingService = activityReceivingService;
- this.tagsService = tagsService;
+ this.filtersService = filtersService;
}
/**
@@ -106,16 +106,16 @@ public class StreamsWebController {
}
/**
- * this method is the entry point for updating tags
+ * this method is the entry point for updating filters
*
* @param subscriberID the id of the subscriber
* @return a message indicating whether the update was successful
*/
- @RequestMapping(value = "/updateTags/{subscriberID}", method = RequestMethod.POST)
+ @RequestMapping(value = "/updateFilters/{subscriberID}", method = RequestMethod.POST)
@ResponseBody
public ResponseEntity<String> updateTags(@PathVariable("subscriberID") String subscriberID, @RequestBody String payload) {
try {
- return new ResponseEntity<String>(tagsService.updateTags(subscriberID, payload), HttpStatus.OK);
+ return new ResponseEntity<String>(filtersService.updateFilters(subscriberID, payload), HttpStatus.OK);
} catch (Exception e) {
log.error(e);
return new ResponseEntity<String>(e.getMessage(), HttpStatus.BAD_REQUEST);
@@ -123,16 +123,16 @@ public class StreamsWebController {
}
/**
- * this method is the entry point for updating tags
+ * this method is the entry point for getting filters
*
* @param subscriberID the id of the subscriber
- * @return an array of tags for this subscriber
+ * @return an array of filters for this subscriber
*/
- @RequestMapping(value = "/getTags/{subscriberID}", method = RequestMethod.GET)
+ @RequestMapping(value = "/getFilters/{subscriberID}", method = RequestMethod.GET)
@ResponseBody
public ResponseEntity<String> getTags(@PathVariable("subscriberID") String subscriberID) {
try {
- return new ResponseEntity<String>(tagsService.getTags(subscriberID), HttpStatus.OK);
+ return new ResponseEntity<String>(filtersService.getFilters(subscriberID), HttpStatus.OK);
} catch (Exception e) {
log.error(e);
return new ResponseEntity<String>(e.getMessage(), HttpStatus.BAD_REQUEST);
Modified: incubator/streams/branches/webservice/streams-web/src/test/java/org/apache/streams/mvc/integration/integration/IntegrationTest.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/webservice/streams-web/src/test/java/org/apache/streams/mvc/integration/integration/IntegrationTest.java?rev=1541121&r1=1541120&r2=1541121&view=diff
==============================================================================
--- incubator/streams/branches/webservice/streams-web/src/test/java/org/apache/streams/mvc/integration/integration/IntegrationTest.java (original)
+++ incubator/streams/branches/webservice/streams-web/src/test/java/org/apache/streams/mvc/integration/integration/IntegrationTest.java Tue Nov 12 16:01:41 2013
@@ -35,25 +35,19 @@ public class IntegrationTest {
private final String publisherRegistrationJson = "{\n" +
- " \"authToken\": \"token\",\n" +
- " \"@class\":\"org.apache.streams.osgi.components.activityconsumer.impl.PushActivityConsumer\", \n" +
" \"src\": \"www.providerexample.com\"\n" +
" }";
private final String subscriberRegistrationJson = "{\n" +
- " \"authToken\": \"token\",\n" +
- " \"@class\":\"org.apache.streams.osgi.components.activitysubscriber.impl.ActivityStreamsSubscriptionImpl\",\n" +
- " \"filters\": [\n" +
- " \"tags\"\n" +
- " ]\n" +
+ " \"username\": \"myUsername\"\n" +
" }";
private final String validActivity = "{\n" +
" \"id\": \"id\",\n" +
" \"verb\": \"verb\",\n" +
- " \"tags\": \"tags\",\n" +
" \"provider\": {\n" +
- " \"url\": \"www.providerexample.com\"\n" +
+ " \"url\": \"www.providerexample.com\",\n" +
+ " \"displayName\": \"myProvider\"\n" +
" },\n" +
" \"actor\": {\n" +
" \"id\": \"actorid\",\n" +