You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sf...@apache.org on 2015/04/21 17:55:23 UTC
[14/16] incubator-usergrid git commit: Fixes several wiring bugs
Fixes several wiring bugs
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/ee1a66fa
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/ee1a66fa
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/ee1a66fa
Branch: refs/heads/USERGRID-578
Commit: ee1a66fabd61f032a8748db2d5914171fd8ae59a
Parents: 1c207d7
Author: Todd Nine <tn...@apigee.com>
Authored: Mon Apr 20 19:09:46 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Mon Apr 20 19:09:46 2015 -0600
----------------------------------------------------------------------
.../usergrid/corepersistence/CoreModule.java | 7 ++-
.../corepersistence/index/IndexService.java | 6 +-
.../corepersistence/index/IndexServiceImpl.java | 13 +++--
.../rx/impl/AllEntityIdsObservableImpl.java | 5 ++
.../corepersistence/index/PublishRxtest.java | 61 ++++++++++++++++++++
.../index/impl/EsEntityIndexImpl.java | 2 +-
6 files changed, 84 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ee1a66fa/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
index f49e066..bb936b5 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
@@ -34,6 +34,8 @@ import org.apache.usergrid.corepersistence.migration.MigrationModuleVersionPlugi
import org.apache.usergrid.corepersistence.rx.impl.AllApplicationsObservable;
import org.apache.usergrid.corepersistence.rx.impl.AllApplicationsObservableImpl;
import org.apache.usergrid.corepersistence.rx.impl.AllEntitiesInSystemImpl;
+import org.apache.usergrid.corepersistence.rx.impl.AllEntityIdsObservable;
+import org.apache.usergrid.corepersistence.rx.impl.AllEntityIdsObservableImpl;
import org.apache.usergrid.corepersistence.rx.impl.AllNodesInGraphImpl;
import org.apache.usergrid.persistence.core.rx.RxSchedulerFig;
import org.apache.usergrid.persistence.core.rx.RxTaskScheduler;
@@ -102,7 +104,7 @@ public class CoreModule extends AbstractModule {
@Override
public void configureMigrationProvider() {
bind( new TypeLiteral<MigrationDataProvider<ApplicationScope>>() {} ).to(
- AllApplicationsObservable.class );
+ AllApplicationsObservableImpl.class );
}
});
// install(new MapModule()); TODO, re-enable when index module doesn't depend on queue
@@ -141,6 +143,7 @@ public class CoreModule extends AbstractModule {
plugins.addBinding().to( MigrationModuleVersionPlugin.class );
bind( AllApplicationsObservable.class ).to( AllApplicationsObservableImpl.class );
+ bind( AllEntityIdsObservable.class).to( AllEntityIdsObservableImpl.class );
/*****
@@ -148,7 +151,7 @@ public class CoreModule extends AbstractModule {
*****/
- bind(IndexService.class).to( IndexServiceImpl.class );
+ bind( IndexService.class ).to( IndexServiceImpl.class );
//bind the queue provider
bind( AsyncIndexService.class).toProvider( AsyncIndexProvider.class );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ee1a66fa/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexService.java
index 2bf073c..41fe323 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexService.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexService.java
@@ -41,9 +41,11 @@ public interface IndexService {
* @param applicationScope The scope of the entity
* @param entity The entity
*
- * @return An observable with the count of every
+ * @return An observable with the count of every batch executed to index the entity. Note that this a cold observable
+ * and must be subscribed to in order to perform the operation. This also makes no assumptions on scheduling. It is up to the caller
+ * to assign the correct scheduler to the observable
*/
- Observable<Integer> indexEntity( final ApplicationScope applicationScope, final Entity entity );
+ Observable<Long> indexEntity( final ApplicationScope applicationScope, final Entity entity );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ee1a66fa/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
index 873e2b6..29dd04a 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
@@ -20,6 +20,8 @@
package org.apache.usergrid.corepersistence.index;
+import java.util.Collection;
+
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.entities.Application;
import org.apache.usergrid.persistence.graph.Edge;
@@ -40,6 +42,7 @@ import com.google.inject.Inject;
import com.google.inject.Singleton;
import rx.Observable;
+import rx.observables.ConnectableObservable;
import static org.apache.usergrid.corepersistence.util.CpNamingUtils.generateScopeFromSource;
import static org.apache.usergrid.corepersistence.util.CpNamingUtils.generateScopeToTarget;
@@ -69,7 +72,7 @@ public class IndexServiceImpl implements IndexService {
@Override
- public Observable<Integer> indexEntity( final ApplicationScope applicationScope, final Entity entity ) {
+ public Observable<Long> indexEntity( final ApplicationScope applicationScope, final Entity entity ) {
//bootstrap the lower modules from their caches
@@ -93,7 +96,8 @@ public class IndexServiceImpl implements IndexService {
final Observable<IndexEdge> targetSizes = getIndexEdgesToTarget( gm, entityId );
- final Observable<IndexOperationMessage> observable =
+ //start the observable via publish
+ final ConnectableObservable<IndexOperationMessage> observable =
//try to send a whole batch if we can
Observable.merge( sourceEdgesToIndex, targetSizes ).buffer( indexFig.getIndexBatchSize() )
@@ -101,12 +105,11 @@ public class IndexServiceImpl implements IndexService {
.flatMap( buffer -> Observable.from( buffer ).collect( () -> ei.createBatch(),
( batch, indexEdge ) -> batch.index( indexEdge, entity ) )
//return the future from the batch execution
- .flatMap( batch -> Observable.from( batch.execute() ) ) );
+ .flatMap( batch -> Observable.from( batch.execute() ) ) ).publish();
- observable.toBlocking().last();
- return Observable.just( 0 );
+ return observable.countLong();
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ee1a66fa/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservableImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservableImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservableImpl.java
index 1f74420..f9df1f5 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservableImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservableImpl.java
@@ -21,7 +21,10 @@ package org.apache.usergrid.corepersistence.rx.impl;
+
import com.google.common.base.Optional;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
@@ -37,12 +40,14 @@ import rx.Observable;
/**
* An implementation that will provide all entityId scopes in the system
*/
+@Singleton
public class AllEntityIdsObservableImpl implements AllEntityIdsObservable {
private final GraphManagerFactory graphManagerFactory;
private final TargetIdObservable targetIdObservable;
private final EdgesObservable edgesObservable;
+ @Inject
public AllEntityIdsObservableImpl( final GraphManagerFactory graphManagerFactory,
final TargetIdObservable targetIdObservable,
final EdgesObservable edgesObservable ) {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ee1a66fa/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/PublishRxtest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/PublishRxtest.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/PublishRxtest.java
new file mode 100644
index 0000000..1eb00c2
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/PublishRxtest.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.index;
+
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.Subscription;
+import rx.observables.ConnectableObservable;
+import rx.schedulers.Schedulers;
+
+import static org.junit.Assert.assertTrue;
+
+
+/**
+ * Validates the Rx scheduler works as expected with publish
+ */
+public class PublishRxTest {
+
+ @Test
+ public void testPublish() throws InterruptedException {
+
+ final int count = 10;
+
+ final CountDownLatch latch = new CountDownLatch( count );
+
+ final Subscription connectedObservable =
+ Observable.range( 0, count ).doOnNext( integer -> latch.countDown() ).subscribeOn( Schedulers.io() )
+ .subscribe();
+
+
+ final boolean completed = latch.await( 5, TimeUnit.SECONDS );
+
+ assertTrue( "publish1 behaves as expected", completed );
+
+ final boolean completedSubscription = connectedObservable.isUnsubscribed();;
+
+ assertTrue("Subscription complete", completedSubscription);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ee1a66fa/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
index fcfba15..dcda045 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
@@ -275,7 +275,7 @@ public class EsEntityIndexImpl implements AliasedEntityIndex,VersionedData {
// delete all types, this way if we miss one it will get cleaned up
esProvider.getClient().prepareDelete( alias.getWriteAlias(), VERIFY_TYPE, tempId).get();
- logger.info( "Successfully deleted documents in index {} read {} write {} and type {} with id {}",
+ logger.info( "Successfully deleted documents in read {} write {} and type {} with id {}",
alias.getReadAlias(), alias.getWriteAlias(), VERIFY_TYPE, tempId );
return true;