You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by to...@apache.org on 2015/03/25 01:13:07 UTC
[2/4] incubator-usergrid git commit: WIP squash
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerify.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerify.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerify.java
index 548127c..a881cbb 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerify.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerify.java
@@ -28,6 +28,9 @@ import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixThreadPoolProperties;
+import org.apache.usergrid.persistence.collection.ScopeSet;
+import org.apache.usergrid.persistence.collection.serialization.impl.ScopeSetImpl;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.util.EntityUtils;
import org.apache.usergrid.persistence.core.astyanax.CassandraConfig;
@@ -98,6 +101,7 @@ public class WriteUniqueVerify implements Action1<CollectionIoEvent<MvccEntity>>
final Entity entity = mvccEntity.getEntity().get();
final CollectionScope scope = ioevent.getEntityCollection();
+ final ApplicationScope applicationScope = ioevent.getApplicationScope();
final MutationBatch batch = keyspace.prepareMutationBatch();
//allocate our max size, worst case
@@ -117,7 +121,7 @@ public class WriteUniqueVerify implements Action1<CollectionIoEvent<MvccEntity>>
final UniqueValue written = new UniqueValueImpl( field, mvccEntity.getId(), mvccEntity.getVersion() );
// use TTL in case something goes wrong before entity is finally committed
- final MutationBatch mb = uniqueValueStrat.write( scope, written, serializationFig.getTimeout() );
+ final MutationBatch mb = uniqueValueStrat.write( applicationScope, scope, written, serializationFig.getTimeout() );
batch.mergeShallow( mb );
uniqueFields.add(field);
@@ -137,12 +141,12 @@ public class WriteUniqueVerify implements Action1<CollectionIoEvent<MvccEntity>>
}
// use simple thread pool to verify fields in parallel
- ConsistentReplayCommand cmd = new ConsistentReplayCommand(uniqueValueStrat,cassandraFig,scope, uniqueFields,entity);
+ ConsistentReplayCommand cmd = new ConsistentReplayCommand(uniqueValueStrat,cassandraFig, applicationScope, scope, uniqueFields,entity);
Map<String,Field> uniquenessViolations = cmd.execute();
cmd.getFailedExecutionException();
//We have violations, throw an exception
if ( !uniquenessViolations.isEmpty() ) {
- throw new WriteUniqueVerifyException( mvccEntity, ioevent.getEntityCollection(), uniquenessViolations );
+ throw new WriteUniqueVerifyException( mvccEntity, applicationScope, ioevent.getEntityCollection(), uniquenessViolations );
}
}
@@ -150,16 +154,19 @@ public class WriteUniqueVerify implements Action1<CollectionIoEvent<MvccEntity>>
private final UniqueValueSerializationStrategy uniqueValueSerializationStrategy;
private final CassandraConfig fig;
- private final CollectionScope scope;
- private final List<Field> uniqueFields;
+ private final ApplicationScope applicationScope;
+// private final CollectionScope scope;
+ private final ScopeSet<Field> uniqueFields;
private final Entity entity;
- public ConsistentReplayCommand(UniqueValueSerializationStrategy uniqueValueSerializationStrategy, CassandraConfig fig, CollectionScope scope, List<Field> uniqueFields, Entity entity){
+ public ConsistentReplayCommand( UniqueValueSerializationStrategy uniqueValueSerializationStrategy,
+ CassandraConfig fig, final ApplicationScope applicationScope, CollectionScope
+ scope, List<Field> uniqueFields, Entity entity ){
super(REPLAY_GROUP);
this.uniqueValueSerializationStrategy = uniqueValueSerializationStrategy;
this.fig = fig;
- this.scope = scope;
- this.uniqueFields = uniqueFields;
+ this.applicationScope = applicationScope;
+ uniqueFields = new ScopeSetImpl<Field>(scope, uniqueFields);
this.entity = entity;
}
@@ -178,7 +185,7 @@ public class WriteUniqueVerify implements Action1<CollectionIoEvent<MvccEntity>>
//now get the set of fields back
final UniqueValueSet uniqueValues;
try {
- uniqueValues = uniqueValueSerializationStrategy.load( scope,consistencyLevel, uniqueFields );
+ uniqueValues = uniqueValueSerializationStrategy.load( applicationScope, scope,consistencyLevel, uniqueFields );
}
catch ( ConnectionException e ) {
throw new RuntimeException( "Unable to read from cassandra", e );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/MvccEntitySerializationStrategy.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/MvccEntitySerializationStrategy.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/MvccEntitySerializationStrategy.java
index 7028620..65be537 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/MvccEntitySerializationStrategy.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/MvccEntitySerializationStrategy.java
@@ -26,8 +26,10 @@ import java.util.UUID;
import org.apache.usergrid.persistence.collection.CollectionScope;
import org.apache.usergrid.persistence.collection.EntitySet;
import org.apache.usergrid.persistence.collection.MvccEntity;
+import org.apache.usergrid.persistence.collection.ScopeSet;
import org.apache.usergrid.persistence.core.migration.data.VersionedData;
import org.apache.usergrid.persistence.core.migration.schema.Migration;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.entity.Id;
import com.google.common.base.Optional;
@@ -45,17 +47,17 @@ public interface MvccEntitySerializationStrategy extends Migration, VersionedDat
* @param entity The entity to persist
* @return The MutationBatch operations for this update
*/
- public MutationBatch write(CollectionScope context, MvccEntity entity);
+ public MutationBatch write(ApplicationScope applicationScope, CollectionScope context, MvccEntity entity);
/**
* Load the entities into the entitySet from the specified Ids. Loads versions <= the maxVersion
*
- * @param scope
+ * @param applicationScope
* @param entityIds
* @return
*/
- public EntitySet load(CollectionScope scope, Collection<Id> entityIds, UUID maxVersion);
+ public EntitySet load(ApplicationScope applicationScope, Collection<ScopeSet<Id>> entityIds, UUID maxVersion);
/**
* Load a list, from highest to lowest of the entity with versions <= version up to maxSize elements
@@ -69,7 +71,7 @@ public interface MvccEntitySerializationStrategy extends Migration, VersionedDat
*/
@Deprecated
//this has been made obsolete in the latest version, only use the load methods
- public Iterator<MvccEntity> loadDescendingHistory( CollectionScope context, Id entityId, UUID version,
+ public Iterator<MvccEntity> loadDescendingHistory( ApplicationScope applicationScope, CollectionScope context, Id entityId, UUID version,
int fetchSize );
/**
@@ -85,7 +87,7 @@ public interface MvccEntitySerializationStrategy extends Migration, VersionedDat
*/
@Deprecated
//this has been made obsolete in the latest version, only use the load methods
- public Iterator<MvccEntity> loadAscendingHistory( CollectionScope context, Id entityId, UUID version,
+ public Iterator<MvccEntity> loadAscendingHistory( ApplicationScope applicationScope, CollectionScope context, Id entityId, UUID version,
int fetchSize );
@@ -96,7 +98,7 @@ public interface MvccEntitySerializationStrategy extends Migration, VersionedDat
* @param entityId
* @return The MvccEntity if it exists. Null otherwise
*/
- public Optional<MvccEntity> load(CollectionScope scope, Id entityId);
+ public Optional<MvccEntity> load(ApplicationScope applicationScope, CollectionScope scope, Id entityId);
/**
@@ -104,7 +106,7 @@ public interface MvccEntitySerializationStrategy extends Migration, VersionedDat
* can be used in a mark+sweep system. The entity with the given version will exist in the context, but no data
* will be stored
*/
- public MutationBatch mark(CollectionScope context, Id entityId, UUID version);
+ public MutationBatch mark(ApplicationScope applicationScope, CollectionScope context, Id entityId, UUID version);
/**
@@ -114,6 +116,6 @@ public interface MvccEntitySerializationStrategy extends Migration, VersionedDat
* @param entityId The entity id to delete
* @param version The version to delete
*/
- public MutationBatch delete(CollectionScope context, Id entityId, UUID version);
+ public MutationBatch delete(ApplicationScope applicationScope, CollectionScope context, Id entityId, UUID version);
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSerializationStrategy.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSerializationStrategy.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSerializationStrategy.java
index d7a6407..ab1f24b 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSerializationStrategy.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSerializationStrategy.java
@@ -25,7 +25,9 @@ import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import com.netflix.astyanax.model.ConsistencyLevel;
import org.apache.usergrid.persistence.collection.CollectionScope;
+import org.apache.usergrid.persistence.collection.ScopeSet;
import org.apache.usergrid.persistence.core.migration.schema.Migration;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.field.Field;
@@ -42,7 +44,7 @@ public interface UniqueValueSerializationStrategy extends Migration {
* @param uniqueValue Object to be written
* @return MutatationBatch that encapsulates operation, caller may or may not execute.
*/
- public MutationBatch write( CollectionScope collectionScope, UniqueValue uniqueValue );
+ MutationBatch write( ApplicationScope applicationScope, CollectionScope collectionScope, UniqueValue uniqueValue );
/**
* Write the specified UniqueValue to Cassandra with optional timeToLive in milliseconds.
@@ -51,30 +53,30 @@ public interface UniqueValueSerializationStrategy extends Migration {
* @param timeToLive How long object should live in seconds. -1 implies store forever
* @return MutatationBatch that encapsulates operation, caller may or may not execute.
*/
- public MutationBatch write( CollectionScope collectionScope, UniqueValue uniqueValue, int timeToLive );
+ MutationBatch write( ApplicationScope applicationScope, CollectionScope collectionScope, UniqueValue uniqueValue, int timeToLive );
/**
* Load UniqueValue that matches field from collection or null if that value does not exist.
*
- * @param collectionScope scope in which to look for field name/value
+ * @param applicationScope scope in which to look for field name/value
* @param fields Field name/value to search for
*
* @return UniqueValueSet containing fields from the collection that exist in cassandra
*
* @throws ConnectionException on error connecting to Cassandra
*/
- public UniqueValueSet load( CollectionScope collectionScope, Collection<Field> fields ) throws ConnectionException;
+ UniqueValueSet load(ApplicationScope applicationScope, Collection<ScopeSet<Field>> fields ) throws ConnectionException;
/**
* Load UniqueValue that matches field from collection or null if that value does not exist.
*
- * @param colScope Collection scope in which to look for field name/value
+ * @param applicationScope Collection scope in which to look for field name/value
* @param consistencyLevel Consistency level of query
* @param fields Field name/value to search for
* @return UniqueValueSet containing fields from the collection that exist in cassandra
* @throws ConnectionException on error connecting to Cassandra
*/
- public UniqueValueSet load( CollectionScope colScope, ConsistencyLevel consistencyLevel, Collection<Field> fields ) throws ConnectionException;
+ UniqueValueSet load(ApplicationScope applicationScope, ConsistencyLevel consistencyLevel, Collection<ScopeSet<Field>> fields ) throws ConnectionException;
/**
@@ -85,7 +87,7 @@ public interface UniqueValueSerializationStrategy extends Migration {
* @param entityId
* @return
*/
- public Iterator<UniqueValue> getAllUniqueFields(CollectionScope collectionScope, Id entityId);
+ Iterator<UniqueValue> getAllUniqueFields(ApplicationScope applicationScope, CollectionScope collectionScope, Id entityId);
/**
@@ -94,7 +96,7 @@ public interface UniqueValueSerializationStrategy extends Migration {
* @param uniqueValue Object to be deleted.
* @return MutatationBatch that encapsulates operation, caller may or may not execute.
*/
- public MutationBatch delete( CollectionScope scope, UniqueValue uniqueValue );
+ MutationBatch delete(ApplicationScope applicationScope, CollectionScope scope, UniqueValue uniqueValue );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSet.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSet.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSet.java
index 702e908..a988f23 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSet.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSet.java
@@ -20,6 +20,8 @@ package org.apache.usergrid.persistence.collection.serialization;/*
import java.util.Set;
+import org.apache.usergrid.persistence.collection.CollectionScope;
+
/**
* A read only view of unique values
@@ -31,6 +33,6 @@ public interface UniqueValueSet extends Iterable<UniqueValue> {
* @param fieldName
* @return
*/
- public UniqueValue getValue(final String fieldName);
+ public UniqueValue getValue(final CollectionScope scope, final String fieldName);
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/LogEntryIterator.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/LogEntryIterator.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/LogEntryIterator.java
index 2b99033..5c44ff4 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/LogEntryIterator.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/LogEntryIterator.java
@@ -9,6 +9,7 @@ import java.util.UUID;
import org.apache.usergrid.persistence.collection.CollectionScope;
import org.apache.usergrid.persistence.collection.mvcc.MvccLogEntrySerializationStrategy;
import org.apache.usergrid.persistence.collection.MvccLogEntry;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.entity.Id;
import com.google.common.base.Preconditions;
@@ -22,6 +23,7 @@ public class LogEntryIterator implements Iterator<MvccLogEntry> {
private final MvccLogEntrySerializationStrategy logEntrySerializationStrategy;
+ private final ApplicationScope applicationScope;
private final CollectionScope scope;
private final Id entityId;
private final int pageSize;
@@ -40,13 +42,14 @@ public class LogEntryIterator implements Iterator<MvccLogEntry> {
* @param pageSize The fetch size to get when querying the serialization strategy
*/
public LogEntryIterator( final MvccLogEntrySerializationStrategy logEntrySerializationStrategy,
- final CollectionScope scope, final Id entityId, final UUID maxVersion,
+ final ApplicationScope applicationScope, final CollectionScope scope, final Id entityId, final UUID maxVersion,
final int pageSize ) {
Preconditions.checkArgument( pageSize > 0, "pageSize must be > 0" );
this.logEntrySerializationStrategy = logEntrySerializationStrategy;
this.scope = scope;
+ this.applicationScope = applicationScope;
this.entityId = entityId;
this.nextStart = maxVersion;
this.pageSize = pageSize;
@@ -92,7 +95,7 @@ public class LogEntryIterator implements Iterator<MvccLogEntry> {
final int requestedSize = pageSize + 1;
//loop through even entry that's < this one and remove it
- List<MvccLogEntry> results = logEntrySerializationStrategy.load( scope, entityId, nextStart, requestedSize );
+ List<MvccLogEntry> results = logEntrySerializationStrategy.load( applicationScope, scope, entityId, nextStart, requestedSize );
//we always remove the first version if it's equal since it's returned
if ( results.size() > 0 && results.get( 0 ).getVersion().equals( nextStart ) ) {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
index ad1d91a..bfabf60 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
@@ -25,7 +25,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.UUID;
-import org.apache.usergrid.persistence.collection.serialization.MvccEntitySerializationStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,19 +35,22 @@ import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.usergrid.persistence.collection.CollectionScope;
import org.apache.usergrid.persistence.collection.EntitySet;
import org.apache.usergrid.persistence.collection.MvccEntity;
+import org.apache.usergrid.persistence.collection.ScopeSet;
import org.apache.usergrid.persistence.collection.exception.CollectionRuntimeException;
import org.apache.usergrid.persistence.collection.exception.DataCorruptionException;
import org.apache.usergrid.persistence.collection.mvcc.entity.impl.MvccEntityImpl;
+import org.apache.usergrid.persistence.collection.serialization.MvccEntitySerializationStrategy;
import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
-import org.apache.usergrid.persistence.model.util.EntityUtils;
import org.apache.usergrid.persistence.core.astyanax.CassandraFig;
import org.apache.usergrid.persistence.core.astyanax.ColumnNameIterator;
import org.apache.usergrid.persistence.core.astyanax.ColumnParser;
import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamily;
import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamilyDefinition;
import org.apache.usergrid.persistence.core.astyanax.ScopedRowKey;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.entity.Id;
+import org.apache.usergrid.persistence.model.util.EntityUtils;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
import com.google.common.base.Optional;
@@ -61,14 +63,11 @@ import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import com.netflix.astyanax.model.Column;
import com.netflix.astyanax.model.ColumnList;
import com.netflix.astyanax.model.Row;
-import com.netflix.astyanax.model.Rows;
import com.netflix.astyanax.query.RowQuery;
import com.netflix.astyanax.serializers.AbstractSerializer;
import rx.Observable;
import rx.Scheduler;
-import rx.functions.Func1;
-import rx.functions.Func2;
import rx.schedulers.Schedulers;
@@ -97,14 +96,14 @@ public abstract class MvccEntitySerializationStrategyImpl implements MvccEntityS
@Override
- public MutationBatch write( final CollectionScope collectionScope, final MvccEntity entity ) {
+ public MutationBatch write(final ApplicationScope applicationScope, final CollectionScope collectionScope, final MvccEntity entity ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
Preconditions.checkNotNull( entity, "entity is required" );
final UUID colName = entity.getVersion();
final Id entityId = entity.getId();
- return doWrite( collectionScope, entityId, new RowOp() {
+ return doWrite( applicationScope, collectionScope, entityId, new RowOp() {
@Override
public void doOp( final ColumnListMutation<UUID> colMutation ) {
colMutation.putColumn( colName, getEntitySerializer()
@@ -115,11 +114,11 @@ public abstract class MvccEntitySerializationStrategyImpl implements MvccEntityS
@Override
- public EntitySet load( final CollectionScope collectionScope, final Collection<Id> entityIds,
+ public EntitySet load(final ApplicationScope applicationScope, final Collection<ScopeSet<Id>> entityIds,
final UUID maxVersion ) {
- Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
+ Preconditions.checkNotNull( entityIds, "entityIds is required" );
Preconditions.checkNotNull( entityIds, "entityIds is required" );
Preconditions.checkArgument( entityIds.size() > 0, "entityIds is required" );
Preconditions.checkNotNull( maxVersion, "version is required" );
@@ -130,25 +129,13 @@ public abstract class MvccEntitySerializationStrategyImpl implements MvccEntityS
}
- final Id applicationId = collectionScope.getApplication();
- final Id ownerId = collectionScope.getOwner();
- final String collectionName = collectionScope.getName();
+ final Id applicationId = applicationScope.getApplication();
- final List<ScopedRowKey<CollectionPrefixedKey<Id>>> rowKeys = new ArrayList<>( entityIds.size() );
-
-
- for ( final Id entityId : entityIds ) {
- final CollectionPrefixedKey<Id> collectionPrefixedKey =
- new CollectionPrefixedKey<>( collectionName, ownerId, entityId );
-
- final ScopedRowKey<CollectionPrefixedKey<Id>> rowKey =
- ScopedRowKey.fromKey( applicationId, collectionPrefixedKey );
+ final List<ScopedRowKey<CollectionPrefixedKey<Id>>> rowKeys = new ArrayList<>( entityIds.size() );
- rowKeys.add( rowKey );
- }
/**
* Our settings may mean we exceed our maximum thrift buffer size. If we do, we have to make multiple requests, not just one.
@@ -190,7 +177,7 @@ public abstract class MvccEntitySerializationStrategyImpl implements MvccEntityS
.withColumnRange( maxVersion, null, false, 1 ).execute().getResult();
}
catch ( ConnectionException e ) {
- throw new CollectionRuntimeException( null, collectionScope,
+ throw new CollectionRuntimeException( null, applicationScope,
"An error occurred connecting to cassandra", e );
}
} ).subscribeOn( scheduler );
@@ -230,7 +217,7 @@ public abstract class MvccEntitySerializationStrategyImpl implements MvccEntityS
@Override
- public Iterator<MvccEntity> loadDescendingHistory( final CollectionScope collectionScope, final Id entityId,
+ public Iterator<MvccEntity> loadDescendingHistory( final ApplicationScope applicationScope, final CollectionScope collectionScope, final Id entityId,
final UUID version, final int fetchSize ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
@@ -239,7 +226,7 @@ public abstract class MvccEntitySerializationStrategyImpl implements MvccEntityS
Preconditions.checkArgument( fetchSize > 0, "max Size must be greater than 0" );
- final Id applicationId = collectionScope.getApplication();
+ final Id applicationId = applicationScope.getApplication();
final Id ownerId = collectionScope.getOwner();
final String collectionName = collectionScope.getName();
@@ -260,7 +247,7 @@ public abstract class MvccEntitySerializationStrategyImpl implements MvccEntityS
@Override
- public Iterator<MvccEntity> loadAscendingHistory( final CollectionScope collectionScope, final Id entityId,
+ public Iterator<MvccEntity> loadAscendingHistory( final ApplicationScope applicationScope, final CollectionScope collectionScope, final Id entityId,
final UUID version, final int fetchSize ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
@@ -269,7 +256,7 @@ public abstract class MvccEntitySerializationStrategyImpl implements MvccEntityS
Preconditions.checkArgument( fetchSize > 0, "max Size must be greater than 0" );
- final Id applicationId = collectionScope.getApplication();
+ final Id applicationId = applicationScope.getApplication();
final Id ownerId = collectionScope.getOwner();
final String collectionName = collectionScope.getName();
@@ -290,20 +277,22 @@ public abstract class MvccEntitySerializationStrategyImpl implements MvccEntityS
@Override
- public Optional<MvccEntity> load( final CollectionScope scope, final Id entityId ) {
- final EntitySet results = load( scope, Collections.singleton( entityId ), UUIDGenerator.newTimeUUID() );
+ public Optional<MvccEntity> load( final ApplicationScope applicationScope, final CollectionScope scope, final Id entityId ) {
+
+ final ScopeSet<Id> scopeSet = new ScopeSetImpl<>( scope, Collections.singleton( entityId ) );
+ final EntitySet results = load(applicationScope, Collections.singleton( scopeSet ) , UUIDGenerator.newTimeUUID() );
return Optional.fromNullable( results.getEntity( entityId ));
}
@Override
- public MutationBatch mark( final CollectionScope collectionScope, final Id entityId, final UUID version ) {
+ public MutationBatch mark( final ApplicationScope applicationScope, final CollectionScope collectionScope, final Id entityId, final UUID version ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
Preconditions.checkNotNull( entityId, "entity id is required" );
Preconditions.checkNotNull( version, "version is required" );
- return doWrite( collectionScope, entityId, new RowOp() {
+ return doWrite(applicationScope, collectionScope, entityId, new RowOp() {
@Override
public void doOp( final ColumnListMutation<UUID> colMutation ) {
colMutation.putColumn( version, getEntitySerializer()
@@ -314,13 +303,13 @@ public abstract class MvccEntitySerializationStrategyImpl implements MvccEntityS
@Override
- public MutationBatch delete( final CollectionScope collectionScope, final Id entityId, final UUID version ) {
+ public MutationBatch delete( final ApplicationScope applicationScope, final CollectionScope collectionScope, final Id entityId, final UUID version ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
Preconditions.checkNotNull( entityId, "entity id is required" );
Preconditions.checkNotNull( version, "version is required" );
- return doWrite( collectionScope, entityId, new RowOp() {
+ return doWrite( applicationScope, collectionScope, entityId, new RowOp() {
@Override
public void doOp( final ColumnListMutation<UUID> colMutation ) {
colMutation.deleteColumn( version );
@@ -347,10 +336,10 @@ public abstract class MvccEntitySerializationStrategyImpl implements MvccEntityS
/**
* Do the write on the correct row for the entity id with the operation
*/
- private MutationBatch doWrite( final CollectionScope collectionScope, final Id entityId, final RowOp op ) {
+ private MutationBatch doWrite(final ApplicationScope applicationScope, final CollectionScope collectionScope, final Id entityId, final RowOp op ) {
final MutationBatch batch = keyspace.prepareMutationBatch();
- final Id applicationId = collectionScope.getApplication();
+ final Id applicationId = applicationScope.getApplication();
final Id ownerId = collectionScope.getOwner();
final String collectionName = collectionScope.getName();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyProxyImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyProxyImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyProxyImpl.java
index b815ef4..4d60f5d 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyProxyImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyProxyImpl.java
@@ -25,12 +25,14 @@ import java.util.UUID;
import org.apache.usergrid.persistence.collection.CollectionScope;
import org.apache.usergrid.persistence.collection.EntitySet;
import org.apache.usergrid.persistence.collection.MvccEntity;
+import org.apache.usergrid.persistence.collection.ScopeSet;
import org.apache.usergrid.persistence.collection.serialization.MvccEntitySerializationStrategy;
import org.apache.usergrid.persistence.collection.serialization.impl.migration.CollectionMigrationPlugin;
import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamilyDefinition;
import org.apache.usergrid.persistence.core.migration.data.MigrationInfoCache;
import org.apache.usergrid.persistence.core.migration.data.MigrationRelationship;
import org.apache.usergrid.persistence.core.migration.data.VersionedMigrationSet;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.entity.Id;
import com.google.common.base.Optional;
@@ -64,110 +66,110 @@ public class MvccEntitySerializationStrategyProxyImpl implements MvccEntitySeria
@Override
- public MutationBatch write( final CollectionScope context, final MvccEntity entity ) {
+ public MutationBatch write( final ApplicationScope applicationScope, final CollectionScope context, final MvccEntity entity ) {
final MigrationRelationship<MvccEntitySerializationStrategy> migration = getMigrationRelationShip();
if ( migration.needsMigration() ) {
final MutationBatch aggregateBatch = keyspace.prepareMutationBatch();
- aggregateBatch.mergeShallow( migration.from.write( context, entity ) );
- aggregateBatch.mergeShallow( migration.to.write( context, entity ) );
+ aggregateBatch.mergeShallow( migration.from.write( applicationScope, context, entity ) );
+ aggregateBatch.mergeShallow( migration.to.write( applicationScope, context, entity ) );
return aggregateBatch;
}
- return migration.to.write( context, entity );
+ return migration.to.write(applicationScope, context, entity );
}
@Override
- public EntitySet load( final CollectionScope scope, final Collection<Id> entityIds, final UUID maxVersion ) {
+ public EntitySet load( final ApplicationScope applicationScope, final Collection<ScopeSet<Id>> entityIds, final UUID maxVersion ) {
final MigrationRelationship<MvccEntitySerializationStrategy> migration = getMigrationRelationShip();
if ( migration.needsMigration() ) {
- return migration.from.load( scope, entityIds, maxVersion );
+ return migration.from.load(applicationScope, entityIds, maxVersion );
}
- return migration.to.load( scope, entityIds, maxVersion );
+ return migration.to.load( applicationScope, entityIds, maxVersion );
}
@Override
- public Iterator<MvccEntity> loadDescendingHistory( final CollectionScope context, final Id entityId,
+ public Iterator<MvccEntity> loadDescendingHistory( final ApplicationScope applicationScope, final CollectionScope context, final Id entityId,
final UUID version, final int fetchSize ) {
final MigrationRelationship<MvccEntitySerializationStrategy> migration = getMigrationRelationShip();
if ( migration.needsMigration()) {
- return migration.from.loadDescendingHistory( context, entityId, version, fetchSize );
+ return migration.from.loadDescendingHistory(applicationScope, context, entityId, version, fetchSize );
}
- return migration.to.loadDescendingHistory( context, entityId, version, fetchSize );
+ return migration.to.loadDescendingHistory(applicationScope, context, entityId, version, fetchSize );
}
@Override
- public Iterator<MvccEntity> loadAscendingHistory( final CollectionScope context, final Id entityId,
+ public Iterator<MvccEntity> loadAscendingHistory(final ApplicationScope applicationScope, final CollectionScope context, final Id entityId,
final UUID version, final int fetchSize ) {
final MigrationRelationship<MvccEntitySerializationStrategy> migration = getMigrationRelationShip();
if ( migration.needsMigration() ) {
- return migration.from.loadAscendingHistory( context, entityId, version, fetchSize );
+ return migration.from.loadAscendingHistory(applicationScope, context, entityId, version, fetchSize );
}
- return migration.to.loadAscendingHistory( context, entityId, version, fetchSize );
+ return migration.to.loadAscendingHistory(applicationScope, context, entityId, version, fetchSize );
}
@Override
- public Optional<MvccEntity> load( final CollectionScope scope, final Id entityId ) {
+ public Optional<MvccEntity> load(final ApplicationScope applicationScope, final CollectionScope scope, final Id entityId ) {
final MigrationRelationship<MvccEntitySerializationStrategy> migration = getMigrationRelationShip();
if ( migration.needsMigration() ) {
- return migration.from.load( scope, entityId );
+ return migration.from.load(applicationScope, scope, entityId );
}
- return migration.to.load( scope, entityId );
+ return migration.to.load(applicationScope, scope, entityId );
}
@Override
- public MutationBatch mark( final CollectionScope context, final Id entityId, final UUID version ) {
+ public MutationBatch mark(final ApplicationScope applicationScope, final CollectionScope context, final Id entityId, final UUID version ) {
final MigrationRelationship<MvccEntitySerializationStrategy> migration = getMigrationRelationShip();
if ( migration.needsMigration() ) {
final MutationBatch aggregateBatch = keyspace.prepareMutationBatch();
- aggregateBatch.mergeShallow( migration.from.mark( context, entityId, version ) );
- aggregateBatch.mergeShallow( migration.to.mark( context, entityId, version ) );
+ aggregateBatch.mergeShallow( migration.from.mark( applicationScope, context, entityId, version ) );
+ aggregateBatch.mergeShallow( migration.to.mark(applicationScope, context, entityId, version ) );
return aggregateBatch;
}
- return migration.to.mark( context, entityId, version );
+ return migration.to.mark(applicationScope, context, entityId, version );
}
@Override
- public MutationBatch delete( final CollectionScope context, final Id entityId, final UUID version ) {
+ public MutationBatch delete(final ApplicationScope applicationScope, final CollectionScope context, final Id entityId, final UUID version ) {
final MigrationRelationship<MvccEntitySerializationStrategy> migration = getMigrationRelationShip();
if ( migration.needsMigration() ) {
final MutationBatch aggregateBatch = keyspace.prepareMutationBatch();
- aggregateBatch.mergeShallow( migration.from.delete( context, entityId, version ) );
- aggregateBatch.mergeShallow( migration.to.delete( context, entityId, version ) );
+ aggregateBatch.mergeShallow( migration.from.delete( applicationScope, context, entityId, version ) );
+ aggregateBatch.mergeShallow( migration.to.delete(applicationScope, context, entityId, version ) );
return aggregateBatch;
}
- return migration.to.delete( context, entityId, version );
+ return migration.to.delete(applicationScope, context, entityId, version );
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyV3Impl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyV3Impl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyV3Impl.java
index de959b5..7b6124e 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyV3Impl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyV3Impl.java
@@ -18,12 +18,14 @@ import org.apache.cassandra.db.marshal.BytesType;
import org.apache.usergrid.persistence.collection.CollectionScope;
import org.apache.usergrid.persistence.collection.EntitySet;
import org.apache.usergrid.persistence.collection.MvccEntity;
+import org.apache.usergrid.persistence.collection.ScopeSet;
import org.apache.usergrid.persistence.collection.exception.CollectionRuntimeException;
import org.apache.usergrid.persistence.collection.exception.DataCorruptionException;
import org.apache.usergrid.persistence.collection.exception.EntityTooLargeException;
import org.apache.usergrid.persistence.collection.mvcc.entity.impl.MvccEntityImpl;
import org.apache.usergrid.persistence.collection.serialization.MvccEntitySerializationStrategy;
import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.util.EntityUtils;
import org.apache.usergrid.persistence.core.astyanax.CassandraFig;
import org.apache.usergrid.persistence.core.astyanax.ColumnParser;
@@ -103,14 +105,14 @@ public class MvccEntitySerializationStrategyV3Impl implements MvccEntitySerializ
@Override
- public MutationBatch write( final CollectionScope collectionScope, final MvccEntity entity ) {
+ public MutationBatch write( final ApplicationScope applicationScope, final CollectionScope collectionScope, final MvccEntity entity ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
Preconditions.checkNotNull( entity, "entity is required" );
final Id entityId = entity.getId();
final UUID version = entity.getVersion();
- return doWrite( collectionScope, entityId, version, new RowOp() {
+ return doWrite(applicationScope, collectionScope, entityId, version, new RowOp() {
@Override
public void doOp( final ColumnListMutation<Boolean> colMutation ) {
colMutation.putColumn( COL_VALUE,
@@ -121,11 +123,11 @@ public class MvccEntitySerializationStrategyV3Impl implements MvccEntitySerializ
@Override
- public EntitySet load( final CollectionScope collectionScope, final Collection<Id> entityIds,
+ public EntitySet load( final ApplicationScope applicationScope, final Collection<ScopeSet<Id>> entityIds,
final UUID maxVersion ) {
- Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
+ Preconditions.checkNotNull( applicationScope, "applicationScope is required" );
Preconditions.checkNotNull( entityIds, "entityIds is required" );
Preconditions.checkArgument( entityIds.size() > 0, "entityIds is required" );
Preconditions.checkNotNull( maxVersion, "version is required" );
@@ -137,24 +139,29 @@ public class MvccEntitySerializationStrategyV3Impl implements MvccEntitySerializ
}
- final Id applicationId = collectionScope.getApplication();
- final Id ownerId = collectionScope.getOwner();
- final String collectionName = collectionScope.getName();
+ final Id applicationId = applicationScope.getApplication();
final List<ScopedRowKey<CollectionPrefixedKey<Id>>> rowKeys = new ArrayList<>( entityIds.size() );
+ for(ScopeSet<Id> collectionScope: entityIds) {
+
+ final Id ownerId = collectionScope.getScope().getOwner();
+ final String collectionName = collectionScope.getScope().getName();
+
- for ( final Id entityId : entityIds ) {
- final CollectionPrefixedKey<Id> collectionPrefixedKey =
+
+ for ( final Id entityId : collectionScope.getIdentifiers() ) {
+ final CollectionPrefixedKey<Id> collectionPrefixedKey =
new CollectionPrefixedKey<>( collectionName, ownerId, entityId );
- final ScopedRowKey<CollectionPrefixedKey<Id>> rowKey =
+ final ScopedRowKey<CollectionPrefixedKey<Id>> rowKey =
ScopedRowKey.fromKey( applicationId, collectionPrefixedKey );
- rowKeys.add( rowKey );
+ rowKeys.add( rowKey );
+ }
}
/**
@@ -199,7 +206,7 @@ public class MvccEntitySerializationStrategyV3Impl implements MvccEntitySerializ
.withColumnSlice( COL_VALUE ).execute().getResult();
}
catch ( ConnectionException e ) {
- throw new CollectionRuntimeException( null, collectionScope,
+ throw new CollectionRuntimeException( null, applicationScope,
"An error occurred connecting to cassandra", e );
}
} ).subscribeOn( scheduler );
@@ -240,7 +247,7 @@ public class MvccEntitySerializationStrategyV3Impl implements MvccEntitySerializ
@Override
- public Iterator<MvccEntity> loadDescendingHistory( final CollectionScope collectionScope, final Id entityId,
+ public Iterator<MvccEntity> loadDescendingHistory( final ApplicationScope applicationScope, final CollectionScope collectionScope, final Id entityId,
final UUID version, final int fetchSize ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
@@ -255,7 +262,7 @@ public class MvccEntitySerializationStrategyV3Impl implements MvccEntitySerializ
@Override
- public Iterator<MvccEntity> loadAscendingHistory( final CollectionScope collectionScope, final Id entityId,
+ public Iterator<MvccEntity> loadAscendingHistory( final ApplicationScope applicationScope, final CollectionScope collectionScope, final Id entityId,
final UUID version, final int fetchSize ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
@@ -268,21 +275,25 @@ public class MvccEntitySerializationStrategyV3Impl implements MvccEntitySerializ
@Override
- public Optional<MvccEntity> load( final CollectionScope scope, final Id entityId ) {
- final EntitySet results = load( scope, Collections.singleton( entityId ), UUIDGenerator.newTimeUUID() );
+ public Optional<MvccEntity> load( final ApplicationScope applicationScope, final CollectionScope scope, final Id entityId ) {
+
+
+ final ScopeSet<Id> scopeSet = new ScopeSetImpl<>( scope, Collections.singleton( entityId ) );
+
+ final EntitySet results = load( applicationScope, Collections.singleton( scopeSet ), UUIDGenerator.newTimeUUID() );
return Optional.fromNullable( results.getEntity( entityId ));
}
@Override
- public MutationBatch mark( final CollectionScope collectionScope, final Id entityId, final UUID version ) {
+ public MutationBatch mark( final ApplicationScope applicationScope, final CollectionScope collectionScope, final Id entityId, final UUID version ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
Preconditions.checkNotNull( entityId, "entity id is required" );
Preconditions.checkNotNull( version, "version is required" );
- return doWrite( collectionScope, entityId, version, new RowOp() {
+ return doWrite( applicationScope, collectionScope, entityId, version, new RowOp() {
@Override
public void doOp( final ColumnListMutation<Boolean> colMutation ) {
colMutation.putColumn( COL_VALUE, entitySerializer.toByteBuffer(
@@ -293,13 +304,13 @@ public class MvccEntitySerializationStrategyV3Impl implements MvccEntitySerializ
@Override
- public MutationBatch delete( final CollectionScope collectionScope, final Id entityId, final UUID version ) {
+ public MutationBatch delete( final ApplicationScope applicationScope, final CollectionScope collectionScope, final Id entityId, final UUID version ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
Preconditions.checkNotNull( entityId, "entity id is required" );
Preconditions.checkNotNull( version, "version is required" );
- return doWrite( collectionScope, entityId, version, new RowOp() {
+ return doWrite(applicationScope, collectionScope, entityId, version, new RowOp() {
@Override
public void doOp( final ColumnListMutation<Boolean> colMutation ) {
colMutation.deleteColumn( Boolean.TRUE );
@@ -326,10 +337,10 @@ public class MvccEntitySerializationStrategyV3Impl implements MvccEntitySerializ
/**
* Do the write on the correct row for the entity id with the operation
*/
- private MutationBatch doWrite( final CollectionScope collectionScope, final Id entityId, final UUID version, final RowOp op ) {
+ private MutationBatch doWrite( final ApplicationScope applicationScope, final CollectionScope collectionScope, final Id entityId, final UUID version, final RowOp op ) {
final MutationBatch batch = keyspace.prepareMutationBatch();
- final Id applicationId = collectionScope.getApplication();
+ final Id applicationId = applicationScope.getApplication();
final Id ownerId = collectionScope.getOwner();
final String collectionName = collectionScope.getName();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccLogEntrySerializationStrategyImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccLogEntrySerializationStrategyImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccLogEntrySerializationStrategyImpl.java
index 831091d..4e71e38 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccLogEntrySerializationStrategyImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccLogEntrySerializationStrategyImpl.java
@@ -38,6 +38,7 @@ import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.usergrid.persistence.collection.CollectionScope;
import org.apache.usergrid.persistence.collection.MvccLogEntry;
+import org.apache.usergrid.persistence.collection.ScopeSet;
import org.apache.usergrid.persistence.collection.VersionSet;
import org.apache.usergrid.persistence.collection.exception.CollectionRuntimeException;
import org.apache.usergrid.persistence.collection.mvcc.MvccLogEntrySerializationStrategy;
@@ -49,6 +50,7 @@ import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamily;
import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamilyDefinition;
import org.apache.usergrid.persistence.core.astyanax.ScopedRowKey;
import org.apache.usergrid.persistence.core.migration.schema.Migration;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.entity.Id;
import com.google.common.base.Preconditions;
@@ -98,7 +100,7 @@ public class MvccLogEntrySerializationStrategyImpl implements MvccLogEntrySerial
@Override
- public MutationBatch write( final CollectionScope collectionScope, final MvccLogEntry entry ) {
+ public MutationBatch write( final ApplicationScope applicationScope, final CollectionScope collectionScope, final MvccLogEntry entry ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
Preconditions.checkNotNull( entry, "entry is required" );
@@ -108,7 +110,7 @@ public class MvccLogEntrySerializationStrategyImpl implements MvccLogEntrySerial
final UUID colName = entry.getVersion();
final StageStatus stageStatus = new StageStatus( stage, entry.getState() );
- return doWrite( collectionScope, entry.getEntityId(), entry.getVersion(), new RowOp() {
+ return doWrite(applicationScope, collectionScope, entry.getEntityId(), entry.getVersion(), new RowOp() {
@Override
public void doOp( final ColumnListMutation<UUID> colMutation ) {
@@ -126,9 +128,9 @@ public class MvccLogEntrySerializationStrategyImpl implements MvccLogEntrySerial
@Override
- public VersionSet load( final CollectionScope collectionScope, final Collection<Id> entityIds,
+ public VersionSet load( final ApplicationScope applicationScope, final Collection<ScopeSet<Id>> entityIds,
final UUID maxVersion ) {
- Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
+ Preconditions.checkNotNull( applicationScope, "applicationScope is required" );
Preconditions.checkNotNull( entityIds, "entityIds is required" );
Preconditions.checkArgument( entityIds.size() > 0, "You must specify an Id" );
Preconditions.checkNotNull( maxVersion, "maxVersion is required" );
@@ -139,24 +141,28 @@ public class MvccLogEntrySerializationStrategyImpl implements MvccLogEntrySerial
"requested size cannot be over configured maximum" );
- final Id applicationId = collectionScope.getApplication();
- final Id ownerId = collectionScope.getOwner();
- final String collectionName = collectionScope.getName();
+ final Id applicationId = applicationScope.getApplication();
+
final List<ScopedRowKey<CollectionPrefixedKey<Id>>> rowKeys = new ArrayList<>( entityIds.size() );
- for ( final Id entityId : entityIds ) {
- final CollectionPrefixedKey<Id> collectionPrefixedKey =
+ for(ScopeSet<Id> scopeSet: entityIds) {
+ final Id ownerId = scopeSet.getScope().getOwner();
+ final String collectionName = scopeSet.getScope().getName();
+
+ for ( final Id entityId : scopeSet.getIdentifiers() ) {
+ final CollectionPrefixedKey<Id> collectionPrefixedKey =
new CollectionPrefixedKey<>( collectionName, ownerId, entityId );
- final ScopedRowKey<CollectionPrefixedKey<Id>> rowKey =
+ final ScopedRowKey<CollectionPrefixedKey<Id>> rowKey =
ScopedRowKey.fromKey( applicationId, collectionPrefixedKey );
- rowKeys.add( rowKey );
+ rowKeys.add( rowKey );
+ }
}
@@ -169,7 +175,7 @@ public class MvccLogEntrySerializationStrategyImpl implements MvccLogEntrySerial
.iterator();
}
catch ( ConnectionException e ) {
- throw new CollectionRuntimeException( null, collectionScope, "An error occurred connecting to cassandra",
+ throw new CollectionRuntimeException( null, applicationScope, "An error occurred connecting to cassandra",
e );
}
@@ -207,8 +213,9 @@ public class MvccLogEntrySerializationStrategyImpl implements MvccLogEntrySerial
@Override
- public List<MvccLogEntry> load( final CollectionScope collectionScope, final Id entityId, final UUID version,
+ public List<MvccLogEntry> load( final ApplicationScope applicationScope, final CollectionScope collectionScope, final Id entityId, final UUID version,
final int maxSize ) {
+ Preconditions.checkNotNull( applicationScope, "applicationScope is required" );
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
Preconditions.checkNotNull( entityId, "entity id is required" );
Preconditions.checkNotNull( version, "version is required" );
@@ -218,7 +225,7 @@ public class MvccLogEntrySerializationStrategyImpl implements MvccLogEntrySerial
ColumnList<UUID> columns;
try {
- final Id applicationId = collectionScope.getApplication();
+ final Id applicationId = applicationScope.getApplication();
final Id ownerId = collectionScope.getOwner();
final String collectionName = collectionScope.getName();
@@ -253,13 +260,14 @@ public class MvccLogEntrySerializationStrategyImpl implements MvccLogEntrySerial
@Override
- public MutationBatch delete( final CollectionScope context, final Id entityId, final UUID version ) {
+ public MutationBatch delete(final ApplicationScope applicationScope, final CollectionScope context, final Id entityId, final UUID version ) {
+ Preconditions.checkNotNull( applicationScope, "applicationScope is required" );
Preconditions.checkNotNull( context, "context is required" );
Preconditions.checkNotNull( entityId, "entityId is required" );
Preconditions.checkNotNull( version, "version context is required" );
- return doWrite( context, entityId, version, new RowOp() {
+ return doWrite( applicationScope, context, entityId, version, new RowOp() {
@Override
public void doOp( final ColumnListMutation<UUID> colMutation ) {
colMutation.deleteColumn( version );
@@ -299,7 +307,7 @@ public class MvccLogEntrySerializationStrategyImpl implements MvccLogEntrySerial
*
* @param collectionScope We need to use this when getting the keyspace
*/
- private MutationBatch doWrite( CollectionScope collectionScope, Id entityId, UUID version, RowOp op ) {
+ private MutationBatch doWrite(final ApplicationScope applicationScope, CollectionScope collectionScope, Id entityId, UUID version, RowOp op ) {
final MutationBatch batch = keyspace.prepareMutationBatch();
@@ -307,7 +315,7 @@ public class MvccLogEntrySerializationStrategyImpl implements MvccLogEntrySerial
LOG.debug( "Writing version with timestamp '{}'", timestamp );
- final Id applicationId = collectionScope.getApplication();
+ final Id applicationId = applicationScope.getApplication();
final Id ownerId = collectionScope.getOwner();
final String collectionName = collectionScope.getName();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/ScopeSetImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/ScopeSetImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/ScopeSetImpl.java
new file mode 100644
index 0000000..98abb4b
--- /dev/null
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/ScopeSetImpl.java
@@ -0,0 +1,67 @@
+/*
+ * 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.persistence.collection.serialization.impl;
+
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.usergrid.persistence.collection.CollectionScope;
+import org.apache.usergrid.persistence.collection.ScopeSet;
+
+
+public class ScopeSetImpl<T> implements ScopeSet<T> {
+
+
+ public final CollectionScope scope;
+ public final Collection<T> identifiers;
+
+
+ public ScopeSetImpl( final CollectionScope scope, final Collection<T> identifiers ) {
+ this.scope = scope;
+ this.identifiers = identifiers;
+ }
+
+
+ public ScopeSetImpl( final CollectionScope scope, final T identifier ) {
+ this.scope = scope;
+ this.identifiers = Collections.singleton(identifier);
+ }
+
+
+
+ @Override
+ public CollectionScope getScope() {
+ return scope;
+ }
+
+
+ @Override
+ public Collection<T> getIdentifiers() {
+ return identifiers;
+ }
+
+
+ @Override
+ public void addIdentifier( final T identifier ) {
+ this.identifiers.add( identifier );
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSerializationStrategyImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSerializationStrategyImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSerializationStrategyImpl.java
index c95650c..d07ec24 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSerializationStrategyImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSerializationStrategyImpl.java
@@ -25,8 +25,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.UUID;
-import com.netflix.astyanax.model.ConsistencyLevel;
-import org.apache.usergrid.persistence.core.astyanax.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,6 +35,7 @@ import org.apache.usergrid.persistence.collection.serialization.SerializationFig
import org.apache.usergrid.persistence.collection.serialization.UniqueValue;
import org.apache.usergrid.persistence.collection.serialization.UniqueValueSerializationStrategy;
import org.apache.usergrid.persistence.collection.serialization.UniqueValueSet;
+import org.apache.usergrid.persistence.core.astyanax.CassandraFig;
import org.apache.usergrid.persistence.core.astyanax.ColumnNameIterator;
import org.apache.usergrid.persistence.core.astyanax.ColumnParser;
import org.apache.usergrid.persistence.core.astyanax.ColumnTypes;
@@ -44,6 +43,7 @@ import org.apache.usergrid.persistence.core.astyanax.IdRowCompositeSerializer;
import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamily;
import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamilyDefinition;
import org.apache.usergrid.persistence.core.astyanax.ScopedRowKey;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.core.util.ValidationUtils;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.field.Field;
@@ -55,6 +55,7 @@ import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import com.netflix.astyanax.model.Column;
+import com.netflix.astyanax.model.ConsistencyLevel;
import com.netflix.astyanax.model.Row;
import com.netflix.astyanax.query.RowQuery;
import com.netflix.astyanax.util.RangeBuilder;
@@ -113,8 +114,11 @@ public class UniqueValueSerializationStrategyImpl implements UniqueValueSerializ
}
- public MutationBatch write( final CollectionScope collectionScope, UniqueValue value ) {
+ public MutationBatch write( final ApplicationScope applicationScope, final CollectionScope collectionScope, final UniqueValue value ) {
+
+ Preconditions.checkNotNull( applicationScope, "applicationScope is required" );
+ Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
Preconditions.checkNotNull( value, "value is required" );
@@ -133,7 +137,7 @@ public class UniqueValueSerializationStrategyImpl implements UniqueValueSerializ
final EntityVersion ev = new EntityVersion( entityId, entityVersion );
final UniqueFieldEntry uniqueFieldEntry = new UniqueFieldEntry( entityVersion, field );
- return doWrite( collectionScope, value, new UniqueValueSerializationStrategyImpl.RowOp() {
+ return doWrite(applicationScope, collectionScope, value, new UniqueValueSerializationStrategyImpl.RowOp() {
@Override
public void doLookup( final ColumnListMutation<EntityVersion> colMutation ) {
@@ -150,7 +154,7 @@ public class UniqueValueSerializationStrategyImpl implements UniqueValueSerializ
@Override
- public MutationBatch write( final CollectionScope collectionScope, final UniqueValue value, final int timeToLive ) {
+ public MutationBatch write(final ApplicationScope applicationScope, final CollectionScope collectionScope, final UniqueValue value, final int timeToLive ) {
Preconditions.checkNotNull( value, "value is required" );
Preconditions.checkArgument( timeToLive > 0, "timeToLive must be greater than 0 is required" );
@@ -165,7 +169,7 @@ public class UniqueValueSerializationStrategyImpl implements UniqueValueSerializ
final EntityVersion ev = new EntityVersion( entityId, entityVersion );
final UniqueFieldEntry uniqueFieldEntry = new UniqueFieldEntry( entityVersion, field );
- return doWrite( collectionScope, value, new UniqueValueSerializationStrategyImpl.RowOp() {
+ return doWrite(applicationScope, collectionScope, value, new UniqueValueSerializationStrategyImpl.RowOp() {
@Override
public void doLookup( final ColumnListMutation<EntityVersion> colMutation ) {
@@ -184,7 +188,7 @@ public class UniqueValueSerializationStrategyImpl implements UniqueValueSerializ
@Override
- public MutationBatch delete( final CollectionScope scope, UniqueValue value ) {
+ public MutationBatch delete(final ApplicationScope applicationScope, final CollectionScope scope, UniqueValue value ) {
Preconditions.checkNotNull( value, "value is required" );
@@ -200,7 +204,7 @@ public class UniqueValueSerializationStrategyImpl implements UniqueValueSerializ
final EntityVersion ev = new EntityVersion( entityId, entityVersion );
final UniqueFieldEntry uniqueFieldEntry = new UniqueFieldEntry( entityVersion, field );
- return doWrite( scope, value, new UniqueValueSerializationStrategyImpl.RowOp() {
+ return doWrite(applicationScope, scope, value, new UniqueValueSerializationStrategyImpl.RowOp() {
@Override
public void doLookup( final ColumnListMutation<EntityVersion> colMutation ) {
@@ -221,14 +225,14 @@ public class UniqueValueSerializationStrategyImpl implements UniqueValueSerializ
*
* @param collectionScope We need to use this when getting the keyspace
*/
- private MutationBatch doWrite( CollectionScope collectionScope, UniqueValue uniqueValue, RowOp op ) {
+ private MutationBatch doWrite( final ApplicationScope applicationScope, CollectionScope collectionScope, UniqueValue uniqueValue, RowOp op ) {
final MutationBatch batch = keyspace.prepareMutationBatch();
final CollectionPrefixedKey<Field> uniquePrefixedKey =
new CollectionPrefixedKey<>( collectionScope.getName(), collectionScope.getOwner(), uniqueValue.getField() );
op.doLookup( batch
- .withRow( CF_UNIQUE_VALUES, ScopedRowKey.fromKey( collectionScope.getApplication(), uniquePrefixedKey ) ) );
+ .withRow( CF_UNIQUE_VALUES, ScopedRowKey.fromKey( applicationScope.getApplication(), uniquePrefixedKey ) ) );
final Id ownerId = collectionScope.getOwner();
@@ -239,7 +243,7 @@ public class UniqueValueSerializationStrategyImpl implements UniqueValueSerializ
op.doLog( batch.withRow( CF_ENTITY_UNIQUE_VALUES,
- ScopedRowKey.fromKey( collectionScope.getApplication(), collectionPrefixedEntityKey ) ) );
+ ScopedRowKey.fromKey( applicationScope.getApplication(), collectionPrefixedEntityKey ) ) );
return batch;
@@ -248,11 +252,13 @@ public class UniqueValueSerializationStrategyImpl implements UniqueValueSerializ
@Override
- public UniqueValueSet load(final CollectionScope colScope, final Collection<Field> fields ) throws ConnectionException{
- return load(colScope,ConsistencyLevel.valueOf(cassandraFig.getReadCL()), fields);
+ public UniqueValueSet load(final ApplicationScope applicationScope, final CollectionScope colScope, final Collection<Field> fields ) throws ConnectionException{
+ return load(applicationScope, colScope,ConsistencyLevel.valueOf(cassandraFig.getReadCL()), fields);
}
+
+
@Override
- public UniqueValueSet load(final CollectionScope colScope, final ConsistencyLevel consistencyLevel, final Collection<Field> fields )
+ public UniqueValueSet load(final ApplicationScope applicationScope, final CollectionScope colScope, final ConsistencyLevel consistencyLevel, final Collection<Field> fields )
throws ConnectionException {
Preconditions.checkNotNull( fields, "fields are required" );
@@ -261,7 +267,7 @@ public class UniqueValueSerializationStrategyImpl implements UniqueValueSerializ
final List<ScopedRowKey<CollectionPrefixedKey<Field>>> keys = new ArrayList<>( fields.size() );
- final Id applicationId = colScope.getApplication();
+ final Id applicationId = applicationScope.getApplication();
final Id ownerId = colScope.getOwner();
final String collectionName = colScope.getName();
@@ -312,14 +318,14 @@ public class UniqueValueSerializationStrategyImpl implements UniqueValueSerializ
@Override
- public Iterator<UniqueValue> getAllUniqueFields( final CollectionScope collectionScope, final Id entityId ) {
+ public Iterator<UniqueValue> getAllUniqueFields( final ApplicationScope applicationScope,final CollectionScope collectionScope, final Id entityId ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
Preconditions.checkNotNull( entityId, "entity id is required" );
- final Id applicationId = collectionScope.getApplication();
+ final Id applicationId = applicationScope.getApplication();
final Id ownerId = collectionScope.getOwner();
final String collectionName = collectionScope.getName();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSetImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSetImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSetImpl.java
index 8dd9528..cc9d5bf 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSetImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSetImpl.java
@@ -23,26 +23,27 @@ import java.util.Iterator;
import java.util.Map;
import java.util.Set;
+import org.apache.usergrid.persistence.collection.CollectionScope;
import org.apache.usergrid.persistence.collection.serialization.UniqueValue;
import org.apache.usergrid.persistence.collection.serialization.UniqueValueSet;
public class UniqueValueSetImpl implements UniqueValueSet {
- private final Map<String, UniqueValue> values;
+ private final Map<MapKey, UniqueValue> values;
public UniqueValueSetImpl(final int expectedMaxSize) {
values = new HashMap<>(expectedMaxSize);
}
- public void addValue(UniqueValue value){
- values.put( value.getField().getName(), value );
+ public void addValue(final CollectionScope scope, final UniqueValue value){
+ values.put( new MapKey( scope, value.getField().getName()), value );
}
@Override
- public UniqueValue getValue( final String fieldName ) {
- return values.get( fieldName );
+ public UniqueValue getValue(final CollectionScope scope, final String fieldName ) {
+ return values.get( new MapKey(scope, fieldName ));
}
@@ -58,9 +59,9 @@ public class UniqueValueSetImpl implements UniqueValueSet {
private static final class
UniqueValueIterator implements Iterator<UniqueValue>{
- private final Iterator<Map.Entry<String, UniqueValue>> sourceIterator;
+ private final Iterator<Map.Entry<MapKey, UniqueValue>> sourceIterator;
- public UniqueValueIterator( final Set<Map.Entry<String, UniqueValue>> entries ) {
+ public UniqueValueIterator( final Set<Map.Entry<MapKey, UniqueValue>> entries ) {
this.sourceIterator = entries.iterator();
}
@@ -82,4 +83,20 @@ public class UniqueValueSetImpl implements UniqueValueSet {
throw new UnsupportedOperationException( "Remove is unsupported" );
}
}
+
+
+ /**
+ * The key to look up unique fields by scope and field name
+ */
+ private static final class MapKey{
+ private final CollectionScope scope;
+ private final String fieldName;
+
+
+ private MapKey( final CollectionScope scope, final String fieldName ) {
+ this.scope = scope;
+ this.fieldName = fieldName;
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/EntityIdScope.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/EntityIdScope.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/EntityIdScope.java
index 0e70e75..a67f49c 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/EntityIdScope.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/EntityIdScope.java
@@ -29,10 +29,12 @@ import org.apache.usergrid.persistence.model.entity.Id;
* Tuple containing collectionscope and entityid
*/
public class EntityIdScope{
- private final Id id;
+ private final ApplicationScope applicationScope;
private final CollectionScope collectionScope;
+ private final Id id;
- public EntityIdScope(CollectionScope collectionScope, Id id){
+ public EntityIdScope( final ApplicationScope applicationScope, CollectionScope collectionScope, Id id ){
+ this.applicationScope = applicationScope;
this.id = id;
this.collectionScope = collectionScope;
}
@@ -46,4 +48,9 @@ public class EntityIdScope{
public CollectionScope getCollectionScope() {
return collectionScope;
}
+
+
+ public ApplicationScope getApplicationScope() {
+ return applicationScope;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java
index 6982857..2a96a9d 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java
@@ -38,6 +38,7 @@ import org.apache.usergrid.persistence.collection.serialization.UniqueValue;
import org.apache.usergrid.persistence.collection.serialization.UniqueValueSerializationStrategy;
import org.apache.usergrid.persistence.collection.serialization.impl.MvccEntitySerializationStrategyV3Impl;
import org.apache.usergrid.persistence.collection.serialization.impl.UniqueValueImpl;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.util.EntityUtils;
import org.apache.usergrid.persistence.core.migration.data.DataMigrationException;
import org.apache.usergrid.persistence.core.migration.data.DataMigration;
@@ -58,8 +59,6 @@ import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import rx.Observable;
import rx.Subscriber;
-import rx.functions.Action1;
-import rx.functions.Func1;
import rx.schedulers.Schedulers;
@@ -126,6 +125,7 @@ public class MvccEntityDataMigrationImpl implements DataMigration<EntityIdScope>
final Observable<List<EntityToSaveMessage>> migrated =
migrationDataProvider.getData().subscribeOn( Schedulers.io() ).flatMap( entityToSaveList -> Observable.just( entityToSaveList ).flatMap( entityIdScope -> {
+ final ApplicationScope applicationScope = entityIdScope.getApplicationScope();
//load the entity
final CollectionScope currentScope = entityIdScope.getCollectionScope();
@@ -138,7 +138,7 @@ public class MvccEntityDataMigrationImpl implements DataMigration<EntityIdScope>
//won't support anything beyond V2
final Iterator<MvccEntity> allVersions =
- migration.from.loadAscendingHistory( currentScope, entityIdScope.getId(), startTime, 100 );
+ migration.from.loadAscendingHistory( applicationScope, currentScope, entityIdScope.getId(), startTime, 100 );
//emit all the entity versions
return Observable.create( new Observable.OnSubscribe<EntityToSaveMessage>() {
@@ -148,7 +148,7 @@ public class MvccEntityDataMigrationImpl implements DataMigration<EntityIdScope>
while ( allVersions.hasNext() ) {
final EntityToSaveMessage message =
- new EntityToSaveMessage( currentScope, allVersions.next() );
+ new EntityToSaveMessage( applicationScope, currentScope, allVersions.next() );
subscriber.onNext( message );
}
@@ -163,7 +163,7 @@ public class MvccEntityDataMigrationImpl implements DataMigration<EntityIdScope>
List<EntityVersionCleanupTask> entityVersionCleanupTasks = new ArrayList( entities.size() );
for ( EntityToSaveMessage message : entities ) {
- final MutationBatch entityRewrite = migration.to.write( message.scope, message.entity );
+ final MutationBatch entityRewrite = migration.to.write(message.applicationScope, message.scope, message.entity );
//add to
// the
@@ -195,7 +195,7 @@ public class MvccEntityDataMigrationImpl implements DataMigration<EntityIdScope>
UniqueValue written = new UniqueValueImpl( field, entityId, version );
- MutationBatch mb = uniqueValueSerializationStrategy.write( message.scope, written );
+ MutationBatch mb = uniqueValueSerializationStrategy.write(message.applicationScope, message.scope, written );
// merge into our
@@ -247,11 +247,14 @@ public class MvccEntityDataMigrationImpl implements DataMigration<EntityIdScope>
private static final class EntityToSaveMessage {
+ private final ApplicationScope applicationScope;
private final CollectionScope scope;
private final MvccEntity entity;
- private EntityToSaveMessage( final CollectionScope scope, final MvccEntity entity ) {
+ private EntityToSaveMessage( final ApplicationScope applicationScope, final CollectionScope scope, final
+ MvccEntity entity ) {
+ this.applicationScope = applicationScope;
this.scope = scope;
this.entity = entity;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/adf19ed6/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTaskTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTaskTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTaskTest.java
index 787ab30..f18c984 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTaskTest.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTaskTest.java
@@ -19,7 +19,6 @@
package org.apache.usergrid.persistence.collection.impl;
-import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -29,16 +28,12 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Semaphore;
import org.junit.AfterClass;
-import org.junit.Ignore;
import org.junit.Test;
import org.apache.usergrid.persistence.collection.CollectionScope;
-import org.apache.usergrid.persistence.collection.MvccEntity;
import org.apache.usergrid.persistence.collection.MvccLogEntry;
import org.apache.usergrid.persistence.collection.event.EntityVersionDeleted;
import org.apache.usergrid.persistence.collection.mvcc.MvccLogEntrySerializationStrategy;
-import org.apache.usergrid.persistence.collection.mvcc.entity.impl.MvccEntityImpl;
-import org.apache.usergrid.persistence.collection.serialization.MvccEntitySerializationStrategy;
import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
import org.apache.usergrid.persistence.collection.serialization.UniqueValue;
import org.apache.usergrid.persistence.collection.serialization.UniqueValueSerializationStrategy;
@@ -47,12 +42,9 @@ import org.apache.usergrid.persistence.collection.util.UniqueValueEntryMock;
import org.apache.usergrid.persistence.collection.util.VersionGenerator;
import org.apache.usergrid.persistence.core.task.NamedTaskExecutorImpl;
import org.apache.usergrid.persistence.core.task.TaskExecutor;
-import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.entity.SimpleId;
-import org.apache.usergrid.persistence.model.util.UUIDGenerator;
-import com.google.common.base.Optional;
import com.google.common.util.concurrent.ListenableFuture;
import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.MutationBatch;
@@ -63,7 +55,6 @@ import static org.mockito.Matchers.any;
import static org.mockito.Matchers.same;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -127,7 +118,8 @@ public class EntityVersionCleanupTaskTest {
EntityVersionCleanupTask cleanupTask =
- new EntityVersionCleanupTask( serializationFig, less, uvss, keyspace, listeners, appScope, entityId,
+ new EntityVersionCleanupTask( serializationFig, less, uvss, keyspace, listeners, applicationScope,
+ appScope, entityId,
version, false );
final MutationBatch newBatch = mock( MutationBatch.class );
@@ -209,7 +201,8 @@ public class EntityVersionCleanupTaskTest {
EntityVersionCleanupTask cleanupTask =
- new EntityVersionCleanupTask( serializationFig, less, uvss, keyspace, listeners, appScope, entityId, version, false );
+ new EntityVersionCleanupTask( serializationFig, less, uvss, keyspace, listeners, applicationScope,
+ appScope, entityId, version, false );
final MutationBatch newBatch = mock( MutationBatch.class );
@@ -288,7 +281,8 @@ public class EntityVersionCleanupTaskTest {
EntityVersionCleanupTask cleanupTask =
- new EntityVersionCleanupTask( serializationFig, less, uvss, keyspace, listeners, appScope, entityId,
+ new EntityVersionCleanupTask( serializationFig, less, uvss, keyspace, listeners, applicationScope,
+ appScope, entityId,
version, false );
final MutationBatch newBatch = mock( MutationBatch.class );
@@ -388,7 +382,8 @@ public class EntityVersionCleanupTaskTest {
EntityVersionCleanupTask cleanupTask =
- new EntityVersionCleanupTask( serializationFig, less, uvss, keyspace, listeners, appScope, entityId,
+ new EntityVersionCleanupTask( serializationFig, less, uvss, keyspace, listeners, applicationScope,
+ appScope, entityId,
version, false );
final MutationBatch newBatch = mock( MutationBatch.class );
@@ -515,7 +510,8 @@ public class EntityVersionCleanupTaskTest {
EntityVersionCleanupTask cleanupTask =
- new EntityVersionCleanupTask( serializationFig, less, uvss, keyspace, listeners, appScope, entityId,
+ new EntityVersionCleanupTask( serializationFig, less, uvss, keyspace, listeners, applicationScope,
+ appScope, entityId,
version, false);
final MutationBatch newBatch = mock( MutationBatch.class );
@@ -643,7 +639,8 @@ public class EntityVersionCleanupTaskTest {
EntityVersionCleanupTask cleanupTask =
- new EntityVersionCleanupTask( serializationFig, less, uvss, keyspace, listeners, appScope, entityId,
+ new EntityVersionCleanupTask( serializationFig, less, uvss, keyspace, listeners, applicationScope,
+ appScope, entityId,
version, false );
final MutationBatch newBatch = mock( MutationBatch.class );