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/27 22:21:48 UTC
[34/50] [abbrv] git commit: Updated tests. Need to figure a work
around for Mockito
Updated tests. Need to figure a work around for Mockito
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/5ded6b52
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/5ded6b52
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/5ded6b52
Branch: refs/pull/77/head
Commit: 5ded6b52943bf9c8bd6c72090a04337196e9ce0e
Parents: 16721c8
Author: Todd Nine <tn...@apigee.com>
Authored: Wed Mar 26 13:43:37 2014 -0700
Committer: Todd Nine <tn...@apigee.com>
Committed: Wed Mar 26 13:43:37 2014 -0700
----------------------------------------------------------------------
.../graph/consistency/AsyncProcessorImpl.java | 46 +-
.../graph/impl/GraphManagerImpl.java | 49 +-
.../graph/impl/SimpleSearchByEdgeType.java | 38 +
.../graph/impl/stage/AbstractEdgeRepair.java | 2 +
.../persistence/graph/EdgeManagerIT.java | 1498 ------------------
.../graph/EdgeManagerStressTest.java | 317 ----
.../persistence/graph/GraphManagerIT.java | 1497 +++++++++++++++++
.../graph/GraphManagerStressTest.java | 317 ++++
.../graph/GraphManagerTimeoutIT.java | 1336 +---------------
.../graph/impl/NodeDeleteListenerTest.java | 4 +-
10 files changed, 1908 insertions(+), 3196 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5ded6b52/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorImpl.java
index 20411e7..000eaeb 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorImpl.java
@@ -10,14 +10,13 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.usergrid.persistence.graph.GraphFig;
+import org.apache.usergrid.persistence.graph.hystrix.HystrixGraphObservable;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import rx.Observable;
-import rx.Scheduler;
-import rx.functions.Action0;
-import rx.functions.Action1;
+import rx.Subscriber;
import rx.functions.FuncN;
import rx.schedulers.Schedulers;
@@ -45,13 +44,14 @@ public class AsyncProcessorImpl<T> implements AsyncProcessor<T> {
@Inject
- public AsyncProcessorImpl( final TimeoutQueue<T> queue, final GraphFig graphFig ) {
+ public AsyncProcessorImpl( final TimeoutQueue<T> queue, final GraphFig graphFig ) {
this.queue = queue;
this.graphFig = graphFig;
//we purposefully use a new thread. We don't want to use one of the I/O threads to run this task
//in the event the scheduler is full, we'll end up rejecting the reschedule of this task
- Schedulers.newThread().schedulePeriodically( new TimeoutTask<T>(this, graphFig), graphFig.getTaskLoopTime(), graphFig.getTaskLoopTime(), TimeUnit.MILLISECONDS );
+ Schedulers.newThread().schedulePeriodically( new TimeoutTask<T>( this, graphFig ), graphFig.getTaskLoopTime(),
+ graphFig.getTaskLoopTime(), TimeUnit.MILLISECONDS );
}
@@ -70,37 +70,43 @@ public class AsyncProcessorImpl<T> implements AsyncProcessor<T> {
List<Observable<?>> observables = new ArrayList<Observable<?>>( listeners.size() );
for ( MessageListener<T, T> listener : listeners ) {
- observables.add( listener.receive( data ).subscribeOn( Schedulers.io() ) );
+ observables.add( HystrixGraphObservable.async( listener.receive( data ) ) );
}
+ LOG.debug( "About to start {} observables for event {}", listeners.size(), event );
+
//run everything in parallel and zip it up
Observable.zip( observables, new FuncN<AsynchronousMessage<T>>() {
@Override
public AsynchronousMessage<T> call( final Object... args ) {
return event;
}
- } ).doOnError( new Action1<Throwable>() {
- @Override
- public void call( final Throwable throwable ) {
- LOG.error( "Unable to process async event", throwable );
+ } ).subscribe( new Subscriber<AsynchronousMessage<T>>() {
- for ( ErrorListener listener : errorListeners ) {
- listener.onError( event, throwable );
- }
- }
- } ).doOnCompleted( new Action0() {
@Override
- public void call() {
+ public void onCompleted() {
+ LOG.debug( "Successfully completed processing for event {}", event );
queue.remove( event );
for ( CompleteListener<T> listener : completeListeners ) {
listener.onComplete( event );
}
}
- } ).subscribe( new Action1<AsynchronousMessage<T>>() {
+
+
+ @Override
+ public void onError( final Throwable throwable ) {
+ LOG.error( "Unable to process async event", throwable );
+
+ for ( ErrorListener listener : errorListeners ) {
+ listener.onError( event, throwable );
+ }
+ }
+
+
@Override
- public void call( final AsynchronousMessage<T> asynchronousMessage ) {
- //no op
+ public void onNext( final AsynchronousMessage<T> tAsynchronousMessage ) {
+ //no op
}
} );
}
@@ -130,6 +136,4 @@ public class AsyncProcessorImpl<T> implements AsyncProcessor<T> {
public void addCompleteListener( final CompleteListener<T> listener ) {
this.completeListeners.add( listener );
}
-
-
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5ded6b52/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
index ff63a60..ad7f67b 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
@@ -40,6 +40,7 @@ import org.apache.usergrid.persistence.graph.consistency.AsyncProcessor;
import org.apache.usergrid.persistence.graph.consistency.AsynchronousMessage;
import org.apache.usergrid.persistence.graph.guice.EdgeDelete;
import org.apache.usergrid.persistence.graph.guice.NodeDelete;
+import org.apache.usergrid.persistence.graph.hystrix.HystrixGraphObservable;
import org.apache.usergrid.persistence.graph.serialization.EdgeMetadataSerialization;
import org.apache.usergrid.persistence.graph.serialization.EdgeSerialization;
import org.apache.usergrid.persistence.graph.serialization.NodeSerialization;
@@ -108,7 +109,7 @@ public class GraphManagerImpl implements GraphManager {
public Observable<Edge> writeEdge( final Edge edge ) {
EdgeUtils.validateEdge( edge );
- return Observable.from( edge ).subscribeOn( Schedulers.io() ).map( new Func1<Edge, Edge>() {
+ return HystrixGraphObservable.user(Observable.from( edge ).subscribeOn( Schedulers.io() ).map( new Func1<Edge, Edge>() {
@Override
public Edge call( final Edge edge ) {
final MutationBatch mutation = edgeMetadataSerialization.writeEdge( scope, edge );
@@ -126,7 +127,7 @@ public class GraphManagerImpl implements GraphManager {
return edge;
}
- } );
+ } ));
}
@@ -134,7 +135,7 @@ public class GraphManagerImpl implements GraphManager {
public Observable<Edge> deleteEdge( final Edge edge ) {
EdgeUtils.validateEdge( edge );
- return Observable.from( edge ).subscribeOn( Schedulers.io() ).map( new Func1<Edge, Edge>() {
+ return HystrixGraphObservable.user(Observable.from( edge ).subscribeOn( Schedulers.io() ).map( new Func1<Edge, Edge>() {
@Override
public Edge call( final Edge edge ) {
final MutationBatch edgeMutation = edgeSerialization.markEdge( scope, edge );
@@ -154,13 +155,13 @@ public class GraphManagerImpl implements GraphManager {
return edge;
}
- } );
+ } ));
}
@Override
public Observable<Id> deleteNode( final Id node ) {
- return Observable.from( node ).subscribeOn( Schedulers.io() ).map( new Func1<Id, Id>() {
+ return HystrixGraphObservable.user(Observable.from( node ).subscribeOn( Schedulers.io() ).map( new Func1<Id, Id>() {
@Override
public Id call( final Id id ) {
@@ -183,114 +184,114 @@ public class GraphManagerImpl implements GraphManager {
return id;
}
- } );
+ } ));
}
@Override
public Observable<Edge> loadEdgeVersions( final SearchByEdge searchByEdge ) {
- return Observable.create( new ObservableIterator<MarkedEdge>( "getEdgeVersions" ) {
+ return HystrixGraphObservable.user(Observable.create( new ObservableIterator<MarkedEdge>( "getEdgeVersions" ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgeVersions( scope, searchByEdge );
}
} ).buffer( graphFig.getScanPageSize() ).flatMap( new EdgeBufferFilter( searchByEdge.getMaxVersion() ) )
- .cast( Edge.class );
+ .cast( Edge.class ));
}
@Override
public Observable<Edge> loadEdgesFromSource( final SearchByEdgeType search ) {
- return Observable.create( new ObservableIterator<MarkedEdge>( "getEdgesFromSource" ) {
+ return HystrixGraphObservable.user(Observable.create( new ObservableIterator<MarkedEdge>( "getEdgesFromSource" ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesFromSource( scope, search );
}
} ).buffer( graphFig.getScanPageSize() ).flatMap( new EdgeBufferFilter( search.getMaxVersion() ) )
- .cast( Edge.class );
+ .cast( Edge.class ));
}
@Override
public Observable<Edge> loadEdgesToTarget( final SearchByEdgeType search ) {
- return Observable.create( new ObservableIterator<MarkedEdge>( "getEdgesToTarget" ) {
+ return HystrixGraphObservable.user(Observable.create( new ObservableIterator<MarkedEdge>( "getEdgesToTarget" ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesToTarget( scope, search );
}
} ).buffer( graphFig.getScanPageSize() ).flatMap( new EdgeBufferFilter( search.getMaxVersion() ) )
- .cast( Edge.class );
+ .cast( Edge.class ));
}
@Override
public Observable<Edge> loadEdgesFromSourceByType( final SearchByIdType search ) {
- return Observable.create( new ObservableIterator<MarkedEdge>( "getEdgesFromSourceByTargetType" ) {
+ return HystrixGraphObservable.user(Observable.create( new ObservableIterator<MarkedEdge>( "getEdgesFromSourceByTargetType" ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesFromSourceByTargetType( scope, search );
}
} ).buffer( graphFig.getScanPageSize() ).flatMap( new EdgeBufferFilter( search.getMaxVersion() ) )
- .cast( Edge.class );
+ .cast( Edge.class ));
}
@Override
public Observable<Edge> loadEdgesToTargetByType( final SearchByIdType search ) {
- return Observable.create( new ObservableIterator<MarkedEdge>( "getEdgesToTargetBySourceType" ) {
+ return HystrixGraphObservable.user(Observable.create( new ObservableIterator<MarkedEdge>( "getEdgesToTargetBySourceType" ) {
@Override
protected Iterator<MarkedEdge> getIterator() {
return edgeSerialization.getEdgesToTargetBySourceType( scope, search );
}
} ).buffer( graphFig.getScanPageSize() ).flatMap( new EdgeBufferFilter( search.getMaxVersion() ) )
- .cast( Edge.class );
+ .cast( Edge.class ));
}
@Override
public Observable<String> getEdgeTypesFromSource( final SearchEdgeType search ) {
- return Observable.create( new ObservableIterator<String>( "getEdgeTypesFromSource" ) {
+ return HystrixGraphObservable.user(Observable.create( new ObservableIterator<String>( "getEdgeTypesFromSource" ) {
@Override
protected Iterator<String> getIterator() {
return edgeMetadataSerialization.getEdgeTypesFromSource( scope, search );
}
- } );
+ } ));
}
@Override
public Observable<String> getIdTypesFromSource( final SearchIdType search ) {
- return Observable.create( new ObservableIterator<String>( "getIdTypesFromSource" ) {
+ return HystrixGraphObservable.user(Observable.create( new ObservableIterator<String>( "getIdTypesFromSource" ) {
@Override
protected Iterator<String> getIterator() {
return edgeMetadataSerialization.getIdTypesFromSource( scope, search );
}
- } );
+ } ));
}
@Override
public Observable<String> getEdgeTypesToTarget( final SearchEdgeType search ) {
- return Observable.create( new ObservableIterator<String>( "getEdgeTypesToTarget" ) {
+ return HystrixGraphObservable.user(Observable.create( new ObservableIterator<String>( "getEdgeTypesToTarget" ) {
@Override
protected Iterator<String> getIterator() {
return edgeMetadataSerialization.getEdgeTypesToTarget( scope, search );
}
- } );
+ } ));
}
@Override
public Observable<String> getIdTypesToTarget( final SearchIdType search ) {
- return Observable.create( new ObservableIterator<String>( "getIdTypesToTarget" ) {
+ return HystrixGraphObservable.user(Observable.create( new ObservableIterator<String>( "getIdTypesToTarget" ) {
@Override
protected Iterator<String> getIterator() {
return edgeMetadataSerialization.getIdTypesToTarget( scope, search );
}
- } );
+ } ));
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5ded6b52/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleSearchByEdgeType.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleSearchByEdgeType.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleSearchByEdgeType.java
index d56c538..a7a79e3 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleSearchByEdgeType.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/SimpleSearchByEdgeType.java
@@ -84,4 +84,42 @@ public class SimpleSearchByEdgeType implements SearchByEdgeType{
public Optional<Edge> last() {
return last;
}
+
+
+ @Override
+ public boolean equals( final Object o ) {
+ if ( this == o ) {
+ return true;
+ }
+ if ( o == null || getClass() != o.getClass() ) {
+ return false;
+ }
+
+ final SimpleSearchByEdgeType that = ( SimpleSearchByEdgeType ) o;
+
+ if ( !last.equals( that.last ) ) {
+ return false;
+ }
+ if ( !maxVersion.equals( that.maxVersion ) ) {
+ return false;
+ }
+ if ( !node.equals( that.node ) ) {
+ return false;
+ }
+ if ( !type.equals( that.type ) ) {
+ return false;
+ }
+
+ return true;
+ }
+
+
+ @Override
+ public int hashCode() {
+ int result = node.hashCode();
+ result = 31 * result + type.hashCode();
+ result = 31 * result + maxVersion.hashCode();
+ result = 31 * result + last.hashCode();
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5ded6b52/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 898a04d..7fc571a 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
@@ -49,6 +49,8 @@ import rx.schedulers.Schedulers;
/**
* SimpleRepair operation
*
+ * TODO T.N. this is still valid code, just not used ATM. DO NOT REMOVE IT. Needs to refactor it to read all
+ * versions of an edge and remove them.
*/
@Singleton
public abstract class AbstractEdgeRepair {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/5ded6b52/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
deleted file mode 100644
index 379f56a..0000000
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerIT.java
+++ /dev/null
@@ -1,1498 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.usergrid.persistence.graph;
-
-
-import java.util.Iterator;
-import java.util.UUID;
-
-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.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 static org.apache.usergrid.persistence.graph.test.util.EdgeTestUtils.createEdge;
-import static org.apache.usergrid.persistence.graph.test.util.EdgeTestUtils.createGetByEdge;
-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.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-
-@RunWith(JukitoRunner.class)
-@UseModules({ TestGraphModule.class })
-//@UseModules( { TestGraphModule.class, EdgeManagerIT.InvalidInput.class } )
-public class EdgeManagerIT {
-
-
- @ClassRule
- public static CassandraRule rule = new CassandraRule();
-
-
- @Inject
- @Rule
- public MigrationManagerRule migrationManagerRule;
-
-
- @Inject
- protected GraphManagerFactory emf;
-
- 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 );
- }
-
-
- @Test
- public void testWriteReadEdgeTypeSource() {
-
- GraphManager 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().last();
-
- assertEquals( "Correct edge returned", edge, returned );
-
- //change edge type to be invalid, shouldn't get a result
- search = createSearchByEdge( edge.getSourceNode(), edge.getType() + "invalid", edge.getVersion(), null );
-
- edges = em.loadEdgesFromSource( 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 testWriteReadEdgeTypeTarget() {
-
- GraphManager 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() {
-
- GraphManager 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() {
-
- GraphManager 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() {
-
- GraphManager 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() );
- assertEquals( "Correct edge returned", edge2, returned.next() );
- assertEquals( "Correct edge returned", edge1, 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() );
- assertEquals( "Correct edge returned", edge1, 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() {
-
-
- GraphManager 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() );
- assertEquals( "Correct edge returned", edge2, returned.next() );
- assertEquals( "Correct edge returned", edge1, 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() );
- assertEquals( "Correct edge returned", edge1, 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() {
-
- GraphManager 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(), 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();
-
-
- //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", edge3, returned.next() );
-
- assertEquals( "Correct edge returned", edge2, returned.next() );
-
- assertEquals( "Correct edge returned", edge1, returned.next() );
-
- assertFalse( "No more edges", returned.hasNext() );
-
- //still edge 3 is our max version, but we start with edge 2 as our last read
- 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", edge1, returned.next() );
-
- assertFalse( "End of stream", returned.hasNext() );
- }
-
-
- @Test
- public void testWriteReadEdgeTypePagingTarget() {
-
-
- GraphManager 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(), 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();
-
-
- //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", edge3, returned.next() );
-
- assertEquals( "Correct edge returned", edge2, returned.next() );
-
- 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", edge1, returned.next() );
-
- assertFalse( "End of stream", returned.hasNext() );
- }
-
-
- @Test
- public void testWriteReadEdgeTypeTargetTypeSource() {
-
- GraphManager 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() {
-
- GraphManager 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() {
-
- GraphManager 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 );
-
- final SearchByEdge searchByEdge = createGetByEdge(edge.getSourceNode(), edge.getType(), edge.getTargetNode(), edge.getVersion(), null);
-
- returned = em.loadEdgeVersions(searchByEdge).toBlockingObservable().single();
-
-
- assertEquals( "Correct edge returned", edge, returned );
-
-
- //now delete it
- returned = em.deleteEdge( edge ).toBlockingObservable().last();
-
-
- assertEquals( "Correct edge returned", edge, returned );
-
-
- //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 );
-
- returned = em.loadEdgeVersions(searchByEdge).toBlockingObservable().singleOrDefault(null);
-
- assertNull( "No edge returned", returned );
- }
-
-
- @Test
- public void testWriteReadEdgeDeleteTarget() {
-
- GraphManager 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 GraphManager 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 GraphManager 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 GraphManager 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 GraphManager 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 GraphManager 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", edge2, results.next() );
-
- 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.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 GraphManager 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", edge2, results.next() );
-
- 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.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 GraphManager 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 GraphManager 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 GraphManager 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", edge2, results.next() );
-
- assertEquals( "Edge found", edge1, 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 GraphManager 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", edge2, results.next() );
-
- assertEquals( "Edge found", edge1, 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 GraphManager em = emf.createEdgeManager( scope );
-
- em.writeEdge( edge );
- }
-
-
- @Test(expected = NullPointerException.class)
- public void invalidEdgeTypesDelete( @All Edge edge ) {
- final GraphManager 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/5ded6b52/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerStressTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerStressTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerStressTest.java
deleted file mode 100644
index 9d8db6e..0000000
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/EdgeManagerStressTest.java
+++ /dev/null
@@ -1,317 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.usergrid.persistence.graph;
-
-
-import java.util.HashSet;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.CountDownLatch;
-
-import org.jukito.JukitoRunner;
-import org.jukito.UseModules;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Ignore;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.commons.lang3.time.StopWatch;
-
-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.SimpleSearchByEdgeType;
-import org.apache.usergrid.persistence.model.entity.Id;
-import org.apache.usergrid.persistence.model.util.UUIDGenerator;
-
-import com.google.inject.Inject;
-
-import rx.Observable;
-import rx.Subscriber;
-
-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.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.fail;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-
-@RunWith(JukitoRunner.class)
-@UseModules(TestGraphModule.class)
-public class EdgeManagerStressTest {
- private static final Logger log = LoggerFactory.getLogger( EdgeManagerStressTest.class );
-
- @Inject
- private GraphManagerFactory factory;
-
- @ClassRule
- public static CassandraRule rule = new CassandraRule();
-
- @Inject
- @Rule
- public MigrationManagerRule migrationManagerRule;
-
-
- 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 );
- }
-
-
- @Test
- @Ignore
- public void writeThousands() throws InterruptedException {
- EdgeGenerator generator = new EdgeGenerator() {
-
- private Set<Id> sourceIds = new HashSet<Id>();
-
-
- @Override
- public Edge newEdge() {
- Edge edge = createEdge( "source", "test", "target" );
-
- sourceIds.add( edge.getSourceNode() );
-
- return edge;
- }
-
-
- @Override
- public Observable<Edge> doSearch( final GraphManager manager ) {
-
-
- final UUID uuid = UUIDGenerator.newTimeUUID();
-
-
- return Observable.create( new Observable.OnSubscribe<Edge>() {
-
- @Override
- public void call( final Subscriber<? super Edge> subscriber ) {
- try {
- for ( Id sourceId : sourceIds ) {
-
- final Iterable<Edge> edges = manager.loadEdgesFromSource(
- new SimpleSearchByEdgeType( sourceId, "test", uuid, null ) )
- .toBlockingObservable().toIterable();
-
- for ( Edge edge : edges ) {
- log.debug( "Firing on next for edge {}", edge );
-
- subscriber.onNext( edge );
- }
- }
- }
- catch ( Throwable throwable ) {
- subscriber.onError( throwable );
- }
- }
- } );
-
-
- //TODO T.N keep this code it's exhibiting a failure /exception swallowing with RX when our scheduler
- // is full
- //
- // return Observable.create( new Observable.OnSubscribe<Edge>() {
- //
- // @Override
- // public void call( final Subscriber<? super Edge> subscriber ) {
- // for ( Id sourceId : sourceIds ) {
- //
- // final Observable<Edge> edges =
- // manager.loadEdgesFromSource( new
- // SimpleSearchByEdgeType( sourceId, "test", uuid, null ) );
- //
- // edges.subscribe( new Action1<Edge>() {
- // @Override
- // public void call( final Edge edge ) {
- // subscriber.onNext( edge );
- // }
- // },
- //
- // new Action1<Throwable>() {
- // @Override
- // public void call( final Throwable throwable ) {
- // subscriber.onError( throwable );
- // }
- // });
- // }
- // }
- // } ) ;
-
-
- }
- };
-
- doTest( generator );
- }
-
-
- @Ignore
- @Test
- public void writeThousandsSingleSource() throws InterruptedException {
- EdgeGenerator generator = new EdgeGenerator() {
-
- private Id sourceId = createId( "source" );
-
-
- @Override
- public Edge newEdge() {
- Edge edge = createEdge( sourceId, "test", createId( "target" ) );
-
-
- return edge;
- }
-
-
- @Override
- public Observable<Edge> doSearch( final GraphManager manager ) {
- UUID uuid = UUIDGenerator.newTimeUUID();
-
- return manager.loadEdgesFromSource( new SimpleSearchByEdgeType( sourceId, "test", uuid, null ) );
- }
- };
-
- doTest( generator );
- }
-
-
- @Test
- @Ignore
- public void writeThousandsSingleTarget() throws InterruptedException {
- EdgeGenerator generator = new EdgeGenerator() {
-
- private Id targetId = createId( "target" );
-
-
- @Override
- public Edge newEdge() {
- Edge edge = createEdge( createId( "source" ), "test", targetId );
-
-
- return edge;
- }
-
-
- @Override
- public Observable<Edge> doSearch( final GraphManager manager ) {
- UUID uuid = UUIDGenerator.newTimeUUID();
-
- return manager.loadEdgesToTarget( new SimpleSearchByEdgeType( targetId, "test", uuid, null ) );
- }
- };
-
- doTest( generator );
- }
-
-
- /**
- * Execute the test with the generator
- */
- private void doTest( EdgeGenerator generator ) throws InterruptedException {
- GraphManager manager = factory.createEdgeManager( scope );
-
- int limit = 10000;
-
- final StopWatch timer = new StopWatch();
- timer.start();
- final Set<Edge> ids = new HashSet<Edge>( limit );
-
- for ( int i = 0; i < limit; i++ ) {
-
- Edge edge = generator.newEdge();
-
- Edge returned = manager.writeEdge( edge ).toBlockingObservable().last();
-
-
- assertNotNull( "Returned has a version", returned.getVersion() );
-
- ids.add( returned );
-
- if ( i % 1000 == 0 ) {
- log.info( " Wrote: " + i );
- }
- }
-
- timer.stop();
- log.info( "Total time to write {} entries {}ms", limit, timer.getTime() );
- timer.reset();
-
- timer.start();
-
- final CountDownLatch latch = new CountDownLatch( 1 );
-
-
- generator.doSearch( manager ).subscribe( new Subscriber<Edge>() {
- @Override
- public void onCompleted() {
- timer.stop();
- latch.countDown();
- }
-
-
- @Override
- public void onError( final Throwable throwable ) {
- fail( "Exception occurced " + throwable );
- }
-
-
- @Override
- public void onNext( final Edge edge ) {
- ids.remove( edge );
- }
- } );
-
-
- latch.await();
-
-
- assertEquals( 0, ids.size() );
-
-
- log.info( "Total time to read {} entries {}ms", limit, timer.getTime() );
- }
-
-
- private interface EdgeGenerator {
-
- /**
- * Create a new edge to persiste
- */
- public Edge newEdge();
-
- public Observable<Edge> doSearch( final GraphManager manager );
- }
-}