usergrid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sfeld...@apache.org
Subject [2/2] usergrid git commit: seperate concerns so indexing and event handling are distinct
Date Fri, 09 Oct 2015 14:54:41 GMT
seperate concerns so indexing and event handling are distinct


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

Branch: refs/heads/2.1-release
Commit: 80324de9e606ef558f129620a242ce4b0a3a25f9
Parents: 3cb0a0e
Author: Shawn Feldman <sfeldman@apache.org>
Authored: Fri Oct 9 08:47:25 2015 -0600
Committer: Shawn Feldman <sfeldman@apache.org>
Committed: Fri Oct 9 08:47:25 2015 -0600

----------------------------------------------------------------------
 .../asyncevents/AmazonAsyncEventService.java    | 95 +++++++++++---------
 1 file changed, 54 insertions(+), 41 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/usergrid/blob/80324de9/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 b384b80..4ee2094 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
@@ -23,7 +23,6 @@ package org.apache.usergrid.corepersistence.asyncevents;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 
@@ -73,7 +72,6 @@ import com.google.inject.Singleton;
 import rx.Observable;
 import rx.Subscriber;
 import rx.Subscription;
-import rx.functions.Func1;
 import rx.schedulers.Schedulers;
 
 
@@ -251,9 +249,14 @@ public class AmazonAsyncEventService implements AsyncEventService {
     }
 
 
-    private List<QueueMessage> handleMessages( final List<QueueMessage> messages
) {
+    /**
+     * calls the event handlers and returns a result with information on whether it needs
to be ack'd and whether it needs to be indexed
+     * @param messages
+     * @return
+     */
+    private List<IndexEventResult> callEventHandlers(final List<QueueMessage>
messages) {
         if (logger.isDebugEnabled()) {
-            logger.debug("handleMessages with {} message", messages.size());
+            logger.debug("callEventHandlers with {} message", messages.size());
         }
 
         Observable<IndexEventResult> masterObservable = Observable.from(messages).map(message
-> {
@@ -310,42 +313,9 @@ public class AmazonAsyncEventService implements AsyncEventService {
         final List<IndexEventResult> indexEventResults = masterObservable
             .collect(() -> new ArrayList<IndexEventResult>(), (list,indexEventResult)
-> list.add(indexEventResult) )
             .toBlocking().lastOrDefault(null);
-        //if nothing came back then return null
-        if(indexEventResults==null){
-            return null;
-        }
 
-        final IndexOperationMessage combined = new IndexOperationMessage();
 
-        //stream and filer the messages
-        List<QueueMessage> messagesToAck = indexEventResults.stream()
-            .map(indexEventResult -> {
-                //collect into the index submission
-                if (indexEventResult.getIndexOperationMessage().isPresent()) {
-                    combined.ingest(indexEventResult.getIndexOperationMessage().get());
-                }
-                return indexEventResult;
-            })
-            //filter out the ones that need to be ack'd
-            .filter(indexEventResult -> indexEventResult.getQueueMessage().isPresent())
-            .map(indexEventResult -> {
-                //record the cycle time
-                messageCycle.update(System.currentTimeMillis() - indexEventResult.getCreationTime());
-                return indexEventResult;
-            })
-                //ack after successful completion of the operation.
-            .map(result -> result.getQueueMessage().get())
-            .collect(Collectors.toList());
-
-        //send the batch
-        //TODO: should retry?
-        try {
-            indexProducer.put(combined).toBlocking().lastOrDefault(null);
-        }catch (Exception e){
-            logger.error("Failed to submit to index producer",messages,e);
-            throw e;
-        }
-        return messagesToAck;
+        return indexEventResults;
     }
 
 
@@ -582,9 +552,8 @@ public class AmazonAsyncEventService implements AsyncEventService {
                                 }
 
                                 try {
-
-                                    List<QueueMessage> messagesToAck = handleMessages(messages);
-
+                                    List<IndexEventResult> indexEventResults = callEventHandlers(messages);
+                                    List<QueueMessage> messagesToAck = submitToIndex(indexEventResults);
                                     if (messagesToAck == null || messagesToAck.size() ==
0) {
                                         logger.error("No messages came back from the queue
operation",messages);
                                         return messagesToAck;
@@ -610,6 +579,50 @@ public class AmazonAsyncEventService implements AsyncEventService {
         }
     }
 
+    /**
+     * Submit results to index and return the queue messages to be ack'd
+     * @param indexEventResults
+     * @return
+     */
+    private List<QueueMessage> submitToIndex( List<IndexEventResult> indexEventResults)
{
+        //if nothing came back then return null
+        if(indexEventResults==null){
+            return null;
+        }
+
+        final IndexOperationMessage combined = new IndexOperationMessage();
+
+        //stream and filer the messages
+        List<QueueMessage> messagesToAck = indexEventResults.stream()
+            .map(indexEventResult -> {
+                //collect into the index submission
+                if (indexEventResult.getIndexOperationMessage().isPresent()) {
+                    combined.ingest(indexEventResult.getIndexOperationMessage().get());
+                }
+                return indexEventResult;
+            })
+                //filter out the ones that need to be ack'd
+            .filter(indexEventResult -> indexEventResult.getQueueMessage().isPresent())
+            .map(indexEventResult -> {
+                //record the cycle time
+                messageCycle.update(System.currentTimeMillis() - indexEventResult.getCreationTime());
+                return indexEventResult;
+            })
+                //ack after successful completion of the operation.
+            .map(result -> result.getQueueMessage().get())
+            .collect(Collectors.toList());
+
+        //send the batch
+        //TODO: should retry?
+        try {
+            indexProducer.put(combined).toBlocking().lastOrDefault(null);
+        }catch (Exception e){
+            logger.error("Failed to submit to index producer",e);
+            throw e;
+        }
+        return messagesToAck;
+    }
+
     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 ), updatedSince
) );


Mime
View raw message