Skip to content

Commit 4615152

Browse files
author
Zhen Li
committed
Added big data tests too
1 parent c8035ce commit 4615152

File tree

8 files changed

+71
-68
lines changed

8 files changed

+71
-68
lines changed

‎driver/src/test/java/org/neo4j/driver/stress/AbstractStressTestBase.java

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,11 +22,14 @@
2222
import org.junit.jupiter.api.AfterEach;
2323
import org.junit.jupiter.api.BeforeEach;
2424
import org.junit.jupiter.api.Test;
25+
import org.reactivestreams.Publisher;
26+
import reactor.core.publisher.Flux;
2527

2628
import java.lang.management.ManagementFactory;
2729
import java.lang.management.OperatingSystemMXBean;
2830
import java.lang.reflect.Method;
2931
import java.net.URI;
32+
import java.time.Duration;
3033
import java.util.ArrayList;
3134
import java.util.Arrays;
3235
import java.util.HashMap;
@@ -66,6 +69,8 @@
6669
import org.neo4j.driver.internal.util.Futures;
6770
import org.neo4j.driver.internal.util.Iterables;
6871
import org.neo4j.driver.internal.util.Neo4jFeature;
72+
import org.neo4j.driver.reactive.RxSession;
73+
import org.neo4j.driver.reactive.RxTransaction;
6974
import org.neo4j.driver.types.Node;
7075
import org.neo4j.driver.util.DaemonThreadFactory;
7176

@@ -91,6 +96,7 @@ abstract class AbstractStressTestBase<C extends AbstractContext>
9196

9297
private static final int BIG_DATA_TEST_NODE_COUNT = Integer.getInteger( "bigDataTestNodeCount", 30_000 );
9398
private static final int BIG_DATA_TEST_BATCH_SIZE = Integer.getInteger( "bigDataTestBatchSize", 10_000 );
99+
private static final Duration DEFAULT_BLOCKING_TIME_OUT = Duration.ofMinutes( 5 );
94100

95101
private LoggerNameTrackingLogging logging;
96102
private ExecutorService executor;
@@ -159,6 +165,14 @@ void asyncApiBigDataTest() throws Throwable
159165
readNodesAsync( driver, bookmark, BIG_DATA_TEST_NODE_COUNT );
160166
}
161167

