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/02/18 01:22:15 UTC
[1/4] git commit: Updated filters for versions as well as de-dup to
skip versions
Repository: incubator-usergrid
Updated Branches:
refs/heads/two-dot-o 3e55e902f -> 4e5b3b25e
Updated filters for versions as well as de-dup to skip versions
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/909d1d84
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/909d1d84
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/909d1d84
Branch: refs/heads/two-dot-o
Commit: 909d1d8450792f7fdea08ab95c654cbf5902b7e3
Parents: d9e0bf7
Author: Todd Nine <tn...@apigee.com>
Authored: Mon Feb 17 15:34:14 2014 -0700
Committer: Todd Nine <tn...@apigee.com>
Committed: Mon Feb 17 15:34:14 2014 -0700
----------------------------------------------------------------------
.../persistence/graph/impl/EdgeManagerImpl.java | 251 ++++++++++---------
.../persistence/graph/EdgeManagerIT.java | 212 +++++++++++++---
2 files changed, 310 insertions(+), 153 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/909d1d84/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java
index 58d3e17..2fc61b0 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java
@@ -23,8 +23,6 @@ package org.apache.usergrid.persistence.graph.impl;
import java.util.Iterator;
import java.util.UUID;
-import org.apache.cassandra.thrift.Mutation;
-
import org.apache.usergrid.persistence.collection.OrganizationScope;
import org.apache.usergrid.persistence.collection.mvcc.entity.ValidationUtils;
import org.apache.usergrid.persistence.graph.Edge;
@@ -39,16 +37,15 @@ import org.apache.usergrid.persistence.graph.serialization.impl.parse.Observable
import org.apache.usergrid.persistence.graph.serialization.stage.GraphIoEvent;
import org.apache.usergrid.persistence.graph.serialization.stage.write.EdgeWriteStage;
import org.apache.usergrid.persistence.graph.serialization.util.EdgeUtils;
+import org.apache.usergrid.persistence.model.entity.Id;
import com.fasterxml.uuid.UUIDComparator;
import com.google.inject.Inject;
import com.google.inject.assistedinject.Assisted;
-import com.netflix.astyanax.MutationBatch;
import rx.Observable;
import rx.Scheduler;
import rx.util.functions.Func1;
-import rx.util.functions.Func2;
/**
@@ -95,94 +92,99 @@ public class EdgeManagerImpl implements EdgeManager {
@Override
public Observable<Edge> deleteEdge( final Edge edge ) {
EdgeUtils.validateEdge( edge );
-//
-//
-// Observable<Edge> observable = Observable.from( edge ).subscribeOn( scheduler );
-//
-//
-// final SimpleSearchByEdge search =
-// new SimpleSearchByEdge( edge.getSourceNode(), edge.getType(), edge.getTargetNode(), edge.getVersion(),
-// null );
-//
-//
-// final UUID maxVersion = edge.getVersion();
-//
-// //search on both our source and target types, we need to filter them if there aren't any left
-// observable.flatMap( new Func1<Edge, Observable<Edge>>() {
-//
-// @Override
-// public Observable<Edge> call( final Edge edge ) {
-//
-// //take the first 2 edges and then count them. If there are 2 elements, then
-// //we can't remove the meta data. If there are 2, we can
-//
-// Observable<Edge> sourceEdges = Observable.create( new ObservableIterator<Edge>() {
-// @Override
-// protected Iterator<Edge> getIterator() {
-// return edgeSerialization.getEdgeFromSource( scope, search );
-// }
-// } );
-//
-// Observable<MutationBatch>
-// sourceDeletes = filter(sourceEdges, maxVersion).map( new Func1<Edge, MutationBatch>() {
-// @Override
-// public MutationBatch call( final Edge edge ) {
-// return edgeSerialization.deleteEdge( scope, edge );
-// }
-// } );
-//
-//
-// Observable<Edge> targetEdges = Observable.create( new ObservableIterator<Edge>() {
-// @Override
-// protected Iterator<Edge> getIterator() {
-// return edgeSerialization.getEdgeToTarget( scope, search );
-// }
-// } );
-//
-// Observable<MutationBatch> targetDeletes = filter(targetEdges, maxVersion).map( new Func1<Edge, MutationBatch> (){
-//
-//
-// @Override
-// public MutationBatch call( final Edge edge ) {
-// return edgeSerialization.deleteEdge( scope, edge );
-// }
-// });
-//
-// return Observable.zip( sourceDeletes, targetDeletes, new Func2<MutationBatch, MutationBatch, Edge>() {
-//
-// private MutationBatch batch;
-//
-// @Override
-// public Edge call( final MutationBatch mutationBatch, final MutationBatch mutationBatch2 ) {
-// if(batch == null){
-// batch = mutationBatch;
-// }
-//
-// batch.mergeShallow( mutationBatch );
-// batch.mergeShallow( mutationBatch2 );
-//
-// return batch;
-// }
-// });
-//
-// return Observable.zip( sourceEdges, targetEdges, new Func2<Edge, Edge, Edge>() {
-// @Override
-// public Edge call( final Edge sourceEdge, final Edge targetEdges ) {
-//
-// edgeSerialization.deleteEdge( scope, sourceEdge );
-// edgeSerialization.deleteEdge( scope, targetEdges );
-//
-// return edge;
-// }
-// } );
-//
-//
-//
-// return null;
-//
-//
-// }
-// } );
+ //
+ //
+ // Observable<Edge> observable = Observable.from( edge ).subscribeOn( scheduler );
+ //
+ //
+ // final SimpleSearchByEdge search =
+ // new SimpleSearchByEdge( edge.getSourceNode(), edge.getType(), edge.getTargetNode(),
+ // edge.getVersion(),
+ // null );
+ //
+ //
+ // final UUID maxVersion = edge.getVersion();
+ //
+ // //search on both our source and target types, we need to filter them if there aren't any left
+ // observable.flatMap( new Func1<Edge, Observable<Edge>>() {
+ //
+ // @Override
+ // public Observable<Edge> call( final Edge edge ) {
+ //
+ // //take the first 2 edges and then count them. If there are 2 elements, then
+ // //we can't remove the meta data. If there are 2, we can
+ //
+ // Observable<Edge> sourceEdges = Observable.create( new ObservableIterator<Edge>() {
+ // @Override
+ // protected Iterator<Edge> getIterator() {
+ // return edgeSerialization.getEdgeFromSource( scope, search );
+ // }
+ // } );
+ //
+ // Observable<MutationBatch>
+ // sourceDeletes = filter(sourceEdges, maxVersion).map( new Func1<Edge,
+ // MutationBatch>() {
+ // @Override
+ // public MutationBatch call( final Edge edge ) {
+ // return edgeSerialization.deleteEdge( scope, edge );
+ // }
+ // } );
+ //
+ //
+ // Observable<Edge> targetEdges = Observable.create( new ObservableIterator<Edge>() {
+ // @Override
+ // protected Iterator<Edge> getIterator() {
+ // return edgeSerialization.getEdgeToTarget( scope, search );
+ // }
+ // } );
+ //
+ // Observable<MutationBatch> targetDeletes = filter(targetEdges,
+ // maxVersion).map( new Func1<Edge, MutationBatch> (){
+ //
+ //
+ // @Override
+ // public MutationBatch call( final Edge edge ) {
+ // return edgeSerialization.deleteEdge( scope, edge );
+ // }
+ // });
+ //
+ // return Observable.zip( sourceDeletes, targetDeletes, new Func2<MutationBatch,
+ // MutationBatch, Edge>() {
+ //
+ // private MutationBatch batch;
+ //
+ // @Override
+ // public Edge call( final MutationBatch mutationBatch, final MutationBatch mutationBatch2
+ // ) {
+ // if(batch == null){
+ // batch = mutationBatch;
+ // }
+ //
+ // batch.mergeShallow( mutationBatch );
+ // batch.mergeShallow( mutationBatch2 );
+ //
+ // return batch;
+ // }
+ // });
+ //
+ // return Observable.zip( sourceEdges, targetEdges, new Func2<Edge, Edge, Edge>() {
+ // @Override
+ // public Edge call( final Edge sourceEdge, final Edge targetEdges ) {
+ //
+ // edgeSerialization.deleteEdge( scope, sourceEdge );
+ // edgeSerialization.deleteEdge( scope, targetEdges );
+ //
+ // return edge;
+ // }
+ // } );
+ //
+ //
+ //
+ // return null;
+ //
+ //
+ // }
+ // } );
throw new UnsupportedOperationException( "Not yet implemented" );
@@ -191,53 +193,61 @@ public class EdgeManagerImpl implements EdgeManager {
@Override
public Observable<Edge> loadEdgesFromSource( final SearchByEdgeType search ) {
- Observable<Edge> iterator = Observable.create( new ObservableIterator<Edge>() {
+ return Observable.create( new ObservableIterator<Edge>() {
@Override
protected Iterator<Edge> getIterator() {
return edgeSerialization.getEdgesFromSource( scope, search );
}
- } );
-
- return filter( iterator, search.getMaxVersion() );
+ } ).filter( new MaxFilter( search.getMaxVersion() ) )
+ //we intentionally use distinct until changed. This way we won't store all the keys since this
+ //would hog far too much ram.
+ .distinctUntilChanged( new Func1<Edge, Id>() {
+ @Override
+ public Id call( final Edge edge ) {
+ return edge.getTargetNode();
+ }
+ } );
}
@Override
public Observable<Edge> loadEdgesToTarget( final SearchByEdgeType search ) {
- Observable<Edge> iterator = Observable.create( new ObservableIterator<Edge>() {
+ return Observable.create( new ObservableIterator<Edge>() {
@Override
protected Iterator<Edge> getIterator() {
return edgeSerialization.getEdgesToTarget( scope, search );
}
- } );
-
- return filter( iterator, search.getMaxVersion() );
+ } ).filter( new MaxFilter( search.getMaxVersion() ) )
+ //we intentionally use distinct until changed. This way we won't store all the keys since this
+ //would hog far too much ram.
+ .distinctUntilChanged( new Func1<Edge, Id>() {
+ @Override
+ public Id call( final Edge edge ) {
+ return edge.getSourceNode();
+ }
+ } );
}
@Override
public Observable<Edge> loadEdgesFromSourceByType( final SearchByIdType search ) {
- Observable<Edge> iterator = Observable.create( new ObservableIterator<Edge>() {
+ return Observable.create( new ObservableIterator<Edge>() {
@Override
protected Iterator<Edge> getIterator() {
return edgeSerialization.getEdgesFromSourceByTargetType( scope, search );
}
- } );
-
- return filter( iterator, search.getMaxVersion() );
+ } ).filter( new MaxFilter( search.getMaxVersion() ) ).takeFirst();
}
@Override
public Observable<Edge> loadEdgesToTargetByType( final SearchByIdType search ) {
- Observable<Edge> iterator = Observable.create( new ObservableIterator<Edge>() {
+ return Observable.create( new ObservableIterator<Edge>() {
@Override
protected Iterator<Edge> getIterator() {
return edgeSerialization.getEdgesToTargetBySourceType( scope, search );
}
- } );
-
- return filter( iterator, search.getMaxVersion() );
+ } ).filter( new MaxFilter( search.getMaxVersion() ) ).takeFirst();
}
@@ -288,15 +298,22 @@ public class EdgeManagerImpl implements EdgeManager {
/**
- * If not max version is specified, just return the original observable. If one is
+ * Filter the returned values based on the max uuid
*/
- private Observable<Edge> filter( final Observable<Edge> observable, final UUID maxVersion ) {
- return observable.filter( new Func1<Edge, Boolean>() {
- @Override
- public Boolean call( final Edge edge ) {
- //our edge version needs to be <= max Version
- return UUIDComparator.staticCompare( edge.getVersion(), maxVersion ) < 1;
- }
- } );
+ private static class MaxFilter implements Func1<Edge, Boolean> {
+
+ private final UUID maxVersion;
+
+
+ private MaxFilter( final UUID maxVersion ) {
+ this.maxVersion = maxVersion;
+ }
+
+
+ @Override
+ public Boolean call( final Edge edge ) {
+ //our edge version needs to be <= max Version
+ return UUIDComparator.staticCompare( edge.getVersion(), maxVersion ) < 1;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/909d1d84/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerIT.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerIT.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerIT.java
index f2528b8..a4d6a22 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerIT.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerIT.java
@@ -59,7 +59,7 @@ import static org.mockito.Mockito.when;
@RunWith( JukitoRunner.class )
-@UseModules( { TestGraphModule.class} )
+@UseModules( { TestGraphModule.class } )
//@UseModules( { TestGraphModule.class, EdgeManagerIT.InvalidInput.class } )
public class EdgeManagerIT {
@@ -229,6 +229,146 @@ public class EdgeManagerIT {
}
+ /**
+ * Tests that if multiple versions of an edge exist, only the distinct edges with a version <= max are returned
+ */
+ @Test
+ public void testWriteReadEdgeTypeVersionSourceDistinct() {
+
+ EdgeManager em = emf.createEdgeManager( scope );
+
+ final UUID earlyVersion = UUIDGenerator.newTimeUUID();
+
+
+ Edge edge1 = createEdge( "source", "test", "target" );
+
+ final Id sourceId = edge1.getSourceNode();
+ final Id targetId = edge1.getTargetNode();
+
+
+ em.writeEdge( edge1 ).toBlockingObservable().last();
+
+ Edge edge2 = createEdge( sourceId, edge1.getType(), targetId );
+
+ em.writeEdge( edge2 ).toBlockingObservable().last();
+
+ Edge edge3 = createEdge( sourceId, edge1.getType(), targetId );
+
+ em.writeEdge( edge3 ).toBlockingObservable().last();
+
+
+ //now test retrieving it, we should only get edge3, since it's the latest
+
+ SearchByEdgeType search =
+ createSearchByEdge( edge1.getSourceNode(), edge1.getType(), edge3.getVersion(), null );
+
+ Observable<Edge> edges = em.loadEdgesFromSource( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ Iterator<Edge> returned = edges.toBlockingObservable().getIterator();
+
+ assertEquals( "Correct edge returned", edge3, returned.next() );
+ assertFalse( "No more edges", returned.hasNext() );
+
+ //now test with an earlier version, we shouldn't get the edge back
+ search = createSearchByEdge( edge1.getSourceNode(), edge1.getType(), edge2.getVersion(), null );
+
+ edges = em.loadEdgesFromSource( search );
+
+ returned = edges.toBlockingObservable().getIterator();
+
+ assertEquals( "Correct edge returned", edge2, returned.next() );
+ assertFalse( "No more edges", returned.hasNext() );
+
+ search = createSearchByEdge( edge1.getSourceNode(), edge1.getType(), edge1.getVersion(), null );
+
+ edges = em.loadEdgesFromSource( search );
+
+ returned = edges.toBlockingObservable().getIterator();
+
+ assertEquals( "Correct edge returned", edge1, returned.next() );
+ assertFalse( "No more edges", returned.hasNext() );
+
+
+ search = createSearchByEdge( edge1.getSourceNode(), edge1.getType(), earlyVersion, null );
+
+ edges = em.loadEdgesFromSource( search );
+
+ returned = edges.toBlockingObservable().getIterator();
+
+ assertFalse( "No more edges", returned.hasNext() );
+ }
+
+
+ @Test
+ public void testWriteReadEdgeTypeVersionTargetDistinct() {
+
+
+ EdgeManager em = emf.createEdgeManager( scope );
+
+ final UUID earlyVersion = UUIDGenerator.newTimeUUID();
+
+
+ Edge edge1 = createEdge( "source", "test", "target" );
+
+ final Id sourceId = edge1.getSourceNode();
+ final Id targetId = edge1.getTargetNode();
+
+
+ em.writeEdge( edge1 ).toBlockingObservable().last();
+
+ Edge edge2 = createEdge( sourceId, edge1.getType(), targetId );
+
+ em.writeEdge( edge2 ).toBlockingObservable().last();
+
+ Edge edge3 = createEdge( sourceId, edge1.getType(), targetId );
+
+ em.writeEdge( edge3 ).toBlockingObservable().last();
+
+
+ //now test retrieving it, we should only get edge3, since it's the latest
+
+ SearchByEdgeType search =
+ createSearchByEdge( edge1.getTargetNode(), edge1.getType(), edge3.getVersion(), null );
+
+ Observable<Edge> edges = em.loadEdgesToTarget( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ Iterator<Edge> returned = edges.toBlockingObservable().getIterator();
+
+ assertEquals( "Correct edge returned", edge3, returned.next() );
+ assertFalse( "No more edges", returned.hasNext() );
+
+ //now test with an earlier version, we shouldn't get the edge back
+ search = createSearchByEdge( edge1.getTargetNode(), edge1.getType(), edge2.getVersion(), null );
+
+ edges = em.loadEdgesToTarget( search );
+
+ returned = edges.toBlockingObservable().getIterator();
+
+ assertEquals( "Correct edge returned", edge2, returned.next() );
+ assertFalse( "No more edges", returned.hasNext() );
+
+ search = createSearchByEdge( edge1.getTargetNode(), edge1.getType(), edge1.getVersion(), null );
+
+ edges = em.loadEdgesToTarget( search );
+
+ returned = edges.toBlockingObservable().getIterator();
+
+ assertEquals( "Correct edge returned", edge1, returned.next() );
+ assertFalse( "No more edges", returned.hasNext() );
+
+
+ search = createSearchByEdge( edge1.getTargetNode(), edge1.getType(), earlyVersion, null );
+
+ edges = em.loadEdgesToTarget( search );
+
+ returned = edges.toBlockingObservable().getIterator();
+
+ assertFalse( "No more edges", returned.hasNext() );
+ }
+
+
@Test
public void testWriteReadEdgeTypePagingSource() {
@@ -239,15 +379,15 @@ public class EdgeManagerIT {
Edge edge1 = createEdge( sourceId, "test", createId( "target" ) );
- em.writeEdge( edge1 ).toBlockingObservable().singleOrDefault( null );
+ em.writeEdge( edge1 ).toBlockingObservable().last();
Edge edge2 = createEdge( sourceId, "test", createId( "target" ) );
- em.writeEdge( edge2 ).toBlockingObservable().singleOrDefault( null );
+ em.writeEdge( edge2 ).toBlockingObservable().last();
Edge edge3 = createEdge( sourceId, "test", createId( "target" ) );
- em.writeEdge( edge3 ).toBlockingObservable().singleOrDefault( null );
+ em.writeEdge( edge3 ).toBlockingObservable().last();
//now test retrieving it
@@ -290,15 +430,15 @@ public class EdgeManagerIT {
Edge edge1 = createEdge( createId( "source" ), "test", targetId );
- em.writeEdge( edge1 ).toBlockingObservable().singleOrDefault( null );
+ em.writeEdge( edge1 ).toBlockingObservable().last();
Edge edge2 = createEdge( createId( "source" ), "test", targetId );
- em.writeEdge( edge2 ).toBlockingObservable().singleOrDefault( null );
+ em.writeEdge( edge2 ).toBlockingObservable().last();
Edge edge3 = createEdge( createId( "source" ), "test", targetId );
- em.writeEdge( edge3 ).toBlockingObservable().singleOrDefault( null );
+ em.writeEdge( edge3 ).toBlockingObservable().last();
//now test retrieving it
@@ -826,7 +966,7 @@ public class EdgeManagerIT {
@Test( expected = NullPointerException.class )
- public void invalidEdgeTypesWrite( @All Edge edge) {
+ public void invalidEdgeTypesWrite( @All Edge edge ) {
final EdgeManager em = emf.createEdgeManager( scope );
em.writeEdge( edge );
@@ -834,10 +974,10 @@ public class EdgeManagerIT {
@Test( expected = NullPointerException.class )
- public void invalidEdgeTypesDelete(@All Edge edge) {
+ public void invalidEdgeTypesDelete( @All Edge edge ) {
final EdgeManager em = emf.createEdgeManager( scope );
- em.deleteEdge( edge);
+ em.deleteEdge( edge );
}
@@ -846,32 +986,32 @@ public class EdgeManagerIT {
@Override
protected void configureTest() {
//create all edge types of junk input
-//
-// final UUID version = UUIDGenerator.newTimeUUID();
-//
-// Id nullUuid = mock( Id.class );
-// when( nullUuid.getUuid() ).thenReturn( null );
-//
-//
-// Id nullType = mock( Id.class );
-// when( nullType.getType() ).thenReturn( "type" );
-//
-// Edge[] edges = new Edge[] {
-// mockEdge( nullUuid, "test", createId( "target" ), version ),
-//
-// mockEdge( nullType, "test", createId( "target" ), version ),
-//
-// mockEdge( createId( "source" ), null, createId( "target" ), version ),
-//
-// mockEdge( createId( "source" ), "test", nullUuid, version ),
-//
-// mockEdge( createId( "source" ), "test", nullType, version ),
-//
-// mockEdge( createId( "source" ), "test", createId( "target" ), null )
-// };
-//
-//
-// bindManyInstances( Edge.class, edges );
+ //
+ // final UUID version = UUIDGenerator.newTimeUUID();
+ //
+ // Id nullUuid = mock( Id.class );
+ // when( nullUuid.getUuid() ).thenReturn( null );
+ //
+ //
+ // Id nullType = mock( Id.class );
+ // when( nullType.getType() ).thenReturn( "type" );
+ //
+ // Edge[] edges = new Edge[] {
+ // mockEdge( nullUuid, "test", createId( "target" ), version ),
+ //
+ // mockEdge( nullType, "test", createId( "target" ), version ),
+ //
+ // mockEdge( createId( "source" ), null, createId( "target" ), version ),
+ //
+ // mockEdge( createId( "source" ), "test", nullUuid, version ),
+ //
+ // mockEdge( createId( "source" ), "test", nullType, version ),
+ //
+ // mockEdge( createId( "source" ), "test", createId( "target" ), null )
+ // };
+ //
+ //
+ // bindManyInstances( Edge.class, edges );
}
[2/4] git commit: Merge branch 'two-dot-o' of
github.com:usergrid/usergrid into two-dot-o
Posted by sn...@apache.org.
Merge branch 'two-dot-o' of github.com:usergrid/usergrid into two-dot-o
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/ccc07106
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/ccc07106
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/ccc07106
Branch: refs/heads/two-dot-o
Commit: ccc071064f5417f840bcff351f5429e0d7964d9a
Parents: 909d1d8 6c35763
Author: Todd Nine <tn...@apigee.com>
Authored: Mon Feb 17 15:50:48 2014 -0700
Committer: Todd Nine <tn...@apigee.com>
Committed: Mon Feb 17 15:50:48 2014 -0700
----------------------------------------------------------------------
.../cassandra/AvailablePortFinder.java | 186 -------------------
.../collection/util/AvailablePortFinder.java | 186 +++++++++++++++++++
.../cassandra/AvailablePortFinderTest.java | 73 --------
.../collection/cassandra/CassandraRule.java | 3 +-
.../util/AvailablePortFinderTest.java | 73 ++++++++
5 files changed, 260 insertions(+), 261 deletions(-)
----------------------------------------------------------------------
[3/4] git commit: Merge branch 'two-dot-o' of
github.com:usergrid/usergrid into two-dot-o
Posted by sn...@apache.org.
Merge branch 'two-dot-o' of github.com:usergrid/usergrid into two-dot-o
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/f157303d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/f157303d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/f157303d
Branch: refs/heads/two-dot-o
Commit: f157303daed84abfe9255bac1b27cf8fd697c829
Parents: ccc0710 3e55e90
Author: Todd Nine <tn...@apigee.com>
Authored: Mon Feb 17 16:21:16 2014 -0700
Committer: Todd Nine <tn...@apigee.com>
Committed: Mon Feb 17 16:21:16 2014 -0700
----------------------------------------------------------------------
.../persistence/collection/util/AvailablePortFinderTest.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
[4/4] git commit: Graph mark completed. Sweep needs implemented
Posted by sn...@apache.org.
Graph mark completed. Sweep needs implemented
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/4e5b3b25
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/4e5b3b25
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/4e5b3b25
Branch: refs/heads/two-dot-o
Commit: 4e5b3b25e87c279ce36f630c8afe9d9cc04f4bc9
Parents: f157303
Author: Todd Nine <tn...@apigee.com>
Authored: Mon Feb 17 17:05:25 2014 -0700
Committer: Todd Nine <tn...@apigee.com>
Committed: Mon Feb 17 17:06:59 2014 -0700
----------------------------------------------------------------------
.../usergrid/persistence/graph/MarkedEdge.java | 35 ++
.../graph/impl/CollectionIndexObserver.java | 6 +-
.../persistence/graph/impl/EdgeManagerImpl.java | 199 ++++-----
.../persistence/graph/impl/SimpleEdge.java | 20 +-
.../graph/impl/SimpleMarkedEdge.java | 71 +++
.../graph/serialization/EdgeSerialization.java | 21 +-
.../impl/EdgeSerializationImpl.java | 95 ++--
.../impl/parse/ObservableIterator.java | 3 +-
.../graph/serialization/stage/GraphIoEvent.java | 32 --
.../stage/write/EdgeWriteStage.java | 59 ---
.../persistence/graph/EdgeManagerIT.java | 442 +++++++++++++++----
.../EdgeSerializationChopTest.java | 4 +-
.../serialization/EdgeSerializationTest.java | 167 ++++++-
.../graph/test/util/EdgeTestUtils.java | 4 +-
14 files changed, 795 insertions(+), 363 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/MarkedEdge.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/MarkedEdge.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/MarkedEdge.java
new file mode 100644
index 0000000..65de44a
--- /dev/null
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/MarkedEdge.java
@@ -0,0 +1,35 @@
+/*
+ * 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.graph;
+
+
+/**
+ *
+ *
+ */
+public interface MarkedEdge extends Edge{
+
+ /**
+ * True if this edge is deleted, false otherwise
+ * @return True if this version is marked as deleted
+ */
+ public boolean isDeleted();
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/CollectionIndexObserver.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/CollectionIndexObserver.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/CollectionIndexObserver.java
index b8b551a..6175b0b 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/CollectionIndexObserver.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/CollectionIndexObserver.java
@@ -64,15 +64,15 @@ public class CollectionIndexObserver implements PostProcessObserver {
*
* We're essentially mapping a tree structure in to a graph edge
*/
- Edge edge = new SimpleEdge(scope.getOwner(), scope.getName(), entity.getId(), entity.getVersion() );
+ Edge edge = new SimpleMarkedEdge( scope.getOwner(), scope.getName(), entity.getId(), entity.getVersion(), false );
//entity exists, write the edge
if(entity.getEntity().isPresent()){
- em.writeEdge( edge );
+ em.writeEdge( edge ).toBlockingObservable().last();
}
//entity does not exist, it's been removed, mark the edge
else{
- em.deleteEdge( edge );
+ em.deleteEdge( edge ).toBlockingObservable().last();
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java
index 2fc61b0..56d9d06 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java
@@ -27,6 +27,7 @@ import org.apache.usergrid.persistence.collection.OrganizationScope;
import org.apache.usergrid.persistence.collection.mvcc.entity.ValidationUtils;
import org.apache.usergrid.persistence.graph.Edge;
import org.apache.usergrid.persistence.graph.EdgeManager;
+import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.SearchByEdgeType;
import org.apache.usergrid.persistence.graph.SearchByIdType;
import org.apache.usergrid.persistence.graph.SearchEdgeType;
@@ -34,14 +35,14 @@ import org.apache.usergrid.persistence.graph.SearchIdType;
import org.apache.usergrid.persistence.graph.serialization.EdgeMetadataSerialization;
import org.apache.usergrid.persistence.graph.serialization.EdgeSerialization;
import org.apache.usergrid.persistence.graph.serialization.impl.parse.ObservableIterator;
-import org.apache.usergrid.persistence.graph.serialization.stage.GraphIoEvent;
-import org.apache.usergrid.persistence.graph.serialization.stage.write.EdgeWriteStage;
import org.apache.usergrid.persistence.graph.serialization.util.EdgeUtils;
import org.apache.usergrid.persistence.model.entity.Id;
import com.fasterxml.uuid.UUIDComparator;
import com.google.inject.Inject;
import com.google.inject.assistedinject.Assisted;
+import com.netflix.astyanax.MutationBatch;
+import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import rx.Observable;
import rx.Scheduler;
@@ -59,18 +60,15 @@ public class EdgeManagerImpl implements EdgeManager {
private final Scheduler scheduler;
- private final EdgeWriteStage edgeWriteStage;
-
private final EdgeMetadataSerialization edgeMetadataSerialization;
+
private final EdgeSerialization edgeSerialization;
@Inject
- public EdgeManagerImpl( final EdgeWriteStage edgeWriteStage, final Scheduler scheduler,
- final EdgeMetadataSerialization edgeMetadataSerialization,
+ public EdgeManagerImpl( final Scheduler scheduler, final EdgeMetadataSerialization edgeMetadataSerialization,
final EdgeSerialization edgeSerialization, @Assisted final OrganizationScope scope ) {
- this.edgeWriteStage = edgeWriteStage;
this.scheduler = scheduler;
this.edgeMetadataSerialization = edgeMetadataSerialization;
this.edgeSerialization = edgeSerialization;
@@ -85,120 +83,64 @@ public class EdgeManagerImpl implements EdgeManager {
@Override
public Observable<Edge> writeEdge( final Edge edge ) {
- return Observable.from( new GraphIoEvent<Edge>( scope, edge ) ).subscribeOn( scheduler ).map( edgeWriteStage );
+ EdgeUtils.validateEdge( edge );
+
+ return Observable.from( edge ).subscribeOn( scheduler ).map( new Func1<Edge, Edge>() {
+ @Override
+ public Edge call( final Edge edge ) {
+ //TODO mark the write in the write ahead log
+ final MutationBatch mutation = edgeMetadataSerialization.writeEdge( scope, edge );
+
+ final MutationBatch edgeMutation = edgeSerialization.writeEdge( scope, edge );
+
+ mutation.mergeShallow( edgeMutation );
+
+ try {
+ mutation.execute();
+ }
+ catch ( ConnectionException e ) {
+ throw new RuntimeException( "Unable to connect to cassandra", e );
+ }
+
+ return edge;
+ }
+ } );
}
@Override
public Observable<Edge> deleteEdge( final Edge edge ) {
EdgeUtils.validateEdge( edge );
- //
- //
- // Observable<Edge> observable = Observable.from( edge ).subscribeOn( scheduler );
- //
- //
- // final SimpleSearchByEdge search =
- // new SimpleSearchByEdge( edge.getSourceNode(), edge.getType(), edge.getTargetNode(),
- // edge.getVersion(),
- // null );
- //
- //
- // final UUID maxVersion = edge.getVersion();
- //
- // //search on both our source and target types, we need to filter them if there aren't any left
- // observable.flatMap( new Func1<Edge, Observable<Edge>>() {
- //
- // @Override
- // public Observable<Edge> call( final Edge edge ) {
- //
- // //take the first 2 edges and then count them. If there are 2 elements, then
- // //we can't remove the meta data. If there are 2, we can
- //
- // Observable<Edge> sourceEdges = Observable.create( new ObservableIterator<Edge>() {
- // @Override
- // protected Iterator<Edge> getIterator() {
- // return edgeSerialization.getEdgeFromSource( scope, search );
- // }
- // } );
- //
- // Observable<MutationBatch>
- // sourceDeletes = filter(sourceEdges, maxVersion).map( new Func1<Edge,
- // MutationBatch>() {
- // @Override
- // public MutationBatch call( final Edge edge ) {
- // return edgeSerialization.deleteEdge( scope, edge );
- // }
- // } );
- //
- //
- // Observable<Edge> targetEdges = Observable.create( new ObservableIterator<Edge>() {
- // @Override
- // protected Iterator<Edge> getIterator() {
- // return edgeSerialization.getEdgeToTarget( scope, search );
- // }
- // } );
- //
- // Observable<MutationBatch> targetDeletes = filter(targetEdges,
- // maxVersion).map( new Func1<Edge, MutationBatch> (){
- //
- //
- // @Override
- // public MutationBatch call( final Edge edge ) {
- // return edgeSerialization.deleteEdge( scope, edge );
- // }
- // });
- //
- // return Observable.zip( sourceDeletes, targetDeletes, new Func2<MutationBatch,
- // MutationBatch, Edge>() {
- //
- // private MutationBatch batch;
- //
- // @Override
- // public Edge call( final MutationBatch mutationBatch, final MutationBatch mutationBatch2
- // ) {
- // if(batch == null){
- // batch = mutationBatch;
- // }
- //
- // batch.mergeShallow( mutationBatch );
- // batch.mergeShallow( mutationBatch2 );
- //
- // return batch;
- // }
- // });
- //
- // return Observable.zip( sourceEdges, targetEdges, new Func2<Edge, Edge, Edge>() {
- // @Override
- // public Edge call( final Edge sourceEdge, final Edge targetEdges ) {
- //
- // edgeSerialization.deleteEdge( scope, sourceEdge );
- // edgeSerialization.deleteEdge( scope, targetEdges );
- //
- // return edge;
- // }
- // } );
- //
- //
- //
- // return null;
- //
- //
- // }
- // } );
-
-
- throw new UnsupportedOperationException( "Not yet implemented" );
+
+ return Observable.from( edge ).subscribeOn( scheduler ).map( new Func1<Edge, Edge>() {
+ @Override
+ public Edge call( final Edge edge ) {
+ //TODO Mark the write in the write ahead log
+ final MutationBatch edgeMutation = edgeSerialization.markEdge( scope, edge );
+
+ try {
+ edgeMutation.execute();
+ }
+ catch ( ConnectionException e ) {
+ throw new RuntimeException( "Unable to connect to cassandra", e );
+ }
+
+ return edge;
+ }
+ } );
+
+ //TODO, fork the background repair scheduling here
}
@Override
public Observable<Edge> loadEdgesFromSource( final SearchByEdgeType search ) {
- return Observable.create( new ObservableIterator<Edge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>() {
@Override
- protected Iterator<Edge> getIterator() {
+ protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesFromSource( scope, search );
}
- } ).filter( new MaxFilter( search.getMaxVersion() ) )
+ } ).filter( new EdgeFilter( search.getMaxVersion() ) )
//we intentionally use distinct until changed. This way we won't store all the keys since this
//would hog far too much ram.
.distinctUntilChanged( new Func1<Edge, Id>() {
@@ -206,18 +148,18 @@ public class EdgeManagerImpl implements EdgeManager {
public Id call( final Edge edge ) {
return edge.getTargetNode();
}
- } );
+ } ).map( mapper );
}
@Override
public Observable<Edge> loadEdgesToTarget( final SearchByEdgeType search ) {
- return Observable.create( new ObservableIterator<Edge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>() {
@Override
- protected Iterator<Edge> getIterator() {
+ protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesToTarget( scope, search );
}
- } ).filter( new MaxFilter( search.getMaxVersion() ) )
+ } ).filter( new EdgeFilter( search.getMaxVersion() ) )
//we intentionally use distinct until changed. This way we won't store all the keys since this
//would hog far too much ram.
.distinctUntilChanged( new Func1<Edge, Id>() {
@@ -225,29 +167,29 @@ public class EdgeManagerImpl implements EdgeManager {
public Id call( final Edge edge ) {
return edge.getSourceNode();
}
- } );
+ } ).map( mapper );
}
@Override
public Observable<Edge> loadEdgesFromSourceByType( final SearchByIdType search ) {
- return Observable.create( new ObservableIterator<Edge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>() {
@Override
- protected Iterator<Edge> getIterator() {
+ protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesFromSourceByTargetType( scope, search );
}
- } ).filter( new MaxFilter( search.getMaxVersion() ) ).takeFirst();
+ } ).filter( new EdgeFilter( search.getMaxVersion() ) ).takeFirst().map( mapper );
}
@Override
public Observable<Edge> loadEdgesToTargetByType( final SearchByIdType search ) {
- return Observable.create( new ObservableIterator<Edge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>() {
@Override
- protected Iterator<Edge> getIterator() {
+ protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesToTargetBySourceType( scope, search );
}
- } ).filter( new MaxFilter( search.getMaxVersion() ) ).takeFirst();
+ } ).filter( new EdgeFilter( search.getMaxVersion() ) ).takeFirst().map( mapper );
}
@@ -298,22 +240,33 @@ public class EdgeManagerImpl implements EdgeManager {
/**
- * Filter the returned values based on the max uuid
+ * Filter the returned values based on the max uuid and if it's been marked for deletion or not
*/
- private static class MaxFilter implements Func1<Edge, Boolean> {
+ private static class EdgeFilter implements Func1<MarkedEdge, Boolean> {
private final UUID maxVersion;
- private MaxFilter( final UUID maxVersion ) {
+ private EdgeFilter( final UUID maxVersion ) {
this.maxVersion = maxVersion;
}
@Override
- public Boolean call( final Edge edge ) {
+ public Boolean call( final MarkedEdge edge ) {
//our edge version needs to be <= max Version
- return UUIDComparator.staticCompare( edge.getVersion(), maxVersion ) < 1;
+ return !edge.isDeleted() && UUIDComparator.staticCompare( edge.getVersion(), maxVersion ) < 1;
}
}
+
+
+ /**
+ * Simple function that maps MarkedEdge to edges
+ */
+ private static final Func1<MarkedEdge, Edge> mapper = new Func1<MarkedEdge, Edge>() {
+ @Override
+ public Edge call( final MarkedEdge markedEdge ) {
+ return markedEdge;
+ }
+ };
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleEdge.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleEdge.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleEdge.java
index 81a6005..279499b 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleEdge.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleEdge.java
@@ -33,10 +33,10 @@ import org.apache.usergrid.persistence.model.entity.Id;
*/
public class SimpleEdge implements Edge {
- private final Id sourceNode;
- private final String type;
- private final Id targetNode;
- private final UUID version;
+ protected final Id sourceNode;
+ protected final String type;
+ protected final Id targetNode;
+ protected final UUID version;
public SimpleEdge( final Id sourceNode, final String type, final Id targetNode, final UUID version ) {
@@ -75,17 +75,25 @@ public class SimpleEdge implements Edge {
}
+
+ /**
+ * Test if the 2 edges are equal to one another. Note that this is an edge comparison, not a marked edge comparison
+ * @param o
+ * @return
+ */
+
@Override
public boolean equals( final Object o ) {
if ( this == o ) {
return true;
}
- if ( !( o instanceof Edge ) ) {
+ if ( !( o instanceof SimpleEdge ) ) {
return false;
}
final Edge that = ( Edge ) o;
+
if ( !sourceNode.equals( that.getSourceNode() ) ) {
return false;
}
@@ -115,7 +123,7 @@ public class SimpleEdge implements Edge {
@Override
public String toString() {
- return "SimpleEdge{" +
+ return "SimpleMarkedEdge{" +
"sourceNode=" + sourceNode +
", type='" + type + '\'' +
", targetNode=" + targetNode +
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleMarkedEdge.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleMarkedEdge.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleMarkedEdge.java
new file mode 100644
index 0000000..7830ce6
--- /dev/null
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleMarkedEdge.java
@@ -0,0 +1,71 @@
+/*
+ * 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.graph.impl;
+
+
+import java.util.UUID;
+
+import org.apache.usergrid.persistence.collection.mvcc.entity.ValidationUtils;
+import org.apache.usergrid.persistence.graph.Edge;
+import org.apache.usergrid.persistence.graph.MarkedEdge;
+import org.apache.usergrid.persistence.model.entity.Id;
+
+
+/**
+ * Simple bean to represent our edge
+ * @author tnine
+ */
+public class SimpleMarkedEdge extends SimpleEdge implements MarkedEdge {
+
+ private final boolean deleted;
+
+
+ public SimpleMarkedEdge( final Id sourceNode, final String type, final Id targetNode, final UUID version, final boolean deleted) {
+
+ super(sourceNode, type, targetNode, version);
+ this.deleted = deleted;
+ }
+
+
+
+ @Override
+ public boolean isDeleted() {
+ return deleted;
+ }
+
+
+ /**
+ * We purposefully don't implement hashcode or equals. The only additional field is deleted. We don't want to compare
+ * edge equality based on deleted true/false, since this does not define relevant indexing information for a marked edge
+ *
+ */
+
+
+ @Override
+ public String toString() {
+ return "SimpleMarkedEdge{" +
+ "sourceNode=" + sourceNode +
+ ", type='" + type + '\'' +
+ ", targetNode=" + targetNode +
+ ", version=" + version +
+ ", deleted=" + deleted +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerialization.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerialization.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerialization.java
index adb5e9e..1575121 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerialization.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerialization.java
@@ -24,6 +24,7 @@ import java.util.Iterator;
import org.apache.usergrid.persistence.collection.OrganizationScope;
import org.apache.usergrid.persistence.graph.Edge;
+import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.SearchByEdge;
import org.apache.usergrid.persistence.graph.SearchByEdgeType;
import org.apache.usergrid.persistence.graph.SearchByIdType;
@@ -47,6 +48,14 @@ public interface EdgeSerialization {
/**
+ * Mark this edge for deletion
+ * @param scope
+ * @param edge
+ * @return
+ */
+ MutationBatch markEdge( OrganizationScope scope, Edge edge);
+
+ /**
* Write both the source -->target edge and the target<--- source edge into the mutation
*
* @param scope The org scope of the graph
@@ -61,7 +70,7 @@ public interface EdgeSerialization {
* @param search
* @return
*/
- Iterator<Edge> getEdgeFromSource( OrganizationScope scope, SearchByEdge search );
+ Iterator<MarkedEdge> getEdgeFromSource( OrganizationScope scope, SearchByEdge search );
/**
* Get an iterator of all edges by edge type originating from source node
@@ -69,7 +78,7 @@ public interface EdgeSerialization {
* @param scope The org scope of the graph
* @param edgeType The search edge
*/
- Iterator<Edge> getEdgesFromSource( OrganizationScope scope, SearchByEdgeType edgeType );
+ Iterator<MarkedEdge> getEdgesFromSource( OrganizationScope scope, SearchByEdgeType edgeType );
/**
@@ -78,7 +87,7 @@ public interface EdgeSerialization {
* @param scope The org scope of the graph
* @param edgeType The search edge
*/
- Iterator<Edge> getEdgesFromSourceByTargetType( OrganizationScope scope, SearchByIdType edgeType );
+ Iterator<MarkedEdge> getEdgesFromSourceByTargetType( OrganizationScope scope, SearchByIdType edgeType );
/**
* Get an iterator of all edges by edge type pointing to the target node. Returns all versions
@@ -86,7 +95,7 @@ public interface EdgeSerialization {
* @param scope The org scope of the graph
* @param edgeType The search edge
*/
- Iterator<Edge> getEdgesToTarget( OrganizationScope scope, SearchByEdgeType edgeType );
+ Iterator<MarkedEdge> getEdgesToTarget( OrganizationScope scope, SearchByEdgeType edgeType );
/**
* Search for specific versions of the edge from source->target. Will return all versions
@@ -94,7 +103,7 @@ public interface EdgeSerialization {
* @param search
* @return
*/
- Iterator<Edge> getEdgeToTarget( OrganizationScope scope, SearchByEdge search );
+ Iterator<MarkedEdge> getEdgeToTarget( OrganizationScope scope, SearchByEdge search );
/**
@@ -104,5 +113,5 @@ public interface EdgeSerialization {
* @param scope The org scope of the graph
* @param edgeType The search edge
*/
- Iterator<Edge> getEdgesToTargetBySourceType( OrganizationScope scope, SearchByIdType edgeType );
+ Iterator<MarkedEdge> getEdgesToTargetBySourceType( OrganizationScope scope, SearchByIdType edgeType );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeSerializationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeSerializationImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeSerializationImpl.java
index f587060..70ce87e 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeSerializationImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeSerializationImpl.java
@@ -42,10 +42,11 @@ import org.apache.usergrid.persistence.collection.migration.Migration;
import org.apache.usergrid.persistence.collection.mvcc.entity.ValidationUtils;
import org.apache.usergrid.persistence.graph.Edge;
import org.apache.usergrid.persistence.graph.GraphFig;
+import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.SearchByEdge;
import org.apache.usergrid.persistence.graph.SearchByEdgeType;
import org.apache.usergrid.persistence.graph.SearchByIdType;
-import org.apache.usergrid.persistence.graph.impl.SimpleEdge;
+import org.apache.usergrid.persistence.graph.impl.SimpleMarkedEdge;
import org.apache.usergrid.persistence.graph.serialization.EdgeSerialization;
import org.apache.usergrid.persistence.graph.serialization.impl.parse.ColumnNameIterator;
import org.apache.usergrid.persistence.graph.serialization.impl.parse.ColumnParser;
@@ -78,9 +79,6 @@ import com.netflix.astyanax.util.RangeBuilder;
public class EdgeSerializationImpl implements EdgeSerialization, Migration {
- //holder to put data into col value
- private static final byte[] HOLDER = new byte[] { 0 };
-
//Row key with no type
private static final RowSerializer ROW_SERIALIZER = new RowSerializer();
@@ -147,7 +145,26 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
public void doWrite( final MultiTennantColumnFamily columnFamily, final Object rowKey,
final DirectedEdge edge ) {
- batch.withRow( columnFamily, ScopedRowKey.fromKey( scope, rowKey ) ).putColumn( edge, HOLDER );
+ batch.withRow( columnFamily, ScopedRowKey.fromKey( scope, rowKey ) ).putColumn( edge, false );
+ }
+ } );
+
+
+ return batch;
+ }
+
+
+ @Override
+ public MutationBatch markEdge( final OrganizationScope scope, final Edge edge ) {
+ final MutationBatch batch = keyspace.prepareMutationBatch();
+
+
+ doWrite( scope, edge, new RowOp() {
+ @Override
+ public void doWrite( final MultiTennantColumnFamily columnFamily, final Object rowKey,
+ final DirectedEdge edge ) {
+
+ batch.withRow( columnFamily, ScopedRowKey.fromKey( scope, rowKey ) ).putColumn( edge, true );
}
} );
@@ -175,6 +192,12 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
}
+ /**
+ * Write the edges internally
+ * @param scope The scope to encapsulate
+ * @param edge The edge to write
+ * @param op The row operation to invoke
+ */
private void doWrite( final OrganizationScope scope, final Edge edge, final RowOp op ) {
ValidationUtils.validateOrganizationScope( scope );
EdgeUtils.validateEdge( edge );
@@ -219,7 +242,7 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
@Override
- public Iterator<Edge> getEdgeFromSource( final OrganizationScope scope, final SearchByEdge search ) {
+ public Iterator<MarkedEdge> getEdgeFromSource( final OrganizationScope scope, final SearchByEdge search ) {
ValidationUtils.validateOrganizationScope( scope );
EdgeUtils.validateSearchByEdge( search );
@@ -242,7 +265,7 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
//start seeking at a value < our max version
final DirectedEdge last = new DirectedEdge( targetId, maxVersion );
- builder.setStart( last, EDGE_SERIALIZER);
+ builder.setStart( last, EDGE_SERIALIZER );
}
@@ -259,15 +282,15 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
@Override
- protected Edge createEdge( final DirectedEdge edge ) {
- return new SimpleEdge( sourceId, type, edge.id, edge.version );
+ protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
+ return new SimpleMarkedEdge( sourceId, type, edge.id, edge.version, marked );
}
} );
}
@Override
- public Iterator<Edge> getEdgesFromSource( final OrganizationScope scope, final SearchByEdgeType edgeType ) {
+ public Iterator<MarkedEdge> getEdgesFromSource( final OrganizationScope scope, final SearchByEdgeType edgeType ) {
ValidationUtils.validateOrganizationScope( scope );
EdgeUtils.validateSearchByEdgeType( edgeType );
@@ -289,16 +312,16 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
@Override
- protected Edge createEdge( final DirectedEdge edge ) {
- return new SimpleEdge( sourceId, type, edge.id, edge.version );
+ protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
+ return new SimpleMarkedEdge( sourceId, type, edge.id, edge.version, marked );
}
} );
}
@Override
- public Iterator<Edge> getEdgesFromSourceByTargetType( final OrganizationScope scope,
- final SearchByIdType edgeType ) {
+ public Iterator<MarkedEdge> getEdgesFromSourceByTargetType( final OrganizationScope scope,
+ final SearchByIdType edgeType ) {
ValidationUtils.validateOrganizationScope( scope );
EdgeUtils.validateSearchByEdgeType( edgeType );
@@ -321,15 +344,15 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
@Override
- protected Edge createEdge( final DirectedEdge edge ) {
- return new SimpleEdge( targetId, type, edge.id, edge.version );
+ protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
+ return new SimpleMarkedEdge( targetId, type, edge.id, edge.version, marked );
}
} );
}
@Override
- public Iterator<Edge> getEdgesToTarget( final OrganizationScope scope, final SearchByEdgeType edgeType ) {
+ public Iterator<MarkedEdge> getEdgesToTarget( final OrganizationScope scope, final SearchByEdgeType edgeType ) {
ValidationUtils.validateOrganizationScope( scope );
EdgeUtils.validateSearchByEdgeType( edgeType );
@@ -353,15 +376,15 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
@Override
- protected Edge createEdge( final DirectedEdge edge ) {
- return new SimpleEdge( edge.id, type, targetId, edge.version );
+ protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
+ return new SimpleMarkedEdge( edge.id, type, targetId, edge.version, marked );
}
} );
}
@Override
- public Iterator<Edge> getEdgeToTarget( final OrganizationScope scope, final SearchByEdge search ) {
+ public Iterator<MarkedEdge> getEdgeToTarget( final OrganizationScope scope, final SearchByEdge search ) {
ValidationUtils.validateOrganizationScope( scope );
EdgeUtils.validateSearchByEdge( search );
@@ -375,7 +398,7 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
@Override
public void setRange( final RangeBuilder builder ) {
- if(last.isPresent()){
+ if ( last.isPresent() ) {
super.setRange( builder );
return;
}
@@ -399,15 +422,16 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
@Override
- protected Edge createEdge( final DirectedEdge edge ) {
- return new SimpleEdge( edge.id, type, targetId, edge.version );
+ protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
+ return new SimpleMarkedEdge( edge.id, type, targetId, edge.version, marked );
}
} );
}
@Override
- public Iterator<Edge> getEdgesToTargetBySourceType( final OrganizationScope scope, final SearchByIdType edgeType ) {
+ public Iterator<MarkedEdge> getEdgesToTargetBySourceType( final OrganizationScope scope,
+ final SearchByIdType edgeType ) {
ValidationUtils.validateOrganizationScope( scope );
EdgeUtils.validateSearchByEdgeType( edgeType );
@@ -430,8 +454,8 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
@Override
- protected Edge createEdge( final DirectedEdge edge ) {
- return new SimpleEdge( edge.id, type, targetId, edge.version );
+ protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
+ return new SimpleMarkedEdge( edge.id, type, targetId, edge.version, marked );
}
} );
}
@@ -445,8 +469,8 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
*
* @return An iterator of all Edge instance that match the criteria
*/
- private <R> Iterator<Edge> getEdges( MultiTennantColumnFamily<OrganizationScope, R, DirectedEdge> cf,
- final EdgeSearcher<R> searcher ) {
+ private <R> Iterator<MarkedEdge> getEdges( MultiTennantColumnFamily<OrganizationScope, R, DirectedEdge> cf,
+ final EdgeSearcher<R> searcher ) {
/**
@@ -468,7 +492,7 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
try {
- return new ColumnNameIterator<DirectedEdge, Edge>( query.execute().getResult().iterator(), searcher,
+ return new ColumnNameIterator<DirectedEdge, MarkedEdge>( query.execute().getResult().iterator(), searcher,
searcher.hasPage() );
}
catch ( ConnectionException e ) {
@@ -546,8 +570,6 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
}
-
-
/**
* Create the dynamic composite for this directed edge
*/
@@ -627,7 +649,7 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
/**
* Searcher to be used when performing the search. Performs I/O transformation as well as parsing for the iterator
*/
- private static abstract class EdgeSearcher<R> implements ColumnParser<DirectedEdge, Edge> {
+ private static abstract class EdgeSearcher<R> implements ColumnParser<DirectedEdge, MarkedEdge> {
protected final Optional<Edge> last;
protected final OrganizationScope scope;
@@ -668,10 +690,10 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
@Override
- public Edge parseColumn( final Column<DirectedEdge> column ) {
+ public MarkedEdge parseColumn( final Column<DirectedEdge> column ) {
final DirectedEdge edge = column.getName();
- return createEdge( edge );
+ return createEdge( edge, column.getBooleanValue() );
}
@@ -689,8 +711,11 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
/**
* Create an edge to return to the user based on the directed edge provided
+ *
+ * @param edge The edge in the column name
+ * @param marked The marked flag in the column value
*/
- protected abstract Edge createEdge( DirectedEdge edge );
+ protected abstract MarkedEdge createEdge( DirectedEdge edge, boolean marked );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/parse/ObservableIterator.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/parse/ObservableIterator.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/parse/ObservableIterator.java
index 33efecc..5bbe091 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/parse/ObservableIterator.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/parse/ObservableIterator.java
@@ -26,7 +26,8 @@ public abstract class ObservableIterator<T> implements Observable.OnSubscribeFun
Iterator<T> itr = getIterator();
- while ( itr.hasNext() ) {
+ //TODO T.N. when > 0.17 comes out, we need to implement the check with each loop as described here https://github.com/Netflix/RxJava/issues/802
+ while ( itr.hasNext()) {
observer.onNext( itr.next() );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/stage/GraphIoEvent.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/stage/GraphIoEvent.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/stage/GraphIoEvent.java
deleted file mode 100644
index b05159e..0000000
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/stage/GraphIoEvent.java
+++ /dev/null
@@ -1,32 +0,0 @@
-package org.apache.usergrid.persistence.graph.serialization.stage;
-
-
-import org.apache.usergrid.persistence.collection.OrganizationScope;
-
-
-/**
- * Simple class for passing I/O events
- * @param <T>
- */
-public class GraphIoEvent<T> {
-
- private OrganizationScope scope;
-
- private T event;
-
-
- public GraphIoEvent( final OrganizationScope scope, final T event ) {
- this.scope = scope;
- this.event = event;
- }
-
-
- public OrganizationScope getOrganization() {
- return scope;
- }
-
-
- public T getEvent() {
- return event;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/stage/write/EdgeWriteStage.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/stage/write/EdgeWriteStage.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/stage/write/EdgeWriteStage.java
deleted file mode 100644
index 03deec6..0000000
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/stage/write/EdgeWriteStage.java
+++ /dev/null
@@ -1,59 +0,0 @@
-package org.apache.usergrid.persistence.graph.serialization.stage.write;
-
-
-import org.apache.usergrid.persistence.graph.Edge;
-import org.apache.usergrid.persistence.graph.serialization.EdgeMetadataSerialization;
-import org.apache.usergrid.persistence.graph.serialization.EdgeSerialization;
-import org.apache.usergrid.persistence.graph.serialization.stage.GraphIoEvent;
-
-import com.google.common.base.Preconditions;
-import com.google.inject.Inject;
-import com.google.inject.Singleton;
-import com.netflix.astyanax.MutationBatch;
-import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
-
-import rx.util.functions.Func1;
-
-
-/**
- * Stage for writing the edge data
- */
-@Singleton
-public class EdgeWriteStage implements Func1<GraphIoEvent<Edge>, Edge> {
-
-
- private final EdgeSerialization edgeSerialization;
- private final EdgeMetadataSerialization edgeMetadata;
-
-
- @Inject
- public EdgeWriteStage( final EdgeSerialization edgeSerialization, final EdgeMetadataSerialization edgeMetadata ) {
- Preconditions.checkNotNull( edgeSerialization, "edgeSerialization is required" );
- Preconditions.checkNotNull( edgeMetadata, "edgeMetaData is required" );
-
- this.edgeMetadata = edgeMetadata;
-
- this.edgeSerialization = edgeSerialization;
- }
-
-
- @Override
- public Edge call( final GraphIoEvent<Edge> edgeGraphIoEvent ) {
- final MutationBatch mutation =
- edgeMetadata.writeEdge( edgeGraphIoEvent.getOrganization(), edgeGraphIoEvent.getEvent() );
-
- final MutationBatch edgeMutation =
- edgeSerialization.writeEdge( edgeGraphIoEvent.getOrganization(), edgeGraphIoEvent.getEvent() );
-
- mutation.mergeShallow( edgeMutation );
-
- try {
- mutation.execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( "Unable to connect to cassandra", e );
- }
-
- return edgeGraphIoEvent.getEvent();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerIT.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerIT.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerIT.java
index a4d6a22..a4c770e 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerIT.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerIT.java
@@ -23,7 +23,6 @@ import java.util.Iterator;
import java.util.UUID;
import org.jukito.All;
-import org.jukito.JukitoModule;
import org.jukito.JukitoRunner;
import org.jukito.UseModules;
import org.junit.Before;
@@ -36,7 +35,6 @@ import org.apache.usergrid.persistence.collection.OrganizationScope;
import org.apache.usergrid.persistence.collection.cassandra.CassandraRule;
import org.apache.usergrid.persistence.collection.guice.MigrationManagerRule;
import org.apache.usergrid.persistence.graph.guice.TestGraphModule;
-import org.apache.usergrid.persistence.graph.impl.SimpleEdge;
import org.apache.usergrid.persistence.graph.impl.SimpleSearchEdgeType;
import org.apache.usergrid.persistence.graph.impl.SimpleSearchIdType;
import org.apache.usergrid.persistence.model.entity.Id;
@@ -365,7 +363,7 @@ public class EdgeManagerIT {
returned = edges.toBlockingObservable().getIterator();
- assertFalse( "No more edges", returned.hasNext() );
+ assertFalse( "No more edges", returned.hasNext() );
}
@@ -577,7 +575,7 @@ public class EdgeManagerIT {
//now delete it
- em.deleteEdge( edge );
+ em.deleteEdge( edge ).toBlockingObservable().last();
//now test retrieval, should be null
edges = em.loadEdgesFromSource( search );
@@ -633,7 +631,7 @@ public class EdgeManagerIT {
//now delete it
- em.deleteEdge( edge );
+ em.deleteEdge( edge ).toBlockingObservable().last();
//now test retrieval, should be null
edges = em.loadEdgesToTarget( search );
@@ -664,16 +662,16 @@ public class EdgeManagerIT {
Id targetId1 = new SimpleId( "target" );
Id targetId2 = new SimpleId( "target2" );
- Edge testTargetEdge = new SimpleEdge( sourceId, "test", targetId1, UUIDGenerator.newTimeUUID() );
+ Edge testTargetEdge = createEdge( sourceId, "test", targetId1, UUIDGenerator.newTimeUUID() );
em.writeEdge( testTargetEdge ).toBlockingObservable().singleOrDefault( null );
- Edge testTarget2Edge = new SimpleEdge( sourceId, "test", targetId2, UUIDGenerator.newTimeUUID() );
+ Edge testTarget2Edge = createEdge( sourceId, "test", targetId2, UUIDGenerator.newTimeUUID() );
em.writeEdge( testTarget2Edge ).toBlockingObservable().singleOrDefault( null );
- Edge test2TargetEdge = new SimpleEdge( sourceId, "test2", targetId1, UUIDGenerator.newTimeUUID() );
+ Edge test2TargetEdge = createEdge( sourceId, "test2", targetId1, UUIDGenerator.newTimeUUID() );
em.writeEdge( test2TargetEdge ).toBlockingObservable().singleOrDefault( null );
@@ -714,18 +712,6 @@ public class EdgeManagerIT {
assertEquals( "Types correct", targetId1.getType(), results.next() );
assertFalse( "No results", results.hasNext() );
-
- //now delete our edges, we shouldn't get anything back
- em.deleteEdge( testTargetEdge );
- em.deleteEdge( testTarget2Edge );
- em.deleteEdge( test2TargetEdge );
-
-
- edges = em.getEdgeTypesToTarget( new SimpleSearchEdgeType( testTargetEdge.getSourceNode(), null ) );
-
- results = edges.toBlockingObservable().getIterator();
-
- assertFalse( "No results", results.hasNext() );
}
@@ -739,16 +725,16 @@ public class EdgeManagerIT {
Id targetId1 = new SimpleId( "target" );
- Edge testTargetEdge = new SimpleEdge( sourceId1, "test", targetId1, UUIDGenerator.newTimeUUID() );
+ Edge testTargetEdge = createEdge( sourceId1, "test", targetId1, UUIDGenerator.newTimeUUID() );
em.writeEdge( testTargetEdge ).toBlockingObservable().singleOrDefault( null );
- Edge testTarget2Edge = new SimpleEdge( sourceId2, "test", targetId1, UUIDGenerator.newTimeUUID() );
+ Edge testTarget2Edge = createEdge( sourceId2, "test", targetId1, UUIDGenerator.newTimeUUID() );
em.writeEdge( testTarget2Edge ).toBlockingObservable().singleOrDefault( null );
- Edge test2TargetEdge = new SimpleEdge( sourceId1, "test2", targetId1, UUIDGenerator.newTimeUUID() );
+ Edge test2TargetEdge = createEdge( sourceId1, "test2", targetId1, UUIDGenerator.newTimeUUID() );
em.writeEdge( test2TargetEdge ).toBlockingObservable().singleOrDefault( null );
@@ -793,16 +779,6 @@ public class EdgeManagerIT {
assertFalse( "No more edges", results.hasNext() );
- em.deleteEdge( testTargetEdge );
- em.deleteEdge( testTarget2Edge );
- em.deleteEdge( test2TargetEdge );
-
-
- edges = em.getEdgeTypesFromSource( new SimpleSearchEdgeType( testTargetEdge.getSourceNode(), null ) );
-
- results = edges.toBlockingObservable().getIterator();
-
- assertEquals( "No results", results.hasNext() );
}
@@ -816,17 +792,17 @@ public class EdgeManagerIT {
Id targetId2 = new SimpleId( "target2" );
- Edge testTargetEdge = new SimpleEdge( sourceId1, "test", targetId1, UUIDGenerator.newTimeUUID() );
+ Edge testTargetEdge = createEdge( sourceId1, "test", targetId1, UUIDGenerator.newTimeUUID() );
em.writeEdge( testTargetEdge ).toBlockingObservable().singleOrDefault( null );
- Edge testTargetEdge2 = new SimpleEdge( sourceId1, "test", targetId2, UUIDGenerator.newTimeUUID() );
+ Edge testTargetEdge2 = createEdge( sourceId1, "test", targetId2, UUIDGenerator.newTimeUUID() );
em.writeEdge( testTargetEdge2 ).toBlockingObservable().singleOrDefault( null );
- Edge test2TargetEdge = new SimpleEdge( sourceId1, "test2", targetId2, UUIDGenerator.newTimeUUID() );
+ Edge test2TargetEdge = createEdge( sourceId1, "test2", targetId2, UUIDGenerator.newTimeUUID() );
em.writeEdge( test2TargetEdge ).toBlockingObservable().singleOrDefault( null );
@@ -894,16 +870,16 @@ public class EdgeManagerIT {
Id targetId = new SimpleId( "target" );
- Edge testTargetEdge = new SimpleEdge( sourceId1, "test", targetId, UUIDGenerator.newTimeUUID() );
+ Edge testTargetEdge = createEdge( sourceId1, "test", targetId, UUIDGenerator.newTimeUUID() );
em.writeEdge( testTargetEdge ).toBlockingObservable().singleOrDefault( null );
- Edge testTargetEdge2 = new SimpleEdge( sourceId2, "test", targetId, UUIDGenerator.newTimeUUID() );
+ Edge testTargetEdge2 = createEdge( sourceId2, "test", targetId, UUIDGenerator.newTimeUUID() );
em.writeEdge( testTargetEdge2 ).toBlockingObservable().singleOrDefault( null );
- Edge test2TargetEdge = new SimpleEdge( sourceId2, "test2", targetId, UUIDGenerator.newTimeUUID() );
+ Edge test2TargetEdge = createEdge( sourceId2, "test2", targetId, UUIDGenerator.newTimeUUID() );
em.writeEdge( test2TargetEdge ).toBlockingObservable().singleOrDefault( null );
@@ -965,6 +941,300 @@ public class EdgeManagerIT {
}
+ @Test
+ public void testMarkSourceEdges() {
+
+ final EdgeManager em = emf.createEdgeManager( scope );
+
+ Id sourceId = new SimpleId( "source" );
+ Id targetId1 = new SimpleId( "target" );
+ Id targetId2 = new SimpleId( "target2" );
+
+ Edge edge1 = createEdge( sourceId, "test", targetId1, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( edge1 ).toBlockingObservable().singleOrDefault( null );
+
+ Edge edge2 = createEdge( sourceId, "test", targetId2, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( edge2 ).toBlockingObservable().singleOrDefault( null );
+
+
+ final UUID maxVersion = UUIDGenerator.newTimeUUID();
+
+
+ //get our 2 edges
+ Observable<Edge> edges = em.loadEdgesFromSource(
+ createSearchByEdge( edge1.getSourceNode(), edge1.getType(), maxVersion, null ) );
+
+
+ Iterator<Edge> results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", edge1, results.next() );
+
+ assertEquals( "Edges correct", edge2, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+
+ em.deleteEdge( edge1 ).toBlockingObservable().last();
+
+
+ edges = em.loadEdgesFromSource(
+ createSearchByEdge( edge1.getSourceNode(), edge1.getType(), maxVersion, null ) );
+
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", edge2, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+
+ em.deleteEdge( edge2 ).toBlockingObservable().last();
+
+ edges = em.loadEdgesFromSource(
+ createSearchByEdge( edge1.getSourceNode(), edge1.getType(), maxVersion, null ) );
+
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+
+ }
+
+
+ @Test
+ public void testMarkTargetEdges() {
+
+ final EdgeManager em = emf.createEdgeManager( scope );
+
+ Id sourceId1 = new SimpleId( "source" );
+ Id sourceId2 = new SimpleId( "source2" );
+ Id targetId = new SimpleId( "target" );
+
+ Edge edge1 = createEdge( sourceId1, "test", targetId, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( edge1 ).toBlockingObservable().last();
+
+ Edge edge2 = createEdge( sourceId2, "test", targetId, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( edge2 ).toBlockingObservable().last();
+
+
+ final UUID maxVersion = UUIDGenerator.newTimeUUID();
+
+
+ //get our 2 edges
+ Observable<Edge> edges = em.loadEdgesToTarget(
+ createSearchByEdge( edge1.getTargetNode(), edge1.getType(), maxVersion, null ) );
+
+
+ Iterator<Edge> results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", edge1, results.next() );
+
+ assertEquals( "Edges correct", edge2, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+
+ em.deleteEdge( edge1 ).toBlockingObservable().last();
+
+
+ edges = em.loadEdgesToTarget(
+ createSearchByEdge( edge1.getTargetNode(), edge1.getType(), maxVersion, null ) );
+
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", edge2, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+
+ em.deleteEdge( edge2 ).toBlockingObservable().last();
+
+ edges = em.loadEdgesToTarget(
+ createSearchByEdge( edge1.getTargetNode(), edge1.getType(), maxVersion, null ) );
+
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+
+ }
+
+
+ @Test
+ public void testMarkSourceEdgesType() {
+
+ final EdgeManager em = emf.createEdgeManager( scope );
+
+ Id sourceId = new SimpleId( "source" );
+ Id targetId1 = new SimpleId( "target" );
+ Id targetId2 = new SimpleId( "target2" );
+
+ Edge edge1 = createEdge( sourceId, "test", targetId1, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( edge1 ).toBlockingObservable().singleOrDefault( null );
+
+ Edge edge2 = createEdge( sourceId, "test", targetId2, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( edge2 ).toBlockingObservable().singleOrDefault( null );
+
+
+ final UUID maxVersion = UUIDGenerator.newTimeUUID();
+
+
+ //get our 2 edges
+ Observable<Edge> edges = em.loadEdgesFromSourceByType(
+ createSearchByEdgeAndId( sourceId, edge1.getType(), maxVersion, targetId1.getType(), null ) );
+
+
+ Iterator<Edge> results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", edge1, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+
+ em.deleteEdge( edge1 ).toBlockingObservable().last();
+
+
+ edges = em.loadEdgesFromSourceByType(
+ createSearchByEdgeAndId( sourceId, edge1.getType(), maxVersion, targetId1.getType(), null ) );
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ edges = em.loadEdgesFromSourceByType(
+ createSearchByEdgeAndId( sourceId, edge1.getType(), maxVersion, targetId2.getType(), null ) );
+
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", edge2, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+
+ em.deleteEdge( edge2 ).toBlockingObservable().last();
+
+
+ edges = em.loadEdgesFromSourceByType(
+ createSearchByEdgeAndId( sourceId, edge1.getType(), maxVersion, targetId2.getType(), null ) );
+
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ //now delete one of the edges
+
+ }
+
+
+ @Test
+ public void testMarkTargetEdgesType() {
+
+ final EdgeManager em = emf.createEdgeManager( scope );
+
+ Id sourceId1 = new SimpleId( "source" );
+ Id sourceId2 = new SimpleId( "source2" );
+ Id targetId = new SimpleId( "target" );
+
+ Edge edge1 = createEdge( sourceId1, "test", targetId, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( edge1 ).toBlockingObservable().last();
+
+ Edge edge2 = createEdge( sourceId2, "test", targetId, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( edge2 ).toBlockingObservable().last();
+
+
+ final UUID maxVersion = UUIDGenerator.newTimeUUID();
+
+ //get our 2 edges
+ Observable<Edge> edges = em.loadEdgesToTargetByType(
+ createSearchByEdgeAndId( targetId, edge1.getType(), maxVersion, sourceId1.getType(), null ) );
+
+
+ Iterator<Edge> results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", edge1, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+
+ em.deleteEdge( edge1 ).toBlockingObservable().last();
+
+
+ edges = em.loadEdgesToTargetByType(
+ createSearchByEdgeAndId( edge1.getSourceNode(), edge1.getType(), maxVersion, sourceId1.getType(),
+ null ) );
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ edges = em.loadEdgesToTargetByType(
+ createSearchByEdgeAndId( targetId, edge1.getType(), maxVersion, sourceId2.getType(), null ) );
+
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", edge2, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+
+ em.deleteEdge( edge2 ).toBlockingObservable().last();
+
+
+ edges = em.loadEdgesToTargetByType(
+ createSearchByEdgeAndId( targetId, edge1.getType(), maxVersion, sourceId2.getType(), null ) );
+
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ //now delete one of the edges
+
+ }
+
+
@Test( expected = NullPointerException.class )
public void invalidEdgeTypesWrite( @All Edge edge ) {
final EdgeManager em = emf.createEdgeManager( scope );
@@ -980,53 +1250,53 @@ public class EdgeManagerIT {
em.deleteEdge( edge );
}
-
- public static class InvalidInput extends JukitoModule {
-
- @Override
- protected void configureTest() {
- //create all edge types of junk input
- //
- // final UUID version = UUIDGenerator.newTimeUUID();
- //
- // Id nullUuid = mock( Id.class );
- // when( nullUuid.getUuid() ).thenReturn( null );
- //
- //
- // Id nullType = mock( Id.class );
- // when( nullType.getType() ).thenReturn( "type" );
- //
- // Edge[] edges = new Edge[] {
- // mockEdge( nullUuid, "test", createId( "target" ), version ),
- //
- // mockEdge( nullType, "test", createId( "target" ), version ),
- //
- // mockEdge( createId( "source" ), null, createId( "target" ), version ),
- //
- // mockEdge( createId( "source" ), "test", nullUuid, version ),
- //
- // mockEdge( createId( "source" ), "test", nullType, version ),
- //
- // mockEdge( createId( "source" ), "test", createId( "target" ), null )
- // };
- //
- //
- // bindManyInstances( Edge.class, edges );
-
- }
-
-
- private Edge mockEdge( final Id sourceId, final String type, final Id targetId, final UUID version ) {
- Edge edge = mock( Edge.class );
-
- when( edge.getSourceNode() ).thenReturn( sourceId );
- when( edge.getType() ).thenReturn( type );
- when( edge.getTargetNode() ).thenReturn( targetId );
- when( edge.getVersion() ).thenReturn( version );
-
- return edge;
- }
- }
+ //
+ // public static class InvalidInput extends JukitoModule {
+ //
+ // @Override
+ // protected void configureTest() {
+ //create all edge types of junk input
+ //
+ // final UUID version = UUIDGenerator.newTimeUUID();
+ //
+ // Id nullUuid = mock( Id.class );
+ // when( nullUuid.getUuid() ).thenReturn( null );
+ //
+ //
+ // Id nullType = mock( Id.class );
+ // when( nullType.getType() ).thenReturn( "type" );
+ //
+ // Edge[] edges = new Edge[] {
+ // mockEdge( nullUuid, "test", createId( "target" ), version ),
+ //
+ // mockEdge( nullType, "test", createId( "target" ), version ),
+ //
+ // mockEdge( createId( "source" ), null, createId( "target" ), version ),
+ //
+ // mockEdge( createId( "source" ), "test", nullUuid, version ),
+ //
+ // mockEdge( createId( "source" ), "test", nullType, version ),
+ //
+ // mockEdge( createId( "source" ), "test", createId( "target" ), null )
+ // };
+ //
+ //
+ // bindManyInstances( Edge.class, edges );
+ //
+ // }
+ //
+ //
+ // private Edge mockEdge( final Id sourceId, final String type, final Id targetId, final UUID version ) {
+ // Edge edge = mock( Edge.class );
+ //
+ // when( edge.getSourceNode() ).thenReturn( sourceId );
+ // when( edge.getType() ).thenReturn( type );
+ // when( edge.getTargetNode() ).thenReturn( targetId );
+ // when( edge.getVersion() ).thenReturn( version );
+ //
+ // return edge;
+ // }
+ // }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerializationChopTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerializationChopTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerializationChopTest.java
index 40db841..7d97f0e 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerializationChopTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerializationChopTest.java
@@ -17,6 +17,7 @@ import org.apache.usergrid.persistence.collection.OrganizationScope;
import org.apache.usergrid.persistence.collection.cassandra.CassandraRule;
import org.apache.usergrid.persistence.collection.guice.MigrationManagerRule;
import org.apache.usergrid.persistence.graph.Edge;
+import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.SearchByEdge;
import org.apache.usergrid.persistence.graph.guice.GraphModule;
import org.apache.usergrid.persistence.graph.guice.TestGraphModule;
@@ -109,7 +110,8 @@ public class EdgeSerializationChopTest {
//get our edges out by name
- Iterator<Edge> results = serialization.getEdgesFromSource( scope, createSearchByEdge( sourceId, "edge", now, null ) );
+ Iterator<MarkedEdge>
+ results = serialization.getEdgesFromSource( scope, createSearchByEdge( sourceId, "edge", now, null ) );
boolean found = false;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerializationTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerializationTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerializationTest.java
index 32f60e4..e90722a 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerializationTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerializationTest.java
@@ -16,8 +16,8 @@ import org.apache.usergrid.persistence.collection.OrganizationScope;
import org.apache.usergrid.persistence.collection.cassandra.CassandraRule;
import org.apache.usergrid.persistence.collection.guice.MigrationManagerRule;
import org.apache.usergrid.persistence.graph.Edge;
+import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.SearchByEdge;
-import org.apache.usergrid.persistence.graph.guice.GraphModule;
import org.apache.usergrid.persistence.graph.guice.TestGraphModule;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
@@ -96,7 +96,7 @@ public class EdgeSerializationTest {
//get our edges out by name
- Iterator<Edge> results =
+ Iterator<MarkedEdge> results =
serialization.getEdgesFromSource( scope, createSearchByEdge( sourceId, "edge1", now, null ) );
assertEquals( edge1, results.next() );
@@ -143,7 +143,7 @@ public class EdgeSerializationTest {
//get our edges out by name
- Iterator<Edge> results =
+ Iterator<MarkedEdge> results =
serialization.getEdgesFromSource( scope, createSearchByEdge( sourceId, "edge", now, edge2 ) );
assertEquals( edge1, results.next() );
@@ -181,8 +181,8 @@ public class EdgeSerializationTest {
final Edge edgev2 = createEdge( sourceId, "edge1", targetId );
- assertTrue( "Edge version 1 has lower time uuid", UUIDComparator
- .staticCompare( edgev1.getVersion(), edgev2.getVersion() ) < 0 );
+ assertTrue( "Edge version 1 has lower time uuid",
+ UUIDComparator.staticCompare( edgev1.getVersion(), edgev2.getVersion() ) < 0 );
//create edge type 2 to ensure we don't get it in results
final Edge edgeType2V1 = createEdge( sourceId, "edge2", targetId );
@@ -196,7 +196,7 @@ public class EdgeSerializationTest {
SearchByEdge search = createGetByEdge( sourceId, "edge1", targetId, now, null );
- Iterator<Edge> results = serialization.getEdgeFromSource( scope, search );
+ Iterator<MarkedEdge> results = serialization.getEdgeFromSource( scope, search );
assertEquals( edgev2, results.next() );
assertEquals( edgev1, results.next() );
@@ -279,7 +279,7 @@ public class EdgeSerializationTest {
//get our edges out by name
- Iterator<Edge> results = serialization.getEdgesFromSourceByTargetType( scope,
+ Iterator<MarkedEdge> results = serialization.getEdgesFromSourceByTargetType( scope,
createSearchByEdgeAndId( sourceId, "edge", now, targetId1.getType(), null ) );
assertEquals( edge1, results.next() );
@@ -330,7 +330,7 @@ public class EdgeSerializationTest {
//get our edges out by name
- Iterator<Edge> results = serialization.getEdgesFromSourceByTargetType( scope,
+ Iterator<MarkedEdge> results = serialization.getEdgesFromSourceByTargetType( scope,
createSearchByEdgeAndId( sourceId, "edge", now, targetId1.getType(), null ) );
assertEquals( edge1, results.next() );
@@ -384,7 +384,7 @@ public class EdgeSerializationTest {
UUID now = UUIDGenerator.newTimeUUID();
//get our edges out by name
- Iterator<Edge> results = serialization.getEdgesFromSourceByTargetType( scope,
+ Iterator<MarkedEdge> results = serialization.getEdgesFromSourceByTargetType( scope,
createSearchByEdgeAndId( sourceId, "edge", now, targetId1.getType(), null ) );
assertEquals( edge1, results.next() );
@@ -465,4 +465,153 @@ public class EdgeSerializationTest {
assertFalse( results.hasNext() );
}
+
+
+ /**
+ * Test paging by resuming the search from the edge
+ */
+ @Test
+ public void mark() throws ConnectionException {
+ final Edge edge1 = createEdge( "source", "edge", "target" );
+
+ final Id sourceId = edge1.getSourceNode();
+ final Id targetId1 = edge1.getTargetNode();
+
+
+ final Edge edge2 = createEdge( sourceId, "edge", createId( "target" ) );
+
+ final Id targetId2 = edge2.getTargetNode();
+
+ serialization.writeEdge( scope, edge1 ).execute();
+ serialization.writeEdge( scope, edge2 ).execute();
+
+
+ UUID now = UUIDGenerator.newTimeUUID();
+
+ //get our edges out by name
+ Iterator<MarkedEdge> results = serialization.getEdgesFromSourceByTargetType( scope,
+ createSearchByEdgeAndId( sourceId, "edge", now, targetId1.getType(), null ) );
+
+ assertEquals( edge1, results.next() );
+ assertEquals( edge2, results.next() );
+ assertFalse( results.hasNext() );
+
+ //get them out by type
+ results = serialization.getEdgesFromSource( scope, createSearchByEdge( sourceId, "edge", now, null ) );
+
+ assertEquals( edge1, results.next() );
+ assertEquals( edge2, results.next() );
+ assertFalse( results.hasNext() );
+
+
+ //validate we get from target
+ results = serialization.getEdgesToTargetBySourceType( scope,
+ createSearchByEdgeAndId( targetId1, "edge", now, sourceId.getType(), null ) );
+
+ assertEquals( edge1, results.next() );
+ assertFalse( results.hasNext() );
+
+ results = serialization.getEdgesToTargetBySourceType( scope,
+ createSearchByEdgeAndId( targetId2, "edge", now, sourceId.getType(), null ) );
+
+ assertEquals( edge2, results.next() );
+ assertFalse( results.hasNext() );
+
+
+ //validate we get from target
+ results = serialization.getEdgesToTarget( scope, createSearchByEdge( targetId1, "edge", now, null ) );
+
+ assertEquals( edge1, results.next() );
+ assertFalse( results.hasNext() );
+
+
+ results = serialization.getEdgesToTarget( scope, createSearchByEdge( targetId2, "edge", now, null ) );
+
+ assertEquals( edge2, results.next() );
+ assertFalse( results.hasNext() );
+
+ //now we've validated everything exists, lets blitz the data and ensure it's removed
+
+ serialization.markEdge( scope, edge1 ).execute();
+ serialization.markEdge( scope, edge2 ).execute();
+
+
+ results = serialization.getEdgesFromSourceByTargetType( scope,
+ createSearchByEdgeAndId( sourceId, "edge", now, targetId1.getType(), null ) );
+
+
+ MarkedEdge edge = results.next();
+
+ assertEquals( edge1, edge );
+ assertTrue(edge.isDeleted());
+
+
+ edge = results.next();
+
+ assertEquals( edge2, edge );
+ assertTrue(edge.isDeleted());
+
+ assertFalse( results.hasNext() );
+
+ //get them out by type
+ results = serialization.getEdgesFromSource( scope, createSearchByEdge( sourceId, "edge", now, null ) );
+
+ edge = results.next();
+
+ assertEquals( edge1, edge );
+ assertTrue(edge.isDeleted());
+
+ edge = results.next();
+
+ assertEquals( edge2, edge );
+ assertTrue(edge.isDeleted());
+
+ assertFalse( results.hasNext() );
+
+
+ //validate we get from target
+ results = serialization.getEdgesToTargetBySourceType( scope,
+ createSearchByEdgeAndId( targetId1, "edge", now, sourceId.getType(), null ) );
+
+ edge = results.next();
+
+ assertEquals( edge1, edge );
+ assertTrue(edge.isDeleted());
+
+ assertFalse( results.hasNext() );
+
+ results = serialization.getEdgesToTargetBySourceType( scope,
+ createSearchByEdgeAndId( targetId2, "edge", now, sourceId.getType(), null ) );
+
+ edge = results.next();
+
+ assertEquals( edge2, edge );
+ assertTrue(edge.isDeleted());
+
+ assertFalse( results.hasNext() );
+
+
+ //validate we get from target
+ results = serialization.getEdgesToTarget( scope, createSearchByEdge( targetId1, "edge", now, null ) );
+
+ edge = results.next();
+
+ assertEquals( edge1, edge );
+ assertTrue(edge.isDeleted());
+
+ assertFalse( results.hasNext() );
+
+
+ results = serialization.getEdgesToTarget( scope, createSearchByEdge( targetId2, "edge", now, null ) );
+
+ edge = results.next();
+
+ assertEquals( edge2, edge );
+ assertTrue(edge.isDeleted());
+
+ assertFalse( results.hasNext() );
+
+ //now we've validated everything exists
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/4e5b3b25/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/test/util/EdgeTestUtils.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/test/util/EdgeTestUtils.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/test/util/EdgeTestUtils.java
index 012c8b8..e9f05e0 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/test/util/EdgeTestUtils.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/test/util/EdgeTestUtils.java
@@ -27,7 +27,7 @@ import org.apache.usergrid.persistence.graph.SearchByEdge;
import org.apache.usergrid.persistence.graph.SearchByEdgeType;
import org.apache.usergrid.persistence.graph.SearchByIdType;
import org.apache.usergrid.persistence.graph.SearchEdgeType;
-import org.apache.usergrid.persistence.graph.impl.SimpleEdge;
+import org.apache.usergrid.persistence.graph.impl.SimpleMarkedEdge;
import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdge;
import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdgeType;
import org.apache.usergrid.persistence.graph.impl.SimpleSearchByIdType;
@@ -70,7 +70,7 @@ public class EdgeTestUtils {
* Create an edge with the specified params
*/
public static Edge createEdge( final Id sourceId, final String edgeType, final Id targetId, final UUID version ) {
- return new SimpleEdge( sourceId, edgeType, targetId, version );
+ return new SimpleMarkedEdge( sourceId, edgeType, targetId, version, false );
}