You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sn...@apache.org on 2014/03/26 22:20:46 UTC
[02/35] git commit: Added hystrix integration. Not working as expected
Added hystrix integration. Not working as expected
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/5cadb9f5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/5cadb9f5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/5cadb9f5
Branch: refs/heads/two-dot-o
Commit: 5cadb9f58b9936580373c1db5fe72ce800a7d017
Parents: 803096e
Author: Todd Nine <tn...@apigee.com>
Authored: Wed Mar 19 14:52:28 2014 -0700
Committer: Todd Nine <tn...@apigee.com>
Committed: Wed Mar 19 14:52:28 2014 -0700
----------------------------------------------------------------------
stack/corepersistence/collection/pom.xml | 6 +
.../usergrid/persistence/graph/GraphFig.java | 42 +-
.../graph/consistency/TimeoutTask.java | 2 +-
.../graph/impl/EdgeDeleteListener.java | 6 +-
.../persistence/graph/impl/EdgeManagerImpl.java | 18 +-
.../graph/impl/EdgeWriteListener.java | 2 +-
.../graph/impl/NodeDeleteListener.java | 22 +-
.../graph/impl/stage/AbstractEdgeRepair.java | 10 +-
.../graph/impl/stage/EdgeMetaRepairImpl.java | 9 +-
.../impl/parse/ObservableIterator.java | 48 +-
.../persistence/graph/EdgeManagerTimeoutIT.java | 1491 ++++++++++++++++++
stack/corepersistence/pom.xml | 1 +
12 files changed, 1594 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5cadb9f5/stack/corepersistence/collection/pom.xml
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/pom.xml b/stack/corepersistence/collection/pom.xml
index 7b34b0b..808347e 100644
--- a/stack/corepersistence/collection/pom.xml
+++ b/stack/corepersistence/collection/pom.xml
@@ -203,5 +203,11 @@
<version>${log4j.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.netflix.hystrix</groupId>
+ <artifactId>hystrix-core</artifactId>
+ <version>${hystrix.version}</version>
+ </dependency>
+
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5cadb9f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/GraphFig.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/GraphFig.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/GraphFig.java
index 64ca397..de7f72f 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/GraphFig.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/GraphFig.java
@@ -17,6 +17,8 @@ public interface GraphFig extends GuicyFig {
public static final String REPAIR_CONCURRENT_SIZE = "usergrid.graph.repair.concurrent.size";
+ public static final String REPAIR_TIMEOUT = "usergrid.graph.repair.timeout";
+
public static final String TIMEOUT_SIZE = "usergrid.graph.timeout.page.size";
@@ -26,37 +28,47 @@ public interface GraphFig extends GuicyFig {
public static final String WRITE_CL = "usergrid.graph.write.cl";
- public static final String WRITE_TIMEOUT = "usergrid.graph.write.timeout";
+ public static final String WRITE_TIMEOUT = "usergrid.graph.write.timeout";
+
+ public static final String READ_TIMEOUT = "usergrid.graph.read.timeout";
- @Default( "1000" )
- @Key( SCAN_PAGE_SIZE )
+ @Default("1000")
+ @Key(SCAN_PAGE_SIZE)
int getScanPageSize();
- @Default( "CL_ONE" )
- @Key( READ_CL )
+ @Default("CL_ONE")
+ @Key(READ_CL)
String getReadCL();
- @Default( "CL_QUORUM" )
- @Key( WRITE_CL )
+ @Default("CL_QUORUM")
+ @Key(WRITE_CL)
String getWriteCL();
- @Default("10000")
- @Key( WRITE_TIMEOUT )
- long getWriteTimeout();
+// @Default("10000")
+// @Key(WRITE_TIMEOUT)
+// int getWriteTimeout();
+
+ /**
+ * Get the read timeout (in milliseconds) that we should allow when reading from the data source
+ */
+ @Default( "10000" )
+ @Key( READ_TIMEOUT )
+ int getReadTimeout();
@Default( "100" )
@Key( TIMEOUT_SIZE )
int getTimeoutReadSize();
- @Default("500")
+ @Default( "500" )
@Key( TIMEOUT_TASK_TIME )
long getTaskLoopTime();
- @Default("10")
- @Key(REPAIR_CONCURRENT_SIZE)
+ @Default( "10" )
+ @Key( REPAIR_CONCURRENT_SIZE )
int getRepairConcurrentSize();
-
-
+ @Default("10000")
+ @Key(WRITE_TIMEOUT)
+ int getRepairTimeout();
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5cadb9f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/TimeoutTask.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/TimeoutTask.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/TimeoutTask.java
index fbfad3e..7555ce2 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/TimeoutTask.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/TimeoutTask.java
@@ -56,6 +56,6 @@ public class TimeoutTask<T> implements Action1<Scheduler.Inner> {
* @return
*/
private Iterator<AsynchronousMessage<T>> getTimeouts() {
- return processor.getTimeouts( graphFig.getTimeoutReadSize(), graphFig.getWriteTimeout() * 2 ).iterator();
+ return processor.getTimeouts( graphFig.getTimeoutReadSize(), graphFig.getRepairTimeout() * 2 ).iterator();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5cadb9f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeDeleteListener.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeDeleteListener.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeDeleteListener.java
index e0ceafc..d377882 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeDeleteListener.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeDeleteListener.java
@@ -8,6 +8,7 @@ import org.apache.usergrid.persistence.collection.OrganizationScope;
import org.apache.usergrid.persistence.graph.Edge;
import org.apache.usergrid.persistence.graph.EdgeManager;
import org.apache.usergrid.persistence.graph.EdgeManagerFactory;
+import org.apache.usergrid.persistence.graph.GraphFig;
import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.consistency.AsyncProcessor;
import org.apache.usergrid.persistence.graph.consistency.MessageListener;
@@ -39,17 +40,20 @@ public class EdgeDeleteListener implements MessageListener<EdgeEvent<Edge>, Edge
private final EdgeMetadataSerialization edgeMetadataSerialization;
private final EdgeManagerFactory edgeManagerFactory;
private final Keyspace keyspace;
+ private final GraphFig graphFig;
@Inject
public EdgeDeleteListener( final EdgeSerialization edgeSerialization,
final EdgeMetadataSerialization edgeMetadataSerialization,
final EdgeManagerFactory edgeManagerFactory, final Keyspace keyspace,
+ final GraphFig graphFig,
@EdgeDelete final AsyncProcessor edgeDelete ) {
this.edgeSerialization = edgeSerialization;
this.edgeMetadataSerialization = edgeMetadataSerialization;
this.edgeManagerFactory = edgeManagerFactory;
this.keyspace = keyspace;
+ this.graphFig = graphFig;
edgeDelete.addListener( this );
}
@@ -72,7 +76,7 @@ public class EdgeDeleteListener implements MessageListener<EdgeEvent<Edge>, Edge
//go through every version of this edge <= the current version and remove it
- Observable<MarkedEdge> edges = Observable.create( new ObservableIterator<MarkedEdge>() {
+ Observable<MarkedEdge> edges = Observable.create( new ObservableIterator<MarkedEdge>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgeToTarget( scope,
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5cadb9f5/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 fe23222..9e529c5 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
@@ -201,7 +201,7 @@ public class EdgeManagerImpl implements EdgeManager {
@Override
public Observable<Edge> loadEdgesFromSource( final SearchByEdgeType search ) {
- return Observable.create( new ObservableIterator<MarkedEdge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesFromSource( scope, search );
@@ -220,7 +220,7 @@ public class EdgeManagerImpl implements EdgeManager {
@Override
public Observable<Edge> loadEdgesToTarget( final SearchByEdgeType search ) {
- return Observable.create( new ObservableIterator<MarkedEdge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesToTarget( scope, search );
@@ -240,7 +240,7 @@ public class EdgeManagerImpl implements EdgeManager {
@Override
public Observable<Edge> loadEdgesFromSourceByType( final SearchByIdType search ) {
- return Observable.create( new ObservableIterator<MarkedEdge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesFromSourceByTargetType( scope, search );
@@ -258,7 +258,7 @@ public class EdgeManagerImpl implements EdgeManager {
@Override
public Observable<Edge> loadEdgesToTargetByType( final SearchByIdType search ) {
- return Observable.create( new ObservableIterator<MarkedEdge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesToTargetBySourceType( scope, search );
@@ -276,7 +276,7 @@ public class EdgeManagerImpl implements EdgeManager {
@Override
public Observable<String> getEdgeTypesFromSource( final SearchEdgeType search ) {
- return Observable.create( new ObservableIterator<String>() {
+ return Observable.create( new ObservableIterator<String>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<String> getIterator() {
return edgeMetadataSerialization.getEdgeTypesFromSource( scope, search );
@@ -287,7 +287,7 @@ public class EdgeManagerImpl implements EdgeManager {
@Override
public Observable<String> getIdTypesFromSource( final SearchIdType search ) {
- return Observable.create( new ObservableIterator<String>() {
+ return Observable.create( new ObservableIterator<String>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<String> getIterator() {
return edgeMetadataSerialization.getIdTypesFromSource( scope, search );
@@ -299,7 +299,7 @@ public class EdgeManagerImpl implements EdgeManager {
@Override
public Observable<String> getEdgeTypesToTarget( final SearchEdgeType search ) {
- return Observable.create( new ObservableIterator<String>() {
+ return Observable.create( new ObservableIterator<String>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<String> getIterator() {
return edgeMetadataSerialization.getEdgeTypesToTarget( scope, search );
@@ -310,7 +310,7 @@ public class EdgeManagerImpl implements EdgeManager {
@Override
public Observable<String> getIdTypesToTarget( final SearchIdType search ) {
- return Observable.create( new ObservableIterator<String>() {
+ return Observable.create( new ObservableIterator<String>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<String> getIterator() {
return edgeMetadataSerialization.getIdTypesToTarget( scope, search );
@@ -323,7 +323,7 @@ public class EdgeManagerImpl implements EdgeManager {
* Get our timeout for write consistency
*/
private long getTimeout() {
- return graphFig.getWriteTimeout() * 2;
+ return graphFig.getRepairTimeout() * 2;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5cadb9f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeWriteListener.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeWriteListener.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeWriteListener.java
index 271375a..ed8e861 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeWriteListener.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeWriteListener.java
@@ -52,7 +52,7 @@ public class EdgeWriteListener implements MessageListener<EdgeEvent<Edge>, EdgeE
final OrganizationScope scope = write.getOrganizationScope();
final UUID maxVersion = edge.getVersion();
- return Observable.create( new ObservableIterator<MarkedEdge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5cadb9f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListener.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListener.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListener.java
index 9fcd3a9..631ac50 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListener.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListener.java
@@ -9,6 +9,7 @@ import org.slf4j.LoggerFactory;
import org.apache.usergrid.persistence.collection.OrganizationScope;
import org.apache.usergrid.persistence.graph.Edge;
+import org.apache.usergrid.persistence.graph.GraphFig;
import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.SearchByEdgeType;
import org.apache.usergrid.persistence.graph.SearchEdgeType;
@@ -31,7 +32,6 @@ import rx.Observable;
import rx.Scheduler;
import rx.functions.Action0;
import rx.functions.Func1;
-import rx.observables.MathObservable;
/**
@@ -47,6 +47,7 @@ public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, Intege
private final EdgeMetadataSerialization edgeMetadataSerialization;
private final EdgeDeleteRepair edgeDeleteRepair;
private final EdgeMetaRepair edgeMetaRepair;
+ private final GraphFig graphFig;
private final Scheduler scheduler;
@@ -55,11 +56,11 @@ public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, Intege
* Wire the serialization dependencies
*/
@Inject
- public NodeDeleteListener( final NodeSerialization nodeSerialization, final EdgeSerialization edgeSerialization,final Scheduler scheduler,
- @NodeDelete final AsyncProcessor nodeDelete,
- final EdgeMetadataSerialization edgeMetadataSerialization, final EdgeDeleteRepair
- edgeDeleteRepair,
- final EdgeMetaRepair edgeMetaRepair ) {
+ public NodeDeleteListener( final NodeSerialization nodeSerialization, final EdgeSerialization edgeSerialization,
+ final Scheduler scheduler, @NodeDelete final AsyncProcessor nodeDelete,
+ final EdgeMetadataSerialization edgeMetadataSerialization,
+ final EdgeDeleteRepair edgeDeleteRepair, final EdgeMetaRepair edgeMetaRepair,
+ final GraphFig graphFig ) {
this.nodeSerialization = nodeSerialization;
@@ -68,6 +69,7 @@ public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, Intege
this.edgeMetadataSerialization = edgeMetadataSerialization;
this.edgeDeleteRepair = edgeDeleteRepair;
this.edgeMetaRepair = edgeMetaRepair;
+ this.graphFig = graphFig;
nodeDelete.addListener( this );
}
@@ -208,7 +210,7 @@ public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, Intege
*/
private Observable<String> getEdgesTypesToTarget( final OrganizationScope scope, final SearchEdgeType search ) {
- return Observable.create( new ObservableIterator<String>() {
+ return Observable.create( new ObservableIterator<String>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<String> getIterator() {
return edgeMetadataSerialization.getEdgeTypesToTarget( scope, search );
@@ -222,7 +224,7 @@ public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, Intege
*/
private Observable<String> getEdgesTypesFromSource( final OrganizationScope scope, final SearchEdgeType search ) {
- return Observable.create( new ObservableIterator<String>() {
+ return Observable.create( new ObservableIterator<String>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<String> getIterator() {
return edgeMetadataSerialization.getEdgeTypesFromSource( scope, search );
@@ -236,7 +238,7 @@ public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, Intege
*/
private Observable<MarkedEdge> loadEdgesToTarget( final OrganizationScope scope, final SearchByEdgeType search ) {
- return Observable.create( new ObservableIterator<MarkedEdge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesToTarget( scope, search );
@@ -250,7 +252,7 @@ public class NodeDeleteListener implements MessageListener<EdgeEvent<Id>, Intege
*/
private Observable<MarkedEdge> loadEdgesFromSource( final OrganizationScope scope, final SearchByEdgeType search ) {
- return Observable.create( new ObservableIterator<MarkedEdge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesFromSource( scope, search );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5cadb9f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/AbstractEdgeRepair.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/AbstractEdgeRepair.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/AbstractEdgeRepair.java
index 28d0622..e09d884 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/AbstractEdgeRepair.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/AbstractEdgeRepair.java
@@ -20,10 +20,8 @@
package org.apache.usergrid.persistence.graph.impl.stage;
-import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
-import java.util.PriorityQueue;
import java.util.UUID;
import org.slf4j.Logger;
@@ -36,10 +34,7 @@ import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdge;
import org.apache.usergrid.persistence.graph.serialization.EdgeSerialization;
import org.apache.usergrid.persistence.graph.serialization.impl.parse.ObservableIterator;
-import org.apache.usergrid.persistence.model.entity.Id;
-import com.fasterxml.uuid.UUIDComparator;
-import com.fasterxml.uuid.impl.UUIDUtil;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.netflix.astyanax.Keyspace;
@@ -49,7 +44,6 @@ import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import rx.Observable;
import rx.Scheduler;
import rx.functions.Func1;
-import rx.functions.Func2;
/**
@@ -131,7 +125,7 @@ public abstract class AbstractEdgeRepair {
*/
private Observable<MarkedEdge> getEdgeVersionsFromSource( final OrganizationScope scope, final Edge edge ) {
- return Observable.create( new ObservableIterator<MarkedEdge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
@@ -148,7 +142,7 @@ public abstract class AbstractEdgeRepair {
*/
private Observable<MarkedEdge> getEdgeVersionsToTarget( final OrganizationScope scope, final Edge edge ) {
- return Observable.create( new ObservableIterator<MarkedEdge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5cadb9f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java
index 5c497a5..c54f485 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeMetaRepairImpl.java
@@ -48,7 +48,6 @@ import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import rx.Observable;
import rx.Scheduler;
-import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.observables.MathObservable;
@@ -252,7 +251,7 @@ public class EdgeMetaRepairImpl implements EdgeMetaRepair {
@Override
public Observable<String> loadEdgeSubTypes( final OrganizationScope scope, final Id nodeId,
final String edgeType, final UUID version ) {
- return Observable.create( new ObservableIterator<String>() {
+ return Observable.create( new ObservableIterator<String>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<String> getIterator() {
return edgeMetadataSerialization
@@ -265,7 +264,7 @@ public class EdgeMetaRepairImpl implements EdgeMetaRepair {
@Override
public Observable<MarkedEdge> loadEdges( final OrganizationScope scope, final Id nodeId, final String edgeType,
final String subType, final UUID version ) {
- return Observable.create( new ObservableIterator<MarkedEdge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesToTargetBySourceType( scope,
@@ -297,7 +296,7 @@ public class EdgeMetaRepairImpl implements EdgeMetaRepair {
@Override
public Observable<String> loadEdgeSubTypes( final OrganizationScope scope, final Id nodeId,
final String edgeType, final UUID version ) {
- return Observable.create( new ObservableIterator<String>() {
+ return Observable.create( new ObservableIterator<String>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<String> getIterator() {
return edgeMetadataSerialization
@@ -310,7 +309,7 @@ public class EdgeMetaRepairImpl implements EdgeMetaRepair {
@Override
public Observable<MarkedEdge> loadEdges( final OrganizationScope scope, final Id nodeId, final String edgeType,
final String subType, final UUID version ) {
- return Observable.create( new ObservableIterator<MarkedEdge>() {
+ return Observable.create( new ObservableIterator<MarkedEdge>( graphFig.getReadTimeout() ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesFromSourceByTargetType( scope,
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5cadb9f5/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 2274868..c70dc4c 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
@@ -3,43 +3,65 @@ package org.apache.usergrid.persistence.graph.serialization.impl.parse;
import java.util.Iterator;
+import com.netflix.hystrix.HystrixCommand;
+import com.netflix.hystrix.HystrixCommandGroupKey;
+import com.netflix.hystrix.HystrixCommandProperties;
+
import rx.Observable;
-import rx.Observer;
import rx.Subscriber;
-import rx.Subscription;
-import rx.subscriptions.Subscriptions;
/**
* Converts an iterator to an observable. Subclasses need to only implement getting the iterator from the data source.
- * This is used in favor of "Observable.just" when the initial fetch of the iterator will require I/O. This allows
- * us to wrap the iterator in a deferred invocation to avoid the blocking on construction.
+ * This is used in favor of "Observable.just" when the initial fetch of the iterator will require I/O. This allows us
+ * to wrap the iterator in a deferred invocation to avoid the blocking on construction.
*/
public abstract class ObservableIterator<T> implements Observable.OnSubscribe<T> {
+ private static final HystrixCommandGroupKey GROUP_KEY = HystrixCommandGroupKey.Factory.asKey( "CassRead" );
+
+ private final int executionTimeout;
+
+
+ protected ObservableIterator( final int executionTimeout ) {
+ this.executionTimeout = executionTimeout;
+ }
+
@Override
public void call( final Subscriber<? super T> subscriber ) {
try {
- //get our iterator and push data to the observer
- Iterator<T> itr = getIterator();
+ //run producing the values within a hystrix command. This way we'll time out if the read takes too long
+ new HystrixCommand<Void>( HystrixCommand.Setter.withGroupKey( GROUP_KEY ).andCommandPropertiesDefaults(
+ HystrixCommandProperties.Setter()
+ .withExecutionIsolationThreadTimeoutInMilliseconds( executionTimeout ) ) ) {
- //while we have items to emit and our subscriber is subscribed, we want to keep emitting items
- while ( itr.hasNext() && !subscriber.isUnsubscribed()) {
- subscriber.onNext( itr.next() );
- }
+ @Override
+ protected Void run() throws Exception {
+ //get our iterator and push data to the observer
+ final Iterator<T> itr = getIterator();
- subscriber.onCompleted();
+
+ //while we have items to emit and our subscriber is subscribed, we want to keep emitting items
+ while ( itr.hasNext() && !subscriber.isUnsubscribed() ) {
+ subscriber.onNext( itr.next() );
+ }
+
+
+ subscriber.onCompleted();
+
+ return null;
+ }
+ }.execute();
}
//if any error occurs, we need to notify the observer so it can perform it's own error handling
catch ( Throwable t ) {
subscriber.onError( t );
}
-
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5cadb9f5/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerTimeoutIT.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerTimeoutIT.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerTimeoutIT.java
new file mode 100644
index 0000000..e4591cd
--- /dev/null
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerTimeoutIT.java
@@ -0,0 +1,1491 @@
+/*
+ * 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;
+
+
+import java.util.Iterator;
+import java.util.UUID;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeoutException;
+
+import org.jukito.All;
+import org.jukito.JukitoRunner;
+import org.jukito.UseModules;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.safehaus.guicyfig.GuicyFig;
+
+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.SimpleSearchEdgeType;
+import org.apache.usergrid.persistence.graph.impl.SimpleSearchIdType;
+import org.apache.usergrid.persistence.model.entity.Id;
+import org.apache.usergrid.persistence.model.entity.SimpleId;
+import org.apache.usergrid.persistence.model.util.UUIDGenerator;
+
+import com.google.inject.Inject;
+
+import rx.Observable;
+import rx.functions.Action1;
+
+import static org.apache.usergrid.persistence.graph.test.util.EdgeTestUtils.createEdge;
+import static org.apache.usergrid.persistence.graph.test.util.EdgeTestUtils.createId;
+import static org.apache.usergrid.persistence.graph.test.util.EdgeTestUtils.createSearchByEdge;
+import static org.apache.usergrid.persistence.graph.test.util.EdgeTestUtils.createSearchByEdgeAndId;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+@RunWith(JukitoRunner.class)
+@UseModules({ TestGraphModule.class })
+//@UseModules( { TestGraphModule.class, EdgeManagerIT.InvalidInput.class } )
+public class EdgeManagerTimeoutIT {
+
+ /**
+ * Test timeout in millis
+ */
+ private static final long TIMEOUT = 30000;
+
+ @ClassRule
+ public static CassandraRule rule = new CassandraRule();
+
+
+ @Inject
+ @Rule
+ public MigrationManagerRule migrationManagerRule;
+
+
+ @Inject
+ protected EdgeManagerFactory emf;
+
+ @Inject
+ protected GraphFig graphFig;
+
+ protected OrganizationScope scope;
+
+
+ @Before
+ public void setup() {
+ scope = mock( OrganizationScope.class );
+
+ Id orgId = mock( Id.class );
+
+ when( orgId.getType() ).thenReturn( "organization" );
+ when( orgId.getUuid() ).thenReturn( UUIDGenerator.newTimeUUID() );
+
+ when( scope.getOrganization() ).thenReturn( orgId );
+
+ if(graphFig.getReadTimeout() > TIMEOUT){
+ fail("Graph read timeout must be <= " + TIMEOUT + ". Otherwise tests are invalid");
+ }
+ }
+
+
+ @Test(timeout = TIMEOUT, expected = TimeoutException.class)
+ public void testWriteReadEdgeTypeSource() throws InterruptedException {
+
+ EdgeManager em = emf.createEdgeManager( scope );
+
+
+ Edge edge = createEdge( "source", "test", "target" );
+
+ em.writeEdge( edge ).toBlockingObservable().last();
+
+ //now test retrieving it
+
+ SearchByEdgeType search = createSearchByEdge( edge.getSourceNode(), edge.getType(), edge.getVersion(), null );
+
+ Observable<Edge> edges = em.loadEdgesFromSource( search );
+
+ //retrieve the edge, ensure that if we block indefinitely, it times out
+
+ final Semaphore blocker = new Semaphore(0);
+
+ edges.subscribe( new Action1<Edge>() {
+ @Override
+ public void call( final Edge edge ) {
+ //block indefinitely, we want to ensure we timeout
+ try {
+ blocker.acquire();
+ }
+ catch ( InterruptedException e ) {
+ throw new RuntimeException(e);
+ }
+ }
+ } );
+
+ blocker.acquire();
+
+ }
+
+
+
+ @Test
+ public void testWriteReadEdgeTypeTarget() {
+
+ EdgeManager em = emf.createEdgeManager( scope );
+
+
+ Edge edge = createEdge( "source", "test", "target" );
+
+ em.writeEdge( edge ).toBlockingObservable().last();
+
+ //now test retrieving it
+
+ SearchByEdgeType search = createSearchByEdge( edge.getTargetNode(), edge.getType(), edge.getVersion(), null );
+
+ Observable<Edge> edges = em.loadEdgesToTarget( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ Edge returned = edges.toBlockingObservable().single();
+
+ assertEquals( "Correct edge returned", edge, returned );
+
+ //change edge type to be invalid, shouldn't get a result
+ search = createSearchByEdge( edge.getTargetNode(), edge.getType() + "invalid", edge.getVersion(), null );
+
+ edges = em.loadEdgesToTarget( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().singleOrDefault( null );
+
+ assertNull( "Invalid type should not be returned", returned );
+ }
+
+
+ @Test
+ public void testWriteReadEdgeTypeVersionSource() {
+
+ EdgeManager em = emf.createEdgeManager( scope );
+
+ final UUID earlyVersion = UUIDGenerator.newTimeUUID();
+
+
+ Edge edge = createEdge( "source", "test", "target" );
+
+ em.writeEdge( edge ).toBlockingObservable().last();
+
+ //now test retrieving it
+
+ SearchByEdgeType search = createSearchByEdge( edge.getSourceNode(), edge.getType(), edge.getVersion(), null );
+
+ Observable<Edge> edges = em.loadEdgesFromSource( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ Edge returned = edges.toBlockingObservable().single();
+
+ assertEquals( "Correct edge returned", edge, returned );
+
+ //now test with an earlier version, we shouldn't get the edge back
+ search = createSearchByEdge( edge.getSourceNode(), edge.getType(), earlyVersion, null );
+
+ edges = em.loadEdgesFromSource( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().singleOrDefault( null );
+
+ assertNull( "Earlier version should not be returned", returned );
+ }
+
+
+ @Test
+ public void testWriteReadEdgeTypeVersionTarget() {
+
+ EdgeManager em = emf.createEdgeManager( scope );
+
+
+ final UUID earlyVersion = UUIDGenerator.newTimeUUID();
+
+
+ Edge edge = createEdge( "source", "test", "target" );
+
+ em.writeEdge( edge ).toBlockingObservable().last();
+
+ //now test retrieving it
+
+ SearchByEdgeType search = createSearchByEdge( edge.getTargetNode(), edge.getType(), edge.getVersion(), null );
+
+ Observable<Edge> edges = em.loadEdgesToTarget( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ Edge returned = edges.toBlockingObservable().single();
+
+ assertEquals( "Correct edge returned", edge, returned );
+
+ //change edge type to be invalid, shouldn't get a result
+ search = createSearchByEdge( edge.getTargetNode(), edge.getType(), earlyVersion, null );
+
+ edges = em.loadEdgesToTarget( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().singleOrDefault( null );
+
+ assertNull( "Earlier version should not be returned", returned );
+ }
+
+
+ /**
+ * 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() {
+
+ EdgeManager em = emf.createEdgeManager( scope );
+
+ final Id sourceId = createId( "source" );
+
+
+ Edge edge1 = createEdge( sourceId, "test", createId( "target" ) );
+
+ em.writeEdge( edge1 ).toBlockingObservable().last();
+
+ Edge edge2 = createEdge( sourceId, "test", createId( "target" ) );
+
+ em.writeEdge( edge2 ).toBlockingObservable().last();
+
+ Edge edge3 = createEdge( sourceId, "test", createId( "target" ) );
+
+ em.writeEdge( edge3 ).toBlockingObservable().last();
+
+
+ //now test retrieving it
+
+ SearchByEdgeType search =
+ createSearchByEdge( edge1.getSourceNode(), edge1.getType(), edge1.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();
+
+
+ //we have 3 edges, but we specified our first edge as the max, we shouldn't get any more results than the first
+ assertEquals( "Correct edge returned", edge1, returned.next() );
+
+ assertFalse( "No more edges", returned.hasNext() );
+
+ search = createSearchByEdge( edge1.getSourceNode(), edge1.getType(), edge3.getVersion(), edge2 );
+
+ edges = em.loadEdgesFromSource( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().getIterator();
+
+ assertEquals( "Paged correctly", edge3, returned.next() );
+
+ assertFalse( "End of stream", returned.hasNext() );
+ }
+
+
+ @Test
+ public void testWriteReadEdgeTypePagingTarget() {
+
+
+ EdgeManager em = emf.createEdgeManager( scope );
+
+
+ final Id targetId = createId( "target" );
+
+ Edge edge1 = createEdge( createId( "source" ), "test", targetId );
+
+ em.writeEdge( edge1 ).toBlockingObservable().last();
+
+ Edge edge2 = createEdge( createId( "source" ), "test", targetId );
+
+ em.writeEdge( edge2 ).toBlockingObservable().last();
+
+ Edge edge3 = createEdge( createId( "source" ), "test", targetId );
+
+ em.writeEdge( edge3 ).toBlockingObservable().last();
+
+
+ //now test retrieving it
+
+ SearchByEdgeType search =
+ createSearchByEdge( edge1.getTargetNode(), edge1.getType(), edge1.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();
+
+
+ //we have 3 edges, but we specified our first edge as the max, we shouldn't get any more results than the first
+ assertEquals( "Correct edge returned", edge1, returned.next() );
+
+
+ assertFalse( "No more edges", returned.hasNext() );
+
+ search = createSearchByEdge( edge1.getTargetNode(), edge1.getType(), edge3.getVersion(), edge2 );
+
+ edges = em.loadEdgesToTarget( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().getIterator();
+
+ assertEquals( "Paged correctly", edge3, returned.next() );
+
+ assertFalse( "End of stream", returned.hasNext() );
+ }
+
+
+ @Test
+ public void testWriteReadEdgeTypeTargetTypeSource() {
+
+ EdgeManager em = emf.createEdgeManager( scope );
+
+
+ Edge edge = createEdge( "source", "test", "target" );
+
+ em.writeEdge( edge ).toBlockingObservable().last();
+
+ //now test retrieving it
+
+ SearchByIdType search = createSearchByEdgeAndId( edge.getSourceNode(), edge.getType(), edge.getVersion(),
+ edge.getTargetNode().getType(), null );
+
+ Observable<Edge> edges = em.loadEdgesFromSourceByType( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ Edge returned = edges.toBlockingObservable().single();
+
+ assertEquals( "Correct edge returned", edge, returned );
+
+
+ //change edge type to be invalid, shouldn't get a result
+ search = createSearchByEdgeAndId( edge.getSourceNode(), edge.getType(), edge.getVersion(),
+ edge.getTargetNode().getType() + "invalid", null );
+
+ edges = em.loadEdgesFromSourceByType( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().singleOrDefault( null );
+
+ assertNull( "Invalid type should not be returned", returned );
+ }
+
+
+ @Test
+ public void testWriteReadEdgeTypeTargetTypeTarget() {
+
+ EdgeManager em = emf.createEdgeManager( scope );
+
+
+ Edge edge = createEdge( "source", "test", "target" );
+
+ em.writeEdge( edge ).toBlockingObservable().last();
+
+ //now test retrieving it
+
+ SearchByIdType search = createSearchByEdgeAndId( edge.getTargetNode(), edge.getType(), edge.getVersion(),
+ edge.getSourceNode().getType(), null );
+
+ Observable<Edge> edges = em.loadEdgesToTargetByType( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ Edge returned = edges.toBlockingObservable().single();
+
+ assertEquals( "Correct edge returned", edge, returned );
+
+
+ //change edge type to be invalid, shouldn't get a result
+ search = createSearchByEdgeAndId( edge.getTargetNode(), edge.getType(), edge.getVersion(),
+ edge.getSourceNode().getType() + "invalid", null );
+
+ edges = em.loadEdgesToTargetByType( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().singleOrDefault( null );
+
+ assertNull( "Invalid type should not be returned", returned );
+ }
+
+
+ @Test
+ public void testWriteReadEdgeDeleteSource() {
+
+ EdgeManager em = emf.createEdgeManager( scope );
+
+
+ Edge edge = createEdge( "source", "test", "target" );
+
+ em.writeEdge( edge ).toBlockingObservable().last();
+
+ //now test retrieving it
+
+
+ SearchByEdgeType search = createSearchByEdge( edge.getSourceNode(), edge.getType(), edge.getVersion(), null );
+
+ Observable<Edge> edges = em.loadEdgesFromSource( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ Edge returned = edges.toBlockingObservable().single();
+
+ assertEquals( "Correct edge returned", edge, returned );
+
+ SearchByIdType searchById = createSearchByEdgeAndId( edge.getSourceNode(), edge.getType(), edge.getVersion(),
+ edge.getTargetNode().getType(), null );
+
+ edges = em.loadEdgesFromSourceByType( searchById );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().single();
+
+ assertEquals( "Correct edge returned", edge, returned );
+
+
+ //now delete it
+ em.deleteEdge( edge ).toBlockingObservable().last();
+
+ //now test retrieval, should be null
+ edges = em.loadEdgesFromSource( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().singleOrDefault( null );
+
+ assertNull( "No edge returned", returned );
+
+
+ //no search by type, should be null as well
+
+ edges = em.loadEdgesFromSourceByType( searchById );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().singleOrDefault( null );
+
+ assertNull( "No edge returned", returned );
+ }
+
+
+ @Test
+ public void testWriteReadEdgeDeleteTarget() {
+
+ EdgeManager em = emf.createEdgeManager( scope );
+
+
+ Edge edge = createEdge( "source", "test", "target" );
+
+ em.writeEdge( edge ).toBlockingObservable().last();
+
+ //now test retrieving it
+
+
+ SearchByEdgeType search = createSearchByEdge( edge.getTargetNode(), edge.getType(), edge.getVersion(), null );
+
+ Observable<Edge> edges = em.loadEdgesToTarget( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ Edge returned = edges.toBlockingObservable().single();
+
+ assertEquals( "Correct edge returned", edge, returned );
+
+ SearchByIdType searchById = createSearchByEdgeAndId( edge.getTargetNode(), edge.getType(), edge.getVersion(),
+ edge.getSourceNode().getType(), null );
+
+ edges = em.loadEdgesToTargetByType( searchById );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().single();
+
+ assertEquals( "Correct edge returned", edge, returned );
+
+
+ //now delete it
+ em.deleteEdge( edge ).toBlockingObservable().last();
+
+ //now test retrieval, should be null
+ edges = em.loadEdgesToTarget( search );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().singleOrDefault( null );
+
+ assertNull( "No edge returned", returned );
+
+
+ //no search by type, should be null as well
+
+ edges = em.loadEdgesToTargetByType( searchById );
+
+ //implicitly blows up if more than 1 is returned from "single"
+ returned = edges.toBlockingObservable().singleOrDefault( null );
+
+ assertNull( "No edge returned", returned );
+ }
+
+
+ @Test
+ public void testWriteReadEdgeTypesSourceTypes() {
+
+ final EdgeManager em = emf.createEdgeManager( scope );
+
+ Id sourceId = new SimpleId( "source" );
+ Id targetId1 = new SimpleId( "target" );
+ Id targetId2 = new SimpleId( "target2" );
+
+ Edge testTargetEdge = createEdge( sourceId, "test", targetId1, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( testTargetEdge ).toBlockingObservable().singleOrDefault( null );
+
+ Edge testTarget2Edge = createEdge( sourceId, "test", targetId2, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( testTarget2Edge ).toBlockingObservable().singleOrDefault( null );
+
+
+ Edge test2TargetEdge = createEdge( sourceId, "test2", targetId1, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( test2TargetEdge ).toBlockingObservable().singleOrDefault( null );
+
+
+ //get our 2 edge types
+ Observable<String> edges =
+ em.getEdgeTypesFromSource( new SimpleSearchEdgeType( testTargetEdge.getSourceNode(), null ) );
+
+
+ Iterator<String> results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", "test", results.next() );
+
+ assertEquals( "Edges correct", "test2", results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ //now test sub edges
+
+ edges = em.getIdTypesFromSource( new SimpleSearchIdType( testTargetEdge.getSourceNode(), "test", null ) );
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Types correct", targetId1.getType(), results.next() );
+
+ assertEquals( "Types correct", targetId2.getType(), results.next() );
+
+ assertFalse( "No results", results.hasNext() );
+
+ //now get types for test2
+ edges = em.getIdTypesFromSource( new SimpleSearchIdType( testTargetEdge.getSourceNode(), "test2", null ) );
+
+ results = edges.toBlockingObservable().getIterator();
+
+ assertEquals( "Types correct", targetId1.getType(), results.next() );
+
+ assertFalse( "No results", results.hasNext() );
+ }
+
+
+ @Test
+ public void testWriteReadEdgeTypesTargetTypes() {
+
+ final EdgeManager em = emf.createEdgeManager( scope );
+
+ Id sourceId1 = new SimpleId( "source" );
+ Id sourceId2 = new SimpleId( "source2" );
+ Id targetId1 = new SimpleId( "target" );
+
+
+ Edge testTargetEdge = createEdge( sourceId1, "test", targetId1, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( testTargetEdge ).toBlockingObservable().singleOrDefault( null );
+
+ Edge testTarget2Edge = createEdge( sourceId2, "test", targetId1, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( testTarget2Edge ).toBlockingObservable().singleOrDefault( null );
+
+
+ Edge test2TargetEdge = createEdge( sourceId1, "test2", targetId1, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( test2TargetEdge ).toBlockingObservable().singleOrDefault( null );
+
+
+ //get our 2 edge types
+ final SearchEdgeType edgeTypes = new SimpleSearchEdgeType( testTargetEdge.getTargetNode(), null );
+
+ Observable<String> edges = em.getEdgeTypesToTarget( edgeTypes );
+
+
+ Iterator<String> results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", "test", results.next() );
+
+ assertEquals( "Edges correct", "test2", results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ //now test sub edges
+
+ edges = em.getIdTypesToTarget( new SimpleSearchIdType( testTargetEdge.getTargetNode(), "test", null ) );
+
+ results = edges.toBlockingObservable().getIterator();
+
+ assertEquals( "Types correct", sourceId1.getType(), results.next() );
+
+ assertEquals( "Types correct", sourceId2.getType(), results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ //now get types for test2
+ edges = em.getIdTypesToTarget( new SimpleSearchIdType( testTargetEdge.getTargetNode(), "test2", null ) );
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Types correct", sourceId1.getType(), results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+ }
+
+
+ @Test
+ public void testWriteReadEdgeTypesSourceTypesPaging() {
+
+ final EdgeManager em = emf.createEdgeManager( scope );
+
+ Id sourceId1 = new SimpleId( "source" );
+ Id targetId1 = new SimpleId( "target" );
+ Id targetId2 = new SimpleId( "target2" );
+
+
+ Edge testTargetEdge = createEdge( sourceId1, "test", targetId1, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( testTargetEdge ).toBlockingObservable().singleOrDefault( null );
+
+
+ Edge testTargetEdge2 = createEdge( sourceId1, "test", targetId2, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( testTargetEdge2 ).toBlockingObservable().singleOrDefault( null );
+
+
+ Edge test2TargetEdge = createEdge( sourceId1, "test2", targetId2, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( test2TargetEdge ).toBlockingObservable().singleOrDefault( null );
+
+
+ //get our 2 edge types
+ SearchEdgeType edgeTypes = new SimpleSearchEdgeType( testTargetEdge.getSourceNode(), null );
+
+ Observable<String> edges = em.getEdgeTypesFromSource( edgeTypes );
+
+
+ Iterator<String> results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", "test", results.next() );
+ assertEquals( "Edges correct", "test2", results.next() );
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now load the next page
+
+ edgeTypes = new SimpleSearchEdgeType( testTargetEdge.getSourceNode(), "test" );
+
+ edges = em.getEdgeTypesFromSource( edgeTypes );
+
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", "test2", results.next() );
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ //now test sub edges
+
+ edges = em.getIdTypesFromSource( new SimpleSearchIdType( testTargetEdge.getSourceNode(), "test", null ) );
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Types correct", targetId1.getType(), results.next() );
+ assertEquals( "Types correct", targetId2.getType(), results.next() );
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ //now get the next page
+
+ edges = em.getIdTypesFromSource(
+ new SimpleSearchIdType( testTargetEdge.getSourceNode(), "test", targetId1.getType() ) );
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Types correct", targetId2.getType(), results.next() );
+
+ assertFalse( "No more results", results.hasNext() );
+ }
+
+
+ @Test
+ public void testWriteReadEdgeTypesTargetTypesPaging() {
+
+ final EdgeManager em = emf.createEdgeManager( scope );
+
+ Id sourceId1 = new SimpleId( "source" );
+ Id sourceId2 = new SimpleId( "source2" );
+ Id targetId = new SimpleId( "target" );
+
+
+ Edge testTargetEdge = createEdge( sourceId1, "test", targetId, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( testTargetEdge ).toBlockingObservable().singleOrDefault( null );
+
+
+ Edge testTargetEdge2 = createEdge( sourceId2, "test", targetId, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( testTargetEdge2 ).toBlockingObservable().singleOrDefault( null );
+
+ Edge test2TargetEdge = createEdge( sourceId2, "test2", targetId, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( test2TargetEdge ).toBlockingObservable().singleOrDefault( null );
+
+
+ //get our 2 edge types
+ SearchEdgeType edgeTypes = new SimpleSearchEdgeType( testTargetEdge.getTargetNode(), null );
+
+ Observable<String> edges = em.getEdgeTypesToTarget( edgeTypes );
+
+
+ Iterator<String> results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", "test", results.next() );
+ assertEquals( "Edges correct", "test2", results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ //now load the next page
+
+ edgeTypes = new SimpleSearchEdgeType( testTargetEdge2.getTargetNode(), "test" );
+
+ edges = em.getEdgeTypesToTarget( edgeTypes );
+
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", "test2", results.next() );
+
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now test sub edges
+
+ edges = em.getIdTypesToTarget( new SimpleSearchIdType( testTargetEdge.getTargetNode(), "test", null ) );
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Types correct", sourceId1.getType(), results.next() );
+
+ assertEquals( "Types correct", sourceId2.getType(), results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now get the next page
+
+ edges = em.getIdTypesToTarget(
+ new SimpleSearchIdType( testTargetEdge.getTargetNode(), "test", sourceId1.getType() ) );
+
+ results = edges.toBlockingObservable().getIterator();
+
+
+ assertEquals( "Types correct", sourceId2.getType(), results.next() );
+
+ assertFalse( "No more results", results.hasNext() );
+ }
+
+
+ @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
+ public void markSourceNode() {
+
+ 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();
+
+ Iterator<Edge> results =
+ em.loadEdgesFromSource( createSearchByEdge( sourceId, edge1.getType(), maxVersion, null ) )
+ .toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edge found", edge1, results.next() );
+
+ assertEquals( "Edge found", edge2, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ //get our 2 edges
+ results = em.loadEdgesFromSourceByType(
+ createSearchByEdgeAndId( sourceId, edge1.getType(), maxVersion, targetId1.getType(), null ) )
+ .toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", edge1, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+ results = em.loadEdgesFromSourceByType(
+ createSearchByEdgeAndId( sourceId, edge2.getType(), maxVersion, targetId2.getType(), null ) )
+ .toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", edge2, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //mark the source node
+ em.deleteNode( sourceId ).toBlockingObservable().last();
+
+
+ //now re-read, nothing should be there since they're marked
+
+ results = em.loadEdgesFromSource( createSearchByEdge( sourceId, edge1.getType(), maxVersion, null ) )
+ .toBlockingObservable().getIterator();
+
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ //get our 2 edges
+ results = em.loadEdgesFromSourceByType(
+ createSearchByEdgeAndId( sourceId, edge1.getType(), maxVersion, targetId1.getType(), null ) )
+ .toBlockingObservable().getIterator();
+
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+ results = em.loadEdgesFromSourceByType(
+ createSearchByEdgeAndId( sourceId, edge2.getType(), maxVersion, targetId2.getType(), null ) )
+ .toBlockingObservable().getIterator();
+
+
+ assertFalse( "No more edges", results.hasNext() );
+ }
+
+
+
+
+ @Test
+ public void markTargetNode() {
+
+ 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().singleOrDefault( null );
+
+ Edge edge2 = createEdge( sourceId2, "test", targetId, UUIDGenerator.newTimeUUID() );
+
+ em.writeEdge( edge2 ).toBlockingObservable().singleOrDefault( null );
+
+
+ final UUID maxVersion = UUIDGenerator.newTimeUUID();
+
+ Iterator<Edge> results =
+ em.loadEdgesToTarget( createSearchByEdge( targetId, edge1.getType(), maxVersion, null ) )
+ .toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edge found", edge1, results.next() );
+
+ assertEquals( "Edge found", edge2, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ //get our 2 edges
+ results = em.loadEdgesToTargetByType(
+ createSearchByEdgeAndId( targetId, edge1.getType(), maxVersion, sourceId1.getType(), null ) )
+ .toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", edge1, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+ results = em.loadEdgesToTargetByType(
+ createSearchByEdgeAndId( targetId, edge2.getType(), maxVersion, sourceId2.getType(), null ) )
+ .toBlockingObservable().getIterator();
+
+
+ assertEquals( "Edges correct", edge2, results.next() );
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //mark the source node
+ em.deleteNode( targetId ).toBlockingObservable().last();
+
+
+ //now re-read, nothing should be there since they're marked
+
+ results = em.loadEdgesToTarget( createSearchByEdge( targetId, edge1.getType(), maxVersion, null ) )
+ .toBlockingObservable().getIterator();
+
+ assertFalse( "No more edges", results.hasNext() );
+
+
+ //get our 2 edges
+ results = em.loadEdgesToTargetByType(
+ createSearchByEdgeAndId( targetId, edge1.getType(), maxVersion, sourceId1.getType(), null ) )
+ .toBlockingObservable().getIterator();
+
+
+ assertFalse( "No more edges", results.hasNext() );
+
+ //now delete one of the edges
+ results = em.loadEdgesToTargetByType(
+ createSearchByEdgeAndId( targetId, edge2.getType(), maxVersion, sourceId2.getType(), null ) )
+ .toBlockingObservable().getIterator();
+
+
+ assertFalse( "No more edges", results.hasNext() );
+ }
+
+
+
+ @Test(expected = NullPointerException.class)
+ public void invalidEdgeTypesWrite( @All Edge edge ) {
+ final EdgeManager em = emf.createEdgeManager( scope );
+
+ em.writeEdge( edge );
+ }
+
+
+ @Test(expected = NullPointerException.class)
+ public void invalidEdgeTypesDelete( @All Edge edge ) {
+ final EdgeManager em = emf.createEdgeManager( scope );
+
+ 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;
+ // }
+ // }
+}
+
+
+
+
+
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5cadb9f5/stack/corepersistence/pom.xml
----------------------------------------------------------------------
diff --git a/stack/corepersistence/pom.xml b/stack/corepersistence/pom.xml
index f538d0e..5ffab73 100644
--- a/stack/corepersistence/pom.xml
+++ b/stack/corepersistence/pom.xml
@@ -39,6 +39,7 @@
<slf4j.version>1.7.2</slf4j.version>
<surefire.version>2.16</surefire.version>
<rx.version>0.17.1</rx.version>
+ <hystrix.version>1.4.0-RC1</hystrix.version>
</properties>