You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sn...@apache.org on 2014/03/18 02:20:53 UTC
[1/4] git commit: Fixed delete process. Need to refactor the creation
of observables from iterators into utilities.
Repository: incubator-usergrid
Updated Branches:
refs/heads/asyncqueue 5c014edbf -> 79fa0cb02
refs/pull/74/head [created] b6304a516
refs/pull/74/merge [created] f9f8b6fea
Fixed delete process. Need to refactor the creation of observables from iterators into utilities.
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/79fa0cb0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/79fa0cb0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/79fa0cb0
Branch: refs/heads/asyncqueue
Commit: 79fa0cb02692e49ad89aedca602e3de3727d5ac7
Parents: 5c014ed
Author: Todd Nine <tn...@apigee.com>
Authored: Mon Mar 17 17:30:31 2014 -0700
Committer: Todd Nine <tn...@apigee.com>
Committed: Mon Mar 17 17:30:31 2014 -0700
----------------------------------------------------------------------
.../src/test/resources/log4j.properties | 1 +
.../graph/consistency/MessageListener.java | 2 +-
.../graph/impl/NodeDeleteListener.java | 242 +++++++------------
.../graph/impl/stage/AbstractEdgeRepair.java | 7 +
.../graph/impl/stage/EdgeMetaRepairImpl.java | 75 +++---
.../graph/impl/NodeDeleteListenerTest.java | 12 +-
.../graph/impl/stage/EdgeMetaRepairTest.java | 14 +-
stack/corepersistence/pom.xml | 1 +
8 files changed, 146 insertions(+), 208 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/79fa0cb0/stack/corepersistence/collection/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/resources/log4j.properties b/stack/corepersistence/collection/src/test/resources/log4j.properties
index f6f50ef..f6a9c18 100644
--- a/stack/corepersistence/collection/src/test/resources/log4j.properties
+++ b/stack/corepersistence/collection/src/test/resources/log4j.properties
@@ -10,5 +10,6 @@ log4j.logger.org.safehaus.guicyfig=ERROR
log4j.logger.org.safehaus.chop.api.store.amazon=DEBUG
log4j.logger.org.apache.http=ERROR
log4j.logger.com.amazonaws.request=ERROR
+log4j.logger.cassandra.db=ERROR
log4j.logger.org.apache.usergrid=DEBUG
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/79fa0cb0/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/MessageListener.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/MessageListener.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/MessageListener.java
index 466e4ed..392a19f 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/MessageListener.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/MessageListener.java
@@ -19,6 +19,6 @@ public interface MessageListener<T, R> {
* @param event The input event
* @return The observable that performs the operations
*/
- Observable<T> receive(final T event);
+ Observable<R> receive(final T event);
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/79fa0cb0/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListener.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListener.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListener.java
index 58862f4..c860fcb 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListener.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListener.java
@@ -2,18 +2,21 @@ package org.apache.usergrid.persistence.graph.impl;
import java.util.Iterator;
-import java.util.List;
import java.util.UUID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.usergrid.persistence.collection.OrganizationScope;
-import org.apache.usergrid.persistence.graph.EdgeManagerFactory;
-import org.apache.usergrid.persistence.graph.GraphFig;
+import org.apache.usergrid.persistence.graph.Edge;
import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.SearchByEdgeType;
import org.apache.usergrid.persistence.graph.SearchEdgeType;
import org.apache.usergrid.persistence.graph.consistency.AsyncProcessor;
import org.apache.usergrid.persistence.graph.consistency.MessageListener;
import org.apache.usergrid.persistence.graph.guice.NodeDelete;
+import org.apache.usergrid.persistence.graph.impl.stage.EdgeDeleteRepair;
+import org.apache.usergrid.persistence.graph.impl.stage.EdgeMetaRepair;
import org.apache.usergrid.persistence.graph.serialization.EdgeMetadataSerialization;
import org.apache.usergrid.persistence.graph.serialization.EdgeSerialization;
import org.apache.usergrid.persistence.graph.serialization.NodeSerialization;
@@ -22,58 +25,61 @@ import org.apache.usergrid.persistence.model.entity.Id;
import com.google.common.base.Optional;
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.Scheduler;
import rx.functions.Action0;
-import rx.functions.Action1;
import rx.functions.Func1;
+import rx.observables.MathObservable;
/**
* Construct the asynchronous node delete from the q
*/
-public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, EdgeEvent<Id>> {
+public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, Integer> {
+
+ private static final Logger LOG = LoggerFactory.getLogger( NodeDeleteListener.class );
private final NodeSerialization nodeSerialization;
private final EdgeSerialization edgeSerialization;
private final EdgeMetadataSerialization edgeMetadataSerialization;
- private final GraphFig graphFig;
- private final Keyspace keyspace;
+ private final EdgeDeleteRepair edgeDeleteRepair;
+ private final EdgeMetaRepair edgeMetaRepair;
private final Scheduler scheduler;
- private final EdgeManagerFactory edgeManagerFactory;
/**
* Wire the serialization dependencies
*/
@Inject
- public NodeDeleteListener( final NodeSerialization nodeSerialization, final EdgeSerialization edgeSerialization,
- final EdgeMetadataSerialization edgeMetadataSerialization, final GraphFig graphFig,
- final Keyspace keyspace, final Scheduler scheduler,
- final EdgeManagerFactory edgeManagerFactory,
- @NodeDelete final AsyncProcessor nodeDelete ) {
+ public NodeDeleteListener( final NodeSerialization nodeSerialization, final EdgeSerialization edgeSerialization,final Scheduler scheduler,
+ @NodeDelete final AsyncProcessor nodeDelete,
+ final EdgeMetadataSerialization edgeMetadataSerialization, final EdgeDeleteRepair
+ edgeDeleteRepair,
+ final EdgeMetaRepair edgeMetaRepair ) {
this.nodeSerialization = nodeSerialization;
this.edgeSerialization = edgeSerialization;
- this.edgeMetadataSerialization = edgeMetadataSerialization;
- this.graphFig = graphFig;
- this.keyspace = keyspace;
this.scheduler = scheduler;
- this.edgeManagerFactory = edgeManagerFactory;
+ this.edgeMetadataSerialization = edgeMetadataSerialization;
+ this.edgeDeleteRepair = edgeDeleteRepair;
+ this.edgeMetaRepair = edgeMetaRepair;
nodeDelete.addListener( this );
}
+ /**
+ * Removes this node from the graph.
+ * @param edgeEvent The edge event that was fired.
+ * @return An observable that emits the total number of edges that have been removed with this node both as the target and source
+ */
@Override
- public Observable<EdgeEvent<Id>> receive( final EdgeEvent<Id> edgeEvent ) {
+ public Observable<Integer> receive( final EdgeEvent<Id> edgeEvent ) {
final Id node = edgeEvent.getData();
final OrganizationScope scope = edgeEvent.getOrganizationScope();
@@ -87,16 +93,20 @@ public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, EdgeEv
}
} ) //only continue if it's present. Otherwise stop
.takeWhile( new Func1<Optional<UUID>, Boolean>() {
+
@Override
public Boolean call( final Optional<UUID> uuidOptional ) {
+
+ LOG.debug( "Node with id {} has max version of {}", node, uuidOptional.orNull() );
+
return uuidOptional.isPresent();
}
} )
//delete source and targets in parallel and merge them into a single observable
- .flatMap( new Func1<Optional<UUID>, Observable<List<MarkedEdge>>>() {
+ .flatMap( new Func1<Optional<UUID>, Observable<MarkedEdge>>() {
@Override
- public Observable<List<MarkedEdge>> call( final Optional<UUID> uuidOptional ) {
+ public Observable<MarkedEdge> call( final Optional<UUID> uuidOptional ) {
//get all edges pointing to the target node and buffer then into groups for deletion
Observable<MarkedEdge> targetEdges =
@@ -105,10 +115,17 @@ public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, EdgeEv
@Override
public Observable<MarkedEdge> call( final String edgeType ) {
return loadEdgesToTarget( scope,
- new SimpleSearchByEdgeType( node, edgeType, uuidOptional.get(),
+ new SimpleSearchByEdgeType( node, edgeType, version,
null ) );
}
- } );
+ } )
+ //filter "old" edges, since we'll be deleting them in 1 shot
+ .distinctUntilChanged( new Func1<Edge, Id>() {
+ @Override
+ public Id call( final Edge edge ) {
+ return edge.getSourceNode();
+ }
+ } );
//get all edges pointing to the source node and buffer them into groups for deletion
@@ -118,124 +135,66 @@ public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, EdgeEv
@Override
public Observable<MarkedEdge> call( final String edgeType ) {
return loadEdgesFromSource( scope,
- new SimpleSearchByEdgeType( node, edgeType, uuidOptional.get(),
+ new SimpleSearchByEdgeType( node, edgeType, version,
null ) );
}
- } );
+ } )
+ //filter "old" edges, since we'll be deleting them in 1 shot
+ .distinctUntilChanged( new Func1<Edge, Id>() {
+ @Override
+ public Id call( final Edge edge ) {
+ return edge.getTargetNode();
+ }
+ } );
//each time an edge is emitted, delete it via batch mutation since we'll already be buffered
- return Observable.merge( targetEdges, sourceEdges ).buffer( graphFig.getScanPageSize() )
- .doOnNext( new Action1<List<MarkedEdge>>() {
- @Override
- public void call( final List<MarkedEdge> markedEdges ) {
- MutationBatch batch = keyspace.prepareMutationBatch();
-
-
- for ( MarkedEdge marked : markedEdges ) {
- batch.mergeShallow(
- edgeSerialization.deleteEdge( scope, marked ) );
- }
-
- try {
- batch.execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( "Unable to execute mutation" );
- }
- }
- } );
+ return Observable.merge( targetEdges, sourceEdges );
}
- } ).last().flatMap( new Func1<List<MarkedEdge>, Observable<String>>() {
+ } ).flatMap( new Func1<MarkedEdge, Observable<MarkedEdge>>() {
@Override
- public Observable<String> call( final List<MarkedEdge> markedEdges ) {
-
- //delete all meta for edges to this node
- Observable<String> targets =
- getEdgesTypesToTarget( scope, new SimpleSearchEdgeType( node, null ) )
- .doOnNext( new Action1<String>() {
- @Override
- public void call( final String type ) {
+ public Observable<MarkedEdge> call( final MarkedEdge edge ) {
- final MutationBatch batch = keyspace.prepareMutationBatch();
+ LOG.debug( "Deleting edge {}", edge );
+ return edgeDeleteRepair.repair( scope, edge );
- final Iterator<String> types = edgeMetadataSerialization
- .getIdTypesToTarget( scope,
- new SimpleSearchIdType( node, type, null ) );
-
- while ( types.hasNext() ) {
- batch.mergeShallow( edgeMetadataSerialization
- .removeIdTypeToTarget( scope, node, type, types.next(),
- version ) );
- }
-
- batch.mergeShallow( edgeMetadataSerialization
- .removeEdgeTypeToTarget( scope, node, type, version ) );
-
- try {
- batch.execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( "Unable to execute cassandra call" );
- }
- }
- } );
-
- //delete all meta for edges from this node
- Observable<String> sources =
- getEdgesTypesFromSource( scope, new SimpleSearchEdgeType( node, null ) )
- .doOnNext( new Action1<String>() {
- @Override
- public void call( final String type ) {
- final MutationBatch batch = keyspace.prepareMutationBatch();
+ }
+ } ).flatMap( new Func1<MarkedEdge, Observable<MarkedEdge>>() {
+ @Override
+ public Observable<MarkedEdge> call( final MarkedEdge edge ) {
- final Iterator<String> types = edgeMetadataSerialization
- .getIdTypesFromSource( scope,
- new SimpleSearchIdType( node, type, null ) );
- while ( types.hasNext() ) {
- batch.mergeShallow( edgeMetadataSerialization
- .removeIdTypeFromSource( scope, node, type, types.next(),
- version ) );
- }
+ //delete both the source and target meta data in parallel.
+ Observable<Integer> sourceMetaRepaired =
+ edgeMetaRepair.repairSources( scope, edge.getSourceNode(), edge.getType(), version );
- batch.mergeShallow( edgeMetadataSerialization
- .removeEdgeTypeFromSource( scope, node, type, version ) );
+ Observable<Integer> targetMetaRepaired = edgeMetaRepair.repairTargets( scope,
+ edge.getTargetNode(), edge.getType(), version );
- try {
- batch.execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( "Unable to execute cassandra call" );
- }
- }
- } );
+ //sum up the number of subtypes we retain
+ return Observable.merge(sourceMetaRepaired, targetMetaRepaired ).last().map( new Func1
+ <Integer, MarkedEdge>() {
+ @Override
+ public MarkedEdge call( final Integer integer ) {
+ LOG.debug( "Retained {} subtypes for edge {}", integer, edge );
- //no op, just zip them up so we can do our action on our last
- return Observable.merge( targets, sources )
- //when we're done emitting, merge them
- .doOnCompleted( new Action0() {
- @Override
- public void call() {
- try {
- nodeSerialization.delete( scope, node, edgeEvent.getVersion() ).execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( "Unable to execute mutation" );
- }
- }
- } );
+ return edge;
+ }
+ } );
}
- } )
- //when we're completed, we need to delete our id from types, then remove our mark
-
- //return our event to the caller
- .map( new Func1<String, EdgeEvent<Id>>() {
+ } ).count()
+ //if nothing is ever emitted, emit 0 so that we know no operations took place. Finally remove the target node in the mark
+ .defaultIfEmpty( 0 ).doOnCompleted( new Action0() {
@Override
- public EdgeEvent<Id> call( final String mark ) {
- return edgeEvent;
+ public void call() {
+ try {
+ nodeSerialization.delete( scope, node, version ).execute();
+ }
+ catch ( ConnectionException e ) {
+ throw new RuntimeException("Unable to delete marked graph node " + node, e);
+ }
}
} );
}
@@ -297,39 +256,4 @@ public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, EdgeEv
}
- private class Delete implements Action1<List<MarkedEdge>> {
-
- private final OrganizationScope scope;
-
-
- private Delete( final OrganizationScope scope ) {
- this.scope = scope;
- }
-
-
- @Override
- public void call( final List<MarkedEdge> markedEdges ) {
-
- //do nothing
- if ( markedEdges.size() == 0 ) {
- return;
- }
-
- Iterator<MarkedEdge> marked = markedEdges.iterator();
-
- MutationBatch batch = edgeSerialization.deleteEdge( scope, marked.next() );
-
- while ( marked.hasNext() ) {
- final MutationBatch newDelete = edgeSerialization.deleteEdge( scope, marked.next() );
- batch.mergeShallow( newDelete );
- }
-
- try {
- batch.execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( "Unable to execute batch mutation", e );
- }
- }
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/79fa0cb0/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/AbstractEdgeRepair.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/AbstractEdgeRepair.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/AbstractEdgeRepair.java
index 838de69..4b06760 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/AbstractEdgeRepair.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/AbstractEdgeRepair.java
@@ -24,6 +24,9 @@ import java.util.Iterator;
import java.util.List;
import java.util.UUID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.usergrid.persistence.collection.OrganizationScope;
import org.apache.usergrid.persistence.graph.Edge;
import org.apache.usergrid.persistence.graph.GraphFig;
@@ -51,6 +54,9 @@ import rx.functions.Func1;
@Singleton
public abstract class AbstractEdgeRepair {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractEdgeRepair.class);
+
protected final EdgeSerialization edgeSerialization;
protected final GraphFig graphFig;
protected final Keyspace keyspace;
@@ -87,6 +93,7 @@ public abstract class AbstractEdgeRepair {
final MutationBatch batch = keyspace.prepareMutationBatch();
for ( MarkedEdge edge : markedEdges ) {
+ LOG.debug( "Deleting edge {}" , edge );
final MutationBatch delete = edgeSerialization.deleteEdge( scope, edge );
batch.mergeShallow( delete );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/79fa0cb0/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java
index 51777ab..5c497a5 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java
@@ -25,6 +25,9 @@ import java.util.Iterator;
import java.util.List;
import java.util.UUID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.usergrid.persistence.collection.OrganizationScope;
import org.apache.usergrid.persistence.collection.mvcc.entity.ValidationUtils;
import org.apache.usergrid.persistence.graph.GraphFig;
@@ -45,6 +48,7 @@ import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import rx.Observable;
import rx.Scheduler;
+import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.observables.MathObservable;
@@ -57,6 +61,7 @@ import rx.observables.MathObservable;
public class EdgeMetaRepairImpl implements EdgeMetaRepair {
+ private static final Logger LOG = LoggerFactory.getLogger( EdgeMetaRepairImpl.class );
private final EdgeMetadataSerialization edgeMetadataSerialization;
private final EdgeSerialization edgeSerialization;
private final Keyspace keyspace;
@@ -115,39 +120,39 @@ public class EdgeMetaRepairImpl implements EdgeMetaRepair {
final List<Observable<Integer>> checks = new ArrayList<Observable<Integer>>( types.size() );
//for each id type, check if the exist in parallel to increase processing speed
- for ( final String sourceIdType : types ) {
+ for ( final String subType : types ) {
+
+ LOG.debug( "Checking for edges with nodeId {}, type {}, and subtype {}", node, edgeType, subType );
Observable<Integer> search =
- serialization.loadEdges( scope, node, edgeType, sourceIdType, version )
- .distinctUntilChanged( new Func1<MarkedEdge, Id>() {
+ serialization.loadEdges( scope, node, edgeType, subType, version ).take( 1 ).count()
+ .doOnNext( new Action1<Integer>() {
- //get distinct by source node
@Override
- public Id call( final MarkedEdge markedEdge ) {
- return markedEdge.getSourceNode();
+ public void call( final Integer count ) {
+ /**
+ * we only want to delete if no edges are in this class. If
+ * there
+ * are
+ * still edges
+ * we must retain the information in order to keep our index
+ * structure
+ * correct for edge
+ * iteration
+ **/
+ if ( count != 0 ) {
+ LOG.debug( "Found edge with nodeId {}, type {}, and subtype {}. Not removing subtype. ", node, edgeType, subType );
+ return;
+ }
+
+
+
+ LOG.debug( "No edges with nodeId {}, type {}, and subtype {}. Removing", node, edgeType, subType );
+ batch.mergeShallow( serialization
+ .removeEdgeSubType( scope, node, edgeType, subType,
+ version ) );
}
- } ).take( 1 ).count().doOnNext( new Action1<Integer>() {
-
- @Override
- public void call( final Integer count ) {
- /**
- * we only want to delete if no edges are in this class. If there
- * are
- * still edges
- * we must retain the information in order to keep our index
- * structure
- * correct for edge
- * iteration
- **/
- if ( count != 0 ) {
- return;
- }
-
- batch.mergeShallow( serialization
- .removeEdgeSubType( scope, node, edgeType, sourceIdType,
- version ) );
- }
- } );
+ } );
checks.add( search );
}
@@ -162,9 +167,8 @@ public class EdgeMetaRepairImpl implements EdgeMetaRepair {
@Override
public void call( final Integer count ) {
- if ( batch.isEmpty() ) {
- return;
- }
+
+ LOG.debug( "Executing batch for subtype deletion with type {}. Mutation has {} rows to mutate ", edgeType, batch.getRowCount() );
try {
batch.execute();
@@ -182,26 +186,27 @@ public class EdgeMetaRepairImpl implements EdgeMetaRepair {
//sum up everything emitted by sub types. If there's no edges existing for all sub types,
// then we can safely remove them
- return MathObservable.sumInteger( deleteCounts ).last().doOnNext( new Action1<Integer>() {
+ return MathObservable.sumInteger( deleteCounts ).lastOrDefault( 0 ).doOnNext( new Action1<Integer>() {
@Override
- public void call( final Integer subTypes ) {
+ public void call( final Integer subTypeUsedCount ) {
/**
* We can only execute deleting this type if no sub types were deleted
*/
- if ( subTypes != 0 ) {
+ if ( subTypeUsedCount != 0 ) {
+ LOG.debug( "Type {} has {} subtypes in use as of version {}. Not deleting type.", edgeType, subTypeUsedCount, version );
return;
}
try {
+ LOG.debug( "Type {} has no subtypes in use as of version {}. Deleting type.", edgeType, version );
serialization.removeEdgeType( scope, node, edgeType, version ).execute();
}
catch ( ConnectionException e ) {
throw new RuntimeException( "Unable to execute mutation" );
}
-
}
} );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/79fa0cb0/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListenerTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListenerTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListenerTest.java
index 82aecff..dcddb27 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListenerTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListenerTest.java
@@ -114,16 +114,16 @@ public class NodeDeleteListenerTest {
EdgeEvent<Id> deleteEvent = new EdgeEvent<Id>( scope, version, sourceNode );
- EdgeEvent<Id> event = deleteListener.receive( deleteEvent ).toBlockingObservable().lastOrDefault( null );
+ int count = deleteListener.receive( deleteEvent ).toBlockingObservable().last();
- assertNull( "Mark was not set, no delete should be executed", event );
+ assertEquals( "Mark was not set, no delete should be executed", 0, count );
deleteEvent = new EdgeEvent<Id>( scope, version, targetNode );
- event = deleteListener.receive( deleteEvent ).toBlockingObservable().lastOrDefault( null );
+ count = deleteListener.receive( deleteEvent ).toBlockingObservable().last();
- assertNull( "Mark was not set, no delete should be executed", event );
+ assertEquals( "Mark was not set, no delete should be executed", 0, count );
}
@@ -157,9 +157,9 @@ public class NodeDeleteListenerTest {
EdgeEvent<Id> deleteEvent = new EdgeEvent<Id>( scope, deleteVersion, sourceNode );
- EdgeEvent<Id> event = deleteListener.receive( deleteEvent ).toBlockingObservable().last();
+ int count = deleteListener.receive( deleteEvent ).toBlockingObservable().last();
- assertEquals( deleteEvent, event );
+ assertEquals( 1, count);
//now verify we can't get any of the info back
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/79fa0cb0/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairTest.java
index 7bd1a68..e8835c3 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairTest.java
@@ -264,7 +264,7 @@ public class EdgeMetaRepairTest {
value = edgeMetaRepair.repairTargets( scope, targetId, edgeType, cleanupVersion )
- .toBlockingObservable().single();
+ .toBlockingObservable().last();
assertEquals( "Subtypes removed", 0, value );
@@ -309,13 +309,13 @@ public class EdgeMetaRepairTest {
//now verify they're gone
Iterator<String> edgeTypes = edgeMetadataSerialization
- .getEdgeTypesToTarget( scope, new SimpleSearchEdgeType( edge.getSourceNode(), null ) );
+ .getEdgeTypesFromSource( scope, new SimpleSearchEdgeType( edge.getSourceNode(), null ) );
assertFalse( "No edge types exist", edgeTypes.hasNext() );
Iterator<String> sourceTypes = edgeMetadataSerialization
- .getIdTypesToTarget( scope, new SimpleSearchIdType( edge.getSourceNode(), edge.getType(), null ) );
+ .getIdTypesFromSource( scope, new SimpleSearchIdType( edge.getSourceNode(), edge.getType(), null ) );
assertFalse( "No edge types exist", sourceTypes.hasNext() );
}
@@ -381,13 +381,13 @@ public class EdgeMetaRepairTest {
//now verify they're gone
Iterator<String> edgeTypes = edgeMetadataSerialization
- .getEdgeTypesToTarget( scope, new SimpleSearchEdgeType( edge1.getSourceNode(), null ) );
+ .getEdgeTypesFromSource( scope, new SimpleSearchEdgeType( edge1.getSourceNode(), null ) );
assertFalse( "No edge types exist", edgeTypes.hasNext() );
Iterator<String> sourceTypes = edgeMetadataSerialization
- .getIdTypesToTarget( scope, new SimpleSearchIdType( edge1.getSourceNode(), edge1.getType(), null ) );
+ .getIdTypesFromSource( scope, new SimpleSearchIdType( edge1.getSourceNode(), edge1.getType(), null ) );
assertFalse( "No edge types exist", sourceTypes.hasNext() );
}
@@ -438,13 +438,13 @@ public class EdgeMetaRepairTest {
//now verify they're gone
Iterator<String> edgeTypes = edgeMetadataSerialization
- .getEdgeTypesToTarget( scope, new SimpleSearchEdgeType( sourceId, null ) );
+ .getEdgeTypesFromSource( scope, new SimpleSearchEdgeType( sourceId, null ) );
assertFalse( "No edge types exist", edgeTypes.hasNext() );
Iterator<String> sourceTypes = edgeMetadataSerialization
- .getIdTypesToTarget( scope, new SimpleSearchIdType( sourceId, edgeType, null ) );
+ .getIdTypesFromSource( scope, new SimpleSearchIdType( sourceId, edgeType, null ) );
assertFalse( "No edge types exist", sourceTypes.hasNext() );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/79fa0cb0/stack/corepersistence/pom.xml
----------------------------------------------------------------------
diff --git a/stack/corepersistence/pom.xml b/stack/corepersistence/pom.xml
index f8b071d..2643d00 100644
--- a/stack/corepersistence/pom.xml
+++ b/stack/corepersistence/pom.xml
@@ -38,6 +38,7 @@
<log4j.version>1.2.17</log4j.version>
<slf4j.version>1.7.2</slf4j.version>
<surefire.version>2.16</surefire.version>
+ <rx.version>0.17.0</rx.version>
</properties>
[3/4] git commit: Fixing tests so that they properly check obj
serialization.
Posted by sn...@apache.org.
Fixing tests so that they properly check obj serialization.
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/b6304a51
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/b6304a51
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/b6304a51
Branch: refs/pull/74/merge
Commit: b6304a516b7bb9e93eacf0e4475f5f055364f312
Parents: 6979ae6
Author: grey <gr...@apigee.com>
Authored: Mon Mar 17 18:14:43 2014 -0700
Committer: grey <gr...@apigee.com>
Committed: Mon Mar 17 18:14:43 2014 -0700
----------------------------------------------------------------------
...MvccEntitySerializationStrategyImplTest.java | 57 ++++++++++----------
1 file changed, 28 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/b6304a51/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
index 5375ead..f8a51b3 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
@@ -52,7 +52,6 @@ import static junit.framework.TestCase.assertEquals;
import static junit.framework.TestCase.assertFalse;
import static junit.framework.TestCase.assertNotNull;
import static junit.framework.TestCase.assertNull;
-import static junit.framework.TestCase.assertSame;
import static junit.framework.TestCase.assertTrue;
import static org.mockito.Mockito.mock;
@@ -160,36 +159,36 @@ public class MvccEntitySerializationStrategyImplTest {
assertEquals( "Mvcc entities are the same", saved, returned );
- assertEquals( id, entity.getId() );
+ assertEquals( id, returned.getId() );
- Field<Boolean> boolFieldReturned = entity.getField( boolField.getName() );
+ Field<Boolean> boolFieldReturned = returned.getEntity().get().getField( boolField.getName() );
- assertSame( boolField, boolFieldReturned );
+ assertEquals( boolField, boolFieldReturned );
- Field<Double> doubleFieldReturned = entity.getField( doubleField.getName() );
+ Field<Double> doubleFieldReturned = returned.getEntity().get().getField( doubleField.getName() );
- assertSame( doubleField, doubleFieldReturned );
+ assertEquals( doubleField, doubleFieldReturned );
- Field<Integer> intFieldReturned = entity.getField( intField.getName() );
+ Field<Integer> intFieldReturned = returned.getEntity().get().getField( intField.getName() );
- assertSame( intField, intFieldReturned );
+ assertEquals( intField, intFieldReturned );
- Field<Long> longFieldReturned = entity.getField( longField.getName() );
+ Field<Long> longFieldReturned = returned.getEntity().get().getField( longField.getName() );
- assertSame( longField, longFieldReturned );
+ assertEquals( longField, longFieldReturned );
- Field<String> stringFieldReturned = entity.getField( stringField.getName() );
+ Field<String> stringFieldReturned = returned.getEntity().get().getField( stringField.getName() );
- assertSame( stringField, stringFieldReturned );
+ assertEquals( stringField, stringFieldReturned );
- Field<UUID> uuidFieldReturned = entity.getField( uuidField.getName() );
+ Field<UUID> uuidFieldReturned = returned.getEntity().get().getField( uuidField.getName() );
- assertSame( uuidField, uuidFieldReturned );
+ assertEquals( uuidField, uuidFieldReturned );
Set<Field> results = new HashSet<Field>();
- results.addAll( entity.getFields() );
+ results.addAll( returned.getEntity().get().getFields());
assertTrue( results.contains( boolField ) );
@@ -435,36 +434,36 @@ public class MvccEntitySerializationStrategyImplTest {
assertEquals( "Mvcc entities are the same", saved, returned );
- assertEquals( id, entity.getId() );
+ assertEquals( id, returned.getId() );
- Field<Boolean> boolFieldReturned = entity.getField( boolField.getName() );
+ Field<Boolean> boolFieldReturned = returned.getEntity().get().getField( boolField.getName() );
- assertSame( boolField, boolFieldReturned );
+ assertEquals( boolField, boolFieldReturned );
- Field<Double> doubleFieldReturned = entity.getField( doubleField.getName() );
+ Field<Double> doubleFieldReturned = returned.getEntity().get().getField( doubleField.getName() );
- assertSame( doubleField, doubleFieldReturned );
+ assertEquals( doubleField, doubleFieldReturned );
- Field<Integer> intFieldReturned = entity.getField( intField.getName() );
+ Field<Integer> intFieldReturned = returned.getEntity().get().getField( intField.getName() );
- assertSame( intField, intFieldReturned );
+ assertEquals( intField, intFieldReturned );
- Field<Long> longFieldReturned = entity.getField( longField.getName() );
+ Field<Long> longFieldReturned = returned.getEntity().get().getField( longField.getName() );
- assertSame( longField, longFieldReturned );
+ assertEquals( longField, longFieldReturned );
- Field<String> stringFieldReturned = entity.getField( stringField.getName() );
+ Field<String> stringFieldReturned = returned.getEntity().get().getField( stringField.getName() );
- assertSame( stringField, stringFieldReturned );
+ assertEquals( stringField, stringFieldReturned );
- Field<UUID> uuidFieldReturned = entity.getField( uuidField.getName() );
+ Field<UUID> uuidFieldReturned = returned.getEntity().get().getField( uuidField.getName() );
- assertSame( uuidField, uuidFieldReturned );
+ assertEquals( uuidField, uuidFieldReturned );
Set<Field> results = new HashSet<Field>();
- results.addAll( entity.getFields() );
+ results.addAll( returned.getEntity().get().getFields());
assertTrue( results.contains( boolField ) );
[4/4] git commit: Merge b6304a516b7bb9e93eacf0e4475f5f055364f312 into
6979ae6bfc0acbbc936abfd0c7a8bed657e75948
Posted by sn...@apache.org.
Merge b6304a516b7bb9e93eacf0e4475f5f055364f312 into 6979ae6bfc0acbbc936abfd0c7a8bed657e75948
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/f9f8b6fe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/f9f8b6fe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/f9f8b6fe
Branch: refs/pull/74/merge
Commit: f9f8b6fea8fca67ea248c2ba47d9c3cc8c9cf870
Parents: 6979ae6 b6304a5
Author: GERey <gr...@apigee.com>
Authored: Tue Mar 18 01:17:53 2014 +0000
Committer: GERey <gr...@apigee.com>
Committed: Tue Mar 18 01:17:53 2014 +0000
----------------------------------------------------------------------
...MvccEntitySerializationStrategyImplTest.java | 57 ++++++++++----------
1 file changed, 28 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
[2/4] git commit: Fixing tests so that they properly check obj
serialization.
Posted by sn...@apache.org.
Fixing tests so that they properly check obj serialization.
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/b6304a51
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/b6304a51
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/b6304a51
Branch: refs/pull/74/head
Commit: b6304a516b7bb9e93eacf0e4475f5f055364f312
Parents: 6979ae6
Author: grey <gr...@apigee.com>
Authored: Mon Mar 17 18:14:43 2014 -0700
Committer: grey <gr...@apigee.com>
Committed: Mon Mar 17 18:14:43 2014 -0700
----------------------------------------------------------------------
...MvccEntitySerializationStrategyImplTest.java | 57 ++++++++++----------
1 file changed, 28 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/b6304a51/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
index 5375ead..f8a51b3 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
@@ -52,7 +52,6 @@ import static junit.framework.TestCase.assertEquals;
import static junit.framework.TestCase.assertFalse;
import static junit.framework.TestCase.assertNotNull;
import static junit.framework.TestCase.assertNull;
-import static junit.framework.TestCase.assertSame;
import static junit.framework.TestCase.assertTrue;
import static org.mockito.Mockito.mock;
@@ -160,36 +159,36 @@ public class MvccEntitySerializationStrategyImplTest {
assertEquals( "Mvcc entities are the same", saved, returned );
- assertEquals( id, entity.getId() );
+ assertEquals( id, returned.getId() );
- Field<Boolean> boolFieldReturned = entity.getField( boolField.getName() );
+ Field<Boolean> boolFieldReturned = returned.getEntity().get().getField( boolField.getName() );
- assertSame( boolField, boolFieldReturned );
+ assertEquals( boolField, boolFieldReturned );
- Field<Double> doubleFieldReturned = entity.getField( doubleField.getName() );
+ Field<Double> doubleFieldReturned = returned.getEntity().get().getField( doubleField.getName() );
- assertSame( doubleField, doubleFieldReturned );
+ assertEquals( doubleField, doubleFieldReturned );
- Field<Integer> intFieldReturned = entity.getField( intField.getName() );
+ Field<Integer> intFieldReturned = returned.getEntity().get().getField( intField.getName() );
- assertSame( intField, intFieldReturned );
+ assertEquals( intField, intFieldReturned );
- Field<Long> longFieldReturned = entity.getField( longField.getName() );
+ Field<Long> longFieldReturned = returned.getEntity().get().getField( longField.getName() );
- assertSame( longField, longFieldReturned );
+ assertEquals( longField, longFieldReturned );
- Field<String> stringFieldReturned = entity.getField( stringField.getName() );
+ Field<String> stringFieldReturned = returned.getEntity().get().getField( stringField.getName() );
- assertSame( stringField, stringFieldReturned );
+ assertEquals( stringField, stringFieldReturned );
- Field<UUID> uuidFieldReturned = entity.getField( uuidField.getName() );
+ Field<UUID> uuidFieldReturned = returned.getEntity().get().getField( uuidField.getName() );
- assertSame( uuidField, uuidFieldReturned );
+ assertEquals( uuidField, uuidFieldReturned );
Set<Field> results = new HashSet<Field>();
- results.addAll( entity.getFields() );
+ results.addAll( returned.getEntity().get().getFields());
assertTrue( results.contains( boolField ) );
@@ -435,36 +434,36 @@ public class MvccEntitySerializationStrategyImplTest {
assertEquals( "Mvcc entities are the same", saved, returned );
- assertEquals( id, entity.getId() );
+ assertEquals( id, returned.getId() );
- Field<Boolean> boolFieldReturned = entity.getField( boolField.getName() );
+ Field<Boolean> boolFieldReturned = returned.getEntity().get().getField( boolField.getName() );
- assertSame( boolField, boolFieldReturned );
+ assertEquals( boolField, boolFieldReturned );
- Field<Double> doubleFieldReturned = entity.getField( doubleField.getName() );
+ Field<Double> doubleFieldReturned = returned.getEntity().get().getField( doubleField.getName() );
- assertSame( doubleField, doubleFieldReturned );
+ assertEquals( doubleField, doubleFieldReturned );
- Field<Integer> intFieldReturned = entity.getField( intField.getName() );
+ Field<Integer> intFieldReturned = returned.getEntity().get().getField( intField.getName() );
- assertSame( intField, intFieldReturned );
+ assertEquals( intField, intFieldReturned );
- Field<Long> longFieldReturned = entity.getField( longField.getName() );
+ Field<Long> longFieldReturned = returned.getEntity().get().getField( longField.getName() );
- assertSame( longField, longFieldReturned );
+ assertEquals( longField, longFieldReturned );
- Field<String> stringFieldReturned = entity.getField( stringField.getName() );
+ Field<String> stringFieldReturned = returned.getEntity().get().getField( stringField.getName() );
- assertSame( stringField, stringFieldReturned );
+ assertEquals( stringField, stringFieldReturned );
- Field<UUID> uuidFieldReturned = entity.getField( uuidField.getName() );
+ Field<UUID> uuidFieldReturned = returned.getEntity().get().getField( uuidField.getName() );
- assertSame( uuidField, uuidFieldReturned );
+ assertEquals( uuidField, uuidFieldReturned );
Set<Field> results = new HashSet<Field>();
- results.addAll( entity.getFields() );
+ results.addAll( returned.getEntity().get().getFields());
assertTrue( results.contains( boolField ) );