usergrid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From g...@apache.org
Subject [1/5] incubator-usergrid git commit: Updates amazon async service wiring to wire to event builder
Date Tue, 04 Aug 2015 23:34:30 GMT
Repository: incubator-usergrid
Updated Branches:
  refs/heads/two-dot-o-dev ba5c49141 -> 9427b4ce4


Updates amazon async service wiring to wire to event builder


Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/d67c220d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/d67c220d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/d67c220d

Branch: refs/heads/two-dot-o-dev
Commit: d67c220dd7221131a5c7bbcfebcd60f78cddffb1
Parents: b9d2d35
Author: Todd Nine <tnine@apigee.com>
Authored: Tue Aug 4 12:30:27 2015 -0600
Committer: Todd Nine <tnine@apigee.com>
Committed: Tue Aug 4 12:30:27 2015 -0600

----------------------------------------------------------------------
 .../asyncevents/AmazonAsyncEventService.java    | 132 +++++++++----------
 .../asyncevents/AsyncIndexProvider.java         |  11 +-
 .../asyncevents/EventBuilder.java               |  11 +-
 .../asyncevents/EventBuilderImpl.java           |  12 +-
 .../asyncevents/InMemoryAsyncEventService.java  |  14 +-
 .../asyncevents/model/EntityIndexEvent.java     |  16 ++-
 .../index/AmazonAsyncEventServiceTest.java      |   7 +-
 7 files changed, 107 insertions(+), 96 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d67c220d/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
index ec43ab7..0429af3 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
@@ -27,14 +27,15 @@ import java.util.concurrent.atomic.AtomicLong;
 
 import com.codahale.metrics.Histogram;
 import com.google.common.base.Preconditions;
-import org.apache.usergrid.corepersistence.CpEntityManager;
+
 import org.apache.usergrid.corepersistence.asyncevents.model.*;
 import org.apache.usergrid.corepersistence.index.*;
 import org.apache.usergrid.corepersistence.rx.impl.EdgeScope;
 import org.apache.usergrid.persistence.index.EntityIndex;
 import org.apache.usergrid.persistence.index.EntityIndexFactory;
 import org.apache.usergrid.persistence.index.IndexLocationStrategy;
