You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by to...@apache.org on 2015/04/24 22:45:57 UTC
[03/14] incubator-usergrid git commit: fix observable for empty sets.
fix observable for empty sets.
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/ae1267aa
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/ae1267aa
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/ae1267aa
Branch: refs/heads/USERGRID-593
Commit: ae1267aae5465c51593668fd0781b82ac2fdfabe
Parents: 07b83e3
Author: Shawn Feldman <sf...@apache.org>
Authored: Thu Apr 23 13:38:43 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Thu Apr 23 13:38:43 2015 -0600
----------------------------------------------------------------------
.../index/impl/EsIndexBufferConsumerImpl.java | 18 +++++++++++++++---
1 file changed, 15 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ae1267aa/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java
index 0577f28..3f41eb3 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java
@@ -125,7 +125,7 @@ public class EsIndexBufferConsumerImpl implements IndexBufferConsumer {
private void startSubscription() {
- final Observable<IndexOperationMessage> observable = Observable.create( bufferProducer );
+ final Observable<IndexOperationMessage> observable = Observable.create(bufferProducer);
//buffer on our new thread with a timeout
observable.buffer( indexFig.getIndexBufferSize(), indexFig.getIndexBufferTimeout(), TimeUnit.MILLISECONDS,
@@ -189,11 +189,23 @@ public class EsIndexBufferConsumerImpl implements IndexBufferConsumer {
//now that we've processed them all, ack the futures after our last batch comes through
final Observable<IndexOperationMessage> processedIndexOperations =
- requests.last().flatMap( lastRequest -> Observable.from( batches ) );
+ requests.lastOrDefault(null).flatMap( lastRequest ->{
+ if(lastRequest!=null){
+ return Observable.from( batches ) ;
+ }else{
+ return Observable.empty();
+ }
+ });
//subscribe to the operations that generate requests on a new thread so that we can execute them quickly
//mark this as done
- return processedIndexOperations.doOnNext( processedIndexOp -> processedIndexOp.done() ).doOnError( t -> log.error( "Unable to ack futures", t ) );
+ return processedIndexOperations.doOnNext( processedIndexOp ->
+ {
+ processedIndexOp.done();
+ }
+ ).doOnError(t -> {
+ log.error("Unable to ack futures", t);
+ });
}