|
18 | 18 | */ |
19 | 19 | package org.neo4j.driver.tck.reactive; |
20 | 20 |
|
21 | | -import static reactor.adapter.JdkFlowAdapter.flowPublisherToFlux; |
22 | | - |
23 | 21 | import java.time.Duration; |
| 22 | +import java.util.Map; |
24 | 23 | import org.neo4j.driver.Driver; |
25 | | -import org.neo4j.driver.reactive.ReactiveResult; |
26 | | -import org.neo4j.driver.reactive.ReactiveSession; |
| 24 | +import org.neo4j.driver.Record; |
| 25 | +import org.neo4j.driver.reactivestreams.ReactiveResult; |
| 26 | +import org.neo4j.driver.reactivestreams.ReactiveSession; |
27 | 27 | import org.reactivestreams.Publisher; |
28 | 28 | import org.reactivestreams.tck.PublisherVerification; |
29 | 29 | import org.reactivestreams.tck.TestEnvironment; |
30 | 30 | import org.testng.annotations.AfterClass; |
31 | 31 | import org.testng.annotations.BeforeClass; |
32 | 32 | import reactor.core.publisher.Mono; |
33 | 33 |
|
34 | | -public class ReactiveResultPublisherVerificationIT extends PublisherVerification<ReactiveResult> { |
| 34 | +public class ReactiveResultPublisherVerificationIT extends PublisherVerification<Record> { |
35 | 35 | private final Neo4jManager NEO4J = new Neo4jManager(); |
36 | 36 | private static final Duration TIMEOUT = Duration.ofSeconds(10); |
37 | 37 | private static final Duration TIMEOUT_FOR_NO_SIGNALS = Duration.ofSeconds(1); |
@@ -63,15 +63,14 @@ public long maxElementsFromPublisher() { |
63 | 63 | } |
64 | 64 |
|
65 | 65 | @Override |
66 | | - public Publisher<ReactiveResult> createPublisher(long elements) { |
67 | | - ReactiveSession session = driver.session(ReactiveSession.class); |
68 | | - return Mono.from(flowPublisherToFlux(session.run("RETURN 1"))); |
| 66 | + public Publisher<Record> createPublisher(long elements) { |
| 67 | + var session = driver.session(ReactiveSession.class); |
| 68 | + return Mono.from(session.run("UNWIND range(0, $elements) AS x RETURN x", Map.of("elements", elements - 1))) |
| 69 | + .flatMapMany(ReactiveResult::records); |
69 | 70 | } |
70 | 71 |
|
71 | 72 | @Override |
72 | | - public Publisher<ReactiveResult> createFailedPublisher() { |
73 | | - ReactiveSession session = driver.session(ReactiveSession.class); |
74 | | - // Please note that this publisher fails on run stage. |
75 | | - return Mono.from(flowPublisherToFlux(session.run("RETURN 5/0"))); |
| 73 | + public Publisher<Record> createFailedPublisher() { |
| 74 | + return null; |
76 | 75 | } |
77 | 76 | } |
0 commit comments