-import org.apache.usergrid.utils.UUIDUtils;
+import org.apache.usergrid.persistence.index.impl.IndexOperationMessage;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -78,15 +79,19 @@ public class AmazonAsyncEventService implements AsyncEventService {
     private final QueueManager queue;
     private final QueueScope queueScope;
     private final IndexProcessorFig indexProcessorFig;
-    private final IndexService indexService;
     private final EntityCollectionManagerFactory entityCollectionManagerFactory;
     private final IndexLocationStrategyFactory indexLocationStrategyFactory;
     private final EntityIndexFactory entityIndexFactory;
+    private final EventBuilder eventBuilder;
+    private final RxTaskScheduler rxTaskScheduler;
 
     private final Timer readTimer;
     private final Timer writeTimer;
     private final Timer ackTimer;
 
+    /**
+     * This mutex is used to start/stop workers to ensure we're not concurrently modifying
our subscriptions
+     */
     private final Object mutex = new Object();
 
     private final Counter indexErrorCounter;
@@ -99,19 +104,17 @@ public class AmazonAsyncEventService implements AsyncEventService {
 
 
     @Inject
-    public AmazonAsyncEventService(final QueueManagerFactory queueManagerFactory,
-                                   final IndexProcessorFig indexProcessorFig,
-                                   final MetricsFactory metricsFactory,
-                                   final IndexService indexService,
-                                   final EntityCollectionManagerFactory entityCollectionManagerFactory,
-                                   final IndexLocationStrategyFactory indexLocationStrategyFactory,
-                                   final EntityIndexFactory entityIndexFactory
-    ) {
-
-        this.indexService = indexService;
+    public AmazonAsyncEventService( final QueueManagerFactory queueManagerFactory, final
IndexProcessorFig indexProcessorFig,
+                                    final MetricsFactory metricsFactory,  final EntityCollectionManagerFactory
entityCollectionManagerFactory,
+                                    final IndexLocationStrategyFactory indexLocationStrategyFactory,
final EntityIndexFactory entityIndexFactory,
+                                    final EventBuilder eventBuilder,
+                                    final RxTaskScheduler rxTaskScheduler ) {
+
         this.entityCollectionManagerFactory = entityCollectionManagerFactory;
         this.indexLocationStrategyFactory = indexLocationStrategyFactory;
         this.entityIndexFactory = entityIndexFactory;
+        this.eventBuilder = eventBuilder;
+        this.rxTaskScheduler = rxTaskScheduler;
 
         this.queueScope = new QueueScopeImpl(QUEUE_NAME, QueueScope.RegionImplementation.ALLREGIONS);
         this.queue = queueManagerFactory.getQueueManager(queueScope);
@@ -144,7 +147,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
 
         try {
             //signal to SQS
-            this.queue.sendMessage(operation);
+            this.queue.sendMessage( operation );
         } catch (IOException e) {
             throw new RuntimeException("Unable to queue message", e);
         } finally {
@@ -157,7 +160,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
 
         try {
             //signal to SQS
-            this.queue.sendMessages(operations);
+            this.queue.sendMessages( operations );
         } catch (IOException e) {
             throw new RuntimeException("Unable to queue message", e);
         } finally {
@@ -189,33 +192,6 @@ public class AmazonAsyncEventService implements AsyncEventService {
     /**
      * Ack message in SQS
      */
-    public void ack(final List<QueueMessage> messages) {
-
-        final Timer.Context timer = this.ackTimer.time();
-
-        try{
-            // no op
-            if (messages.size() == 0) {
-                return;
-            }
-            queue.commitMessages(messages);
-
-            //decrement our in-flight counter
-            inFlight.addAndGet(-1 * messages.size());
-
-        }catch(Exception e){
-            throw new RuntimeException("Unable to ack messages", e);
-        }
-        finally {
-            timer.stop();
-        }
-
-
-    }
-
-    /**
-     * Ack message in SQS
-     */
     public void ack(final QueueMessage message) {
 
         final Timer.Context timer = this.ackTimer.time();
@@ -290,7 +266,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
     public void queueEntityIndexUpdate(final ApplicationScope applicationScope,
                                        final Entity entity) {
 
-        offer(new EntityIndexEvent(new EntityIdScope(applicationScope, entity.getId())));
+        offer(new EntityIndexEvent(new EntityIdScope(applicationScope, entity.getId()), 0));
     }
 
 
@@ -298,7 +274,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
 
         Preconditions.checkNotNull(message, "Queue Message cannot be null for handleEntityIndexUpdate");
 
-        final AsyncEvent event = (AsyncEvent) message.getBody();
+        final EntityIndexEvent event = (EntityIndexEvent) message.getBody();
 
         Preconditions.checkNotNull(message, "QueueMessage Body cannot be null for handleEntityIndexUpdate");
         Preconditions.checkArgument(event.getEventType() == AsyncEvent.EventType.ENTITY_INDEX,
String.format("Event Type for handleEntityIndexUpdate must be ENTITY_INDEX, got %s", event.getEventType()));
@@ -307,13 +283,14 @@ public class AmazonAsyncEventService implements AsyncEventService {
         //only process the same version, otherwise ignore
         final EntityIdScope entityIdScope = event.getEntityIdScope();
         final ApplicationScope applicationScope = entityIdScope.getApplicationScope();
+        final Id entityId = entityIdScope.getId();
+        final long updatedAfter = event.getUpdatedAfter();
+
+        final EntityIndexOperation entityIndexOperation = new EntityIndexOperation( applicationScope,
entityId, updatedAfter);
 
-        final EntityCollectionManager ecm = entityCollectionManagerFactory.createCollectionManager(applicationScope);
+        final Observable<IndexOperationMessage> observable = eventBuilder.buildEntityIndex(
entityIndexOperation );
 
-        ecm.load(entityIdScope.getId())
-                .first()
-                .flatMap(entity -> indexService.indexEntity(applicationScope, entity))
-                .doOnNext(ignore -> ack(message)).subscribe();
+        subscibeAndAck( observable, message );
     }
 
 
@@ -324,7 +301,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
 
         EdgeIndexEvent operation = new EdgeIndexEvent(applicationScope, entity.getId(), newEdge);
 
-        offer(operation);
+        offer( operation );
     }
 
     public void handleEdgeIndex(final QueueMessage message) {
@@ -339,18 +316,21 @@ public class AmazonAsyncEventService implements AsyncEventService {
         final ApplicationScope applicationScope = event.getApplicationScope();
         final Edge edge = event.getEdge();
 
-        final EntityCollectionManager ecm = entityCollectionManagerFactory.createCollectionManager(applicationScope);
 
-        ecm.load(event.getEntityId())
-                .flatMap(entity -> indexService.indexEdge(applicationScope, entity, edge))
-                .doOnNext(ignore -> ack(message)).subscribe();
+
+        final EntityCollectionManager ecm = entityCollectionManagerFactory.createCollectionManager(
applicationScope );
+
+        final Observable<IndexOperationMessage> edgeIndexObservable = ecm.load(event.getEntityId()).flatMap(
entity -> eventBuilder.buildNewEdge(
+            applicationScope, entity, edge ) );
+
+        subscibeAndAck( edgeIndexObservable, message );
     }
 
     @Override
     public void queueDeleteEdge(final ApplicationScope applicationScope,
                                 final Edge edge) {
 
-        offer(new EdgeDeleteEvent(applicationScope, edge));
+        offer( new EdgeDeleteEvent( applicationScope, edge ) );
     }
 
     public void handleEdgeDelete(final QueueMessage message) {
@@ -367,15 +347,16 @@ public class AmazonAsyncEventService implements AsyncEventService {
 
         if (logger.isDebugEnabled()) logger.debug("Deleting in app scope {} with edge {}",
applicationScope, edge);
 
-        indexService.deleteIndexEdge(applicationScope, edge)
-                .doOnNext(ignore -> ack(message)).subscribe();
+        final Observable<IndexOperationMessage> observable = eventBuilder.buildDeleteEdge(
applicationScope, edge );
+
+        subscibeAndAck( observable, message );
     }
 
 
     @Override
     public void queueEntityDelete(final ApplicationScope applicationScope, final Id entityId)
{
 
-        offer(new EntityDeleteEvent(new EntityIdScope(applicationScope, entityId)));
+        offer( new EntityDeleteEvent( new EntityIdScope( applicationScope, entityId ) ) );
     }
 
     public void handleEntityDelete(final QueueMessage message) {
@@ -384,7 +365,8 @@ public class AmazonAsyncEventService implements AsyncEventService {
 
         final AsyncEvent event = (AsyncEvent) message.getBody();
         Preconditions.checkNotNull(message, "QueueMessage Body cannot be null for handleEntityDelete");
-        Preconditions.checkArgument(event.getEventType() == AsyncEvent.EventType.ENTITY_DELETE,
String.format("Event Type for handleEntityDelete must be ENTITY_DELETE, got %s", event.getEventType()));
+        Preconditions.checkArgument( event.getEventType() == AsyncEvent.EventType.ENTITY_DELETE,
+            String.format( "Event Type for handleEntityDelete must be ENTITY_DELETE, got
%s", event.getEventType() ) );
 
         final ApplicationScope applicationScope = event.getApplicationScope();
         final Id entityId = event.getEntityId();
@@ -392,10 +374,16 @@ public class AmazonAsyncEventService implements AsyncEventService {
         if (logger.isDebugEnabled())
             logger.debug("Deleting entity id from index in app scope {} with entityId {}",
applicationScope, entityId);
 
-        ack(message);
+        ack( message );
+
+        final EventBuilderImpl.EntityDeleteResults
+            entityDeleteResults = eventBuilder.buildEntityDelete( applicationScope, entityId
);
+
+
+        final Observable merged = Observable.merge( entityDeleteResults.getEntitiesCompacted(),
+            entityDeleteResults.getIndexObservable() );
 
-        indexService.deleteEntityIndexes(applicationScope, entityId, UUIDUtils.maxTimeUUID(Long.MAX_VALUE))
-                .doOnNext(ignore -> ack(message)).subscribe();
+        subscibeAndAck( merged, message );
     }
 
 
@@ -407,9 +395,9 @@ public class AmazonAsyncEventService implements AsyncEventService {
         Preconditions.checkArgument(event.getEventType() == AsyncEvent.EventType.APPLICATION_INDEX,
String.format("Event Type for handleInitializeApplicationIndex must be APPLICATION_INDEX,
got %s", event.getEventType()));
 
         final IndexLocationStrategy indexLocationStrategy = event.getIndexLocationStrategy();
-        final EntityIndex index = entityIndexFactory.createEntityIndex(indexLocationStrategy);
+        final EntityIndex index = entityIndexFactory.createEntityIndex( indexLocationStrategy
);
         index.initialize();
-        ack(message);
+        ack( message );
     }
 
     /**
@@ -494,7 +482,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
 
     public void index(final ApplicationScope applicationScope, final Id id, final long updatedSince)
{
         //change to id scope to avoid serialization issues
-        offer(new EntityIndexEvent(new EntityIdScope(applicationScope, id)));
+        offer(new EntityIndexEvent(new EntityIdScope(applicationScope, id), updatedSince));
     }
 
     public void indexBatch(final List<EdgeScope> edges, final long updatedSince) {
@@ -502,8 +490,18 @@ public class AmazonAsyncEventService implements AsyncEventService {
         List batch = new ArrayList<EdgeScope>();
         for ( EdgeScope e : edges){
             //change to id scope to avoid serialization issues
-            batch.add(new EntityIndexEvent(new EntityIdScope(e.getApplicationScope(), e.getEdge().getTargetNode())));
+            batch.add(new EntityIndexEvent(new EntityIdScope(e.getApplicationScope(), e.getEdge().getTargetNode()),
updatedSince));
         }
-        offerBatch(batch);
+        offerBatch( batch );
+    }
+
+
+    /**
+     * Subscribes to the observable and acks the message via SQS on completion
+     * @param observable
+     * @param message
+     */
+    private void subscibeAndAck( final Observable<?> observable, final QueueMessage
message ){
+       observable.doOnCompleted( ()-> ack(message)  ).subscribeOn( rxTaskScheduler.getAsyncIOScheduler()
).subscribe();
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d67c220d/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
index 0a58369..0e773cf 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
@@ -44,7 +44,6 @@ public class AsyncIndexProvider implements Provider<AsyncEventService>
{
 
     private final QueueManagerFactory queueManagerFactory;
     private final MetricsFactory metricsFactory;
-    private final IndexService indexService;
     private final RxTaskScheduler rxTaskScheduler;
     private final EntityCollectionManagerFactory entityCollectionManagerFactory;
     private final EventBuilder eventBuilder;
@@ -58,7 +57,6 @@ public class AsyncIndexProvider implements Provider<AsyncEventService>
{
     public AsyncIndexProvider(final IndexProcessorFig indexProcessorFig,
                               final QueueManagerFactory queueManagerFactory,
                               final MetricsFactory metricsFactory,
-                              final IndexService indexService,
                               final RxTaskScheduler rxTaskScheduler,
                               final EntityCollectionManagerFactory entityCollectionManagerFactory,
                               final EventBuilder eventBuilder,
@@ -68,7 +66,6 @@ public class AsyncIndexProvider implements Provider<AsyncEventService>
{
         this.indexProcessorFig = indexProcessorFig;
         this.queueManagerFactory = queueManagerFactory;
         this.metricsFactory = metricsFactory;
-        this.indexService = indexService;
         this.rxTaskScheduler = rxTaskScheduler;
         this.entityCollectionManagerFactory = entityCollectionManagerFactory;
         this.eventBuilder = eventBuilder;
@@ -97,11 +94,11 @@ public class AsyncIndexProvider implements Provider<AsyncEventService>
{
             case LOCAL:
                 return new InMemoryAsyncEventService(eventBuilder, rxTaskScheduler, indexProcessorFig.resolveSynchronously());
             case SQS:
-                return new AmazonAsyncEventService(queueManagerFactory, indexProcessorFig,
metricsFactory, indexService,
-                    entityCollectionManagerFactory, indexLocationStrategyFactory,entityIndexFactory
);
+                return new AmazonAsyncEventService(queueManagerFactory, indexProcessorFig,
metricsFactory,
+                    entityCollectionManagerFactory, indexLocationStrategyFactory,entityIndexFactory,
eventBuilder, rxTaskScheduler );
             case SNS:
-                return new AmazonAsyncEventService(queueManagerFactory, indexProcessorFig,
metricsFactory, indexService,
-                    entityCollectionManagerFactory, indexLocationStrategyFactory,entityIndexFactory
);
+                return new AmazonAsyncEventService(queueManagerFactory, indexProcessorFig,
metricsFactory,
+                    entityCollectionManagerFactory, indexLocationStrategyFactory,entityIndexFactory,
eventBuilder, rxTaskScheduler );
             default:
                 throw new IllegalArgumentException("Configuration value of " + getErrorValues()
+ " are allowed");
         }

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d67c220d/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilder.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilder.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilder.java
index f9f157e..d246e2f 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilder.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilder.java
@@ -37,13 +37,14 @@ import rx.Observable;
  * Interface for constructing an observable stream to perform asynchonous events
  */
 public interface EventBuilder {
+
     /**
      * Return the cold observable of entity index update operations
      * @param applicationScope
      * @param entity
      * @return
      */
-    Observable<IndexOperationMessage> queueEntityIndexUpdate( ApplicationScope applicationScope,
Entity entity );
+    Observable<IndexOperationMessage> buildEntityIndexUpdate( ApplicationScope applicationScope,
Entity entity );
 
     /**
      * Return the cold observable of the new edge operation
@@ -52,7 +53,7 @@ public interface EventBuilder {
      * @param newEdge
      * @return
      */
-    Observable<IndexOperationMessage> queueNewEdge( ApplicationScope applicationScope,
Entity entity, Edge newEdge );
+    Observable<IndexOperationMessage> buildNewEdge( ApplicationScope applicationScope,
Entity entity, Edge newEdge );
 
     /**
      * Return the cold observable of the deleted edge operations
@@ -60,7 +61,7 @@ public interface EventBuilder {
      * @param edge
      * @return
      */
-    Observable<IndexOperationMessage> queueDeleteEdge( ApplicationScope applicationScope,
Edge edge );
+    Observable<IndexOperationMessage> buildDeleteEdge( ApplicationScope applicationScope,
Edge edge );
 
     /**
      * Return a ben with 2 obervable streams for entity delete.
@@ -68,14 +69,14 @@ public interface EventBuilder {
      * @param entityId
      * @return
      */
-    EventBuilderImpl.EntityDeleteResults queueEntityDelete( ApplicationScope applicationScope,
Id entityId );
+    EventBuilderImpl.EntityDeleteResults buildEntityDelete( ApplicationScope applicationScope,
Id entityId );
 
     /**
      * Re-index an entity in the scope provided
      * @param entityIndexOperation
      * @return
      */
-    Observable<IndexOperationMessage> index( EntityIndexOperation entityIndexOperation
);
+    Observable<IndexOperationMessage> buildEntityIndex( EntityIndexOperation entityIndexOperation
);
 
     /**
      * A bean to hold both our observables so the caller can choose the subscription mechanism.
 Note that

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d67c220d/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
index e4cd4b5..46cec2e 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
@@ -73,7 +73,7 @@ public class EventBuilderImpl implements EventBuilder {
 
 
     @Override
-    public Observable<IndexOperationMessage> queueEntityIndexUpdate( final ApplicationScope
applicationScope,
+    public Observable<IndexOperationMessage> buildEntityIndexUpdate( final ApplicationScope
applicationScope,
                                                                      final Entity entity
) {
 
         //process the entity immediately
@@ -89,7 +89,7 @@ public class EventBuilderImpl implements EventBuilder {
 
 
     @Override
-    public Observable<IndexOperationMessage> queueNewEdge( final ApplicationScope applicationScope,
final Entity entity,
+    public Observable<IndexOperationMessage> buildNewEdge( final ApplicationScope applicationScope,
final Entity entity,
                                                            final Edge newEdge ) {
 
         log.debug( "Indexing  in app scope {} with entity {} and new edge {}",
@@ -103,8 +103,8 @@ public class EventBuilderImpl implements EventBuilder {
 
 
     @Override
-    public Observable<IndexOperationMessage> queueDeleteEdge( final ApplicationScope
applicationScope,
-                                                              final Edge edge ) {
+    public Observable<IndexOperationMessage> buildDeleteEdge( final ApplicationScope
applicationScope, final Edge
+        edge ) {
         log.debug( "Deleting in app scope {} with edge {} }", applicationScope, edge );
 
         final Observable<IndexOperationMessage> edgeObservable =
@@ -121,7 +121,7 @@ public class EventBuilderImpl implements EventBuilder {
     //it'll need to be pushed up higher so we can do the marking that isn't async or does
it not matter?
 
     @Override
-    public EntityDeleteResults queueEntityDelete( final ApplicationScope applicationScope,
final Id entityId ) {
+    public EntityDeleteResults buildEntityDelete( final ApplicationScope applicationScope,
final Id entityId ) {
         log.debug( "Deleting entity id from index in app scope {} with entityId {} }", applicationScope,
entityId );
 
         final EntityCollectionManager ecm = entityCollectionManagerFactory.createCollectionManager(
applicationScope );
@@ -163,7 +163,7 @@ public class EventBuilderImpl implements EventBuilder {
 
 
     @Override
-    public Observable<IndexOperationMessage> index( final EntityIndexOperation entityIndexOperation
) {
+    public Observable<IndexOperationMessage> buildEntityIndex( final EntityIndexOperation
entityIndexOperation ) {
 
         final ApplicationScope applicationScope = entityIndexOperation.getApplicationScope();
 

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d67c220d/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
index adb4a90..830033d 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
@@ -20,8 +20,6 @@
 package org.apache.usergrid.corepersistence.asyncevents;
 
 
-import com.amazonaws.services.opsworks.model.App;
-import org.apache.usergrid.persistence.index.IndexLocationStrategy;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -77,19 +75,19 @@ public class InMemoryAsyncEventService implements AsyncEventService {
         //only process the same version, otherwise ignore
 
 
-        run( eventBuilder.queueEntityIndexUpdate(applicationScope, entity) );
+        run( eventBuilder.buildEntityIndexUpdate( applicationScope, entity ) );
     }
 
 
     @Override
     public void queueNewEdge( final ApplicationScope applicationScope, final Entity entity,
final Edge newEdge ) {
-        run( eventBuilder.queueNewEdge(applicationScope, entity, newEdge) );
+        run( eventBuilder.buildNewEdge( applicationScope, entity, newEdge ) );
     }
 
 
     @Override
     public void queueDeleteEdge( final ApplicationScope applicationScope, final Edge edge
) {
-        run( eventBuilder.queueDeleteEdge(applicationScope, edge) );
+        run( eventBuilder.buildDeleteEdge( applicationScope, edge ) );
     }
 
 
@@ -97,7 +95,7 @@ public class InMemoryAsyncEventService implements AsyncEventService {
     public void queueEntityDelete( final ApplicationScope applicationScope, final Id entityId
) {
 
         final EventBuilderImpl.EntityDeleteResults results =
-            eventBuilder.queueEntityDelete( applicationScope, entityId );
+            eventBuilder.buildEntityDelete( applicationScope, entityId );
 
         run( results.getIndexObservable() );
         run( results.getEntitiesCompacted() );
@@ -107,7 +105,7 @@ public class InMemoryAsyncEventService implements AsyncEventService {
     public void index( final ApplicationScope applicationScope, final Id id, final long updatedSince
) {
         final EntityIndexOperation entityIndexOperation = new EntityIndexOperation( applicationScope,
id, updatedSince );
 
-        run(eventBuilder.index( entityIndexOperation ));
+        run(eventBuilder.buildEntityIndex( entityIndexOperation ));
     }
 
     public void indexBatch(final List<EdgeScope> edges, final long updatedSince) {
@@ -115,7 +113,7 @@ public class InMemoryAsyncEventService implements AsyncEventService {
             final EntityIndexOperation entityIndexOperation = new EntityIndexOperation(e.getApplicationScope(),
                 e.getEdge().getTargetNode(), updatedSince);
 
-            run(eventBuilder.index (entityIndexOperation));
+            run(eventBuilder.buildEntityIndex( entityIndexOperation ));
         }
 
     }

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d67c220d/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/model/EntityIndexEvent.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/model/EntityIndexEvent.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/model/EntityIndexEvent.java
index 7b79987..81961a0 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/model/EntityIndexEvent.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/model/EntityIndexEvent.java
@@ -27,10 +27,24 @@ import org.apache.usergrid.persistence.collection.serialization.impl.migration.E
  */
 @JsonDeserialize(as = AsyncEvent.class)
 public final class EntityIndexEvent extends AsyncEvent {
+
+    private long updatedAfter;
+
     public EntityIndexEvent() {
     }
 
-    public EntityIndexEvent(EntityIdScope entityIdScope) {
+    public EntityIndexEvent(EntityIdScope entityIdScope, final long updatedAfter ) {
         super(EventType.ENTITY_INDEX, entityIdScope);
+        this.updatedAfter = updatedAfter;
+    }
+
+
+    public long getUpdatedAfter() {
+        return updatedAfter;
+    }
+
+
+    public void setUpdatedAfter( long updatedAfter ) {
+        this.updatedAfter = updatedAfter;
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d67c220d/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/AmazonAsyncEventServiceTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/AmazonAsyncEventServiceTest.java
b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/AmazonAsyncEventServiceTest.java
index 9cf896c..d37701b 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/AmazonAsyncEventServiceTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/AmazonAsyncEventServiceTest.java
@@ -20,6 +20,7 @@
 package org.apache.usergrid.corepersistence.index;
 
 
+import org.apache.usergrid.corepersistence.asyncevents.EventBuilder;
 import org.apache.usergrid.persistence.index.EntityIndexFactory;
 import org.junit.Rule;
 import org.junit.runner.RunWith;
@@ -71,6 +72,9 @@ public class AmazonAsyncEventServiceTest extends AsyncIndexServiceTest {
     @Inject
     public RxTaskScheduler rxTaskScheduler;
 
+    @Inject
+    public EventBuilder eventBuilder;
+
 
     @Inject
     public IndexLocationStrategyFactory indexLocationStrategyFactory;
@@ -81,8 +85,7 @@ public class AmazonAsyncEventServiceTest extends AsyncIndexServiceTest {
 
     @Override
     protected AsyncEventService getAsyncEventService() {
-        return  new AmazonAsyncEventService( queueManagerFactory, indexProcessorFig, metricsFactory,
indexService,
-                    entityCollectionManagerFactory, indexLocationStrategyFactory, entityIndexFactory
);
+        return  new AmazonAsyncEventService( queueManagerFactory, indexProcessorFig, metricsFactory,
 entityCollectionManagerFactory, indexLocationStrategyFactory, entityIndexFactory, eventBuilder,
rxTaskScheduler );
     }
 
 


Mime
View raw message