You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sn...@apache.org on 2015/10/15 16:18:22 UTC
[32/50] [abbrv] usergrid git commit: add better exception handling
add better exception handling
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/3cb0a0e0
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/3cb0a0e0
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/3cb0a0e0
Branch: refs/heads/usergrid-1007-shiro-cache
Commit: 3cb0a0e0c5a8a1f1e06d3bd71b9a2bcef333ca80
Parents: b485517
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Oct 9 08:39:27 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Oct 9 08:39:27 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 19 ++++++++++++++-----
1 file changed, 14 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/3cb0a0e0/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 0ac3860..b384b80 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
@@ -307,7 +307,9 @@ public class AmazonAsyncEventService implements AsyncEventService {
}
});
//resolve the list and return it.
- final List<IndexEventResult> indexEventResults = masterObservable.toList().toBlocking().lastOrDefault(null);
+ 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;
@@ -318,13 +320,16 @@ public class AmazonAsyncEventService implements AsyncEventService {
//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;
})
@@ -333,8 +338,13 @@ public class AmazonAsyncEventService implements AsyncEventService {
.collect(Collectors.toList());
//send the batch
- indexProducer.put(combined).toBlocking().lastOrDefault(null);
-
+ //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;
}
@@ -584,10 +594,9 @@ public class AmazonAsyncEventService implements AsyncEventService {
}
//ack each message, but only if we didn't error.
ack(messagesToAck);
- //messagesToAck.stream().forEach(message -> ack(message));
return messagesToAck;
} catch (Exception e) {
- logger.error("failed to ack messages to sqs", messages.get(0).getMessageId(), e);
+ logger.error("failed to ack messages to sqs", messages, e);
return null;
//do not rethrow so we can process all of them
}