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/03/20 17:24:42 UTC
[1/8] incubator-usergrid git commit: First pass at upgrading to java
8 and latest RX java
Repository: incubator-usergrid
Updated Branches:
refs/heads/USERGRID-486 72ec19d56 -> dcf469378
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
index 26d06ad..ef258f4 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
@@ -188,7 +188,7 @@ public class GraphManagerImpl implements GraphManager {
final Timer.Context timer = writeEdgeTimer.time();
final Meter meter = writeEdgeMeter;
- return Observable.from( markedEdge ).map( new Func1<MarkedEdge, Edge>() {
+ return Observable.just( markedEdge ).map( new Func1<MarkedEdge, Edge>() {
@Override
public Edge call( final MarkedEdge edge ) {
@@ -234,7 +234,7 @@ public class GraphManagerImpl implements GraphManager {
final Timer.Context timer = deleteEdgeTimer.time();
final Meter meter = deleteEdgeMeter;
- return Observable.from(markedEdge).map(new Func1<MarkedEdge, Edge>() {
+ return Observable.just(markedEdge).map(new Func1<MarkedEdge, Edge>() {
@Override
public Edge call(final MarkedEdge edge) {
@@ -281,7 +281,7 @@ public class GraphManagerImpl implements GraphManager {
public Observable<Id> deleteNode( final Id node, final long timestamp ) {
final Timer.Context timer = deleteNodeTimer.time();
final Meter meter = deleteNodeMeter;
- return Observable.from( node ).map( new Func1<Id, Id>() {
+ return Observable.just( node ).map( new Func1<Id, Id>() {
@Override
public Id call( final Id id ) {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java
index ab141f7..bfaeaaa 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java
@@ -176,6 +176,8 @@ public class EdgeMetaRepairImpl implements EdgeMetaRepair {
* Sum up the total number of edges we had, then execute the mutation if we have
* anything to do
*/
+
+
return MathObservable.sumInteger( Observable.merge( checks ) )
.doOnNext( new Action1<Integer>() {
@Override
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
index e8c224e..6236a16 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
@@ -103,7 +103,7 @@ public class NodeDeleteListenerImpl implements NodeDeleteListener {
public Observable<Integer> receive( final ApplicationScope scope, final Id node, final UUID timestamp ) {
- return Observable.from( node )
+ return Observable.just( node )
//delete source and targets in parallel and merge them into a single observable
.flatMap( new Func1<Id, Observable<Integer>>() {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/migration/EdgeDataMigrationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/migration/EdgeDataMigrationImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/migration/EdgeDataMigrationImpl.java
index 2d9b47f..ecb9a9b 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/migration/EdgeDataMigrationImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/migration/EdgeDataMigrationImpl.java
@@ -40,6 +40,7 @@ import org.slf4j.LoggerFactory;
import rx.Observable;
import rx.functions.Action1;
import rx.functions.Func1;
+import rx.schedulers.Schedulers;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
@@ -75,61 +76,49 @@ public class EdgeDataMigrationImpl implements DataMigration<GraphNode> {
}
-
-
@Override
- public int migrate( final int currentVersion, final MigrationDataProvider<GraphNode> migrationDataProvider,
- final ProgressObserver observer ) {
+ public int migrate( final int currentVersion, final MigrationDataProvider<GraphNode> migrationDataProvider,
+ final ProgressObserver observer ) {
final AtomicLong counter = new AtomicLong();
- final MigrationRelationship<EdgeMetadataSerialization>
- migration = allVersions.getMigrationRelationship( currentVersion );
-
- final Observable<List<Edge>> observable = migrationDataProvider.getData().flatMap( new Func1<GraphNode,
- Observable<List<Edge>>>() {
- @Override
- public Observable<List<Edge>> call( final GraphNode graphNode ) {
- final GraphManager gm = graphManagerFactory.createEdgeManager( graphNode.applicationScope );
-
- //get edges from the source
- return edgesFromSourceObservable.edgesFromSource( gm, graphNode.entryNode ).buffer( 1000 ).parallel( new Func1<Observable<List<Edge>>, Observable<List<Edge>>>() {
- @Override
- public Observable<List<Edge>> call( final Observable<List<Edge>> listObservable ) {
- return listObservable.doOnNext( new Action1<List<Edge>>() {
- @Override
- public void call( List<Edge> edges ) {
- final MutationBatch batch = keyspace.prepareMutationBatch();
-
- for ( Edge edge : edges ) {
- logger.info( "Migrating meta for edge {}", edge );
- final MutationBatch edgeBatch =
- migration.to.writeEdge( graphNode.applicationScope, edge );
- batch.mergeShallow( edgeBatch );
- }
-
- try {
- batch.execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( "Unable to perform migration", e );
- }
-
- //update the observer so the admin can see it
- final long newCount = counter.addAndGet( edges.size() );
-
- observer.update( migration.to.getImplementationVersion(),
- String.format( "Currently running. Rewritten %d edge types",
- newCount ) );
- }
- } );
- } } );
- }} );
-
- observable.longCount().toBlocking().last();
+ final MigrationRelationship<EdgeMetadataSerialization> migration =
+ allVersions.getMigrationRelationship( currentVersion );
- return migration.to.getImplementationVersion();
+ final Observable<List<Edge>> observable = migrationDataProvider.getData().flatMap( graphNode -> {
+ final GraphManager gm = graphManagerFactory.createEdgeManager( graphNode.applicationScope );
+
+ //get edges from the source
+ return edgesFromSourceObservable.edgesFromSource( gm, graphNode.entryNode ).buffer( 1000 )
+ .doOnNext( edges -> {
+ final MutationBatch batch = keyspace.prepareMutationBatch();
+
+ for ( Edge edge : edges ) {
+ logger.info( "Migrating meta for edge {}", edge );
+ final MutationBatch edgeBatch =
+ migration.to.writeEdge( graphNode.applicationScope, edge );
+ batch.mergeShallow( edgeBatch );
+ }
+ try {
+ batch.execute();
+ }
+ catch ( ConnectionException e ) {
+ throw new RuntimeException( "Unable to perform migration", e );
+ }
+
+ //update the observer so the admin can see it
+ final long newCount = counter.addAndGet( edges.size() );
+
+ observer.update( migration.to.getImplementationVersion(), String
+ .format( "Currently running. Rewritten %d edge types",
+ newCount ) );
+ } ).subscribeOn( Schedulers.io() );
+ }, 10 );
+
+ observable.countLong().toBlocking().last();
+
+ return migration.to.getImplementationVersion();
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerShardConsistencyIT.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerShardConsistencyIT.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerShardConsistencyIT.java
index 6d30d22..3bbf3e4 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerShardConsistencyIT.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerShardConsistencyIT.java
@@ -516,7 +516,7 @@ public class GraphManagerShardConsistencyIT {
}
} )
- .longCount().toBlocking().last();
+ .countLong().toBlocking().last();
// if(returnedEdgeCount != count[0]-duplicate[0]){
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/SimpleTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/SimpleTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/SimpleTest.java
index 0a27a6b..7b3fafd 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/SimpleTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/SimpleTest.java
@@ -63,23 +63,23 @@ public class SimpleTest {
Edge testTargetEdge = createEdge( sourceId1, "test", targetId1, System.currentTimeMillis() );
- gm.writeEdge( testTargetEdge ).toBlockingObservable().singleOrDefault( null );
+ gm.writeEdge( testTargetEdge ).toBlocking().singleOrDefault( null );
Edge testTarget2Edge = createEdge( sourceId2, "edgeType1", targetId1, System.currentTimeMillis() );
- gm.writeEdge( testTarget2Edge ).toBlockingObservable().singleOrDefault( null );
+ gm.writeEdge( testTarget2Edge ).toBlocking().singleOrDefault( null );
Edge test2TargetEdge = createEdge( sourceId1, "edgeType1", targetId1, System.currentTimeMillis() );
- gm.writeEdge( test2TargetEdge ).toBlockingObservable().singleOrDefault( null );
+ gm.writeEdge( test2TargetEdge ).toBlocking().singleOrDefault( null );
Edge test3TargetEdge = createEdge( sourceId1, "edgeType2", targetId1, System.currentTimeMillis() );
- gm.writeEdge( test3TargetEdge ).toBlockingObservable().singleOrDefault( null );
+ gm.writeEdge( test3TargetEdge ).toBlocking().singleOrDefault( null );
int count = gm.getEdgeTypesToTarget( new SimpleSearchEdgeType(targetId1, null, null) )
- .count().toBlockingObservable().last();
+ .count().toBlocking().last();
assertEquals( 3, count );
count = gm.getEdgeTypesToTarget( new SimpleSearchEdgeType(targetId1, "edgeType", null) )
- .count().toBlockingObservable().last();
+ .count().toBlocking().last();
assertEquals( 2, count );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/migration/EdgeDataMigrationImplTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/migration/EdgeDataMigrationImplTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/migration/EdgeDataMigrationImplTest.java
index a269c15..049c3d2 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/migration/EdgeDataMigrationImplTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/migration/EdgeDataMigrationImplTest.java
@@ -121,7 +121,7 @@ public class EdgeDataMigrationImplTest implements DataMigrationResetRule.DataMig
//walk from s1 and s2
- final Observable<GraphNode> graphNodes = Observable.from( new GraphNode( applicationScope, sourceId1), new GraphNode(applicationScope, sourceId2 ) );
+ final Observable<GraphNode> graphNodes = Observable.just( new GraphNode( applicationScope, sourceId1), new GraphNode(applicationScope, sourceId2 ) );
final MigrationDataProvider<GraphNode> testMigrationProvider = new MigrationDataProvider<GraphNode>() {
@Override
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/pom.xml
----------------------------------------------------------------------
diff --git a/stack/corepersistence/pom.xml b/stack/corepersistence/pom.xml
index 9656e2d..3ec7852 100644
--- a/stack/corepersistence/pom.xml
+++ b/stack/corepersistence/pom.xml
@@ -47,8 +47,8 @@ limitations under the License.
<properties>
- <maven.compiler.source>1.7</maven.compiler.source>
- <maven.compiler.target>1.7</maven.compiler.target>
+ <maven.compiler.source>1.8</maven.compiler.source>
+ <maven.compiler.target>1.8</maven.compiler.target>
<antlr.version>3.4</antlr.version>
<archaius.version>0.5.12</archaius.version>
@@ -64,14 +64,14 @@ limitations under the License.
<guava.version>18.0</guava.version>
<guice.version>4.0-beta5</guice.version>
<guicyfig.version>3.2</guicyfig.version>
- <hystrix.version>1.3.16</hystrix.version>
+ <hystrix.version>1.4.0</hystrix.version>
<jackson-2-version>2.4.1</jackson-2-version>
<jackson-smile.verson>2.4.3</jackson-smile.verson>
<mockito.version>1.10.8</mockito.version>
<junit.version>4.11</junit.version>
<kryo-serializers.version>0.26</kryo-serializers.version>
<log4j.version>1.2.17</log4j.version>
- <rx.version>0.19.6</rx.version>
+ <rx.version>1.0.8</rx.version>
<slf4j.version>1.7.2</slf4j.version>
<surefire.version>2.16</surefire.version>
<aws.version>1.9.0</aws.version>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/IndexLoadTestsIT.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/IndexLoadTestsIT.java b/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/IndexLoadTestsIT.java
index c962d6b..82af950 100644
--- a/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/IndexLoadTestsIT.java
+++ b/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/IndexLoadTestsIT.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.UUID;
import org.junit.ClassRule;
+import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
@@ -32,7 +33,6 @@ import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.core.scope.ApplicationScopeImpl;
import org.apache.usergrid.persistence.core.test.UseModules;
import org.apache.usergrid.persistence.index.EntityIndex;
-import org.apache.usergrid.persistence.index.EntityIndexBatch;
import org.apache.usergrid.persistence.index.EntityIndexFactory;
import org.apache.usergrid.persistence.index.IndexScope;
import org.apache.usergrid.persistence.index.guice.IndexTestFig;
@@ -41,13 +41,11 @@ import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.entity.SimpleId;
import org.apache.usergrid.persistence.model.field.IntegerField;
-import org.apache.usergrid.persistence.model.util.UUIDGenerator;
import com.google.inject.Inject;
import rx.Observable;
import rx.functions.Action1;
-import rx.functions.Action2;
import rx.functions.Func1;
import rx.schedulers.Schedulers;
@@ -57,6 +55,7 @@ import rx.schedulers.Schedulers;
*/
@RunWith( EsRunner.class )
@UseModules( { TestIndexModule.class } )
+@Ignore( "Should only be run during load tests of elasticsearch" )
public class IndexLoadTestsIT extends BaseIT {
private static final Logger log = LoggerFactory.getLogger( IndexLoadTestsIT.class );
@@ -70,13 +69,14 @@ public class IndexLoadTestsIT extends BaseIT {
@Inject
public EntityIndexFactory entityIndexFactory;
+
@Test
- public void testHeavyLoad(){
+ public void testHeavyLoad() {
final UUID applicationUUID = UUID.fromString( indexTestFig.getApplicationId() );
- final Id applicationId = new SimpleId(applicationUUID, "application");
- final ApplicationScope scope = new ApplicationScopeImpl( applicationId );
+ final Id applicationId = new SimpleId( applicationUUID, "application" );
+ final ApplicationScope scope = new ApplicationScopeImpl( applicationId );
final EntityIndex index = entityIndexFactory.createEntityIndex( scope );
@@ -87,83 +87,52 @@ public class IndexLoadTestsIT extends BaseIT {
//run them all
createEntities.toBlocking().last();
-
-
-
-
}
- public Observable<Entity> createStreamFromWorkers(final EntityIndex entityIndex, final Id ownerId){
-
- //create a sequence of observables. Each index will be it's own worker thread using the Schedulers.newthread()
- return Observable.range( 0, indexTestFig.getNumberOfWorkers() ).parallel( new Func1<Observable<Integer>, Observable<Entity>>() {
-
- @Override
- public Observable<Entity> call( final Observable<Integer> integerObservable ) {
- return integerObservable.flatMap( new Func1<Integer, Observable<Entity>>() {
- @Override
- public Observable<Entity> call( final Integer integer ) {
- return createWriteObservable( entityIndex, ownerId, integer );
- }
- } );
+ public Observable<Entity> createStreamFromWorkers( final EntityIndex entityIndex, final Id ownerId ) {
- }
- }, Schedulers.newThread() );
+ //create a sequence of observables. Each index will be it's own worker thread using the Schedulers.newthread()
+ return Observable.range( 0, indexTestFig.getNumberOfWorkers() ).flatMap(
+ integer -> createWriteObservable( entityIndex, ownerId, integer ).subscribeOn( Schedulers.newThread() ) );
}
- private Observable<Entity> createWriteObservable( final EntityIndex entityIndex, final Id ownerId, final int workerIndex){
+ private Observable<Entity> createWriteObservable( final EntityIndex entityIndex, final Id ownerId,
+ final int workerIndex ) {
final IndexScope scope = new IndexScopeImpl( ownerId, "test" );
-
- return Observable.range( 0, indexTestFig.getNumberOfRecords() )
+ return Observable.range( 0, indexTestFig.getNumberOfRecords() )
//create our entity
- .map( new Func1<Integer, Entity>() {
- @Override
- public Entity call( final Integer integer ) {
- final Entity entity = new Entity("test");
-
- entity.setField( new IntegerField("workerIndex", workerIndex));
- entity.setField( new IntegerField( "ordinal", integer ) );
-
- return entity;
- }
- } ).buffer( indexTestFig.getBufferSize() ).doOnNext( new Action1<List<Entity>>() {
- @Override
- public void call( final List<Entity> entities ) {
- //take our entities and roll them into a batch
- Observable.from( entities ).collect( entityIndex.createBatch(), new Action2<EntityIndexBatch, Entity>() {
-
-
- @Override
- public void call( final EntityIndexBatch entityIndexBatch, final Entity entity ) {
- entityIndexBatch.index(scope, entity );
- }
- } ).doOnNext( new Action1<EntityIndexBatch>() {
- @Override
- public void call( final EntityIndexBatch entityIndexBatch ) {
+ .map( new Func1<Integer, Entity>() {
+ @Override
+ public Entity call( final Integer integer ) {
+ final Entity entity = new Entity( "test" );
+
+ entity.setField( new IntegerField( "workerIndex", workerIndex ) );
+ entity.setField( new IntegerField( "ordinal", integer ) );
+
+ return entity;
+ }
+ } ).buffer( indexTestFig.getBufferSize() ).doOnNext( new Action1<List<Entity>>() {
+ @Override
+ public void call( final List<Entity> entities ) {
+ //take our entities and roll them into a batch
+ Observable.from( entities )
+ .collect( () -> entityIndex.createBatch(), ( entityIndexBatch, entity ) -> {
+
+ entityIndexBatch.index( scope, entity );
+ } ).doOnNext( entityIndexBatch -> {
entityIndexBatch.execute();
- }
- } ).toBlocking().last();
- }
- } )
-
- //translate back into a stream of entities for the caller to use
- .flatMap( new Func1<List<Entity>, Observable<Entity>>() {
- @Override
- public Observable<Entity> call( final List<Entity> entities ) {
- return Observable.from( entities );
- }
- } );
+ } ).toBlocking().last();
+ }
+ } )
+ //translate back into a stream of entities for the caller to use
+ .flatMap(entities -> Observable.from( entities ) );
}
-
-
-
-
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/pom.xml
----------------------------------------------------------------------
diff --git a/stack/pom.xml b/stack/pom.xml
index f24917a..efbda2d 100644
--- a/stack/pom.xml
+++ b/stack/pom.xml
@@ -125,7 +125,7 @@
<usergrid.it.threads>8</usergrid.it.threads>
<metrics.version>3.0.0</metrics.version>
- <rx.version>0.19.6</rx.version>
+ <rx.version>1.0.8</rx.version>
<surefire.plugin.artifactName>surefire-junit47</surefire.plugin.artifactName>
<surefire.plugin.version>2.18.1</surefire.plugin.version>
<powermock.version>1.6.1</powermock.version>
@@ -1560,8 +1560,8 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
- <source>1.7</source>
- <target>1.7</target>
+ <source>1.8</source>
+ <target>1.8</target>
<optimize>true</optimize>
<showDeprecation>true</showDeprecation>
<debug>true</debug>
@@ -1583,7 +1583,7 @@
<configuration>
<rules>
<requireJavaVersion>
- <version>1.7.0</version>
+ <version>1.8.0</version>
</requireJavaVersion>
<requireMavenVersion>
<version>[3.0,)</version>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java b/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
index 4f849e0..bebd557 100644
--- a/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
+++ b/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
@@ -937,18 +937,12 @@ public class ImportServiceImpl implements ImportService {
// potentially skip the first n if this is a resume operation
final int entityNumSkip = (int)tracker.getTotalEntityCount();
- // with this code we get asynchronous behavior and testImportWithMultipleFiles will fail
- final int entityCount = entityEventObservable.takeWhile( new Func1<WriteEvent, Boolean>() {
- @Override
- public Boolean call( final WriteEvent writeEvent ) {
- return !tracker.shouldStopProcessingEntities();
- }
- } ).skip(entityNumSkip).parallel(new Func1<Observable<WriteEvent>, Observable<WriteEvent>>() {
- @Override
- public Observable<WriteEvent> call(Observable<WriteEvent> entityWrapperObservable) {
- return entityWrapperObservable.doOnNext(doWork);
- }
- }, Schedulers.io()).reduce(0, heartbeatReducer).toBlocking().last();
+
+ entityEventObservable.takeWhile( writeEvent -> !tracker.shouldStopProcessingEntities() ).skip( entityNumSkip )
+ .flatMap( writeEvent -> {
+ return Observable.just( writeEvent ).doOnNext( doWork );
+ }, 10 ).reduce( 0, heartbeatReducer ).toBlocking().last();
+
jp.close();
@@ -979,17 +973,11 @@ public class ImportServiceImpl implements ImportService {
final int connectionNumSkip = (int)tracker.getTotalConnectionCount();
// with this code we get asynchronous behavior and testImportWithMultipleFiles will fail
- final int connectionCount = otherEventObservable.takeWhile( new Func1<WriteEvent, Boolean>() {
- @Override
- public Boolean call( final WriteEvent writeEvent ) {
- return !tracker.shouldStopProcessingConnections();
- }
- } ).skip(connectionNumSkip).parallel(new Func1<Observable<WriteEvent>, Observable<WriteEvent>>() {
- @Override
- public Observable<WriteEvent> call(Observable<WriteEvent> entityWrapperObservable) {
- return entityWrapperObservable.doOnNext(doWork);
- }
- }, Schedulers.io()).reduce(0, heartbeatReducer).toBlocking().last();
+ final int connectionCount = otherEventObservable.takeWhile(
+ writeEvent -> !tracker.shouldStopProcessingConnections() ).skip(connectionNumSkip).flatMap( entityWrapper ->{
+ return Observable.just(entityWrapper).doOnNext( doWork ).subscribeOn( Schedulers.io() );
+
+ }, 10 ).reduce(0, heartbeatReducer).toBlocking().last();
jp.close();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
index 5b1a6b3..b183daa 100644
--- a/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
@@ -110,84 +110,81 @@ public class ApplicationQueueManagerImpl implements ApplicationQueueManager {
final UUID appId = em.getApplication().getUuid();
final Map<String,Object> payloads = notification.getPayloads();
- final Func1<Entity,Entity> entityListFunct = new Func1<Entity, Entity>() {
- @Override
- public Entity call(Entity entity) {
+ final Func1<Entity,Entity> entityListFunct = entity -> {
- try {
+ try {
- long now = System.currentTimeMillis();
- List<EntityRef> devicesRef = getDevices(entity); // resolve group
+ long now = System.currentTimeMillis();
+ List<EntityRef> devicesRef = getDevices(entity); // resolve group
- LOG.info("notification {} queue {} devices, duration "+(System.currentTimeMillis()-now)+" ms", notification.getUuid(), devicesRef.size());
+ LOG.info("notification {} queue {} devices, duration "+(System.currentTimeMillis()-now)+" ms", notification.getUuid(), devicesRef.size());
- for (EntityRef deviceRef : devicesRef) {
- LOG.info("notification {} starting to queue device {} ", notification.getUuid(), deviceRef.getUuid());
- long hash = MurmurHash.hash(deviceRef.getUuid());
- if (sketch.estimateCount(hash) > 0) { //look for duplicates
- LOG.warn("Maybe Found duplicate device: {}", deviceRef.getUuid());
- continue;
- } else {
- sketch.add(hash, 1);
- }
- String notifierId = null;
- String notifierKey = null;
-
- //find the device notifier info, match it to the payload
- for (Map.Entry<String, Object> entry : payloads.entrySet()) {
- ProviderAdapter adapter = notifierMap.get(entry.getKey().toLowerCase());
- now = System.currentTimeMillis();
- String providerId = getProviderId(deviceRef, adapter.getNotifier());
- if (providerId != null) {
- notifierId = providerId;
- notifierKey = entry.getKey().toLowerCase();
- break;
- }
- LOG.info("Provider query for notification {} device {} took "+(System.currentTimeMillis()-now)+" ms",notification.getUuid(),deviceRef.getUuid());
- }
+ for (EntityRef deviceRef : devicesRef) {
+ LOG.info("notification {} starting to queue device {} ", notification.getUuid(), deviceRef.getUuid());
+ long hash = MurmurHash.hash(deviceRef.getUuid());
+ if (sketch.estimateCount(hash) > 0) { //look for duplicates
+ LOG.warn("Maybe Found duplicate device: {}", deviceRef.getUuid());
+ continue;
+ } else {
+ sketch.add(hash, 1);
+ }
+ String notifierId = null;
+ String notifierKey = null;
- if (notifierId == null) {
- LOG.info("Notifier did not match for device {} ", deviceRef);
- continue;
+ //find the device notifier info, match it to the payload
+ for (Map.Entry<String, Object> entry : payloads.entrySet()) {
+ ProviderAdapter adapter = notifierMap.get(entry.getKey().toLowerCase());
+ now = System.currentTimeMillis();
+ String providerId = getProviderId(deviceRef, adapter.getNotifier());
+ if (providerId != null) {
+ notifierId = providerId;
+ notifierKey = entry.getKey().toLowerCase();
+ break;
}
+ LOG.info("Provider query for notification {} device {} took "+(System.currentTimeMillis()-now)+" ms",notification.getUuid(),deviceRef.getUuid());
+ }
- ApplicationQueueMessage message = new ApplicationQueueMessage(appId, notification.getUuid(), deviceRef.getUuid(), notifierKey, notifierId);
- if (notification.getQueued() == null) {
- // update queued time
- now = System.currentTimeMillis();
- notification.setQueued(System.currentTimeMillis());
- LOG.info("notification {} device {} queue time set. duration "+(System.currentTimeMillis()-now)+" ms", notification.getUuid(), deviceRef.getUuid());
- }
+ if (notifierId == null) {
+ LOG.info("Notifier did not match for device {} ", deviceRef);
+ continue;
+ }
+
+ ApplicationQueueMessage message = new ApplicationQueueMessage(appId, notification.getUuid(), deviceRef.getUuid(), notifierKey, notifierId);
+ if (notification.getQueued() == null) {
+ // update queued time
now = System.currentTimeMillis();
- qm.sendMessage(message);
- LOG.info("notification {} post-queue to device {} duration " + (System.currentTimeMillis() - now) + " ms "+queueName+" queue", notification.getUuid(), deviceRef.getUuid());
- deviceCount.incrementAndGet();
- queueMeter.mark();
+ notification.setQueued(System.currentTimeMillis());
+ LOG.info("notification {} device {} queue time set. duration "+(System.currentTimeMillis()-now)+" ms", notification.getUuid(), deviceRef.getUuid());
}
- } catch (Exception deviceLoopException) {
- LOG.error("Failed to add devices", deviceLoopException);
- errorMessages.add("Failed to add devices for entity: " + entity.getUuid() + " error:" + deviceLoopException);
+ now = System.currentTimeMillis();
+ qm.sendMessage(message);
+ LOG.info("notification {} post-queue to device {} duration " + (System.currentTimeMillis() - now) + " ms "+queueName+" queue", notification.getUuid(), deviceRef.getUuid());
+ deviceCount.incrementAndGet();
+ queueMeter.mark();
}
- return entity;
+ } catch (Exception deviceLoopException) {
+ LOG.error("Failed to add devices", deviceLoopException);
+ errorMessages.add("Failed to add devices for entity: " + entity.getUuid() + " error:" + deviceLoopException);
}
+ return entity;
};
long now = System.currentTimeMillis();
- Observable o = rx.Observable.create(new IteratorObservable<Entity>(iterator))
- .parallel(new Func1<Observable<Entity>, Observable<Entity>>() {
- @Override
- public rx.Observable<Entity> call(rx.Observable<Entity> deviceObservable) {
- return deviceObservable.map(entityListFunct);
- }
- }, Schedulers.io())
- .doOnError(new Action1<Throwable>() {
- @Override
- public void call(Throwable throwable) {
- LOG.error("Failed while writing", throwable);
- }
- });
- o.toBlocking().lastOrDefault(null);
- LOG.info("notification {} done queueing duration {} ms", notification.getUuid(), System.currentTimeMillis() - now);
+
+
+ //process up to 10 concurrently
+ Observable o = rx.Observable.create( new IteratorObservable<Entity>( iterator ) )
+
+ .flatMap( entity -> Observable.just( entity ).map( entityListFunct )
+ .doOnError( throwable -> {
+ LOG.error( "Failed while writing",
+ throwable );
+ } ).subscribeOn( Schedulers.io() )
+
+ , 10 );
+
+ o.toBlocking().lastOrDefault( null );
+ LOG.info( "notification {} done queueing duration {} ms", notification.getUuid(), System.currentTimeMillis() - now);
}
// update queued time
@@ -338,48 +335,39 @@ public class ApplicationQueueManagerImpl implements ApplicationQueueManager {
return message;
}
};
- Observable o = rx.Observable.from(messages)
- .parallel(new Func1<rx.Observable<QueueMessage>, rx.Observable<ApplicationQueueMessage>>() {
- @Override
- public rx.Observable<ApplicationQueueMessage> call(rx.Observable<QueueMessage> messageObservable) {
- return messageObservable.map(func);
+
+ //from each queue message, process them in parallel up to 10 at a time
+ Observable o = rx.Observable.from( messages ).flatMap( queueMessage -> {
+
+
+ return Observable.just( queueMessage ).map( func ).buffer( messages.size() ).map( queueMessages -> {
+ //for gcm this will actually send notification
+ for ( ProviderAdapter providerAdapter : notifierMap.values() ) {
+ try {
+ providerAdapter.doneSendingNotifications();
}
- }, Schedulers.io())
- .buffer(messages.size())
- .map(new Func1<List<ApplicationQueueMessage>, HashMap<UUID, ApplicationQueueMessage>>() {
- @Override
- public HashMap<UUID, ApplicationQueueMessage> call(List<ApplicationQueueMessage> queueMessages) {
- //for gcm this will actually send notification
- for (ProviderAdapter providerAdapter : notifierMap.values()) {
- try {
- providerAdapter.doneSendingNotifications();
- } catch (Exception e) {
- LOG.error("providerAdapter.doneSendingNotifications: ", e);
- }
+ catch ( Exception e ) {
+ LOG.error( "providerAdapter.doneSendingNotifications: ", e );
+ }
+ }
+ //TODO: check if a notification is done and mark it
+ HashMap<UUID, ApplicationQueueMessage> notifications = new HashMap<>();
+ for ( ApplicationQueueMessage message : queueMessages ) {
+ if ( notifications.get( message.getNotificationId() ) == null ) {
+ try {
+ TaskManager taskManager = taskMap.get( message.getNotificationId() );
+ notifications.put( message.getNotificationId(), message );
+ taskManager.finishedBatch();
}
- //TODO: check if a notification is done and mark it
- HashMap<UUID, ApplicationQueueMessage> notifications = new HashMap<UUID, ApplicationQueueMessage>();
- for (ApplicationQueueMessage message : queueMessages) {
- if (notifications.get(message.getNotificationId()) == null) {
- try {
- TaskManager taskManager = taskMap.get(message.getNotificationId());
- notifications.put(message.getNotificationId(), message);
- taskManager.finishedBatch();
- } catch (Exception e) {
- LOG.error("Failed to finish batch", e);
- }
- }
-
+ catch ( Exception e ) {
+ LOG.error( "Failed to finish batch", e );
}
- return notifications;
- }
- })
- .doOnError(new Action1<Throwable>() {
- @Override
- public void call(Throwable throwable) {
- LOG.error("Failed while sending",throwable);
}
- });
+ }
+ return notifications;
+ } ).doOnError( throwable -> LOG.error( "Failed while sending", throwable ) );
+ }, 10 );
+
return o;
}
@@ -400,7 +388,8 @@ public class ApplicationQueueManagerImpl implements ApplicationQueueManager {
* {"winphone":"mymessage","apple":"mymessage"}
* TODO: document this method better
*/
- private Map<String, Object> translatePayloads(Map<String, Object> payloads, Map<Object, ProviderAdapter> notifierMap) throws Exception {
+ private Map<String, Object> translatePayloads(Map<String, Object> payloads, Map<Object, ProviderAdapter>
+ notifierMap) throws Exception {
Map<String, Object> translatedPayloads = new HashMap<String, Object>( payloads.size());
for (Map.Entry<String, Object> entry : payloads.entrySet()) {
String payloadKey = entry.getKey().toLowerCase();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/test-utils/src/main/java/org/apache/usergrid/setup/ConcurrentProcessSingleton.java
----------------------------------------------------------------------
diff --git a/stack/test-utils/src/main/java/org/apache/usergrid/setup/ConcurrentProcessSingleton.java b/stack/test-utils/src/main/java/org/apache/usergrid/setup/ConcurrentProcessSingleton.java
index e8c5ace..6d0419a 100644
--- a/stack/test-utils/src/main/java/org/apache/usergrid/setup/ConcurrentProcessSingleton.java
+++ b/stack/test-utils/src/main/java/org/apache/usergrid/setup/ConcurrentProcessSingleton.java
@@ -20,6 +20,8 @@
package org.apache.usergrid.setup;
+import java.io.IOException;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -106,7 +108,19 @@ public class ConcurrentProcessSingleton {
barrier.await( ONE_MINUTE );
logger.info( "Setup to complete" );
- lock.maybeReleaseLock();
+
+ Runtime.getRuntime().addShutdownHook( new Thread( ){
+ @Override
+ public void run() {
+ try {
+ lock.maybeReleaseLock();
+ }
+ catch ( IOException e ) {
+ throw new RuntimeException( "Unable to release lock" );
+ }
+ }
+ });
+
}
catch ( Exception e ) {
throw new RuntimeException( "Unable to initialize system", e );
[5/8] incubator-usergrid git commit: Resolves class loader problems
during test
Posted by to...@apache.org.
Resolves class loader problems during test
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/0a16033c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/0a16033c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/0a16033c
Branch: refs/heads/USERGRID-486
Commit: 0a16033c3fa4c3f093a58bbdb39f323c2958e5a5
Parents: 8ea46ba
Author: Todd Nine <tn...@apigee.com>
Authored: Thu Mar 19 18:31:34 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Thu Mar 19 19:47:22 2015 -0600
----------------------------------------------------------------------
stack/core/pom.xml | 7 +++----
stack/pom.xml | 6 ++++--
stack/rest/pom.xml | 1 +
stack/services/pom.xml | 1 +
4 files changed, 9 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/0a16033c/stack/core/pom.xml
----------------------------------------------------------------------
diff --git a/stack/core/pom.xml b/stack/core/pom.xml
index 971ee62..d053211 100644
--- a/stack/core/pom.xml
+++ b/stack/core/pom.xml
@@ -88,10 +88,9 @@
<forkCount>${usergrid.it.forkCount}</forkCount>
<threadCount>${usergrid.it.threads}</threadCount>
<reuseForks>true</reuseForks>
- <argLine>-Dtest.barrier.timestamp=${maven.build.timestamp} -Dtest.clean.storage=true -Xmx${ug.heapmax}
- -Xms${ug.heapmin} -Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8
- -javaagent:${settings.localRepository}/com/github/stephenc/jamm/0.2.5/jamm-0.2.5.jar ${ug.argline}
- </argLine>
+ <argLine>-Dtest.barrier.timestamp=${maven.build.timestamp} -Dtest.clean.storage=true -Xmx${ug.heapmax} -Xms${ug.heapmin} -Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8 -javaagent:${settings.localRepository}/com/github/stephenc/jamm/0.2.5/jamm-0.2.5.jar ${ug.argline} </argLine>
+ <!-- see this page for documentation on classloading issues http://maven.apache.org/surefire/maven-surefire-plugin/examples/class-loading.html -->
+ <useSystemClassLoader>false</useSystemClassLoader>
<includes>
<include>**/*IT.java</include>
<include>**/*Test.java</include>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/0a16033c/stack/pom.xml
----------------------------------------------------------------------
diff --git a/stack/pom.xml b/stack/pom.xml
index f24917a..6126046 100644
--- a/stack/pom.xml
+++ b/stack/pom.xml
@@ -109,7 +109,7 @@
<jersey-version>1.18.1</jersey-version>
<junit-version>4.12</junit-version>
<log4j-version>1.2.16</log4j-version>
- <org.springframework.version>3.1.2.RELEASE</org.springframework.version>
+ <org.springframework.version>3.2.13.RELEASE</org.springframework.version>
<shiro-version>1.2.3</shiro-version>
<slf4j-version>1.6.1</slf4j-version>
<snakeyaml-version>1.8</snakeyaml-version>
@@ -1461,6 +1461,8 @@
<forkCount>${usergrid.it.forkCount}</forkCount>
<reuseForks>${usergrid.it.reuseForks}</reuseForks>
<threadCount>${usergrid.it.forkCount}</threadCount>
+ <!-- see this page for documentation on classloading issues http://maven.apache.org/surefire/maven-surefire-plugin/examples/class-loading.html -->
+ <useSystemClassLoader>false</useSystemClassLoader>
<argLine>-Xmx${ug.heapmax} -Xms${ug.heapmin} -javaagent:${settings.localRepository}/com/github/stephenc/jamm/0.2.5/jamm-0.2.5.jar ${ug.argline}</argLine>
<testFailureIgnore>false</testFailureIgnore>
</configuration>
@@ -1471,7 +1473,7 @@
<dependencies>
<dependency>
<groupId>org.apache.maven.surefire</groupId>
- <artifactId>surefire-junit4</artifactId>
+ <artifactId>${surefire.plugin.artifactName}</artifactId>
<version>${surefire.plugin.version}</version>
<!--<exclusions>-->
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/0a16033c/stack/rest/pom.xml
----------------------------------------------------------------------
diff --git a/stack/rest/pom.xml b/stack/rest/pom.xml
index deb11f8..974945f 100644
--- a/stack/rest/pom.xml
+++ b/stack/rest/pom.xml
@@ -102,6 +102,7 @@
<parallel>methods</parallel>
<forkCount>1</forkCount>
<threadCount>${usergrid.rest.threads}</threadCount>
+ <useSystemClassLoader>false</useSystemClassLoader>
<reuseForks>true</reuseForks>
<argLine>-Dwebapp.directory=${basedir}/src/main/webapp -Dtest.barrier.timestamp=${maven.build.timestamp} -Dtest.clean.storage=true -Xmx${ug.heapmax} -Xms${ug.heapmin} -Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8 -javaagent:${settings.localRepository}/com/github/stephenc/jamm/0.2.5/jamm-0.2.5.jar ${ug.argline}
</argLine>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/0a16033c/stack/services/pom.xml
----------------------------------------------------------------------
diff --git a/stack/services/pom.xml b/stack/services/pom.xml
index 68f1a4a..6074e65 100644
--- a/stack/services/pom.xml
+++ b/stack/services/pom.xml
@@ -92,6 +92,7 @@
<forkCount>${usergrid.it.forkCount}</forkCount>
<threadCount>${usergrid.it.threads}</threadCount>
<reuseForks>${usergrid.it.reuseForks}</reuseForks>
+ <useSystemClassLoader>false</useSystemClassLoader>
<argLine>-Dtest.barrier.timestamp=${maven.build.timestamp} -Dtest.clean.storage=true -Xmx${ug.heapmax} -Xms${ug.heapmin} -Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8 -javaagent:${settings.localRepository}/com/github/stephenc/jamm/0.2.5/jamm-0.2.5.jar ${ug.argline}
</argLine>
<includes>
[2/8] incubator-usergrid git commit: First pass at upgrading to java
8 and latest RX java
Posted by to...@apache.org.
First pass at upgrading to java 8 and latest RX java
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/282e2271
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/282e2271
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/282e2271
Branch: refs/heads/USERGRID-486
Commit: 282e22712890cdda0439a5694810cff632526d7b
Parents: 72ec19d
Author: Todd Nine <tn...@apigee.com>
Authored: Thu Mar 19 18:00:57 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Thu Mar 19 18:00:57 2015 -0600
----------------------------------------------------------------------
stack/core/pom.xml | 26 +--
.../corepersistence/CpEntityManager.java | 2 +-
.../corepersistence/CpEntityManagerFactory.java | 7 +-
.../corepersistence/CpRelationManager.java | 24 +--
.../usergrid/corepersistence/CpWalker.java | 77 +++-----
.../events/EntityVersionDeletedHandler.java | 72 +++----
.../migration/EntityTypeMappingMigration.java | 41 ++--
.../migration/EntityTypeMappingMigrationIT.java | 2 +-
.../impl/EntityCollectionManagerImpl.java | 10 +-
.../collection/impl/EntityDeletedTask.java | 20 +-
.../impl/EntityVersionCleanupTask.java | 27 +--
.../impl/EntityVersionCreatedTask.java | 26 +--
.../MvccEntitySerializationStrategyImpl.java | 89 +++------
.../MvccEntitySerializationStrategyV3Impl.java | 91 +++------
.../migration/MvccEntityDataMigrationImpl.java | 169 +++++++---------
.../persistence/collection/rx/ParallelTest.java | 10 +-
...ctMvccEntityDataMigrationV1ToV3ImplTest.java | 2 +-
stack/corepersistence/common/pom.xml | 15 +-
.../astyanax/MultiKeyColumnNameIterator.java | 4 +-
.../MultiKeyColumnNameIteratorTest.java | 187 ++++++++----------
.../astyanax/MultiRowColumnIteratorTest.java | 50 ++---
.../graph/impl/GraphManagerImpl.java | 6 +-
.../graph/impl/stage/EdgeMetaRepairImpl.java | 2 +
.../impl/stage/NodeDeleteListenerImpl.java | 2 +-
.../impl/migration/EdgeDataMigrationImpl.java | 87 ++++-----
.../graph/GraphManagerShardConsistencyIT.java | 2 +-
.../usergrid/persistence/graph/SimpleTest.java | 12 +-
.../migration/EdgeDataMigrationImplTest.java | 2 +-
stack/corepersistence/pom.xml | 8 +-
.../index/impl/IndexLoadTestsIT.java | 105 ++++------
stack/pom.xml | 8 +-
.../management/importer/ImportServiceImpl.java | 34 ++--
.../impl/ApplicationQueueManagerImpl.java | 195 +++++++++----------
.../setup/ConcurrentProcessSingleton.java | 16 +-
34 files changed, 604 insertions(+), 826 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/core/pom.xml
----------------------------------------------------------------------
diff --git a/stack/core/pom.xml b/stack/core/pom.xml
index 971ee62..119a52b 100644
--- a/stack/core/pom.xml
+++ b/stack/core/pom.xml
@@ -130,15 +130,7 @@
</execution>
</executions>
</plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-compiler-plugin</artifactId>
- <version>2.3.2</version>
- <configuration>
- <source>1.7</source>
- <target>1.7</target>
- </configuration>
- </plugin>
+
<!-- <plugin>
<groupId>org.apache.maven.plugins</groupId>
@@ -481,17 +473,11 @@
<version>${metrics.version}</version>
</dependency>
- <dependency>
- <groupId>com.netflix.rxjava</groupId>
- <artifactId>rxjava-core</artifactId>
- <version>${rx.version}</version>
- </dependency>
-
- <dependency>
- <groupId>com.netflix.rxjava</groupId>
- <artifactId>rxjava-math</artifactId>
- <version>${rx.version}</version>
- </dependency>
+ <dependency>
+ <groupId>io.reactivex</groupId>
+ <artifactId>rxjava</artifactId>
+ <version>${rx.version}</version>
+ </dependency>
<dependency>
<groupId>com.clearspring.analytics</groupId>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
index 789e640..9cffdaf 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
@@ -1097,7 +1097,7 @@ public class CpEntityManager implements EntityManager {
} );
//TODO: does this call and others like it need a graphite reporter?
- cpEntity = ecm.write( cpEntity ).toBlockingObservable().last();
+ cpEntity = ecm.write( cpEntity ).toBlocking().last();
logger.debug( "Wrote {}:{} version {}", new Object[] {
cpEntity.getId().getType(), cpEntity.getId().getUuid(), cpEntity.getVersion()
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
index f76b9fc..83c3d85 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
@@ -451,7 +451,8 @@ public class CpEntityManagerFactory implements EntityManagerFactory, Application
fromEntityId, edgeType, Long.MAX_VALUE,
SearchByEdgeType.Order.DESCENDING, null ));
- Iterator<Edge> iter = edges.toBlockingObservable().getIterator();
+ //TODO This is wrong, and will result in OOM if there are too many applications. This needs to stream properly with a buffer
+ Iterator<Edge> iter = edges.toBlocking().getIterator();
while ( iter.hasNext() ) {
Edge edge = iter.next();
@@ -469,7 +470,7 @@ public class CpEntityManagerFactory implements EntityManagerFactory, Application
org.apache.usergrid.persistence.model.entity.Entity e =
managerCache.getEntityCollectionManager( collScope ).load( targetId )
- .toBlockingObservable().lastOrDefault(null);
+ .toBlocking().lastOrDefault(null);
if ( e == null ) {
logger.warn("Applicaion {} in index but not found in collections", targetId );
@@ -624,7 +625,7 @@ public class CpEntityManagerFactory implements EntityManagerFactory, Application
public long performEntityCount() {
//TODO, this really needs to be a task that writes this data somewhere since this will get
//progressively slower as the system expands
- return (Long) getAllEntitiesObservable().longCount().toBlocking().last();
+ return (Long) getAllEntitiesObservable().countLong().toBlocking().last();
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
index 2eeee28..c4e970d 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
@@ -280,7 +280,7 @@ public class CpRelationManager implements RelationManager {
Observable<String> types= gm.getEdgeTypesFromSource(
new SimpleSearchEdgeType( cpHeadEntity.getId(), edgeTypePrefix, null ));
- Iterator<String> iter = types.toBlockingObservable().getIterator();
+ Iterator<String> iter = types.toBlocking().getIterator();
while ( iter.hasNext() ) {
indexes.add( iter.next() );
}
@@ -346,7 +346,7 @@ public class CpRelationManager implements RelationManager {
Observable<Edge> edges = gm.loadEdgesToTarget( new SimpleSearchByEdgeType(
cpHeadEntity.getId(), etype, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, null ));
- Iterator<Edge> iter = edges.toBlockingObservable().getIterator();
+ Iterator<Edge> iter = edges.toBlocking().getIterator();
while ( iter.hasNext() ) {
Edge edge = iter.next();
@@ -383,7 +383,7 @@ public class CpRelationManager implements RelationManager {
final GraphManager gm = managerCache.getGraphManager( applicationScope );
Iterator<String> edgeTypesToTarget = gm.getEdgeTypesToTarget( new SimpleSearchEdgeType(
- cpHeadEntity.getId(), null, null) ).toBlockingObservable().getIterator();
+ cpHeadEntity.getId(), null, null) ).toBlocking().getIterator();
logger.debug("updateContainingCollectionsAndCollections(): "
+ "Searched for edges to target {}:{}\n in scope {}\n found: {}",
@@ -484,7 +484,7 @@ public class CpRelationManager implements RelationManager {
SearchByEdgeType.Order.DESCENDING,
null ) );
- return edges.toBlockingObservable().firstOrDefault( null ) != null;
+ return edges.toBlocking().firstOrDefault( null ) != null;
}
@@ -511,7 +511,7 @@ public class CpRelationManager implements RelationManager {
SearchByEdgeType.Order.DESCENDING,
null ) );
- return edges.toBlockingObservable().firstOrDefault( null ) != null;
+ return edges.toBlocking().firstOrDefault( null ) != null;
}
@@ -528,7 +528,7 @@ public class CpRelationManager implements RelationManager {
SearchByEdgeType.Order.DESCENDING,
null ) ); // last
- Iterator<Edge> iterator = edgesToTarget.toBlockingObservable().getIterator();
+ Iterator<Edge> iterator = edgesToTarget.toBlocking().getIterator();
int count = 0;
while ( iterator.hasNext() ) {
iterator.next();
@@ -569,7 +569,7 @@ public class CpRelationManager implements RelationManager {
Observable<String> str = gm.getEdgeTypesFromSource(
new SimpleSearchEdgeType( cpHeadEntity.getId(), null, null ) );
- Iterator<String> iter = str.toBlockingObservable().getIterator();
+ Iterator<String> iter = str.toBlocking().getIterator();
while ( iter.hasNext() ) {
String edgeType = iter.next();
indexes.add( CpNamingUtils.getCollectionName( edgeType ) );
@@ -692,7 +692,7 @@ public class CpRelationManager implements RelationManager {
// create graph edge connection from head entity to member entity
Edge edge = new SimpleEdge( cpHeadEntity.getId(), edgeType, memberEntity.getId(), uuidHash );
GraphManager gm = managerCache.getGraphManager( applicationScope );
- gm.writeEdge( edge ).toBlockingObservable().last();
+ gm.writeEdge( edge ).toBlocking().last();
logger.debug( "Wrote edgeType {}\n from {}:{}\n to {}:{}\n scope {}:{}",
new Object[] {
@@ -855,7 +855,7 @@ public class CpRelationManager implements RelationManager {
cpHeadEntity.getId(),
CpNamingUtils.getEdgeTypeFromCollectionName( collName ),
memberEntity.getId(), UUIDUtils.getUUIDLong( memberEntity.getId().getUuid() ) );
- gm.deleteEdge( collectionToItemEdge ).toBlockingObservable().last();
+ gm.deleteEdge( collectionToItemEdge ).toBlocking().last();
// remove edge from item to collection
Edge itemToCollectionEdge = new SimpleEdge(
@@ -865,7 +865,7 @@ public class CpRelationManager implements RelationManager {
cpHeadEntity.getId(),
UUIDUtils.getUUIDLong( cpHeadEntity.getId().getUuid() ) );
- gm.deleteEdge( itemToCollectionEdge ).toBlockingObservable().last();
+ gm.deleteEdge( itemToCollectionEdge ).toBlocking().last();
// special handling for roles collection of a group
if ( headEntity.getType().equals( Group.ENTITY_TYPE ) ) {
@@ -1058,7 +1058,7 @@ public class CpRelationManager implements RelationManager {
cpHeadEntity.getId(), edgeType, targetEntity.getId(), System.currentTimeMillis() );
GraphManager gm = managerCache.getGraphManager( applicationScope );
- gm.writeEdge( edge ).toBlockingObservable().last();
+ gm.writeEdge( edge ).toBlocking().last();
EntityIndex ei = managerCache.getEntityIndex( applicationScope );
EntityIndexBatch batch = ei.createBatch();
@@ -1290,7 +1290,7 @@ public class CpRelationManager implements RelationManager {
System.currentTimeMillis() );
GraphManager gm = managerCache.getGraphManager( applicationScope );
- gm.deleteEdge( edge ).toBlockingObservable().last();
+ gm.deleteEdge( edge ).toBlocking().last();
final EntityIndex ei = managerCache.getEntityIndex( applicationScope );
final EntityIndexBatch batch = ei.createBatch();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
index 4b902d8..332d5a8 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
@@ -104,53 +104,38 @@ public class CpWalker {
Observable<String> edgeTypes = gm.getEdgeTypesFromSource(
new SimpleSearchEdgeType( applicationId, edgeType, null ) );
- edgeTypes.flatMap( new Func1<String, Observable<Edge>>() {
- @Override
- public Observable<Edge> call( final String edgeType ) {
-
- logger.debug( "Loading edges of type {} from node {}", edgeType, applicationId );
-
- return gm.loadEdgesFromSource( new SimpleSearchByEdgeType(
- applicationId, edgeType, Long.MAX_VALUE, order , null ) );
-
- }
-
- } ).parallel( new Func1<Observable<Edge>, Observable<Edge>>() {
-
- @Override
- public Observable<Edge> call( final Observable<Edge> edgeObservable ) { // process edges in parallel
- return edgeObservable.doOnNext( new Action1<Edge>() { // visit and update then entity
-
- @Override
- public void call( Edge edge ) {
-
- logger.info( "Re-indexing edge {}", edge );
-
- EntityRef targetNodeEntityRef = new SimpleEntityRef(
- edge.getTargetNode().getType(),
- edge.getTargetNode().getUuid() );
-
- Entity entity;
- try {
- entity = em.get( targetNodeEntityRef );
- }
- catch ( Exception ex ) {
- logger.error( "Error getting sourceEntity {}:{}, continuing",
- targetNodeEntityRef.getType(),
- targetNodeEntityRef.getUuid() );
- return;
- }
- if(entity == null){
- return;
- }
- String collName = CpNamingUtils.getCollectionName( edge.getType() );
- visitor.visitCollectionEntry( em, collName, entity );
- }
- } );
- }
- }, Schedulers.io() )
+ edgeTypes.flatMap( emittedEdgeType -> {
+
+ logger.debug( "Loading edges of type {} from node {}", edgeType, applicationId );
+
+ return gm.loadEdgesFromSource(
+ new SimpleSearchByEdgeType( applicationId, emittedEdgeType, Long.MAX_VALUE, order, null ) );
+ } ).flatMap( edge -> {
+ //run each edge through it's own scheduler, up to 100 at a time
+ return Observable.just( edge ).doOnNext( edgeValue -> {
+ logger.info( "Re-indexing edge {}", edgeValue );
+
+ EntityRef targetNodeEntityRef =
+ new SimpleEntityRef( edgeValue.getTargetNode().getType(), edgeValue.getTargetNode().getUuid() );
+
+ Entity entity;
+ try {
+ entity = em.get( targetNodeEntityRef );
+ }
+ catch ( Exception ex ) {
+ logger.error( "Error getting sourceEntity {}:{}, continuing", targetNodeEntityRef.getType(),
+ targetNodeEntityRef.getUuid() );
+ return;
+ }
+ if ( entity == null ) {
+ return;
+ }
+ String collName = CpNamingUtils.getCollectionName( edgeValue.getType() );
+ visitor.visitCollectionEntry( em, collName, entity );
+ } ).subscribeOn( Schedulers.io() );
+ }, 100 );
// wait for it to complete
- .toBlocking().lastOrDefault( null ); // end foreach on edges
+ edgeTypes.toBlocking().lastOrDefault( null ); // end foreach on edges
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionDeletedHandler.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionDeletedHandler.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionDeletedHandler.java
index c45949b..23f5a32 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionDeletedHandler.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionDeletedHandler.java
@@ -17,53 +17,48 @@
*/
package org.apache.usergrid.corepersistence.events;
-import com.google.inject.Inject;
-import com.google.inject.Singleton;
import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.usergrid.corepersistence.CpEntityManagerFactory;
-import static org.apache.usergrid.corepersistence.CoreModule.EVENTS_DISABLED;
import org.apache.usergrid.persistence.EntityManagerFactory;
import org.apache.usergrid.persistence.collection.CollectionScope;
-import org.apache.usergrid.persistence.collection.MvccEntity;
import org.apache.usergrid.persistence.collection.MvccLogEntry;
import org.apache.usergrid.persistence.collection.event.EntityVersionDeleted;
-import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
import org.apache.usergrid.persistence.index.EntityIndex;
-import org.apache.usergrid.persistence.index.EntityIndexBatch;
import org.apache.usergrid.persistence.index.IndexScope;
import org.apache.usergrid.persistence.index.impl.IndexScopeImpl;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.entity.SimpleId;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
import rx.Observable;
-import rx.functions.Action1;
-import rx.functions.Action2;
-import rx.functions.Func1;
-import rx.schedulers.Schedulers;
+
+import static org.apache.usergrid.corepersistence.CoreModule.EVENTS_DISABLED;
/**
- * Remove Entity index when specific version of Entity is deleted.
- * TODO: do we need this? Don't our version-created and entity-deleted handlers take care of this?
- * If we do need it then it should be wired in via GuiceModule in the corepersistence package.
+ * Remove Entity index when specific version of Entity is deleted. TODO: do we need this? Don't our version-created and
+ * entity-deleted handlers take care of this? If we do need it then it should be wired in via GuiceModule in the
+ * corepersistence package.
*/
@Singleton
public class EntityVersionDeletedHandler implements EntityVersionDeleted {
- private static final Logger logger = LoggerFactory.getLogger(EntityVersionDeletedHandler.class );
-
-
+ private static final Logger logger = LoggerFactory.getLogger( EntityVersionDeletedHandler.class );
private final EntityManagerFactory emf;
+
@Inject
public EntityVersionDeletedHandler( final EntityManagerFactory emf ) {this.emf = emf;}
-
@Override
public void versionDeleted( final CollectionScope scope, final Id entityId,
final List<MvccLogEntry> entityVersions ) {
@@ -71,40 +66,33 @@ public class EntityVersionDeletedHandler implements EntityVersionDeleted {
// This check is for testing purposes and for a test that to be able to dynamically turn
// off and on delete previous versions so that it can test clean-up on read.
- if ( System.getProperty( EVENTS_DISABLED, "false" ).equals( "true" )) {
+ if ( System.getProperty( EVENTS_DISABLED, "false" ).equals( "true" ) ) {
return;
}
- if(logger.isDebugEnabled()) {
- logger.debug( "Handling versionDeleted count={} event for entity {}:{} v {} " + "scope\n name: {}\n owner: {}\n app: {}",
- new Object[] {
+ if ( logger.isDebugEnabled() ) {
+ logger.debug( "Handling versionDeleted count={} event for entity {}:{} v {} "
+ + "scope\n name: {}\n owner: {}\n app: {}", new Object[] {
entityVersions.size(), entityId.getType(), entityId.getUuid(), scope.getName(), scope.getOwner(),
scope.getApplication()
} );
}
- CpEntityManagerFactory cpemf = (CpEntityManagerFactory)emf;
+ CpEntityManagerFactory cpemf = ( CpEntityManagerFactory ) emf;
final EntityIndex ei = cpemf.getManagerCache().getEntityIndex( scope );
- final IndexScope indexScope = new IndexScopeImpl(
- new SimpleId(scope.getOwner().getUuid(), scope.getOwner().getType()),
- scope.getName()
- );
-
- Observable.from( entityVersions )
- .collect( ei.createBatch(), new Action2<EntityIndexBatch, MvccLogEntry>() {
- @Override
- public void call( final EntityIndexBatch entityIndexBatch, final MvccLogEntry mvccLogEntry ) {
- entityIndexBatch.deindex( indexScope, mvccLogEntry.getEntityId(), mvccLogEntry.getVersion() );
- }
- } ).doOnNext( new Action1<EntityIndexBatch>() {
- @Override
- public void call( final EntityIndexBatch entityIndexBatch ) {
+ final IndexScope indexScope =
+ new IndexScopeImpl( new SimpleId( scope.getOwner().getUuid(), scope.getOwner().getType() ),
+ scope.getName() );
+
+ //create our batch, and then collect all of them into a single batch
+ Observable.from( entityVersions ).collect( () -> ei.createBatch(), ( entityIndexBatch, mvccLogEntry ) -> {
+ entityIndexBatch.deindex( indexScope, mvccLogEntry.getEntityId(), mvccLogEntry.getVersion() );
+ } )
+ //after our batch is collected, execute it
+ .doOnNext( entityIndexBatch -> {
entityIndexBatch.execute();
- }
- } ).toBlocking().last();
+ } ).toBlocking().last();
}
-
-
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigration.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigration.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigration.java
index 40ad236..6531d16 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigration.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigration.java
@@ -37,6 +37,7 @@ import com.google.inject.Inject;
import rx.Observable;
import rx.functions.Action1;
import rx.functions.Func1;
+import rx.schedulers.Schedulers;
/**
@@ -63,36 +64,26 @@ public class EntityTypeMappingMigration implements DataMigration<EntityIdScope>
final AtomicLong atomicLong = new AtomicLong();
- allEntitiesInSystemObservable.getData()
- //process the entities in parallel
- .parallel( new Func1<Observable<EntityIdScope>, Observable<EntityIdScope>>() {
+ //migrate up to 100 types simultaneously
+ allEntitiesInSystemObservable.getData().flatMap( entityIdScope -> {
+ return Observable.just( entityIdScope ).doOnNext( entityIdScopeObservable -> {
+ final MapScope ms = CpNamingUtils
+ .getEntityTypeMapScope( entityIdScope.getCollectionScope().getApplication() );
+ final MapManager mapManager = managerCache.getMapManager( ms );
- @Override
- public Observable<EntityIdScope> call( final Observable<EntityIdScope> entityIdScopeObservable ) {
+ final UUID entityUuid = entityIdScope.getId().getUuid();
+ final String entityType = entityIdScope.getId().getType();
- //for each entity observable, get the map scope and write it to the map
- return entityIdScopeObservable.doOnNext( new Action1<EntityIdScope>() {
- @Override
- public void call( final EntityIdScope entityIdScope ) {
- final MapScope ms = CpNamingUtils
- .getEntityTypeMapScope( entityIdScope.getCollectionScope().getApplication() );
+ mapManager.putString( entityUuid.toString(), entityType );
- final MapManager mapManager = managerCache.getMapManager( ms );
+ if ( atomicLong.incrementAndGet() % 100 == 0 ) {
+ observer.update( getMaxVersion(),
+ String.format( "Updated %d entities", atomicLong.get() ) );
+ }
- final UUID entityUuid = entityIdScope.getId().getUuid();
- final String entityType = entityIdScope.getId().getType();
-
- mapManager.putString( entityUuid.toString(), entityType );
-
- if ( atomicLong.incrementAndGet() % 100 == 0 ) {
- observer.update( getMaxVersion(),
- String.format( "Updated %d entities", atomicLong.get() ) );
- }
- }
- } );
- }
- } ).count().toBlocking().last();
+ } ).subscribeOn( Schedulers.io() );
+ }, 100 ).count().toBlocking().last();
return getMaxVersion();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigrationIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigrationIT.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigrationIT.java
index be7cee4..88f56c8 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigrationIT.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigrationIT.java
@@ -77,7 +77,7 @@ public class EntityTypeMappingMigrationIT {
final MapScope mapScope2 = new MapScopeImpl(applicationId, CpNamingUtils.TYPES_BY_UUID_MAP );
- final Observable<EntityIdScope> scopes = Observable.from(idScope1, idScope2);
+ final Observable<EntityIdScope> scopes = Observable.just(idScope1, idScope2);
final TestMigrationDataProvider<EntityIdScope> migrationDataProvider = new TestMigrationDataProvider<>();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
index f565fab..70b5a3a 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
@@ -235,7 +235,7 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
Preconditions.checkNotNull( entityId.getType(), "Entity type is required in this stage" );
final Timer.Context timer = deleteTimer.time();
- Observable<Id> o = Observable.from(new CollectionIoEvent<Id>(collectionScope, entityId))
+ Observable<Id> o = Observable.just( new CollectionIoEvent<Id>( collectionScope, entityId ) )
.map(markStart)
.doOnNext( markCommit )
.map(new Func1<CollectionIoEvent<MvccEntity>, Id>() {
@@ -284,7 +284,7 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
return Observable.empty();
}
- return Observable.from(entity.getEntity().get());
+ return Observable.just( entity.getEntity().get() );
}
})
.doOnNext( new Action1<Entity>() {
@@ -449,19 +449,19 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
public Observable<CollectionIoEvent<MvccEntity>> stageRunner( CollectionIoEvent<Entity> writeData,
WriteStart writeState ) {
- return Observable.from( writeData ).map( writeState ).doOnNext( new Action1<CollectionIoEvent<MvccEntity>>() {
+ return Observable.just( writeData ).map( writeState ).doOnNext( new Action1<CollectionIoEvent<MvccEntity>>() {
@Override
public void call( final CollectionIoEvent<MvccEntity> mvccEntityCollectionIoEvent ) {
Observable<CollectionIoEvent<MvccEntity>> unique =
- Observable.from( mvccEntityCollectionIoEvent ).subscribeOn( Schedulers.io() )
+ Observable.just( mvccEntityCollectionIoEvent ).subscribeOn( Schedulers.io() )
.doOnNext( writeVerifyUnique );
// optimistic verification
Observable<CollectionIoEvent<MvccEntity>> optimistic =
- Observable.from( mvccEntityCollectionIoEvent ).subscribeOn( Schedulers.io() )
+ Observable.just( mvccEntityCollectionIoEvent ).subscribeOn( Schedulers.io() )
.doOnNext( writeOptimisticVerify );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityDeletedTask.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityDeletedTask.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityDeletedTask.java
index 5472645..7620907 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityDeletedTask.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityDeletedTask.java
@@ -127,22 +127,10 @@ public class EntityDeletedTask implements Task<Void> {
LOG.debug( "Started firing {} listeners", listenerSize );
- //if we have more than 1, run them on the rx scheduler for a max of 8 operations at a time
- Observable.from(listeners)
- .parallel( new Func1<Observable<EntityDeleted>, Observable<EntityDeleted>>() {
-
- @Override
- public Observable<EntityDeleted> call(
- final Observable<EntityDeleted> entityVersionDeletedObservable ) {
-
- return entityVersionDeletedObservable.doOnNext( new Action1<EntityDeleted>() {
- @Override
- public void call( final EntityDeleted listener ) {
- listener.deleted(collectionScope, entityId, version);
- }
- } );
- }
- }, Schedulers.io() ).toBlocking().last();
+ //if we have more than 1, run them on the rx scheduler for a max of 10 operations at a time
+ Observable.from(listeners).flatMap( currentListener -> Observable.just( currentListener ).doOnNext( listener -> {
+ listener.deleted( collectionScope, entityId, version );
+ } ).subscribeOn( Schedulers.io() ), 10 ).toBlocking().last();
LOG.debug( "Finished firing {} listeners", listenerSize );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTask.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTask.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTask.java
index b245528..1a7b86b 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTask.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTask.java
@@ -159,7 +159,7 @@ public class EntityVersionCleanupTask implements Task<Void> {
throw new RuntimeException( "Unable to execute batch mutation", e );
}
}
- } ).subscribeOn( Schedulers.io() ).longCount().toBlocking();
+ } ).subscribeOn( Schedulers.io() ).countLong().toBlocking();
//start calling the listeners for remove log entries
@@ -201,7 +201,7 @@ public class EntityVersionCleanupTask implements Task<Void> {
throw new RuntimeException( "Unable to execute batch mutation", e );
}
}
- } ).subscribeOn( Schedulers.io() ).longCount().toBlocking();
+ } ).subscribeOn( Schedulers.io() ).countLong().toBlocking();
//wait or this to complete
final Long removedCount = uniqueValueCleanup.last();
@@ -232,21 +232,14 @@ public class EntityVersionCleanupTask implements Task<Void> {
logger.debug( "Started firing {} listeners", listenerSize );
//if we have more than 1, run them on the rx scheduler for a max of 8 operations at a time
- Observable.from( listeners )
- .parallel( new Func1<Observable<EntityVersionDeleted>, Observable<EntityVersionDeleted>>() {
-
- @Override
- public Observable<EntityVersionDeleted> call(
- final Observable<EntityVersionDeleted> entityVersionDeletedObservable ) {
-
- return entityVersionDeletedObservable.doOnNext( new Action1<EntityVersionDeleted>() {
- @Override
- public void call( final EntityVersionDeleted listener ) {
- listener.versionDeleted( scope, entityId, versions );
- }
- } );
- }
- }, Schedulers.io() ).toBlocking().last();
+
+
+ //if we have more than 1, run them on the rx scheduler for a max of 10 operations at a time
+ Observable.from(listeners).flatMap( currentListener -> Observable.just( currentListener ).doOnNext( listener -> {
+ listener.versionDeleted( scope, entityId, versions );
+ } ).subscribeOn( Schedulers.io() ), 10 ).toBlocking().last();
+
+
logger.debug( "Finished firing {} listeners", listenerSize );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCreatedTask.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCreatedTask.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCreatedTask.java
index 7d3beb1..16a6e77 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCreatedTask.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCreatedTask.java
@@ -64,7 +64,7 @@ public class EntityVersionCreatedTask implements Task<Void> {
@Override
public Void rejected() {
- // Our task was rejected meaning our queue was full.
+ // Our task was rejected meaning our queue was full.
// We need this operation to run, so we'll run it in our current thread
try {
call();
@@ -76,7 +76,7 @@ public class EntityVersionCreatedTask implements Task<Void> {
return null;
}
-
+
@Override
public Void call() throws Exception {
@@ -100,22 +100,12 @@ public class EntityVersionCreatedTask implements Task<Void> {
logger.debug( "Started firing {} listeners", listenerSize );
- //if we have more than 1, run them on the rx scheduler for a max of 8 operations at a time
- Observable.from(listeners).parallel(
- new Func1<Observable<EntityVersionCreated>, Observable<EntityVersionCreated>>() {
-
- @Override
- public Observable<EntityVersionCreated> call(
- final Observable<EntityVersionCreated> entityVersionCreatedObservable ) {
-
- return entityVersionCreatedObservable.doOnNext( new Action1<EntityVersionCreated>() {
- @Override
- public void call( final EntityVersionCreated listener ) {
- listener.versionCreated(collectionScope,entity);
- }
- } );
- }
- }, Schedulers.io() ).toBlocking().last();
+
+ Observable.from( listeners )
+ .flatMap( currentListener -> Observable.just( currentListener ).doOnNext( listener -> {
+ listener.versionCreated( collectionScope, entity );
+ } ).subscribeOn( Schedulers.io() ), 10 ).toBlocking().last();
+
logger.debug( "Finished firing {} listeners", listenerSize );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
index e1445e3..ad1d91a 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
@@ -176,77 +176,52 @@ public abstract class MvccEntitySerializationStrategyImpl implements MvccEntityS
}
- final EntitySetImpl entitySetResults = Observable.from( rowKeys )
- //buffer our entities per request, then for that buffer, execute the query in parallel (if neccessary)
- .buffer(entitiesPerRequest )
- .parallel( new Func1<Observable<List<ScopedRowKey
- <CollectionPrefixedKey<Id>>>>, Observable<Rows<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID>>>() {
+ final EntitySetImpl entitySetResults = Observable.from( rowKeys )
+ //buffer our entities per request, then for that buffer, execute the query in parallel (if neccessary)
+ .buffer( entitiesPerRequest ).flatMap( listObservable -> {
- @Override
- public Observable<Rows<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID>> call(
- final Observable<List<ScopedRowKey<CollectionPrefixedKey<Id>>>> listObservable ) {
-
-
- //here, we execute our query then emit the items either in parallel, or on the current thread if we have more than 1 request
- return listObservable.map( new Func1<List<ScopedRowKey<CollectionPrefixedKey<Id>>>,
- Rows<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID>>() {
-
+ //here, we execute our query then emit the items either in parallel, or on the current thread
+ // if we have more than 1 request
+ return Observable.just( listObservable ).map( scopedRowKeys -> {
- @Override
- public Rows<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID> call(
- final List<ScopedRowKey<CollectionPrefixedKey<Id>>> scopedRowKeys ) {
-
- try {
- return keyspace.prepareQuery( columnFamily ).getKeySlice( rowKeys )
- .withColumnRange( maxVersion, null, false,
- 1 ).execute().getResult();
- }
- catch ( ConnectionException e ) {
- throw new CollectionRuntimeException( null, collectionScope, "An error occurred connecting to cassandra",
- e );
- }
+ try {
+ return keyspace.prepareQuery( columnFamily ).getKeySlice( rowKeys )
+ .withColumnRange( maxVersion, null, false, 1 ).execute().getResult();
}
- } );
-
-
-
- }
- }, scheduler )
-
- //reduce all the output into a single Entity set
- .reduce( new EntitySetImpl( entityIds.size() ),
- new Func2<EntitySetImpl, Rows<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID>, EntitySetImpl>() {
- @Override
- public EntitySetImpl call( final EntitySetImpl entitySet,
- final Rows<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID> rows ) {
-
- final Iterator<Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID>> latestEntityColumns = rows.iterator();
+ catch ( ConnectionException e ) {
+ throw new CollectionRuntimeException( null, collectionScope,
+ "An error occurred connecting to cassandra", e );
+ }
+ } ).subscribeOn( scheduler );
+ }, 10 )
- while ( latestEntityColumns.hasNext() ) {
- final Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID> row = latestEntityColumns.next();
+ .reduce( new EntitySetImpl( entityIds.size() ), ( entitySet, rows ) -> {
+ final Iterator<Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID>> latestEntityColumns =
+ rows.iterator();
- final ColumnList<UUID> columns = row.getColumns();
+ while ( latestEntityColumns.hasNext() ) {
+ final Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID> row = latestEntityColumns.next();
- if ( columns.size() == 0 ) {
- continue;
- }
+ final ColumnList<UUID> columns = row.getColumns();
- final Id entityId = row.getKey().getKey().getSubKey();
+ if ( columns.size() == 0 ) {
+ continue;
+ }
- final Column<UUID> column = columns.getColumnByIndex( 0 );
+ final Id entityId = row.getKey().getKey().getSubKey();
- final MvccEntity parsedEntity =
- new MvccColumnParser( entityId, getEntitySerializer() ).parseColumn( column );
+ final Column<UUID> column = columns.getColumnByIndex( 0 );
- entitySet.addEntity( parsedEntity );
- }
+ final MvccEntity parsedEntity =
+ new MvccColumnParser( entityId, getEntitySerializer() ).parseColumn( column );
+ entitySet.addEntity( parsedEntity );
+ }
- return entitySet;
- }
- } ).toBlocking().last();
+ return entitySet;
+ } ).toBlocking().last();
return entitySetResults;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyV3Impl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyV3Impl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyV3Impl.java
index a5046f6..de959b5 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyV3Impl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyV3Impl.java
@@ -185,82 +185,55 @@ public class MvccEntitySerializationStrategyV3Impl implements MvccEntitySerializ
final EntitySetImpl entitySetResults = Observable.from( rowKeys )
- //buffer our entities per request, then for that buffer, execute the query in parallel (if neccessary)
- .buffer( entitiesPerRequest ).parallel(
- new Func1<Observable<List<ScopedRowKey<CollectionPrefixedKey<Id>>>>,
- Observable<Rows<ScopedRowKey<CollectionPrefixedKey<Id>>, Boolean>>>() {
+ //buffer our entities per request, then for that buffer, execute the query in parallel (if neccessary)
+ .buffer( entitiesPerRequest ).flatMap( listObservable -> {
- @Override
- public Observable<Rows<ScopedRowKey<CollectionPrefixedKey<Id>>, Boolean>> call(
- final Observable<List<ScopedRowKey<CollectionPrefixedKey<Id>>>> listObservable ) {
+ //here, we execute our query then emit the items either in parallel, or on the current thread
+ // if we have more than 1 request
+ return Observable.just( listObservable ).map( scopedRowKeys -> {
- //here, we execute our query then emit
- // the items either in parallel, or on
- // the current thread if we have more
- // than 1 request
- return listObservable
- .map( new Func1<List<ScopedRowKey<CollectionPrefixedKey<Id>>>,
- Rows<ScopedRowKey<CollectionPrefixedKey<Id>>, Boolean>>() {
+ try {
+ return keyspace.prepareQuery( CF_ENTITY_DATA ).getKeySlice( rowKeys )
+ .withColumnSlice( COL_VALUE ).execute().getResult();
+ }
+ catch ( ConnectionException e ) {
+ throw new CollectionRuntimeException( null, collectionScope,
+ "An error occurred connecting to cassandra", e );
+ }
+ } ).subscribeOn( scheduler );
+ }, 10 )
+ .reduce( new EntitySetImpl( entityIds.size() ), ( entitySet, rows ) -> {
+ final Iterator<Row<ScopedRowKey<CollectionPrefixedKey<Id>>, Boolean>> latestEntityColumns =
+ rows.iterator();
- @Override
- public Rows<ScopedRowKey<CollectionPrefixedKey<Id>>, Boolean> call(
- final List<ScopedRowKey<CollectionPrefixedKey<Id>>> scopedRowKeys
- ) {
+ while ( latestEntityColumns.hasNext() ) {
+ final Row<ScopedRowKey<CollectionPrefixedKey<Id>>, Boolean> row = latestEntityColumns.next();
- try {
- return keyspace.prepareQuery( CF_ENTITY_DATA )
- .getKeySlice( rowKeys )
- .withColumnSlice( COL_VALUE )
- .execute().getResult();
- }
- catch ( ConnectionException e ) {
- throw new CollectionRuntimeException( null, collectionScope,
- "An error occurred connecting to cassandra", e );
- }
- }
- } );
- }
- }, scheduler )
+ final ColumnList<Boolean> columns = row.getColumns();
- //reduce all the output into a single Entity set
- .reduce( new EntitySetImpl( entityIds.size() ),
- new Func2<EntitySetImpl, Rows<ScopedRowKey<CollectionPrefixedKey<Id>>, Boolean>, EntitySetImpl>() {
- @Override
- public EntitySetImpl call( final EntitySetImpl entitySet,
- final Rows<ScopedRowKey<CollectionPrefixedKey<Id>>, Boolean> rows
- ) {
+ if ( columns.size() == 0 ) {
+ continue;
+ }
- final Iterator<Row<ScopedRowKey<CollectionPrefixedKey<Id>>, Boolean>> latestEntityColumns =
- rows.iterator();
+ final Id entityId = row.getKey().getKey().getSubKey();
- while ( latestEntityColumns.hasNext() ) {
- final Row<ScopedRowKey<CollectionPrefixedKey<Id>>, Boolean> row =
- latestEntityColumns.next();
+ final Column<Boolean> column = columns.getColumnByIndex( 0 );
- final ColumnList<Boolean> columns = row.getColumns();
+ final MvccEntity parsedEntity =
+ new MvccColumnParser( entityId, entitySerializer ).parseColumn( column );
- if ( columns.size() == 0 ) {
- continue;
- }
- final Id entityId = row.getKey().getKey().getSubKey();
+ entitySet.addEntity( parsedEntity );
+ }
- final Column<Boolean> column = columns.getColumnByIndex( 0 );
- final MvccEntity parsedEntity =
- new MvccColumnParser( entityId, entitySerializer ).parseColumn( column );
+ return entitySet;
+ } ).toBlocking().last();
- entitySet.addEntity( parsedEntity );
- }
-
-
- return entitySet;
- }
- } ).toBlocking().last();
return entitySetResults;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java
index f87b5fd..6982857 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java
@@ -124,130 +124,107 @@ public class MvccEntityDataMigrationImpl implements DataMigration<EntityIdScope>
final Observable<List<EntityToSaveMessage>> migrated =
- migrationDataProvider.getData().subscribeOn( Schedulers.io() ).parallel(
- new Func1<Observable<EntityIdScope>, Observable<List<EntityToSaveMessage>>>() {
+ migrationDataProvider.getData().subscribeOn( Schedulers.io() ).flatMap( entityToSaveList -> Observable.just( entityToSaveList ).flatMap( entityIdScope -> {
+ //load the entity
+ final CollectionScope currentScope = entityIdScope.getCollectionScope();
- //process the ids in parallel
- @Override
- public Observable<List<EntityToSaveMessage>> call(
- final Observable<EntityIdScope> entityIdScopeObservable ) {
-
-
- return entityIdScopeObservable.flatMap(
- new Func1<EntityIdScope, Observable<EntityToSaveMessage>>() {
-
-
- @Override
- public Observable<EntityToSaveMessage> call( final EntityIdScope entityIdScope ) {
- //load the entity
- final CollectionScope currentScope = entityIdScope.getCollectionScope();
+ //for each element in our
+ // history, we need to copy it
+ // to v2.
+ // Note that
+ // this migration
+ //won't support anything beyond V2
+ final Iterator<MvccEntity> allVersions =
+ migration.from.loadAscendingHistory( currentScope, entityIdScope.getId(), startTime, 100 );
- //for each element in our
- // history, we need to copy it
- // to v2.
- // Note that
- // this migration
- //won't support anything beyond V2
-
- final Iterator<MvccEntity> allVersions = migration.from
- .loadAscendingHistory( currentScope, entityIdScope.getId(), startTime, 100 );
+ //emit all the entity versions
+ return Observable.create( new Observable.OnSubscribe<EntityToSaveMessage>() {
+ @Override
+ public void call( final Subscriber<? super
+ EntityToSaveMessage> subscriber ) {
- //emit all the entity versions
- return Observable.create( new Observable.OnSubscribe<EntityToSaveMessage>() {
- @Override
- public void call( final Subscriber<? super
- EntityToSaveMessage> subscriber ) {
+ while ( allVersions.hasNext() ) {
+ final EntityToSaveMessage message =
+ new EntityToSaveMessage( currentScope, allVersions.next() );
+ subscriber.onNext( message );
+ }
- while ( allVersions.hasNext() ) {
- final EntityToSaveMessage message = new EntityToSaveMessage( currentScope, allVersions.next() );
- subscriber.onNext( message );
- }
+ subscriber.onCompleted();
+ }
+ } ).buffer( 100 ).doOnNext( entities -> {
- subscriber.onCompleted();
- }
- } );
- }
- } )
- //buffer 10 versions
- .buffer( 100 ).doOnNext( new Action1<List<EntityToSaveMessage>>() {
- @Override
- public void call( final List<EntityToSaveMessage> entities ) {
+ final MutationBatch totalBatch = keyspace.prepareMutationBatch();
- final MutationBatch totalBatch = keyspace.prepareMutationBatch();
+ atomicLong.addAndGet( entities.size() );
- atomicLong.addAndGet( entities.size() );
+ List<EntityVersionCleanupTask> entityVersionCleanupTasks = new ArrayList( entities.size() );
- List<EntityVersionCleanupTask> entityVersionCleanupTasks = new ArrayList(entities.size());
+ for ( EntityToSaveMessage message : entities ) {
+ final MutationBatch entityRewrite = migration.to.write( message.scope, message.entity );
- for ( EntityToSaveMessage message : entities ) {
- final MutationBatch entityRewrite =
- migration.to.write( message.scope, message.entity );
+ //add to
+ // the
+ // total
+ // batch
+ totalBatch.mergeShallow( entityRewrite );
- //add to
- // the
- // total
- // batch
- totalBatch.mergeShallow( entityRewrite );
+ //write
+ // the
+ // unique values
- //write
- // the
- // unique values
+ if ( !message.entity.getEntity().isPresent() ) {
+ return;
+ }
- if ( !message.entity.getEntity().isPresent() ) {
- return;
- }
+ final Entity entity = message.entity.getEntity().get();
- final Entity entity = message.entity.getEntity().get();
+ final Id entityId = entity.getId();
- final Id entityId = entity.getId();
+ final UUID version = message.entity.getVersion();
- final UUID version = message.entity.getVersion();
+ // re-write the unique
+ // values
+ // but this
+ // time with
+ // no TTL so that cleanup can clean up
+ // older values
+ for ( Field field : EntityUtils.getUniqueFields( message.entity.getEntity().get() ) ) {
- // re-write the unique
- // values
- // but this
- // time with
- // no TTL so that cleanup can clean up
- // older values
- for ( Field field : EntityUtils
- .getUniqueFields( message.entity.getEntity().get() ) ) {
+ UniqueValue written = new UniqueValueImpl( field, entityId, version );
- UniqueValue written = new UniqueValueImpl( field, entityId, version );
+ MutationBatch mb = uniqueValueSerializationStrategy.write( message.scope, written );
- MutationBatch mb =
- uniqueValueSerializationStrategy.write( message.scope, written );
+ // merge into our
+ // existing mutation
+ // batch
+ totalBatch.mergeShallow( mb );
+ }
- // merge into our
- // existing mutation
- // batch
- totalBatch.mergeShallow( mb );
- }
+ final EntityVersionCleanupTask task = entityVersionCleanupFactory
+ .getCleanupTask( message.scope, message.entity.getId(), version, false );
- final EntityVersionCleanupTask task = entityVersionCleanupFactory.getCleanupTask( message.scope, message.entity.getId(), version, false );
+ entityVersionCleanupTasks.add( task );
+ }
- entityVersionCleanupTasks.add( task );
- }
+ executeBatch( migration.to.getImplementationVersion(), totalBatch, observer, atomicLong );
- executeBatch( migration.to.getImplementationVersion(), totalBatch, observer, atomicLong );
+ //now run our cleanup task
- //now run our cleanup task
+ for ( EntityVersionCleanupTask entityVersionCleanupTask : entityVersionCleanupTasks ) {
+ try {
+ entityVersionCleanupTask.call();
+ }
+ catch ( Exception e ) {
+ LOGGER.error( "Unable to run cleanup task", e );
+ }
+ }
+ } ).subscribeOn( Schedulers.io() );
- for(EntityVersionCleanupTask entityVersionCleanupTask: entityVersionCleanupTasks){
- try {
- entityVersionCleanupTask.call();
- }
- catch ( Exception e ) {
- LOGGER.error( "Unable to run cleanup task", e );
- }
- }
- }
- } );
- }
- } );
+ }, 10) );
migrated.toBlocking().lastOrDefault(null);
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/rx/ParallelTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/rx/ParallelTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/rx/ParallelTest.java
index 2d416a4..a49e533 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/rx/ParallelTest.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/rx/ParallelTest.java
@@ -64,7 +64,7 @@ public class ParallelTest {
final int expected = size - 1;
- // QUESTION Using this thread blocks indefinitely. The execution of the Hystrix command
+ // QUESTION Using this thread blocks indefinitely. The execution of the Hystrix command
// happens on the computation Thread if this is used
// final Scheduler scheduler = Schedulers.threadPoolForComputation();
@@ -90,7 +90,7 @@ public class ParallelTest {
* non blocking?
*/
- final Observable<String> observable = Observable.from( input ).observeOn( Schedulers.io() );
+ final Observable<String> observable = Observable.just( input ).observeOn( Schedulers.io() );
Observable<Integer> thing = observable.flatMap( new Func1<String, Observable<Integer>>() {
@@ -99,7 +99,7 @@ public class ParallelTest {
public Observable<Integer> call( final String s ) {
List<Observable<Integer>> functions = new ArrayList<Observable<Integer>>();
- logger.info( "Creating new set of observables in thread {}",
+ logger.info( "Creating new set of observables in thread {}",
Thread.currentThread().getName() );
for ( int i = 0; i < size; i++ ) {
@@ -107,13 +107,13 @@ public class ParallelTest {
final int index = i;
- // create a new observable and execute the function on it.
+ // create a new observable and execute the function on it.
// These should happen in parallel when a subscription occurs
/**
* QUESTION: Should this again be the process thread, not the I/O
*/
- Observable<String> newObservable = Observable.from( input ).subscribeOn( Schedulers.io() );
+ Observable<String> newObservable = Observable.just( input ).subscribeOn( Schedulers.io() );
Observable<Integer> transformed = newObservable.map( new Func1<String, Integer>() {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/AbstractMvccEntityDataMigrationV1ToV3ImplTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/AbstractMvccEntityDataMigrationV1ToV3ImplTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/AbstractMvccEntityDataMigrationV1ToV3ImplTest.java
index 747ea7b..9938caf 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/AbstractMvccEntityDataMigrationV1ToV3ImplTest.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/AbstractMvccEntityDataMigrationV1ToV3ImplTest.java
@@ -119,7 +119,7 @@ public abstract class AbstractMvccEntityDataMigrationV1ToV3ImplTest implements D
assertEquals( "Same entity", entity2, returned2 );
final Observable<EntityIdScope> entityIdScope =
- Observable.from( new EntityIdScope( scope, entity1.getId() ), new EntityIdScope( scope, entity2.getId() ) );
+ Observable.just( new EntityIdScope( scope, entity1.getId() ), new EntityIdScope( scope, entity2.getId() ) );
final MigrationDataProvider<EntityIdScope> migrationProvider = new MigrationDataProvider<EntityIdScope>() {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/common/pom.xml
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/pom.xml b/stack/corepersistence/common/pom.xml
index 82df1d8..2be1c1a 100644
--- a/stack/corepersistence/common/pom.xml
+++ b/stack/corepersistence/common/pom.xml
@@ -101,15 +101,16 @@
<!-- RX java -->
+ <dependency>
+ <groupId>io.reactivex</groupId>
+ <artifactId>rxjava</artifactId>
+ <version>${rx.version}</version>
+ </dependency>
+
<dependency>
- <groupId>com.netflix.rxjava</groupId>
- <artifactId>rxjava-core</artifactId>
- <version>${rx.version}</version>
- </dependency>
- <dependency>
- <groupId>com.netflix.rxjava</groupId>
+ <groupId>io.reactivex</groupId>
<artifactId>rxjava-math</artifactId>
- <version>${rx.version}</version>
+ <version>1.0.0</version>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIterator.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIterator.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIterator.java
index 15f9aab..23661ee 100644
--- a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIterator.java
+++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIterator.java
@@ -77,7 +77,9 @@ public class MultiKeyColumnNameIterator<C, T> implements Iterable<T>, Iterator<T
for ( ColumnNameIterator<C, T> columnNameIterator : columnNameIterators ) {
- observables[i] = Observable.from( columnNameIterator, Schedulers.io() );
+
+
+ observables[i] = Observable.from( columnNameIterator ).subscribeOn( Schedulers.io() );
i++;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIteratorTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIteratorTest.java b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIteratorTest.java
index f4f6f9c..3c56763 100644
--- a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIteratorTest.java
+++ b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIteratorTest.java
@@ -52,6 +52,7 @@ import com.netflix.astyanax.util.RangeBuilder;
import rx.Observable;
import rx.functions.Action1;
import rx.functions.Func1;
+import rx.schedulers.Schedulers;
import static org.junit.Assert.assertEquals;
@@ -125,100 +126,95 @@ public class MultiKeyColumnNameIteratorTest {
final long maxValue = 10000;
+
/**
* Write to both rows in parallel
*/
Observable.from( new String[] { rowKey1, rowKey2, rowKey3 } )
- .parallel( new Func1<Observable<String>, Observable<String>>() {
- @Override
- public Observable<String> call( final Observable<String> stringObservable ) {
- return stringObservable.doOnNext( new Action1<String>() {
- @Override
- public void call( final String key ) {
-
- final MutationBatch batch = keyspace.prepareMutationBatch();
-
- for ( long i = 0; i < maxValue; i++ ) {
- batch.withRow( COLUMN_FAMILY, key ).putColumn( i, TRUE );
-
- if ( i % 1000 == 0 ) {
- try {
- batch.execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( e );
- }
- }
- }
-
- try {
- batch.execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( e );
- }
+ //perform a flatmap
+ .flatMap( stringObservable -> Observable.just( stringObservable ).doOnNext( key -> {
+ final MutationBatch batch = keyspace.prepareMutationBatch();
+
+ for ( long i = 0; i < maxValue; i++ ) {
+ batch.withRow( COLUMN_FAMILY, key ).putColumn( i, TRUE );
+
+ if ( i % 1000 == 0 ) {
+ try {
+ batch.execute();
+ }
+ catch ( ConnectionException e ) {
+ throw new RuntimeException( e );
}
- } );
+ }
}
- } ).toBlocking().last();
+ try {
+ batch.execute();
+ }
+ catch ( ConnectionException e ) {
+ throw new RuntimeException( e );
+ }
+ } ).subscribeOn( Schedulers.io() ) ).toBlocking().last();
- //create 3 iterators
- ColumnNameIterator<Long, Long> row1Iterator = createIterator( rowKey1, false );
- ColumnNameIterator<Long, Long> row2Iterator = createIterator( rowKey2, false );
- ColumnNameIterator<Long, Long> row3Iterator = createIterator( rowKey3, false );
- final Comparator<Long> ascendingComparator = new Comparator<Long>() {
- @Override
- public int compare( final Long o1, final Long o2 ) {
- return Long.compare( o1, o2 );
- }
- };
+ //create 3 iterators
- /**
- * Again, arbitrary buffer size to attempt we buffer at some point
- */
- final MultiKeyColumnNameIterator<Long, Long> ascendingItr =
+ ColumnNameIterator<Long, Long> row1Iterator = createIterator( rowKey1, false );
+ ColumnNameIterator<Long, Long> row2Iterator = createIterator( rowKey2, false );
+ ColumnNameIterator<Long, Long> row3Iterator = createIterator( rowKey3, false );
+
+ final Comparator<Long> ascendingComparator = new Comparator<Long>() {
+
+ @Override
+ public int compare( final Long o1, final Long o2 ) {
+ return Long.compare( o1, o2 );
+ }
+ };
+
+ /**
+ * Again, arbitrary buffer size to attempt we buffer at some point
+ */
+ final MultiKeyColumnNameIterator<Long, Long> ascendingItr =
new MultiKeyColumnNameIterator<>( Arrays.asList( row1Iterator, row2Iterator, row3Iterator ),
- ascendingComparator, 900 );
+ ascendingComparator, 900 );
- //ensure we have to make several trips, purposefully set to a nonsensical value to ensure we make all the
- // trips required
+ //ensure we have to make several trips, purposefully set to a nonsensical value to ensure we make all the
+ // trips required
- for ( long i = 0; i < maxValue; i++ ) {
- assertEquals( i, ascendingItr.next().longValue() );
- }
+ for ( long i = 0; i < maxValue; i++ ) {
+ assertEquals( i, ascendingItr.next().longValue() );
+ }
- //now test it in reverse
+ //now test it in reverse
- ColumnNameIterator<Long, Long> row1IteratorDesc = createIterator( rowKey1, true );
- ColumnNameIterator<Long, Long> row2IteratorDesc = createIterator( rowKey2, true );
- ColumnNameIterator<Long, Long> row3IteratorDesc = createIterator( rowKey3, true );
+ ColumnNameIterator<Long, Long> row1IteratorDesc = createIterator( rowKey1, true );
+ ColumnNameIterator<Long, Long> row2IteratorDesc = createIterator( rowKey2, true );
+ ColumnNameIterator<Long, Long> row3IteratorDesc = createIterator( rowKey3, true );
- final Comparator<Long> descendingComparator = new Comparator<Long>() {
+ final Comparator<Long> descendingComparator = new Comparator<Long>() {
- @Override
- public int compare( final Long o1, final Long o2 ) {
- return ascendingComparator.compare( o1, o2 ) * -1;
- }
- };
+ @Override
+ public int compare( final Long o1, final Long o2 ) {
+ return ascendingComparator.compare( o1, o2 ) * -1;
+ }
+ };
- /**
- * Again, arbitrary buffer size to attempt we buffer at some point
- */
- final MultiKeyColumnNameIterator<Long, Long> descendingItr =
+ /**
+ * Again, arbitrary buffer size to attempt we buffer at some point
+ */
+ final MultiKeyColumnNameIterator<Long, Long> descendingItr =
new MultiKeyColumnNameIterator<>( Arrays.asList( row1IteratorDesc, row2IteratorDesc, row3IteratorDesc ),
- descendingComparator, 900 );
+ descendingComparator, 900 );
- for ( long i = maxValue - 1; i > -1; i-- ) {
- assertEquals( i, descendingItr.next().longValue() );
+ for ( long i = maxValue - 1; i > -1; i-- ) {
+ assertEquals( i, descendingItr.next().longValue() );
+ }
}
- }
@Test
@@ -233,39 +229,28 @@ public class MultiKeyColumnNameIteratorTest {
/**
* Write to both rows in parallel
*/
- Observable.just( rowKey1 )
- .parallel( new Func1<Observable<String>, Observable<String>>() {
- @Override
- public Observable<String> call( final Observable<String> stringObservable ) {
- return stringObservable.doOnNext( new Action1<String>() {
- @Override
- public void call( final String key ) {
-
- final MutationBatch batch = keyspace.prepareMutationBatch();
-
- for ( long i = 0; i < maxValue; i++ ) {
- batch.withRow( COLUMN_FAMILY, key ).putColumn( i, TRUE );
-
- if ( i % 1000 == 0 ) {
- try {
- batch.execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( e );
- }
- }
- }
-
- try {
- batch.execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( e );
- }
- }
- } );
- }
- } ).toBlocking().last();
+ Observable.just( rowKey1 ).flatMap( rowKey -> Observable.just( rowKey ).doOnNext( key -> {
+ final MutationBatch batch = keyspace.prepareMutationBatch();
+
+ for ( long i = 0; i < maxValue; i++ ) {
+ batch.withRow( COLUMN_FAMILY, key ).putColumn( i, TRUE );
+
+ if ( i % 1000 == 0 ) {
+ try {
+ batch.execute();
+ }
+ catch ( ConnectionException e ) {
+ throw new RuntimeException( e );
+ }
+ }
+ }
+
+ try {
+ batch.execute();
+ }
+ catch ( ConnectionException e ) {
+ throw new RuntimeException( e );
+ }} ).subscribeOn( Schedulers.io() ) ).toBlocking().last();
//create 3 iterators
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/282e2271/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/astyanax/MultiRowColumnIteratorTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/astyanax/MultiRowColumnIteratorTest.java b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/astyanax/MultiRowColumnIteratorTest.java
index c32b820..d88ebe5 100644
--- a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/astyanax/MultiRowColumnIteratorTest.java
+++ b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/astyanax/MultiRowColumnIteratorTest.java
@@ -54,6 +54,7 @@ import rx.Observable;
import rx.Observer;
import rx.functions.Action1;
import rx.functions.Func1;
+import rx.schedulers.Schedulers;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -373,38 +374,29 @@ public class MultiRowColumnIteratorTest {
/**
* Write to both rows in parallel
*/
- Observable.just( rowKey1 ).parallel( new Func1<Observable<String>, Observable<String>>() {
- @Override
- public Observable<String> call( final Observable<String> stringObservable ) {
- return stringObservable.doOnNext( new Action1<String>() {
- @Override
- public void call( final String key ) {
-
- final MutationBatch batch = keyspace.prepareMutationBatch();
-
- for ( long i = 0; i < maxValue; i++ ) {
- batch.withRow( COLUMN_FAMILY, key ).putColumn( i, TRUE );
-
- if ( i % 1000 == 0 ) {
- try {
- batch.execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( e );
- }
- }
- }
+ Observable.just( rowKey1 ).flatMap( rowKey -> Observable.just( rowKey ).doOnNext( key -> {
+ final MutationBatch batch = keyspace.prepareMutationBatch();
- try {
- batch.execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( e );
- }
+ for ( long i = 0; i < maxValue; i++ ) {
+ batch.withRow( COLUMN_FAMILY, key ).putColumn( i, TRUE );
+
+ if ( i % 1000 == 0 ) {
+ try {
+ batch.execute();
+ }
+ catch ( ConnectionException e ) {
+ throw new RuntimeException( e );
}
- } );
+ }
+ }
+
+ try {
+ batch.execute();
+ }
+ catch ( ConnectionException e ) {
+ throw new RuntimeException( e );
}
- } ).toBlocking().last();
+ } ).subscribeOn( Schedulers.io() ) ).toBlocking().last();
//create 3 iterators
[3/8] incubator-usergrid git commit: Resolves class loader problems
during test
Posted by to...@apache.org.
Resolves class loader problems during test
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/61067f13
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/61067f13
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/61067f13
Branch: refs/heads/USERGRID-486
Commit: 61067f1300353b5a58f4b40e28faeae3977bdc4a
Parents: 282e227
Author: Todd Nine <tn...@apigee.com>
Authored: Thu Mar 19 18:31:34 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Thu Mar 19 18:31:34 2015 -0600
----------------------------------------------------------------------
stack/core/pom.xml | 7 +++----
stack/pom.xml | 6 ++++--
stack/rest/pom.xml | 1 +
stack/services/pom.xml | 1 +
4 files changed, 9 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/61067f13/stack/core/pom.xml
----------------------------------------------------------------------
diff --git a/stack/core/pom.xml b/stack/core/pom.xml
index 119a52b..ba73150 100644
--- a/stack/core/pom.xml
+++ b/stack/core/pom.xml
@@ -88,10 +88,9 @@
<forkCount>${usergrid.it.forkCount}</forkCount>
<threadCount>${usergrid.it.threads}</threadCount>
<reuseForks>true</reuseForks>
- <argLine>-Dtest.barrier.timestamp=${maven.build.timestamp} -Dtest.clean.storage=true -Xmx${ug.heapmax}
- -Xms${ug.heapmin} -Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8
- -javaagent:${settings.localRepository}/com/github/stephenc/jamm/0.2.5/jamm-0.2.5.jar ${ug.argline}
- </argLine>
+ <argLine>-Dtest.barrier.timestamp=${maven.build.timestamp} -Dtest.clean.storage=true -Xmx${ug.heapmax} -Xms${ug.heapmin} -Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8 -javaagent:${settings.localRepository}/com/github/stephenc/jamm/0.2.5/jamm-0.2.5.jar ${ug.argline} </argLine>
+ <!-- see this page for documentation on classloading issues http://maven.apache.org/surefire/maven-surefire-plugin/examples/class-loading.html -->
+ <useSystemClassLoader>false</useSystemClassLoader>
<includes>
<include>**/*IT.java</include>
<include>**/*Test.java</include>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/61067f13/stack/pom.xml
----------------------------------------------------------------------
diff --git a/stack/pom.xml b/stack/pom.xml
index efbda2d..a107114 100644
--- a/stack/pom.xml
+++ b/stack/pom.xml
@@ -109,7 +109,7 @@
<jersey-version>1.18.1</jersey-version>
<junit-version>4.12</junit-version>
<log4j-version>1.2.16</log4j-version>
- <org.springframework.version>3.1.2.RELEASE</org.springframework.version>
+ <org.springframework.version>3.2.13.RELEASE</org.springframework.version>
<shiro-version>1.2.3</shiro-version>
<slf4j-version>1.6.1</slf4j-version>
<snakeyaml-version>1.8</snakeyaml-version>
@@ -1461,6 +1461,8 @@
<forkCount>${usergrid.it.forkCount}</forkCount>
<reuseForks>${usergrid.it.reuseForks}</reuseForks>
<threadCount>${usergrid.it.forkCount}</threadCount>
+ <!-- see this page for documentation on classloading issues http://maven.apache.org/surefire/maven-surefire-plugin/examples/class-loading.html -->
+ <useSystemClassLoader>false</useSystemClassLoader>
<argLine>-Xmx${ug.heapmax} -Xms${ug.heapmin} -javaagent:${settings.localRepository}/com/github/stephenc/jamm/0.2.5/jamm-0.2.5.jar ${ug.argline}</argLine>
<testFailureIgnore>false</testFailureIgnore>
</configuration>
@@ -1471,7 +1473,7 @@
<dependencies>
<dependency>
<groupId>org.apache.maven.surefire</groupId>
- <artifactId>surefire-junit4</artifactId>
+ <artifactId>${surefire.plugin.artifactName}</artifactId>
<version>${surefire.plugin.version}</version>
<!--<exclusions>-->
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/61067f13/stack/rest/pom.xml
----------------------------------------------------------------------
diff --git a/stack/rest/pom.xml b/stack/rest/pom.xml
index deb11f8..974945f 100644
--- a/stack/rest/pom.xml
+++ b/stack/rest/pom.xml
@@ -102,6 +102,7 @@
<parallel>methods</parallel>
<forkCount>1</forkCount>
<threadCount>${usergrid.rest.threads}</threadCount>
+ <useSystemClassLoader>false</useSystemClassLoader>
<reuseForks>true</reuseForks>
<argLine>-Dwebapp.directory=${basedir}/src/main/webapp -Dtest.barrier.timestamp=${maven.build.timestamp} -Dtest.clean.storage=true -Xmx${ug.heapmax} -Xms${ug.heapmin} -Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8 -javaagent:${settings.localRepository}/com/github/stephenc/jamm/0.2.5/jamm-0.2.5.jar ${ug.argline}
</argLine>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/61067f13/stack/services/pom.xml
----------------------------------------------------------------------
diff --git a/stack/services/pom.xml b/stack/services/pom.xml
index 68f1a4a..6074e65 100644
--- a/stack/services/pom.xml
+++ b/stack/services/pom.xml
@@ -92,6 +92,7 @@
<forkCount>${usergrid.it.forkCount}</forkCount>
<threadCount>${usergrid.it.threads}</threadCount>
<reuseForks>${usergrid.it.reuseForks}</reuseForks>
+ <useSystemClassLoader>false</useSystemClassLoader>
<argLine>-Dtest.barrier.timestamp=${maven.build.timestamp} -Dtest.clean.storage=true -Xmx${ug.heapmax} -Xms${ug.heapmin} -Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8 -javaagent:${settings.localRepository}/com/github/stephenc/jamm/0.2.5/jamm-0.2.5.jar ${ug.argline}
</argLine>
<includes>
[4/8] incubator-usergrid git commit: Added error logging to the
iterator
Posted by to...@apache.org.
Added error logging to the iterator
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/c7e3459e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/c7e3459e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/c7e3459e
Branch: refs/heads/USERGRID-486
Commit: c7e3459ec79ed64e74adba0c5fc9c6dc8cd12252
Parents: 61067f1
Author: Todd Nine <tn...@apigee.com>
Authored: Thu Mar 19 19:46:58 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Thu Mar 19 19:46:58 2015 -0600
----------------------------------------------------------------------
.../java/org/apache/usergrid/persistence/ObservableIterator.java | 1 +
.../usergrid/persistence/PerformanceEntityRebuildIndexTest.java | 4 +---
.../usergrid/persistence/index/impl/EsEntityIndexImpl.java | 2 +-
3 files changed, 3 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c7e3459e/stack/core/src/main/java/org/apache/usergrid/persistence/ObservableIterator.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/ObservableIterator.java b/stack/core/src/main/java/org/apache/usergrid/persistence/ObservableIterator.java
index b91bd22..9befb79 100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/ObservableIterator.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/ObservableIterator.java
@@ -70,6 +70,7 @@ public abstract class ObservableIterator<T> implements Observable.OnSubscribe<T>
//if any error occurs, we need to notify the observer so it can perform it's own error handling
catch ( Throwable t ) {
+ log.error( "Unable to emit items from iterator {}", name, t );
subscriber.onError( t );
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c7e3459e/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java b/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
index f1f165d..52d4a48 100644
--- a/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
@@ -81,7 +81,6 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
@After
public void printReport() {
-
logger.debug("Printing metrics report");
reporter.report();
reporter.stop();
@@ -428,8 +427,7 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
}
if ( expectedEntities != -1 && expectedEntities != count ) {
- throw new RuntimeException("Did not get expected "
- + expectedEntities + " entities, instead got " + count );
+ throw new RuntimeException("Did not get expected " + expectedEntities + " entities, instead got " + count );
}
return count;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c7e3459e/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 8814583..d92ae7d 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
@@ -662,7 +662,7 @@ public class EsEntityIndexImpl implements AliasedEntityIndex {
if ( response.isAcknowledged() ) {
logger.info( "Deleted index: read {} write {}", alias.getReadAlias(), alias.getWriteAlias());
- //invlaidate the alias
+ //invalidate the alias
aliasCache.invalidate(alias);
}
else {
[8/8] incubator-usergrid git commit: Merge branch 'USERGRID-451' into
USERGRID-405
Posted by to...@apache.org.
Merge branch 'USERGRID-451' into USERGRID-405
Upgrade to Java 8
Upgrades RX to latest release
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/dcf46937
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/dcf46937
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/dcf46937
Branch: refs/heads/USERGRID-486
Commit: dcf469378f5d68551cc452368bc6a5378926c7aa
Parents: 0a16033 6dc742e
Author: Todd Nine <tn...@apigee.com>
Authored: Fri Mar 20 07:41:23 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Fri Mar 20 07:41:23 2015 -0600
----------------------------------------------------------------------
stack/core/pom.xml | 26 +-
.../corepersistence/CpEntityManager.java | 2 +-
.../corepersistence/CpEntityManagerFactory.java | 7 +-
.../corepersistence/CpRelationManager.java | 24 +-
.../usergrid/corepersistence/CpWalker.java | 81 ++---
.../events/EntityVersionDeletedHandler.java | 72 ++--
.../migration/EntityTypeMappingMigration.java | 41 +--
.../persistence/ObservableIterator.java | 1 +
.../migration/EntityTypeMappingMigrationIT.java | 2 +-
.../PerformanceEntityRebuildIndexTest.java | 4 +-
stack/corepersistence/collection/pom.xml | 14 +-
.../impl/EntityCollectionManagerImpl.java | 10 +-
.../collection/impl/EntityDeletedTask.java | 20 +-
.../impl/EntityVersionCleanupTask.java | 40 +--
.../impl/EntityVersionCreatedTask.java | 26 +-
.../mvcc/stage/write/WriteCommit.java | 2 +-
.../mvcc/stage/write/WriteUniqueVerify.java | 8 +-
.../MvccEntitySerializationStrategyImpl.java | 92 ++---
.../MvccEntitySerializationStrategyV3Impl.java | 94 ++---
.../UniqueValueSerializationStrategyImpl.java | 8 -
.../migration/MvccEntityDataMigrationImpl.java | 171 ++++------
.../collection/util/EntityUtils.java | 72 ----
.../mvcc/stage/AbstractEntityStageTest.java | 2 +-
.../mvcc/stage/AbstractMvccEntityStageTest.java | 2 +-
.../mvcc/stage/TestEntityGenerator.java | 2 +-
.../persistence/collection/rx/ParallelTest.java | 10 +-
...MvccEntitySerializationStrategyImplTest.java | 4 +-
...ccEntitySerializationStrategyV1ImplTest.java | 4 +-
...ccEntitySerializationStrategyV2ImplTest.java | 2 +-
.../impl/SerializationComparison.java | 4 +-
...ctMvccEntityDataMigrationV1ToV3ImplTest.java | 2 +-
.../collection/util/InvalidEntityGenerator.java | 1 +
stack/corepersistence/common/pom.xml | 15 +-
.../astyanax/MultiKeyColumnNameIterator.java | 4 +-
.../MultiKeyColumnNameIteratorTest.java | 187 +++++-----
.../astyanax/MultiRowColumnIteratorTest.java | 50 ++-
.../graph/impl/GraphManagerImpl.java | 6 +-
.../graph/impl/stage/EdgeMetaRepairImpl.java | 2 +
.../impl/stage/NodeDeleteListenerImpl.java | 2 +-
.../impl/migration/EdgeDataMigrationImpl.java | 87 +++--
.../persistence/graph/GraphManagerIT.java | 17 +-
.../graph/GraphManagerShardConsistencyIT.java | 2 +-
.../usergrid/persistence/graph/SimpleTest.java | 12 +-
.../migration/EdgeDataMigrationImplTest.java | 2 +-
stack/corepersistence/model/pom.xml | 1 -
.../persistence/model/util/EntityUtils.java | 72 ++++
stack/corepersistence/pom.xml | 8 +-
stack/corepersistence/queryindex/pom.xml | 6 -
.../index/impl/EsEntityIndexImpl.java | 2 +-
.../persistence/index/query/EntityResults.java | 108 ------
.../persistence/index/query/Results.java | 148 --------
.../persistence/index/utils/ListUtils.java | 6 +-
.../persistence/index/guice/IndexTestFig.java | 57 ++++
.../index/guice/TestIndexModule.java | 8 +-
.../index/impl/CorePerformanceIT.java | 339 -------------------
.../impl/EntityConnectionIndexImplTest.java | 306 -----------------
.../persistence/index/impl/EntityIndexTest.java | 32 +-
.../index/impl/IndexLoadTestsIT.java | 138 ++++++++
stack/pom.xml | 8 +-
.../management/importer/ImportServiceImpl.java | 34 +-
.../impl/ApplicationQueueManagerImpl.java | 195 +++++------
.../setup/ConcurrentProcessSingleton.java | 16 +-
62 files changed, 900 insertions(+), 1820 deletions(-)
----------------------------------------------------------------------
[7/8] incubator-usergrid git commit: Fixes test that won't work when
the system runs too quickly.
Posted by to...@apache.org.
Fixes test that won't work when the system runs too quickly.
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/6dc742e4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/6dc742e4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/6dc742e4
Branch: refs/heads/USERGRID-486
Commit: 6dc742e402bf75a8cf6bb0f2fa3ffc4c07367263
Parents: 5889a3b
Author: Todd Nine <tn...@apigee.com>
Authored: Fri Mar 20 07:30:55 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Fri Mar 20 07:30:55 2015 -0600
----------------------------------------------------------------------
.../usergrid/persistence/graph/GraphManagerIT.java | 17 +++++++++++------
1 file changed, 11 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/6dc742e4/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerIT.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerIT.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerIT.java
index ef5d69d..05b36fe 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerIT.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerIT.java
@@ -964,20 +964,25 @@ public abstract class GraphManagerIT {
Id targetId2 = new SimpleId( "target2" );
- Edge edge1 = createEdge( sourceId, "test", targetId1, System.currentTimeMillis() );
+ long startTime = System.currentTimeMillis();
+
+ long edge1Time = startTime;
+ long edge2Time = edge1Time+1;
+
+ final long maxVersion= edge2Time;
+
+ Edge edge1 = createEdge( sourceId, "test", targetId1, edge1Time);
gm.writeEdge( edge1 ).toBlocking().singleOrDefault( null );
- Edge edge2 = createEdge( sourceId, "test", targetId2, System.currentTimeMillis() );
+ Edge edge2 = createEdge( sourceId, "test", targetId2, edge2Time );
gm.writeEdge( edge2 ).toBlocking().singleOrDefault( null );
- final long maxVersion = System.currentTimeMillis();
-
- assertTrue( Long.compare( maxVersion, edge2.getTimestamp() ) > 0 );
- assertTrue( Long.compare( maxVersion, edge1.getTimestamp() ) > 0 );
+ assertTrue( Long.compare( maxVersion, edge2.getTimestamp() ) >= 0 );
+ assertTrue( Long.compare( maxVersion, edge1.getTimestamp() ) >= 0 );
//get our 2 edges
[6/8] incubator-usergrid git commit: Fixes rebuild bug
Posted by to...@apache.org.
Fixes rebuild bug
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/5889a3b4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/5889a3b4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/5889a3b4
Branch: refs/heads/USERGRID-486
Commit: 5889a3b4d1ae7afe450a76594f3de8f9ba300f8c
Parents: c7e3459
Author: Todd Nine <tn...@apigee.com>
Authored: Thu Mar 19 19:53:55 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Thu Mar 19 19:53:55 2015 -0600
----------------------------------------------------------------------
.../java/org/apache/usergrid/corepersistence/CpWalker.java | 8 +++-----
1 file changed, 3 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5889a3b4/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
index 332d5a8..c14447d 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
@@ -101,10 +101,8 @@ public class CpWalker {
edgeType = CpNamingUtils.getEdgeTypeFromCollectionName( collectionName );
}
- Observable<String> edgeTypes = gm.getEdgeTypesFromSource(
- new SimpleSearchEdgeType( applicationId, edgeType, null ) );
-
- edgeTypes.flatMap( emittedEdgeType -> {
+ Observable<Edge> edges = gm.getEdgeTypesFromSource(
+ new SimpleSearchEdgeType( applicationId, edgeType, null ) ).flatMap( emittedEdgeType -> {
logger.debug( "Loading edges of type {} from node {}", edgeType, applicationId );
@@ -136,6 +134,6 @@ public class CpWalker {
}, 100 );
// wait for it to complete
- edgeTypes.toBlocking().lastOrDefault( null ); // end foreach on edges
+ edges.toBlocking().lastOrDefault( null ); // end foreach on edges
}
}