You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sf...@apache.org on 2015/02/14 00:05:04 UTC
incubator-usergrid git commit: tests passing again
Repository: incubator-usergrid
Updated Branches:
refs/heads/USERGRID-365 db2585ec2 -> e52246952
tests passing again
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/e5224695
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/e5224695
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/e5224695
Branch: refs/heads/USERGRID-365
Commit: e5224695202fd5533ea027964f7fc0e515f50678
Parents: db2585e
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Feb 13 16:04:59 2015 -0700
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Feb 13 16:04:59 2015 -0700
----------------------------------------------------------------------
.../impl/AllEntitiesInSystemObservableImpl.java | 104 ++++++++++---------
.../rx/ApplicationObservableTestIT.java | 2 +-
.../rx/EdgesFromSourceObservableIT.java | 8 +-
.../rx/TargetIdObservableTestIT.java | 2 +-
.../data/DataMigrationManagerImpl.java | 29 +++---
.../core/rx/AllEntitiesInSystemObservable.java | 9 ++
.../data/DataMigrationManagerImplTest.java | 5 +
.../rx/AllEntitiesInSystemTestObservable.java | 5 +
8 files changed, 92 insertions(+), 72 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e5224695/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntitiesInSystemObservableImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntitiesInSystemObservableImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntitiesInSystemObservableImpl.java
index 4f5909a..a4fbc5d 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntitiesInSystemObservableImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntitiesInSystemObservableImpl.java
@@ -60,59 +60,61 @@ public class AllEntitiesInSystemObservableImpl implements AllEntitiesInSystemObs
this.targetIdObservable = targetIdObservable;
}
-
public Observable<ApplicationEntityGroup<CollectionScope>> getAllEntitiesInSystem(final int bufferSize) {
+ return getAllEntitiesInSystem(applicationObservable.getAllApplicationIds( ),bufferSize);
+
+ }
+
+ public Observable<ApplicationEntityGroup<CollectionScope>> getAllEntitiesInSystem(final Observable<Id> appIdsObservable, final int bufferSize) {
//traverse all nodes in the graph, load all source edges from them, then re-save the meta data
- return applicationObservable.getAllApplicationIds( )
-
- .flatMap( new Func1<Id, Observable<ApplicationEntityGroup<CollectionScope>>>() {
- @Override
- public Observable<ApplicationEntityGroup<CollectionScope>> call( final Id applicationId ) {
-
- //set up our application scope and graph manager
- final ApplicationScope applicationScope = new ApplicationScopeImpl(
- applicationId );
-
- final GraphManager gm = graphManagerFactory.createEdgeManager(applicationScope);
-
- //load all nodes that are targets of our application node. I.E.
- // entities that have been saved
- final Observable<Id> entityNodes =
- targetIdObservable.getTargetNodes(gm, applicationId);
-
-
- //get scope here
-
-
- //emit Scope + ID
-
- //create our application node to emit since it's an entity as well
- final Observable<Id> applicationNode = Observable.just( applicationId );
-
- //merge both the specified application node and the entity node
- // so they all get used
- return Observable
- .merge(applicationNode, entityNodes)
- .buffer(bufferSize)
- .map(new Func1<List<Id>, List<EntityIdScope<CollectionScope>>>() {
- @Override
- public List<EntityIdScope<CollectionScope>> call(List<Id> ids) {
- List<EntityIdScope<CollectionScope>> scopes = new ArrayList<>(ids.size());
- for (Id id : ids) {
- CollectionScope scope = CpNamingUtils.getCollectionScopeNameFromEntityType(applicationId, id.getType());
- EntityIdScope<CollectionScope> idScope = new EntityIdScope<>(id, scope);
- scopes.add(idScope);
- }
- return scopes;
- }
- })
- .map(new Func1<List<EntityIdScope<CollectionScope>>, ApplicationEntityGroup<CollectionScope>>() {
- @Override
- public ApplicationEntityGroup<CollectionScope> call(final List<EntityIdScope<CollectionScope>> scopes) {
- return new ApplicationEntityGroup<>(applicationScope, scopes);
- }
- });
- }
+ return appIdsObservable.flatMap(new Func1<Id, Observable<ApplicationEntityGroup<CollectionScope>>>() {
+ @Override
+ public Observable<ApplicationEntityGroup<CollectionScope>> call(final Id applicationId) {
+
+ //set up our application scope and graph manager
+ final ApplicationScope applicationScope = new ApplicationScopeImpl(
+ applicationId);
+
+ final GraphManager gm = graphManagerFactory.createEdgeManager(applicationScope);
+
+ //load all nodes that are targets of our application node. I.E.
+ // entities that have been saved
+ final Observable<Id> entityNodes =
+ targetIdObservable.getTargetNodes(gm, applicationId);
+
+
+ //get scope here
+
+
+ //emit Scope + ID
+
+ //create our application node to emit since it's an entity as well
+ final Observable<Id> applicationNode = Observable.just(applicationId);
+
+ //merge both the specified application node and the entity node
+ // so they all get used
+ return Observable
+ .merge(applicationNode, entityNodes)
+ .buffer(bufferSize)
+ .map(new Func1<List<Id>, List<EntityIdScope<CollectionScope>>>() {
+ @Override
+ public List<EntityIdScope<CollectionScope>> call(List<Id> ids) {
+ List<EntityIdScope<CollectionScope>> scopes = new ArrayList<>(ids.size());
+ for (Id id : ids) {
+ CollectionScope scope = CpNamingUtils.getCollectionScopeNameFromEntityType(applicationId, id.getType());
+ EntityIdScope<CollectionScope> idScope = new EntityIdScope<>(id, scope);
+ scopes.add(idScope);
+ }
+ return scopes;
+ }
+ })
+ .map(new Func1<List<EntityIdScope<CollectionScope>>, ApplicationEntityGroup<CollectionScope>>() {
+ @Override
+ public ApplicationEntityGroup<CollectionScope> call(final List<EntityIdScope<CollectionScope>> scopes) {
+ return new ApplicationEntityGroup<>(applicationScope, scopes);
+ }
+ });
+ }
} );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e5224695/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/ApplicationObservableTestIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/ApplicationObservableTestIT.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/ApplicationObservableTestIT.java
index 2f29373..081bbed 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/ApplicationObservableTestIT.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/ApplicationObservableTestIT.java
@@ -52,7 +52,7 @@ public class ApplicationObservableTestIT extends AbstractCoreIT {
final Application createdApplication = app.getEntityManager().getApplication();
- ApplicationObservable applicationObservable =SpringResource.getInstance().getBean(ApplicationObservable.class);
+ ApplicationObservable applicationObservable =SpringResource.getInstance().getBean(Injector.class).getInstance(ApplicationObservable.class);
//now our get all apps we expect. There may be more, but we don't care about those.
final Set<UUID> applicationIds = new HashSet<UUID>() {{
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e5224695/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/EdgesFromSourceObservableIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/EdgesFromSourceObservableIT.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/EdgesFromSourceObservableIT.java
index c013c1e..50c2cd9 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/EdgesFromSourceObservableIT.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/EdgesFromSourceObservableIT.java
@@ -60,18 +60,14 @@ public class EdgesFromSourceObservableIT extends AbstractCoreIT {
@Test
public void testEntities() throws Exception {
- EdgesObservable edgesToTargetObservable = SpringResource.getInstance().getBean(EdgesObservable.class);
+ EdgesObservable edgesToTargetObservable = SpringResource.getInstance().getBean(Injector.class).getInstance(EdgesObservable.class);
final EntityManager em = app.getEntityManager();
final Application createdApplication = em.getApplication();
-
-
final String type1 = "targetthings";
final String type2 = "sourcethings";
final int size = 10;
-
-
final Set<Id> sourceIdentities = EntityWriteHelper.createTypes( em, type2, size );
@@ -79,8 +75,6 @@ public class EdgesFromSourceObservableIT extends AbstractCoreIT {
final Id target = new SimpleId( entity.getUuid(), entity.getType() );
-
-
for ( Id source : sourceIdentities ) {
em.createConnection( SimpleEntityRef.fromId( source ), "likes", SimpleEntityRef.fromId( target ) );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e5224695/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/TargetIdObservableTestIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/TargetIdObservableTestIT.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/TargetIdObservableTestIT.java
index 5926d40..270f6d9 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/TargetIdObservableTestIT.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/TargetIdObservableTestIT.java
@@ -54,7 +54,7 @@ public class TargetIdObservableTestIT extends AbstractCoreIT {
@Test
public void testEntities() throws Exception {
- TargetIdObservable targetIdObservable = SpringResource.getInstance().getBean(TargetIdObservable.class);
+ TargetIdObservable targetIdObservable = SpringResource.getInstance().getBean(Injector.class).getInstance(TargetIdObservable.class);
final EntityManager em = app.getEntityManager();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e5224695/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImpl.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImpl.java
index a7fe2a4..b8e4af9 100644
--- a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImpl.java
+++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImpl.java
@@ -97,7 +97,9 @@ public class DataMigrationManagerImpl implements DataMigrationManager {
int currentVersion = migrationInfoSerialization.getCurrentVersion();
if(currentVersion <= 0){
- currentVersion = resetToHighestVersion();
+ resetToHighestVersion();
+ //you have no work to do, just use the latest
+ return;
}
LOG.info( "Saved schema version is {}, max migration version is {}", currentVersion,
@@ -110,6 +112,7 @@ public class DataMigrationManagerImpl implements DataMigrationManager {
final CassandraProgressObserver observer = new CassandraProgressObserver();
Observable<DataMigration> migrations = Observable.from(migrationsToRun.values()).subscribeOn(Schedulers.io());
+ final Observable appIdObservable = applicationObservable.getAllApplicationIds();
Observable entityMigrations = migrations
.filter(new Func1<DataMigration, Boolean>() {
@@ -121,7 +124,7 @@ public class DataMigrationManagerImpl implements DataMigrationManager {
@Override
public Observable<ApplicationEntityGroup> call(final DataMigration dataMigration) {
return allEntitiesInSystemObservable
- .getAllEntitiesInSystem(1000)
+ .getAllEntitiesInSystem(appIdObservable, 1000)
.doOnNext(new Action1<ApplicationEntityGroup>() {
@Override
public void call(ApplicationEntityGroup entityGroup) {
@@ -141,22 +144,24 @@ public class DataMigrationManagerImpl implements DataMigrationManager {
}).flatMap(new Func1<DataMigration, Observable<?>>() {
@Override
public Observable call(final DataMigration dataMigration) {
- return applicationObservable.getAllApplicationIds()
- .doOnNext(new Action1<Id>() {
- @Override
- public void call(Id id) {
- ApplicationScope scope = new ApplicationScopeImpl(id);
- runMigration(dataMigration, observer, new ApplicationEntityGroup(scope, null));
- }
- });
+ return appIdObservable.doOnNext(new Action1<Id>() {
+ @Override
+ public void call(Id id) {
+ ApplicationScope scope = new ApplicationScopeImpl(id);
+ runMigration(dataMigration, observer, new ApplicationEntityGroup(scope, null));
+ }
+ });
}
});
try {
- Observable.merge(entityMigrations, otherMigrations).toBlocking().lastOrDefault(null);
+ Observable
+ .merge(entityMigrations, otherMigrations)
+ .subscribeOn(Schedulers.io())
+ .toBlocking().lastOrDefault(null);
migrationInfoSerialization.setStatusCode( StatusCode.COMPLETE.status );
- }catch (Exception e){
+ } catch (Exception e){
LOG.error("Migration Failed");
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e5224695/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/rx/AllEntitiesInSystemObservable.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/rx/AllEntitiesInSystemObservable.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/rx/AllEntitiesInSystemObservable.java
index e670fbb..141b6c1 100644
--- a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/rx/AllEntitiesInSystemObservable.java
+++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/rx/AllEntitiesInSystemObservable.java
@@ -39,6 +39,15 @@ public interface AllEntitiesInSystemObservable<T extends ApplicationScope> {
*/
public Observable<ApplicationEntityGroup<T>> getAllEntitiesInSystem(final int bufferSize);
+ /**
+ * Return an observable that emits all entities in the system.
+ *
+ * @param appIdObservable list of app ids
+ * @param bufferSize The amount of entityIds to buffer into each ApplicationEntityGroup. Note that if we exceed the buffer size
+ * you may be more than 1 ApplicationEntityGroup with the same application and different ids
+ */
+ public Observable<ApplicationEntityGroup<T>> getAllEntitiesInSystem(Observable<Id> appIdObservable, final int bufferSize);
+
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e5224695/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImplTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImplTest.java b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImplTest.java
index 305ab67..0c7e76b 100644
--- a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImplTest.java
+++ b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImplTest.java
@@ -63,6 +63,11 @@ public class DataMigrationManagerImplTest {
}
});
}
+
+ @Override
+ public Observable<ApplicationEntityGroup> getAllEntitiesInSystem(Observable appIdObservable, int bufferSize) {
+ return this.getAllEntitiesInSystem(bufferSize) ;
+ }
};
ApplicationObservable allApplicationsObservable = new ApplicationObservable() {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e5224695/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/AllEntitiesInSystemTestObservable.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/AllEntitiesInSystemTestObservable.java b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/AllEntitiesInSystemTestObservable.java
index fd19e9c..bf244e0 100644
--- a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/AllEntitiesInSystemTestObservable.java
+++ b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/AllEntitiesInSystemTestObservable.java
@@ -31,4 +31,9 @@ public class AllEntitiesInSystemTestObservable implements AllEntitiesInSystemObs
public Observable<ApplicationEntityGroup> getAllEntitiesInSystem(int bufferSize) {
return Observable.empty();
}
+
+ @Override
+ public Observable<ApplicationEntityGroup> getAllEntitiesInSystem(Observable appIdObservable, int bufferSize) {
+ return this.getAllEntitiesInSystem(bufferSize);
+ }
}