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/10 19:05:20 UTC
[2/4] incubator-usergrid git commit: moving dependencies to core
persistence
moving dependencies to core persistence
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/7daca75e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/7daca75e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/7daca75e
Branch: refs/heads/USERGRID-365
Commit: 7daca75ef9954b6da4d0a549755cf2afe85834dc
Parents: dd5e6ea
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Feb 9 15:59:30 2015 -0700
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Feb 9 15:59:30 2015 -0700
----------------------------------------------------------------------
.../corepersistence/CpEntityManagerFactory.java | 21 +--
.../corepersistence/CpManagerCache.java | 71 ++--------
.../usergrid/corepersistence/GuiceModule.java | 13 +-
.../usergrid/corepersistence/ManagerCache.java | 12 +-
.../migration/EntityDataMigration.java | 28 ++--
.../migration/EntityTypeMappingMigration.java | 14 +-
.../migration/GraphShardVersionMigration.java | 50 +++----
.../rx/AllEntitiesInSystemObservable.java | 90 ------------
.../rx/ApplicationObservable.java | 128 -----------------
.../rx/EdgesFromSourceObservable.java | 63 ---------
.../corepersistence/rx/TargetIdObservable.java | 9 +-
.../impl/AllEntitiesInSystemObservableImpl.java | 95 +++++++++++++
.../rx/impl/ApplicationObservableImpl.java | 135 ++++++++++++++++++
.../migration/EntityDataMigrationIT.java | 8 +-
.../migration/EntityTypeMappingMigrationIT.java | 4 +-
.../migration/GraphShardVersionMigrationIT.java | 6 +-
.../rx/AllEntitiesInSystemObservableIT.java | 3 +-
.../rx/ApplicationObservableTestIT.java | 5 +-
.../collection/EntityCollectionManager.java | 8 +-
.../EntityCollectionManagerFactory.java | 22 +--
.../collection/guice/CollectionModule.java | 8 +-
.../EntityCollectionManagerFactoryImpl.java | 137 +++++++++++++++++++
.../usergrid/persistence/core/CPManager.java | 28 ++++
.../core/entity/ApplicationEntityGroup.java | 40 ++++++
.../migration/schema/MigrationStrategy.java | 2 +-
.../core/rx/AllEntitiesInSystemObservable.java | 40 ++++++
.../core/rx/ApplicationObservable.java | 33 +++++
.../core/scope/ApplicationEntityGroup.java | 39 ------
.../persistence/graph/GraphManager.java | 3 +-
.../persistence/graph/GraphManagerFactory.java | 2 +
.../persistence/graph/guice/GraphModule.java | 16 +--
.../EdgesFromSourceObservable.java | 32 +++++
.../serialization/EdgesToTargetObservable.java | 32 +++++
.../impl/EdgesFromSourceObservableImpl.java | 62 +++++++++
.../impl/EdgesToTargetObservableImpl.java | 65 +++++++++
.../impl/GraphManagerFactoryImpl.java | 88 ++++++++++++
.../usergrid/persistence/map/MapManager.java | 4 +-
.../persistence/map/MapManagerFactory.java | 2 +
.../persistence/map/guice/MapModule.java | 10 +-
.../map/impl/MapManagerFactoryImpl.java | 62 +++++++++
.../usergrid/persistence/index/EntityIndex.java | 9 +-
.../persistence/index/EntityIndexFactory.java | 4 +-
.../persistence/index/guice/IndexModule.java | 6 +-
.../index/impl/EntityIndexFactoryImpl.java | 67 +++++++++
44 files changed, 1077 insertions(+), 499 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/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 93eb15e..5a4fcc1 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
@@ -36,6 +36,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.lang.StringUtils;
import org.apache.usergrid.corepersistence.rx.AllEntitiesInSystemObservable;
+import org.apache.usergrid.corepersistence.rx.impl.AllEntitiesInSystemObservableImpl;
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
import org.apache.usergrid.persistence.AbstractEntity;
import org.apache.usergrid.persistence.Entity;
@@ -61,14 +62,10 @@ import org.apache.usergrid.persistence.entities.Application;
import org.apache.usergrid.persistence.exceptions.ApplicationAlreadyExistsException;
import org.apache.usergrid.persistence.graph.Edge;
import org.apache.usergrid.persistence.graph.GraphManager;
-import org.apache.usergrid.persistence.graph.GraphManagerFactory;
import org.apache.usergrid.persistence.graph.SearchByEdgeType;
import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdgeType;
-import org.apache.usergrid.persistence.index.AliasedEntityIndex;
import org.apache.usergrid.persistence.index.EntityIndex;
-import org.apache.usergrid.persistence.index.EntityIndexFactory;
import org.apache.usergrid.persistence.index.query.Query;
-import org.apache.usergrid.persistence.map.MapManagerFactory;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.entity.SimpleId;
import org.apache.usergrid.utils.UUIDUtils;
@@ -110,8 +107,10 @@ public class CpEntityManagerFactory implements EntityManagerFactory, Application
}
});
-
private ManagerCache managerCache;
+
+ private AllEntitiesInSystemObservable allEntitiesInSystemObservable;
+
private DataMigrationManager dataMigrationManager;
CassandraService cass;
@@ -161,6 +160,12 @@ public class CpEntityManagerFactory implements EntityManagerFactory, Application
return managerCache;
}
+ private AllEntitiesInSystemObservable getAllEntitiesObservable(){
+ if(allEntitiesInSystemObservable==null)
+ allEntitiesInSystemObservable = CpSetup.getInjector().getInstance(AllEntitiesInSystemObservable.class);
+ return allEntitiesInSystemObservable;
+ }
+
@Override
public String getImplementationDescription() throws Exception {
@@ -426,8 +431,8 @@ public class CpEntityManagerFactory implements EntityManagerFactory, Application
if ( e == null ) {
logger.warn("Applicaion {} in index but not found in collections", targetId );
continue;
- }
-
+ }
+
appMap.put(
(String)e.getField( PROPERTY_NAME ).getValue(),
(UUID)e.getField( "applicationUuid" ).getValue());
@@ -576,7 +581,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 AllEntitiesInSystemObservable.getAllEntitiesInSystem( managerCache, 1000 ).longCount().toBlocking().last();
+ return getAllEntitiesObservable().getAllEntitiesInSystem(1000).longCount().toBlocking().last();
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpManagerCache.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpManagerCache.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpManagerCache.java
index e29e5c2..8292170 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpManagerCache.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpManagerCache.java
@@ -18,6 +18,7 @@ package org.apache.usergrid.corepersistence;
import java.util.concurrent.ExecutionException;
+import com.amazonaws.services.cloudfront.model.InvalidArgumentException;
import org.apache.usergrid.persistence.collection.CollectionScope;
import org.apache.usergrid.persistence.collection.EntityCollectionManager;
import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
@@ -45,35 +46,8 @@ public class CpManagerCache implements ManagerCache {
// TODO: consider making these cache sizes and timeouts configurable
- private LoadingCache<CollectionScope, EntityCollectionManager> ecmCache =
- CacheBuilder.newBuilder().maximumSize( 1000 )
- .build( new CacheLoader<CollectionScope, EntityCollectionManager>() {
- public EntityCollectionManager load( CollectionScope scope ) {
- return ecmf.createCollectionManager( scope );
- }
- } );
-
- private LoadingCache<ApplicationScope, EntityIndex> eiCache =
- CacheBuilder.newBuilder().maximumSize( 1000 ).build( new CacheLoader<ApplicationScope, EntityIndex>() {
- public EntityIndex load( ApplicationScope scope ) {
- return eif.createEntityIndex( scope );
- }
- } );
-
- private LoadingCache<ApplicationScope, GraphManager> gmCache =
- CacheBuilder.newBuilder().maximumSize( 1000 ).build( new CacheLoader<ApplicationScope, GraphManager>() {
- public GraphManager load(
- ApplicationScope scope ) {
- return gmf.createEdgeManager( scope );
- }
- } );
-
- private LoadingCache<MapScope, MapManager> mmCache =
- CacheBuilder.newBuilder().maximumSize( 1000 ).build( new CacheLoader<MapScope, MapManager>() {
- public MapManager load( MapScope scope ) {
- return mmf.createMapManager( scope );
- }
- } );
+
+
@Inject
@@ -89,53 +63,34 @@ public class CpManagerCache implements ManagerCache {
@Override
public EntityCollectionManager getEntityCollectionManager( CollectionScope scope ) {
- try {
- return ecmCache.get( scope );
- }
- catch ( ExecutionException ex ) {
- throw new RuntimeException( "Error getting manager", ex );
- }
+ return ecmf.createCollectionManager(scope);
}
@Override
public EntityIndex getEntityIndex( ApplicationScope appScope ) {
- try {
- return eiCache.get( appScope );
- }
- catch ( ExecutionException ex ) {
- throw new RuntimeException( "Error getting manager", ex );
- }
+
+ return eif.createEntityIndex( appScope );
+
}
@Override
public GraphManager getGraphManager( ApplicationScope appScope ) {
- try {
- return gmCache.get( appScope );
- }
- catch ( ExecutionException ex ) {
- throw new RuntimeException( "Error getting manager", ex );
- }
+ return gmf.createEdgeManager(appScope);
}
@Override
public MapManager getMapManager( MapScope mapScope ) {
- try {
- return mmCache.get( mapScope );
- }
- catch ( ExecutionException ex ) {
- throw new RuntimeException( "Error getting manager", ex );
- }
+ return mmf.createMapManager(mapScope);
}
-
@Override
public void invalidate() {
- ecmCache.invalidateAll();
- eiCache.invalidateAll();
- gmCache.invalidateAll();
- mmCache.invalidateAll();
+ ecmf.invalidate();
+ eif.invalidate();
+ gmf.invalidate();
+ mmf.invalidate();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/main/java/org/apache/usergrid/corepersistence/GuiceModule.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/GuiceModule.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/GuiceModule.java
index ed6cba2..0143a89 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/GuiceModule.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/GuiceModule.java
@@ -24,6 +24,10 @@ import org.apache.usergrid.corepersistence.migration.GraphShardVersionMigration;
import org.apache.usergrid.corepersistence.events.EntityDeletedHandler;
import org.apache.usergrid.corepersistence.events.EntityVersionCreatedHandler;
import org.apache.usergrid.corepersistence.events.EntityVersionDeletedHandler;
+import org.apache.usergrid.corepersistence.rx.AllEntitiesInSystemObservable;
+import org.apache.usergrid.corepersistence.rx.ApplicationObservable;
+import org.apache.usergrid.corepersistence.rx.impl.AllEntitiesInSystemObservableImpl;
+import org.apache.usergrid.corepersistence.rx.impl.ApplicationObservableImpl;
import org.apache.usergrid.persistence.EntityManagerFactory;
import org.apache.usergrid.persistence.collection.event.EntityDeleted;
import org.apache.usergrid.persistence.collection.event.EntityVersionCreated;
@@ -69,18 +73,21 @@ public class GuiceModule extends AbstractModule {
install(new QueueModule());
bind(ManagerCache.class).to( CpManagerCache.class );
+ bind(AllEntitiesInSystemObservable.class).to( AllEntitiesInSystemObservableImpl.class );
+ bind(ApplicationObservable.class).to( ApplicationObservableImpl.class );
- Multibinder<DataMigration> dataMigrationMultibinder =
+
+ Multibinder<DataMigration> dataMigrationMultibinder =
Multibinder.newSetBinder( binder(), DataMigration.class );
dataMigrationMultibinder.addBinding().to( EntityTypeMappingMigration.class );
dataMigrationMultibinder.addBinding().to( GraphShardVersionMigration.class );
dataMigrationMultibinder.addBinding().to( EntityDataMigration.class );
- Multibinder<EntityDeleted> entityBinder =
+ Multibinder<EntityDeleted> entityBinder =
Multibinder.newSetBinder(binder(), EntityDeleted.class);
entityBinder.addBinding().to(EntityDeletedHandler.class);
- Multibinder<EntityVersionDeleted> versionBinder =
+ Multibinder<EntityVersionDeleted> versionBinder =
Multibinder.newSetBinder(binder(), EntityVersionDeleted.class);
versionBinder.addBinding().to(EntityVersionDeletedHandler.class);
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/main/java/org/apache/usergrid/corepersistence/ManagerCache.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/ManagerCache.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/ManagerCache.java
index c1b7b95..921fde5 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/ManagerCache.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/ManagerCache.java
@@ -33,6 +33,7 @@ import org.apache.usergrid.persistence.map.MapScope;
* The cache of the manager
*/
public interface ManagerCache {
+
/**
* Get the entity collection manager for the specified scope
* @param scope
@@ -42,27 +43,30 @@ public interface ManagerCache {
/**
* Get the entity index for the specified app scope
+ *
* @param appScope
* @return
*/
- EntityIndex getEntityIndex( ApplicationScope appScope );
+ EntityIndex getEntityIndex(ApplicationScope appScope);
/**
* Get the graph manager for the graph scope
+ *
* @param appScope
* @return
*/
- GraphManager getGraphManager( ApplicationScope appScope );
+ GraphManager getGraphManager(ApplicationScope appScope);
/**
* Get the map manager for the map scope
+ *
* @param mapScope
* @return
*/
- MapManager getMapManager( MapScope mapScope );
+ MapManager getMapManager(MapScope mapScope);
/**
- * Invalidate all cache entries
+ * invalidate the cache
*/
void invalidate();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityDataMigration.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityDataMigration.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityDataMigration.java
index 22bc472..861c343 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityDataMigration.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityDataMigration.java
@@ -20,21 +20,14 @@
package org.apache.usergrid.corepersistence.migration;
-import java.util.*;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.corepersistence.rx.AllEntitiesInSystemObservable;
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
-import org.apache.usergrid.persistence.collection.CollectionScope;
import org.apache.usergrid.persistence.collection.mvcc.MvccEntityMigrationStrategy;
+import org.apache.usergrid.persistence.core.entity.ApplicationEntityGroup;
import org.apache.usergrid.persistence.core.migration.data.DataMigration;
-import org.apache.usergrid.persistence.core.scope.ApplicationEntityGroup;
+import org.apache.usergrid.persistence.core.rx.AllEntitiesInSystemObservable;
import com.google.inject.Inject;
-import org.apache.usergrid.persistence.model.entity.Id;
import rx.functions.Action1;
-import rx.functions.Func1;
/**
@@ -43,24 +36,29 @@ import rx.functions.Func1;
public class EntityDataMigration implements DataMigration {
- private final ManagerCache managerCache;
private final MvccEntityMigrationStrategy migrationStrategy;
+ private final AllEntitiesInSystemObservable allEntitiesInSystemObservable;
@Inject
public EntityDataMigration( final MvccEntityMigrationStrategy migrationStrategy,
- final ManagerCache managerCache ) {
+ final AllEntitiesInSystemObservable allEntitiesInSystemObservable) {
this.migrationStrategy = migrationStrategy;
- this.managerCache = managerCache;
+ this.allEntitiesInSystemObservable = allEntitiesInSystemObservable;
}
@Override
public void migrate( final ProgressObserver observer ) throws Throwable {
- AllEntitiesInSystemObservable
- .getAllEntitiesInSystem(managerCache, 1000)
+ allEntitiesInSystemObservable
+ .getAllEntitiesInSystem(1000)
.doOnNext(new Action1<ApplicationEntityGroup>() {
@Override
public void call( final ApplicationEntityGroup applicationEntityGroup ) {
- migrationStrategy.executeMigration( null,applicationEntityGroup, observer, CpNamingUtils.getCollectionScopeByEntityIdFunc1(applicationEntityGroup.applicationScope));
+ migrationStrategy.executeMigration(
+ null,
+ applicationEntityGroup,
+ observer,
+ CpNamingUtils.getCollectionScopeByEntityIdFunc1(applicationEntityGroup.applicationScope)
+ );
}
})
.toBlocking().last();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/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 a7655ad..08001d1 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
@@ -23,14 +23,13 @@ package org.apache.usergrid.corepersistence.migration;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
-import org.apache.usergrid.persistence.core.scope.ApplicationEntityGroup;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.corepersistence.rx.AllEntitiesInSystemObservable;
+import org.apache.usergrid.corepersistence.rx.impl.AllEntitiesInSystemObservableImpl;
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
+import org.apache.usergrid.persistence.core.entity.ApplicationEntityGroup;
import org.apache.usergrid.persistence.core.migration.data.DataMigration;
+import org.apache.usergrid.persistence.core.rx.AllEntitiesInSystemObservable;
import org.apache.usergrid.persistence.map.MapManager;
import org.apache.usergrid.persistence.map.MapScope;
import org.apache.usergrid.persistence.model.entity.Id;
@@ -46,12 +45,13 @@ import rx.functions.Action1;
public class EntityTypeMappingMigration implements DataMigration {
private final ManagerCache managerCache;
-
+ private final AllEntitiesInSystemObservable allEntitiesInSystemObservable;
@Inject
- public EntityTypeMappingMigration( final ManagerCache managerCache) {
+ public EntityTypeMappingMigration( final ManagerCache managerCache, final AllEntitiesInSystemObservable allEntitiesInSystemObservable) {
this.managerCache = managerCache;
+ this.allEntitiesInSystemObservable = allEntitiesInSystemObservable;
}
@@ -60,7 +60,7 @@ public class EntityTypeMappingMigration implements DataMigration {
final AtomicLong atomicLong = new AtomicLong();
- AllEntitiesInSystemObservable.getAllEntitiesInSystem(managerCache, 1000 )
+ allEntitiesInSystemObservable.getAllEntitiesInSystem(1000)
.doOnNext( new Action1<ApplicationEntityGroup>() {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/GraphShardVersionMigration.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/GraphShardVersionMigration.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/GraphShardVersionMigration.java
index 0a1fbd2..41b390b 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/GraphShardVersionMigration.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/GraphShardVersionMigration.java
@@ -22,32 +22,20 @@
package org.apache.usergrid.corepersistence.migration;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicLong;
-
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
-import org.apache.usergrid.persistence.collection.impl.CollectionScopeImpl;
-import org.apache.usergrid.persistence.core.scope.ApplicationEntityGroup;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.core.entity.ApplicationEntityGroup;
+import org.apache.usergrid.persistence.core.rx.AllEntitiesInSystemObservable;
+import org.apache.usergrid.persistence.graph.GraphManagerFactory;
import org.apache.usergrid.persistence.graph.serialization.EdgeMigrationStrategy;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.usergrid.persistence.graph.serialization.EdgesFromSourceObservable;
-import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.corepersistence.rx.AllEntitiesInSystemObservable;
-import org.apache.usergrid.corepersistence.rx.EdgesFromSourceObservable;
import org.apache.usergrid.persistence.core.migration.data.DataMigration;
import org.apache.usergrid.persistence.graph.Edge;
import org.apache.usergrid.persistence.graph.GraphManager;
-import org.apache.usergrid.persistence.model.entity.Id;
import com.google.inject.Inject;
-import com.netflix.astyanax.Keyspace;
-import com.netflix.astyanax.MutationBatch;
-import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import rx.Observable;
-import rx.functions.Action1;
import rx.functions.Func1;
@@ -57,34 +45,46 @@ import rx.functions.Func1;
public class GraphShardVersionMigration implements DataMigration {
- private final ManagerCache managerCache;
+ private final GraphManagerFactory graphManagerFactory;
+ private final AllEntitiesInSystemObservable allEntitiesInSystemObservable;
+ private final EdgesFromSourceObservable edgesFromSourceObservable;
private final EdgeMigrationStrategy migrationStrategy;
@Inject
public GraphShardVersionMigration( final EdgeMigrationStrategy migrationStrategy,
- final ManagerCache managerCache ) {
+ final GraphManagerFactory graphManagerFactory,
+ final AllEntitiesInSystemObservable allEntitiesInSystemObservable,
+ final EdgesFromSourceObservable edgesFromSourceObservable
+ ) {
this.migrationStrategy = migrationStrategy;
- this.managerCache = managerCache;
+ this.graphManagerFactory = graphManagerFactory;
+ this.allEntitiesInSystemObservable = allEntitiesInSystemObservable;
+ this.edgesFromSourceObservable = edgesFromSourceObservable;
}
@Override
public void migrate( final ProgressObserver observer ) throws Throwable {
- AllEntitiesInSystemObservable.getAllEntitiesInSystem( managerCache, 1000 )
+ allEntitiesInSystemObservable.getAllEntitiesInSystem(1000)
.flatMap(
new Func1<ApplicationEntityGroup, Observable<Long>>() {
@Override
public Observable<Long> call(
final ApplicationEntityGroup applicationEntityGroup) {
- final GraphManager gm = managerCache.getGraphManager(applicationEntityGroup.applicationScope);
- Observable<Edge> edgesFromSource = EdgesFromSourceObservable.edgesFromSource(gm, applicationEntityGroup.applicationScope.getApplication());
+ final GraphManager gm = graphManagerFactory.createEdgeManager(applicationEntityGroup.applicationScope);
+ Observable<Edge> edgesFromSource = edgesFromSourceObservable.edgesFromSource(gm, applicationEntityGroup.applicationScope.getApplication());
//emit a stream of all ids from this group
- return migrationStrategy.executeMigration(edgesFromSource,applicationEntityGroup, observer, CpNamingUtils.getCollectionScopeByEntityIdFunc1(applicationEntityGroup.applicationScope));
+ return migrationStrategy.executeMigration(
+ edgesFromSource,
+ applicationEntityGroup,
+ observer,
+ CpNamingUtils.getCollectionScopeByEntityIdFunc1(applicationEntityGroup.applicationScope)
+ );
}
- } )
- .toBlocking().lastOrDefault( null );
+ })
+ .toBlocking().lastOrDefault(null);
;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/AllEntitiesInSystemObservable.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/AllEntitiesInSystemObservable.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/AllEntitiesInSystemObservable.java
deleted file mode 100644
index d3add56..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/AllEntitiesInSystemObservable.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.corepersistence.rx;
-
-
-import java.util.List;
-
-import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.persistence.core.scope.ApplicationEntityGroup;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.core.scope.ApplicationScopeImpl;
-import org.apache.usergrid.persistence.graph.GraphManager;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import rx.Observable;
-import rx.functions.Func1;
-
-
-/**
- * An observable that will emit every entity Id stored in our entire system across all apps.
- * Note that this only walks each application applicationId graph, and emits edges from the applicationId and it's edges as the s
- * source node
- */
-public class AllEntitiesInSystemObservable {
-
-
- /**
- * Return an observable that emits all entities in the system.
- * @param managerCache the managerCache to use
- * @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 static Observable<ApplicationEntityGroup> getAllEntitiesInSystem( final ManagerCache managerCache, final int bufferSize) {
- //traverse all nodes in the graph, load all source edges from them, then re-save the meta data
- return ApplicationObservable.getAllApplicationIds( managerCache )
-
- .flatMap( new Func1<Id, Observable<ApplicationEntityGroup>>() {
- @Override
- public Observable<ApplicationEntityGroup> call( final Id applicationId ) {
-
- //set up our application scope and graph manager
- final ApplicationScope applicationScope = new ApplicationScopeImpl(
- applicationId );
-
-
- final GraphManager gm =
- managerCache.getGraphManager( 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 );
-
- //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>, ApplicationEntityGroup>() {
- @Override
- public ApplicationEntityGroup call( final List<Id> id ) {
- return new ApplicationEntityGroup( applicationScope, id );
- }
- } );
- }
- } );
- }
-
-
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/ApplicationObservable.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/ApplicationObservable.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/ApplicationObservable.java
deleted file mode 100644
index 6019bca..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/ApplicationObservable.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.corepersistence.rx;
-
-
-import java.util.Arrays;
-import java.util.UUID;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.corepersistence.util.CpNamingUtils;
-import org.apache.usergrid.persistence.collection.CollectionScope;
-import org.apache.usergrid.persistence.collection.EntityCollectionManager;
-import org.apache.usergrid.persistence.collection.impl.CollectionScopeImpl;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.graph.Edge;
-import org.apache.usergrid.persistence.graph.GraphManager;
-import org.apache.usergrid.persistence.graph.SearchByEdgeType;
-import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdgeType;
-import org.apache.usergrid.persistence.model.entity.Entity;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import rx.Observable;
-import rx.functions.Func1;
-
-import static org.apache.usergrid.corepersistence.util.CpNamingUtils.generateApplicationId;
-import static org.apache.usergrid.corepersistence.util.CpNamingUtils.getApplicationScope;
-
-
-/**
- * An observable that will emit all application stored in the system.
- */
-public class ApplicationObservable {
-
- private static final Logger logger = LoggerFactory.getLogger( ApplicationObservable.class );
-
- /**
- * Get all applicationIds as an observable
- */
- public static Observable<Id> getAllApplicationIds( final ManagerCache managerCache ) {
-
- //emit our 3 hard coded applications that are used the manage the system first.
- //this way consumers can perform whatever work they need to on the root system first
-
-
- final Observable<Id> systemIds = Observable.from( Arrays
- .asList( generateApplicationId( CpNamingUtils.DEFAULT_APPLICATION_ID ),
- generateApplicationId( CpNamingUtils.MANAGEMENT_APPLICATION_ID ),
- generateApplicationId( CpNamingUtils.SYSTEM_APP_ID ) ) );
-
-
- final ApplicationScope appScope = getApplicationScope( CpNamingUtils.SYSTEM_APP_ID );
-
- final CollectionScope appInfoCollectionScope =
- new CollectionScopeImpl( appScope.getApplication(), appScope.getApplication(),
- CpNamingUtils.getCollectionScopeNameFromCollectionName( CpNamingUtils.APPINFOS ) );
-
- final EntityCollectionManager collectionManager =
- managerCache.getEntityCollectionManager( appInfoCollectionScope );
-
-
- final GraphManager gm = managerCache.getGraphManager( appScope );
-
-
- String edgeType = CpNamingUtils.getEdgeTypeFromCollectionName( CpNamingUtils.APPINFOS );
-
- Id rootAppId = appScope.getApplication();
-
-
- //we have app infos. For each of these app infos, we have to load the application itself
- Observable<Id> appIds = gm.loadEdgesFromSource(
- new SimpleSearchByEdgeType( rootAppId, edgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING,
- null ) ).flatMap( new Func1<Edge, Observable<Id>>() {
- @Override
- public Observable<Id> call( final Edge edge ) {
- //get the app info and load it
- final Id appInfo = edge.getTargetNode();
-
- return collectionManager.load( appInfo )
- //filter out null entities
- .filter( new Func1<Entity, Boolean>() {
- @Override
- public Boolean call( final Entity entity ) {
- if ( entity == null ) {
- logger.warn( "Encountered a null application info for id {}", appInfo );
- return false;
- }
-
- return true;
- }
- } )
- //get the id from the entity
- .map( new Func1<org.apache.usergrid.persistence.model.entity.Entity, Id>() {
-
-
- @Override
- public Id call( final org.apache.usergrid.persistence.model.entity.Entity entity ) {
-
- final UUID uuid = ( UUID ) entity.getField( "applicationUuid" ).getValue();
-
- return CpNamingUtils.generateApplicationId( uuid );
- }
- } );
- }
- } );
-
- return Observable.merge( systemIds, appIds );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/EdgesFromSourceObservable.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/EdgesFromSourceObservable.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/EdgesFromSourceObservable.java
deleted file mode 100644
index d3e2ee5..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/EdgesFromSourceObservable.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.corepersistence.rx;
-
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.graph.Edge;
-import org.apache.usergrid.persistence.graph.GraphManager;
-import org.apache.usergrid.persistence.graph.SearchByEdgeType;
-import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdgeType;
-import org.apache.usergrid.persistence.graph.impl.SimpleSearchEdgeType;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import rx.Observable;
-import rx.functions.Func1;
-
-
-/**
- * Emits the edges that are edges from the specified source node
- */
-public class EdgesFromSourceObservable {
-
- private static final Logger logger = LoggerFactory.getLogger( EdgesFromSourceObservable.class );
-
-
- /**
- * Get all edges from the source
- */
- public static Observable<Edge> edgesFromSource( final GraphManager gm, final Id sourceNode){
- Observable<String> edgeTypes = gm.getEdgeTypesFromSource( new SimpleSearchEdgeType( sourceNode, null, null ) );
-
- return edgeTypes.flatMap( new Func1<String, Observable<Edge>>() {
- @Override
- public Observable<Edge> call( final String edgeType ) {
-
- logger.debug( "Loading edges of edgeType {} from {}", edgeType, sourceNode );
-
- return gm.loadEdgesFromSource( new SimpleSearchByEdgeType( sourceNode, edgeType, Long.MAX_VALUE,
- SearchByEdgeType.Order.DESCENDING, null ) );
- }
- } );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/TargetIdObservable.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/TargetIdObservable.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/TargetIdObservable.java
index c4b6526..96b638b 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/TargetIdObservable.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/TargetIdObservable.java
@@ -20,6 +20,7 @@
package org.apache.usergrid.corepersistence.rx;
+import org.apache.usergrid.persistence.graph.serialization.EdgesFromSourceObservable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,7 +38,11 @@ import rx.functions.Func1;
public class TargetIdObservable {
private static final Logger logger = LoggerFactory.getLogger( TargetIdObservable.class );
+ private final EdgesFromSourceObservable edgesFromSourceObservable;
+ public TargetIdObservable(final EdgesFromSourceObservable edgesFromSourceObservable){
+ this.edgesFromSourceObservable = edgesFromSourceObservable;
+ }
/**
* Get all nodes that are target nodes from the sourceNode
@@ -46,10 +51,10 @@ public class TargetIdObservable {
*
* @return
*/
- public static Observable<Id> getTargetNodes( final GraphManager gm, final Id sourceNode) {
+ public Observable<Id> getTargetNodes( final GraphManager gm, final Id sourceNode) {
//only search edge types that start with collections
- return EdgesFromSourceObservable.edgesFromSource(gm, sourceNode ).map( new Func1<Edge, Id>() {
+ return edgesFromSourceObservable.edgesFromSource(gm, sourceNode ).map( new Func1<Edge, Id>() {
@Override
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/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
new file mode 100644
index 0000000..4218107
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntitiesInSystemObservableImpl.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.rx.impl;
+
+
+import java.util.List;
+
+import org.apache.usergrid.corepersistence.rx.TargetIdObservable;
+import org.apache.usergrid.persistence.core.rx.AllEntitiesInSystemObservable;
+import org.apache.usergrid.persistence.core.rx.ApplicationObservable;
+import org.apache.usergrid.persistence.core.scope.ApplicationEntityGroup;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.core.scope.ApplicationScopeImpl;
+import org.apache.usergrid.persistence.graph.GraphManager;
+import org.apache.usergrid.persistence.graph.GraphManagerFactory;
+import org.apache.usergrid.persistence.model.entity.Id;
+
+import rx.Observable;
+import rx.functions.Func1;
+
+
+/**
+ * An observable that will emit every entity Id stored in our entire system across all apps.
+ * Note that this only walks each application applicationId graph, and emits edges from the applicationId and it's edges as the s
+ * source node
+ */
+public class AllEntitiesInSystemObservableImpl implements AllEntitiesInSystemObservable {
+
+ private final ApplicationObservable applicationObservable;
+ private final GraphManagerFactory graphManagerFactory;
+ private final TargetIdObservable targetIdObservable;
+
+ public AllEntitiesInSystemObservableImpl(ApplicationObservable applicationObservable, GraphManagerFactory graphManagerFactory, TargetIdObservable targetIdObservable){
+
+ this.applicationObservable = applicationObservable;
+ this.graphManagerFactory = graphManagerFactory;
+ this.targetIdObservable = targetIdObservable;
+ }
+
+
+ public Observable<ApplicationEntityGroup> getAllEntitiesInSystem(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>>() {
+ @Override
+ public Observable<ApplicationEntityGroup> 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);
+
+ //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>, ApplicationEntityGroup>() {
+ @Override
+ public ApplicationEntityGroup call( final List<Id> id ) {
+ return new ApplicationEntityGroup( applicationScope, id );
+ }
+ } );
+ }
+ } );
+ }
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/ApplicationObservableImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/ApplicationObservableImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/ApplicationObservableImpl.java
new file mode 100644
index 0000000..10a043d
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/ApplicationObservableImpl.java
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.rx.impl;
+
+
+import java.util.Arrays;
+import java.util.UUID;
+
+import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
+import org.apache.usergrid.persistence.core.rx.ApplicationObservable;
+import org.apache.usergrid.persistence.graph.GraphManagerFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.usergrid.corepersistence.util.CpNamingUtils;
+import org.apache.usergrid.persistence.collection.CollectionScope;
+import org.apache.usergrid.persistence.collection.EntityCollectionManager;
+import org.apache.usergrid.persistence.collection.impl.CollectionScopeImpl;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.graph.Edge;
+import org.apache.usergrid.persistence.graph.GraphManager;
+import org.apache.usergrid.persistence.graph.SearchByEdgeType;
+import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdgeType;
+import org.apache.usergrid.persistence.model.entity.Entity;
+import org.apache.usergrid.persistence.model.entity.Id;
+
+import rx.Observable;
+import rx.functions.Func1;
+
+import static org.apache.usergrid.corepersistence.util.CpNamingUtils.generateApplicationId;
+import static org.apache.usergrid.corepersistence.util.CpNamingUtils.getApplicationScope;
+
+
+/**
+ * An observable that will emit all application stored in the system.
+ */
+public class ApplicationObservableImpl implements ApplicationObservable {
+
+ private static final Logger logger = LoggerFactory.getLogger( ApplicationObservableImpl.class );
+ private final EntityCollectionManagerFactory entityCollectionManagerFactory;
+ private final GraphManagerFactory graphManagerFactory;
+
+ public ApplicationObservableImpl(EntityCollectionManagerFactory entityCollectionManagerFactory, GraphManagerFactory graphManagerFactory){
+
+ this.entityCollectionManagerFactory = entityCollectionManagerFactory;
+ this.graphManagerFactory = graphManagerFactory;
+ }
+
+
+ @Override
+ public Observable<Id> getAllApplicationIds() {
+
+ //emit our 3 hard coded applications that are used the manage the system first.
+ //this way consumers can perform whatever work they need to on the root system first
+ final Observable<Id> systemIds = Observable.from( Arrays
+ .asList( generateApplicationId( CpNamingUtils.DEFAULT_APPLICATION_ID ),
+ generateApplicationId( CpNamingUtils.MANAGEMENT_APPLICATION_ID ),
+ generateApplicationId( CpNamingUtils.SYSTEM_APP_ID ) ) );
+
+
+ final ApplicationScope appScope = getApplicationScope( CpNamingUtils.SYSTEM_APP_ID );
+
+ final CollectionScope appInfoCollectionScope =
+ new CollectionScopeImpl( appScope.getApplication(), appScope.getApplication(),
+ CpNamingUtils.getCollectionScopeNameFromCollectionName( CpNamingUtils.APPINFOS ) );
+
+ final EntityCollectionManager collectionManager =
+ entityCollectionManagerFactory.createCollectionManager( appInfoCollectionScope );
+
+
+ final GraphManager gm = graphManagerFactory.createEdgeManager(appScope);
+
+
+ String edgeType = CpNamingUtils.getEdgeTypeFromCollectionName( CpNamingUtils.APPINFOS );
+
+ Id rootAppId = appScope.getApplication();
+
+
+ //we have app infos. For each of these app infos, we have to load the application itself
+ Observable<Id> appIds = gm.loadEdgesFromSource(
+ new SimpleSearchByEdgeType( rootAppId, edgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING,
+ null ) ).flatMap( new Func1<Edge, Observable<Id>>() {
+ @Override
+ public Observable<Id> call( final Edge edge ) {
+ //get the app info and load it
+ final Id appInfo = edge.getTargetNode();
+
+ return collectionManager.load( appInfo )
+ //filter out null entities
+ .filter( new Func1<Entity, Boolean>() {
+ @Override
+ public Boolean call( final Entity entity ) {
+ if ( entity == null ) {
+ logger.warn( "Encountered a null application info for id {}", appInfo );
+ return false;
+ }
+
+ return true;
+ }
+ } )
+ //get the id from the entity
+ .map( new Func1<org.apache.usergrid.persistence.model.entity.Entity, Id>() {
+
+
+ @Override
+ public Id call( final org.apache.usergrid.persistence.model.entity.Entity entity ) {
+
+ final UUID uuid = ( UUID ) entity.getField( "applicationUuid" ).getValue();
+
+ return CpNamingUtils.generateApplicationId( uuid );
+ }
+ } );
+ }
+ } );
+
+ return Observable.merge( systemIds, appIds );
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityDataMigrationIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityDataMigrationIT.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityDataMigrationIT.java
index 50b14ec..f58696e 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityDataMigrationIT.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityDataMigrationIT.java
@@ -34,7 +34,7 @@ import org.apache.usergrid.AbstractCoreIT;
import org.apache.usergrid.corepersistence.CpSetup;
import org.apache.usergrid.corepersistence.EntityWriteHelper;
import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.corepersistence.rx.AllEntitiesInSystemObservable;
+import org.apache.usergrid.corepersistence.rx.impl.AllEntitiesInSystemObservableImpl;
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
import org.apache.usergrid.persistence.Entity;
import org.apache.usergrid.persistence.EntityManager;
@@ -132,7 +132,7 @@ public class EntityDataMigrationIT extends AbstractCoreIT {
//read everything in previous version format and put it into our types. Assumes we're
//using a test system, and it's not a huge amount of data, otherwise we'll overflow.
- AllEntitiesInSystemObservable.getAllEntitiesInSystem( managerCache, 1000 )
+ AllEntitiesInSystemObservableImpl.getAllEntitiesInSystem(managerCache, 1000)
.doOnNext( new Action1<ApplicationEntityGroup>() {
@Override
public void call(
@@ -184,7 +184,7 @@ public class EntityDataMigrationIT extends AbstractCoreIT {
//now visit all entities in the system again, load them from v2, and ensure they're the same
- AllEntitiesInSystemObservable.getAllEntitiesInSystem( managerCache, 1000 )
+ AllEntitiesInSystemObservableImpl.getAllEntitiesInSystem(managerCache, 1000)
.doOnNext( new Action1<ApplicationEntityGroup>() {
@Override
public void call(
@@ -220,7 +220,7 @@ public class EntityDataMigrationIT extends AbstractCoreIT {
//now visit all entities in the system again, and load them from the EM,
// ensure we see everything we did in the v1 traversal
- AllEntitiesInSystemObservable.getAllEntitiesInSystem( managerCache, 1000 )
+ AllEntitiesInSystemObservableImpl.getAllEntitiesInSystem(managerCache, 1000)
.doOnNext( new Action1<ApplicationEntityGroup>() {
@Override
public void call(
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/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 ee88cd4..ee95d58 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
@@ -32,7 +32,7 @@ import org.apache.usergrid.AbstractCoreIT;
import org.apache.usergrid.corepersistence.CpSetup;
import org.apache.usergrid.corepersistence.EntityWriteHelper;
import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.corepersistence.rx.AllEntitiesInSystemObservable;
+import org.apache.usergrid.corepersistence.rx.impl.AllEntitiesInSystemObservableImpl;
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
import org.apache.usergrid.persistence.Entity;
import org.apache.usergrid.persistence.EntityManager;
@@ -126,7 +126,7 @@ public class EntityTypeMappingMigrationIT extends AbstractCoreIT {
entityTypeMappingMigration.migrate( progressObserver );
- AllEntitiesInSystemObservable.getAllEntitiesInSystem(managerCache, 1000)
+ AllEntitiesInSystemObservableImpl.getAllEntitiesInSystem(managerCache, 1000)
.doOnNext( new Action1<ApplicationEntityGroup>() {
@Override
public void call(
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/GraphShardVersionMigrationIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/GraphShardVersionMigrationIT.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/GraphShardVersionMigrationIT.java
index 6f87623..45f41aa 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/GraphShardVersionMigrationIT.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/GraphShardVersionMigrationIT.java
@@ -32,7 +32,7 @@ import org.apache.usergrid.AbstractCoreIT;
import org.apache.usergrid.corepersistence.CpSetup;
import org.apache.usergrid.corepersistence.EntityWriteHelper;
import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.corepersistence.rx.AllEntitiesInSystemObservable;
+import org.apache.usergrid.corepersistence.rx.impl.AllEntitiesInSystemObservableImpl;
import org.apache.usergrid.persistence.EntityManager;
import org.apache.usergrid.persistence.core.migration.data.DataMigrationManager;
import org.apache.usergrid.persistence.core.migration.data.DataMigrationManagerImpl;
@@ -115,7 +115,7 @@ public class GraphShardVersionMigrationIT extends AbstractCoreIT {
//read everything in previous version format and put it into our types.
- AllEntitiesInSystemObservable.getAllEntitiesInSystem(managerCache, 1000)
+ AllEntitiesInSystemObservableImpl.getAllEntitiesInSystem(managerCache, 1000)
.doOnNext( new Action1<ApplicationEntityGroup>() {
@Override
public void call(
@@ -172,7 +172,7 @@ public class GraphShardVersionMigrationIT extends AbstractCoreIT {
//now visit all nodes in the system and remove their types from the multi maps, it should be empty at the end
- AllEntitiesInSystemObservable.getAllEntitiesInSystem(managerCache, 1000)
+ AllEntitiesInSystemObservableImpl.getAllEntitiesInSystem(managerCache, 1000)
.doOnNext( new Action1<ApplicationEntityGroup>() {
@Override
public void call(
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/AllEntitiesInSystemObservableIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/AllEntitiesInSystemObservableIT.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/AllEntitiesInSystemObservableIT.java
index fcb9813..b246e04 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/AllEntitiesInSystemObservableIT.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/rx/AllEntitiesInSystemObservableIT.java
@@ -23,6 +23,7 @@ package org.apache.usergrid.corepersistence.rx;
import java.util.HashSet;
import java.util.Set;
+import org.apache.usergrid.corepersistence.rx.impl.AllEntitiesInSystemObservableImpl;
import org.apache.usergrid.persistence.core.scope.ApplicationEntityGroup;
import org.junit.Test;
import org.slf4j.Logger;
@@ -93,7 +94,7 @@ public class AllEntitiesInSystemObservableIT extends AbstractCoreIT {
final GraphManager gm = managerCache.getGraphManager( scope );
- AllEntitiesInSystemObservable.getAllEntitiesInSystem( managerCache, 1000 ).doOnNext( new Action1<ApplicationEntityGroup>() {
+ AllEntitiesInSystemObservableImpl.getAllEntitiesInSystem(managerCache, 1000).doOnNext( new Action1<ApplicationEntityGroup>() {
@Override
public void call( final ApplicationEntityGroup entity ) {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/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 f8f3c50..daeca87 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
@@ -24,15 +24,14 @@ import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
+import org.apache.usergrid.corepersistence.rx.impl.ApplicationObservable;
import org.junit.Test;
import org.apache.usergrid.AbstractCoreIT;
import org.apache.usergrid.corepersistence.CpSetup;
import org.apache.usergrid.corepersistence.ManagerCache;
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
-import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
import org.apache.usergrid.persistence.entities.Application;
-import org.apache.usergrid.persistence.graph.GraphManagerFactory;
import org.apache.usergrid.persistence.model.entity.Id;
import rx.Observable;
@@ -66,7 +65,7 @@ public class ApplicationObservableTestIT extends AbstractCoreIT {
//clean up our wiring
ManagerCache managerCache = CpSetup.getInjector().getInstance( ManagerCache.class );
- Observable<Id> appObservable = ApplicationObservable.getAllApplicationIds( managerCache );
+ Observable<Id> appObservable = ApplicationObservable.getAllApplicationIds(managerCache);
appObservable.doOnNext( new Action1<Id>() {
@Override
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManager.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManager.java
index 4de18fe..90cade0 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManager.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManager.java
@@ -20,6 +20,8 @@ package org.apache.usergrid.persistence.collection;
import java.util.Collection;
+
+import org.apache.usergrid.persistence.core.CPManager;
import org.apache.usergrid.persistence.core.util.Health;
import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.entity.Id;
@@ -30,7 +32,7 @@ import rx.Observable;
/**
* The operations for performing changes on an entity
*/
-public interface EntityCollectionManager {
+public interface EntityCollectionManager extends CPManager {
/**
* Write the entity in the entity collection. This is an entire entity, it's contents will
@@ -68,12 +70,12 @@ public interface EntityCollectionManager {
/**
* Takes the change and reloads an entity with all changes applied in this entity applied.
- * The resulting entity from calling load will be the previous version of this entity plus
+ * The resulting entity from calling load will be the previous version of this entity plus
* the entity in this object applied to it.
*/
public Observable<Entity> update ( Entity entity );
- /**
+ /**
* Returns health of entity data store.
*/
public Health getHealth();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerFactory.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerFactory.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerFactory.java
index ef579f8..9140913 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerFactory.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerFactory.java
@@ -20,40 +20,42 @@ package org.apache.usergrid.persistence.collection;
/**
- * A basic factory that creates a collection manager with the given context.
+ * A basic factory that creates a collection manager with the given context.
* Each instance of this factory should exist for a Single ApplicationScope
*/
public interface EntityCollectionManagerFactory {
/**
- * Create a new EntityCollectionManager for the given context.
- * The EntityCollectionManager can safely be used on the current thread
+ * Create a new EntityCollectionManager for the given context.
+ * The EntityCollectionManager can safely be used on the current thread
* and will shard responses. The returned instance should not be shared
* among threads it will not be guaranteed to be thread safe.
*
- * @param collectionScope The collectionScope collectionScope to use
+ * @param collectionScope The collectionScope collectionScope to use
* when creating the collectionScope manager
*
* @return The collectionScope manager to perform operations within the provided context
*/
- public EntityCollectionManager
+ public EntityCollectionManager
createCollectionManager( CollectionScope collectionScope );
/**
- * Create a new EntityCollectionManagerSync for the given context.
- * The EntityCollectionManager can safely be used on the current thread
+ * Create a new EntityCollectionManagerSync for the given context.
+ * The EntityCollectionManager can safely be used on the current thread
* and will shard responses. The returned instance should not be shared
- * among threads it will not be guaranteed to be thread safe.
+ * among threads it will not be guaranteed to be thread safe.
* This implementation will be synchronous. Try to use the org.apache.usergrid.persistence.core.consistency
* implementation if possible
*
- * @param collectionScope The collectionScope collectionScope to use when
+ * @param collectionScope The collectionScope collectionScope to use when
* creating the collectionScope manager
*
* @return The collectionScope manager to perform operations within the provided context
*/
- public EntityCollectionManagerSync
+ public EntityCollectionManagerSync
createCollectionManagerSync( CollectionScope collectionScope );
+
+ void invalidate();
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/guice/CollectionModule.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/guice/CollectionModule.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/guice/CollectionModule.java
index 1c3e258..c5dd961 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/guice/CollectionModule.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/guice/CollectionModule.java
@@ -19,6 +19,7 @@ package org.apache.usergrid.persistence.collection.guice;
+import org.apache.usergrid.persistence.collection.impl.EntityCollectionManagerFactoryImpl;
import org.safehaus.guicyfig.GuicyFigModule;
import org.apache.usergrid.persistence.collection.EntityCollectionManager;
@@ -81,10 +82,7 @@ public class CollectionModule extends AbstractModule {
Multibinder.newSetBinder( binder(), EntityDeleted.class );
// create a guice factor for getting our collection manager
- install( new FactoryModuleBuilder()
- .implement( EntityCollectionManager.class, EntityCollectionManagerImpl.class )
- .implement( EntityCollectionManagerSync.class, EntityCollectionManagerSyncImpl.class )
- .build( EntityCollectionManagerFactory.class ) );
+ bind(EntityCollectionManagerFactory.class).to(EntityCollectionManagerFactoryImpl.class);
bind( UniqueValueSerializationStrategy.class ).to( UniqueValueSerializationStrategyImpl.class );
bind( ChangeLogGenerator.class).to( ChangeLogGeneratorImpl.class);
@@ -116,7 +114,7 @@ public class CollectionModule extends AbstractModule {
@Provides
@CollectionTaskExecutor
public TaskExecutor collectionTaskExecutor(final SerializationFig serializationFig){
- return new NamedTaskExecutorImpl( "collectiontasks",
+ return new NamedTaskExecutorImpl( "collectiontasks",
serializationFig.getTaskPoolThreadSize(), serializationFig.getTaskPoolQueueSize() );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerFactoryImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerFactoryImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerFactoryImpl.java
new file mode 100644
index 0000000..790be19
--- /dev/null
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerFactoryImpl.java
@@ -0,0 +1,137 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. The ASF licenses this file to You
+ * * under the Apache License, Version 2.0 (the "License"); you may not
+ * * use this file except in compliance with the License.
+ * * You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License. For additional information regarding
+ * * copyright in this work, please see the NOTICE file in the top level
+ * * directory of this distribution.
+ *
+ */
+package org.apache.usergrid.persistence.collection.impl;
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.inject.assistedinject.Assisted;
+import com.netflix.astyanax.Keyspace;
+import org.apache.usergrid.persistence.collection.*;
+import org.apache.usergrid.persistence.collection.guice.CollectionTaskExecutor;
+import org.apache.usergrid.persistence.collection.guice.Write;
+import org.apache.usergrid.persistence.collection.guice.WriteUpdate;
+import org.apache.usergrid.persistence.collection.mvcc.MvccLogEntrySerializationStrategy;
+import org.apache.usergrid.persistence.collection.mvcc.stage.delete.MarkCommit;
+import org.apache.usergrid.persistence.collection.mvcc.stage.delete.MarkStart;
+import org.apache.usergrid.persistence.collection.mvcc.stage.write.*;
+import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
+import org.apache.usergrid.persistence.collection.serialization.UniqueValueSerializationStrategy;
+import org.apache.usergrid.persistence.collection.serialization.impl.MvccEntitySerializationStrategy;
+import org.apache.usergrid.persistence.core.guice.ProxyImpl;
+import org.apache.usergrid.persistence.core.task.TaskExecutor;
+
+import java.util.concurrent.ExecutionException;
+
+/**
+ * Classy class class.
+ */
+public class EntityCollectionManagerFactoryImpl implements EntityCollectionManagerFactory {
+
+
+ private final WriteStart writeStart;
+ private final WriteStart writeUpdate;
+ private final WriteUniqueVerify writeVerifyUnique;
+ private final WriteOptimisticVerify writeOptimisticVerify;
+ private final WriteCommit writeCommit;
+ private final RollbackAction rollback;
+ private final MarkStart markStart;
+ private final MarkCommit markCommit;
+ private final MvccEntitySerializationStrategy entitySerializationStrategy;
+ private final UniqueValueSerializationStrategy uniqueValueSerializationStrategy;
+ private final MvccLogEntrySerializationStrategy mvccLogEntrySerializationStrategy;
+ private final Keyspace keyspace;
+ private final SerializationFig config;
+ private final EntityVersionCleanupFactory entityVersionCleanupFactory;
+ private final EntityVersionCreatedFactory entityVersionCreatedFactory;
+ private final EntityDeletedFactory entityDeletedFactory;
+ private final TaskExecutor taskExecutor;
+ private final CollectionScope collectionScope;
+ private LoadingCache<CollectionScope, EntityCollectionManager> ecmCache =
+ CacheBuilder.newBuilder().maximumSize( 1000 )
+ .build( new CacheLoader<CollectionScope, EntityCollectionManager>() {
+ public EntityCollectionManager load( CollectionScope scope ) {
+ return new EntityCollectionManagerImpl(
+ writeStart,
+ writeUpdate,
+ writeVerifyUnique,
+ writeOptimisticVerify,writeCommit,rollback,markStart,markCommit,entitySerializationStrategy,uniqueValueSerializationStrategy,mvccLogEntrySerializationStrategy,keyspace,config,entityVersionCleanupFactory,entityVersionCreatedFactory,entityDeletedFactory,taskExecutor,collectionScope);
+ }
+ } );
+
+
+
+ public EntityCollectionManagerFactoryImpl( @Write final WriteStart writeStart,
+ @WriteUpdate final WriteStart writeUpdate,
+ final WriteUniqueVerify writeVerifyUnique,
+ final WriteOptimisticVerify writeOptimisticVerify,
+ final WriteCommit writeCommit,
+ final RollbackAction rollback,
+ final MarkStart markStart,
+ final MarkCommit markCommit,
+ @ProxyImpl final MvccEntitySerializationStrategy entitySerializationStrategy,
+ final UniqueValueSerializationStrategy uniqueValueSerializationStrategy,
+ final MvccLogEntrySerializationStrategy mvccLogEntrySerializationStrategy,
+ final Keyspace keyspace,
+ final SerializationFig config,
+ final EntityVersionCleanupFactory entityVersionCleanupFactory,
+ final EntityVersionCreatedFactory entityVersionCreatedFactory,
+ final EntityDeletedFactory entityDeletedFactory,
+ @CollectionTaskExecutor final TaskExecutor taskExecutor,
+ @Assisted final CollectionScope collectionScope){
+
+ this.writeStart = writeStart;
+ this.writeUpdate = writeUpdate;
+ this.writeVerifyUnique = writeVerifyUnique;
+ this.writeOptimisticVerify = writeOptimisticVerify;
+ this.writeCommit = writeCommit;
+ this.rollback = rollback;
+ this.markStart = markStart;
+ this.markCommit = markCommit;
+ this.entitySerializationStrategy = entitySerializationStrategy;
+ this.uniqueValueSerializationStrategy = uniqueValueSerializationStrategy;
+ this.mvccLogEntrySerializationStrategy = mvccLogEntrySerializationStrategy;
+ this.keyspace = keyspace;
+ this.config = config;
+ this.entityVersionCleanupFactory = entityVersionCleanupFactory;
+ this.entityVersionCreatedFactory = entityVersionCreatedFactory;
+ this.entityDeletedFactory = entityDeletedFactory;
+ this.taskExecutor = taskExecutor;
+ this.collectionScope = collectionScope;
+ }
+ @Override
+ public EntityCollectionManager createCollectionManager(CollectionScope collectionScope) {
+ try{
+ return ecmCache.get(collectionScope);
+ }catch (ExecutionException ee){
+ throw new RuntimeException(ee);
+ }
+ }
+
+ @Override
+ public EntityCollectionManagerSync createCollectionManagerSync(CollectionScope collectionScope) {
+ return new EntityCollectionManagerSyncImpl(this,collectionScope);
+ }
+
+ @Override
+ public void invalidate() {
+ ecmCache.invalidateAll();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/CPManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/CPManager.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/CPManager.java
new file mode 100644
index 0000000..347080f
--- /dev/null
+++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/CPManager.java
@@ -0,0 +1,28 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. The ASF licenses this file to You
+ * * under the Apache License, Version 2.0 (the "License"); you may not
+ * * use this file except in compliance with the License.
+ * * You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License. For additional information regarding
+ * * copyright in this work, please see the NOTICE file in the top level
+ * * directory of this distribution.
+ *
+ */
+
+
+package org.apache.usergrid.persistence.core;
+/**
+ * Base Manager Class
+ */
+
+public interface CPManager {
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/entity/ApplicationEntityGroup.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/entity/ApplicationEntityGroup.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/entity/ApplicationEntityGroup.java
new file mode 100644
index 0000000..fd3bba6
--- /dev/null
+++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/entity/ApplicationEntityGroup.java
@@ -0,0 +1,40 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. The ASF licenses this file to You
+ * * under the Apache License, Version 2.0 (the "License"); you may not
+ * * use this file except in compliance with the License.
+ * * You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License. For additional information regarding
+ * * copyright in this work, please see the NOTICE file in the top level
+ * * directory of this distribution.
+ *
+ */
+package org.apache.usergrid.persistence.core.entity;
+
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.model.entity.Id;
+
+import java.util.List;
+
+/**
+ * Get the entity data. Immutable bean for fast access
+ */
+public final class ApplicationEntityGroup {
+ public final ApplicationScope applicationScope;
+ public final List<Id> entityIds;
+
+
+ public ApplicationEntityGroup(final ApplicationScope applicationScope, final List<Id> entityIds) {
+ this.applicationScope = applicationScope;
+ this.entityIds = entityIds;
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7daca75e/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/migration/schema/MigrationStrategy.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/migration/schema/MigrationStrategy.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/migration/schema/MigrationStrategy.java
index cf30016..cb2f2eb 100644
--- a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/migration/schema/MigrationStrategy.java
+++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/migration/schema/MigrationStrategy.java
@@ -19,8 +19,8 @@
*/
package org.apache.usergrid.persistence.core.migration.schema;
+import org.apache.usergrid.persistence.core.entity.ApplicationEntityGroup;
import org.apache.usergrid.persistence.core.migration.data.DataMigration;
-import org.apache.usergrid.persistence.core.scope.ApplicationEntityGroup;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.entity.Id;
import rx.Observable;