168+
@Test
169+
@EnabledOnNeo4jWith( Neo4jFeature.BOLT_V4 )
170+
void rxApiBigDataTest() throws Throwable
171+
{
172+
String bookmark = createNodesRx( bigDataTestBatchCount(), BIG_DATA_TEST_BATCH_SIZE, driver );
173+
readNodesRx( driver, bookmark, BIG_DATA_TEST_NODE_COUNT );
174+
}
175+
162176
private void runStressTest( Function<C,List<Future<?>>> threadLauncher ) throws Throwable
163177
{
164178
C context = createContext();
@@ -603,6 +617,57 @@ private static void readNodesAsync( Driver driver, String bookmark, int expected
603617
System.out.println( "Reading nodes with async API took: " + NANOSECONDS.toMillis( end - start ) + "ms" );
604618
}
605619

620+
private String createNodesRx( int batchCount, int batchSize, InternalDriver driver )
621+
{
622+
long start = System.nanoTime();
623+
624+
RxSession session = driver.rxSession();
625+
626+
Flux.concat( Flux.range( 0, batchCount ).map( batchIndex ->
627+
session.writeTransaction( tx -> createNodesInTxRx( tx, batchIndex, batchSize ) )
628+
) ).blockLast( DEFAULT_BLOCKING_TIME_OUT ); // throw any error if happened
629+
630+
long end = System.nanoTime();
631+
System.out.println( "Node creation with reactive API took: " + NANOSECONDS.toMillis( end - start ) + "ms" );
632+
633+
return session.lastBookmark();
634+
}
635+
636+
private Publisher<Void> createNodesInTxRx( RxTransaction tx, int batchIndex, int batchSize )
637+
{
638+
return Flux.concat( Flux.range( 0, batchSize ).map( index -> batchIndex * batchSize + index ).map( nodeIndex -> {
639+
Statement statement = createNodeInTxStatement( nodeIndex );
640+
return Flux.from( tx.run( statement ).summary() ).then(); // As long as there is no error
641+
} ) );
642+
}
643+
644+
private void readNodesRx( InternalDriver driver, String bookmark, int expectedNodeCount )
645+
{
646+
long start = System.nanoTime();
647+
648+
RxSession session = driver.rxSession( t -> t.withBookmarks( bookmark ) );
649+
AtomicInteger nodesSeen = new AtomicInteger();
650+
651+
Publisher<Void> readQuery = session.readTransaction( tx -> Flux.from( tx.run( "MATCH (n:Node) RETURN n" ).records() ).doOnNext( record -> {
652+
Node node = record.get( 0 ).asNode();
653+
nodesSeen.incrementAndGet();
654+
655+
List<String> labels = Iterables.asList( node.labels() );
656+
assertEquals( 2, labels.size() );
657+
assertTrue( labels.contains( "Test" ) );
658+
assertTrue( labels.contains( "Node" ) );
659+
660+
verifyNodeProperties( node );
661+
} ).then() );
662+
663+
Flux.from( readQuery ).blockLast( DEFAULT_BLOCKING_TIME_OUT );
664+
665+
assertEquals( expectedNodeCount, nodesSeen.get() );
666+
667+
long end = System.nanoTime();
668+
System.out.println( "Reading nodes with async API took: " + NANOSECONDS.toMillis( end - start ) + "ms" );
669+
}
670+
606671
private static Void createNodesInTx( Transaction tx, int batchIndex, int batchSize )
607672
{
608673
for ( int index = 0; index < batchSize; index++ )

‎driver/src/test/java/org/neo4j/driver/stress/RxFailingQuery.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
package org.neo4j.driver.stress;
2020

2121
import reactor.core.publisher.Flux;
22-
import reactor.core.publisher.Mono;
2322

2423
import java.util.concurrent.CompletableFuture;
2524
import java.util.concurrent.CompletionStage;
@@ -46,7 +45,7 @@ public RxFailingQuery( Driver driver )
4645
public CompletionStage<Void> execute( C context )
4746
{
4847
CompletableFuture<Void> queryFinished = new CompletableFuture<>();
49-
Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.READ, context ) ),
48+
Flux.using( () -> newSession( AccessMode.READ, context ),
5049
session -> session.run( "UNWIND [10, 5, 0] AS x RETURN 10 / x" ).records(),
5150
RxSession::close )
5251
.subscribe( record -> {

‎driver/src/test/java/org/neo4j/driver/stress/RxFailingQueryWithRetries.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
package org.neo4j.driver.stress;
2020

2121
import reactor.core.publisher.Flux;
22-
import reactor.core.publisher.Mono;
2322

2423
import java.util.concurrent.CompletableFuture;
2524
import java.util.concurrent.CompletionStage;
@@ -46,7 +45,7 @@ public RxFailingQueryWithRetries( Driver driver )
4645
public CompletionStage<Void> execute( C context )
4746
{
4847
CompletableFuture<Void> queryFinished = new CompletableFuture<>();
49-
Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.READ, context ) ),
48+
Flux.using( () -> newSession( AccessMode.READ, context ),
5049
session -> session.readTransaction( tx -> tx.run( "UNWIND [10, 5, 0] AS x RETURN 10 / x" ).records() ),
5150
RxSession::close )
5251
.subscribe( record -> {

‎driver/src/test/java/org/neo4j/driver/stress/RxReadQuery.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ public RxReadQuery( Driver driver, boolean useBookmark )
4343
public CompletionStage<Void> execute( C context )
4444
{
4545
CompletableFuture<Void> queryFinished = new CompletableFuture<>();
46-
Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.READ, context ) ), this::processAndGetSummary, RxSession::close )
46+
Flux.using( () -> newSession( AccessMode.READ, context ), this::processAndGetSummary, RxSession::close )
4747
.subscribe( summary -> {
4848
context.readCompleted( summary );
4949
queryFinished.complete( null );

‎driver/src/test/java/org/neo4j/driver/stress/RxReadQueryWithRetries.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ public RxReadQueryWithRetries( Driver driver, boolean useBookmark )
4343
public CompletionStage<Void> execute( C context )
4444
{
4545
CompletableFuture<Void> queryFinished = new CompletableFuture<>();
46-
Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.READ, context ) ), this::processAndGetSummary, RxSession::close )
46+
Flux.using( () -> newSession( AccessMode.READ, context ), this::processAndGetSummary, RxSession::close )
4747
.subscribe( summary -> {
4848
queryFinished.complete( null );
4949
context.readCompleted( summary );

‎driver/src/test/java/org/neo4j/driver/stress/RxWriteQuery.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
package org.neo4j.driver.stress;
2020

2121
import reactor.core.publisher.Flux;
22-
import reactor.core.publisher.Mono;
2322

2423
import java.util.concurrent.CompletableFuture;
2524
import java.util.concurrent.CompletionStage;
@@ -45,7 +44,7 @@ public RxWriteQuery( AbstractStressTestBase<C> stressTest, Driver driver, boolea
4544
public CompletionStage<Void> execute( C context )
4645
{
4746
CompletableFuture<Void> queryFinished = new CompletableFuture<>();
48-
Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.READ, context ) ),
47+
Flux.using( () -> newSession( AccessMode.READ, context ),
4948
session -> session.run( "CREATE ()" ).summary(), RxSession::close )
5049
.subscribe( summary -> {
5150
queryFinished.complete( null );

‎driver/src/test/java/org/neo4j/driver/stress/RxWriteQueryWithRetries.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
package org.neo4j.driver.stress;
2020

2121
import reactor.core.publisher.Flux;
22-
import reactor.core.publisher.Mono;
2322

2423
import java.util.concurrent.CompletableFuture;
2524
import java.util.concurrent.CompletionStage;
@@ -45,7 +44,7 @@ public RxWriteQueryWithRetries( AbstractStressTestBase<C> stressTest, Driver dri
4544
public CompletionStage<Void> execute( C context )
4645
{
4746
CompletableFuture<Void> queryFinished = new CompletableFuture<>();
48-
Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.READ, context ) ),
47+
Flux.using( () -> newSession( AccessMode.READ, context ),
4948
session -> session.writeTransaction( tx -> tx.run( "CREATE ()" ).summary() ), RxSession::close )
5049
.subscribe( summary -> {
5150
queryFinished.complete( null );

‎examples/src/main/java/org/neo4j/docs/driver/RxAutocommitTransactionExample.java

Lines changed: 0 additions & 58 deletions
This file was deleted.

0 commit comments

Comments
 (0)