Spring/spring-data-neo4j
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add support for Kotlins Coroutines to the ReactiveNeo4jClient.

Browse Source 
This adds either standard Coroutines support or the experimental Flow support of Kotlin 1.3 to ReactiveNeo4jClient where applicable and closes #38.
This commit is contained in:
Michael Simons
2019-09-25 18:00:45 +02:00
committed by Gerrit Meier
parent 3e0f95b942
commit 78abeaf6b1
10 changed files with 415 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-data-neo4j/pom.xml
View File

@@ -133,6 +133,16 @@
<artifactId>kotlin-reflect</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.jetbrains.kotlinx</groupId>
<artifactId>kotlinx-coroutines-core</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.jetbrains.kotlinx</groupId>
<artifactId>kotlinx-coroutines-reactor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.mockk</groupId>
<artifactId>mockk</artifactId>

18
spring-data-neo4j/src/main/java/org/neo4j/springframework/data/core/DefaultNeo4jClient.java
View File

@@ -148,11 +148,11 @@ class DefaultNeo4jClient implements Neo4jClient {
@Override
public <T> ExecutableQuery<T> toExecutableQuery(PreparedQuery<T> preparedQuery) {
Neo4jClient.MappingSpec<Optional<T>, Collection<T>, T> mappingSpec = this
Neo4jClient.MappingSpec<T> mappingSpec = this
.query(preparedQuery.getCypherQuery())
.bindAll(preparedQuery.getParameters())
.fetchAs(preparedQuery.getResultType());
Neo4jClient.RecordFetchSpec<Optional<T>, Collection<T>, T> fetchSpec = preparedQuery
Neo4jClient.RecordFetchSpec<T> fetchSpec = preparedQuery
.getOptionalMappingFunction()
.map(f -> mappingSpec.mappedBy(f))
.orElse(mappingSpec);
@@ -249,14 +249,14 @@ class DefaultNeo4jClient implements Neo4jClient {
}
@Override
public <T> MappingSpec<Optional<T>, Collection<T>, T> fetchAs(Class<T> targetClass) {
public <T> MappingSpec<T> fetchAs(Class<T> targetClass) {
return new DefaultRecordFetchSpec(this.targetDatabase, this.runnableStatement,
new SingleValueMappingFunction(conversionService, targetClass));
}
@Override
public RecordFetchSpec<Optional<Map<String, Object>>, Collection<Map<String, Object>>, Map<String, Object>> fetch() {
public RecordFetchSpec<Map<String, Object>> fetch() {
return new DefaultRecordFetchSpec<>(
this.targetDatabase,
@@ -273,8 +273,7 @@ class DefaultNeo4jClient implements Neo4jClient {
}
}
class DefaultRecordFetchSpec<T>
implements RecordFetchSpec<Optional<T>, Collection<T>, T>, MappingSpec<Optional<T>, Collection<T>, T> {
class DefaultRecordFetchSpec<T> implements RecordFetchSpec<T>, MappingSpec<T> {
private final String targetDatabase;
@@ -290,7 +289,7 @@ class DefaultNeo4jClient implements Neo4jClient {
}
@Override
public RecordFetchSpec<Optional<T>, Collection<T>, T> mappedBy(
public RecordFetchSpec<T> mappedBy(
@SuppressWarnings("HiddenField") BiFunction<TypeSystem, Record, T> mappingFunction) {
this.mappingFunction = new DelegatingMappingFunctionWithNullCheck<>(mappingFunction);
@@ -368,10 +367,9 @@ class DefaultNeo4jClient implements Neo4jClient {
final class DefaultExecutableQuery<T> implements ExecutableQuery<T> {
private final PreparedQuery<T> preparedQuery;
private final Neo4jClient.RecordFetchSpec<Optional<T>, Collection<T>, T> fetchSpec;
private final Neo4jClient.RecordFetchSpec<T> fetchSpec;
DefaultExecutableQuery(PreparedQuery<T> preparedQuery,
RecordFetchSpec<Optional<T>, Collection<T>, T> fetchSpec) {
DefaultExecutableQuery(PreparedQuery<T> preparedQuery, RecordFetchSpec<T> fetchSpec) {
this.preparedQuery = preparedQuery;
this.fetchSpec = fetchSpec;
}

74
spring-data-neo4j/src/main/java/org/neo4j/springframework/data/core/DefaultReactiveNeo4jClient.java
View File

@@ -38,9 +38,7 @@ import org.neo4j.driver.reactive.RxSession;
import org.neo4j.driver.reactive.RxStatementRunner;
import org.neo4j.driver.summary.ResultSummary;
import org.neo4j.driver.types.TypeSystem;
import org.neo4j.springframework.data.core.Neo4jClient.MappingSpec;
import org.neo4j.springframework.data.core.Neo4jClient.OngoingBindSpec;
import org.neo4j.springframework.data.core.Neo4jClient.RecordFetchSpec;
import org.neo4j.springframework.data.core.convert.Neo4jConversions;
import org.reactivestreams.Publisher;
import org.springframework.core.convert.ConversionService;
@@ -89,8 +87,8 @@ class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
return Mono.usingWhen(retrieveRxStatementRunnerHolder(targetDatabase),
holder -> func.apply(holder.getRxStatementRunner()),
RxStatementRunnerHolder::getCommit,
RxStatementRunnerHolder::getRollback);
(holder, ex) -> holder.getRollback(),
RxStatementRunnerHolder::getCommit);
}
<T> Flux<T> doInStatementRunnerForFlux(final String targetDatabase, Function<RxStatementRunner, Flux<T>> func) {
@@ -98,34 +96,35 @@ class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
return Flux.usingWhen(retrieveRxStatementRunnerHolder(targetDatabase),
holder -> func.apply(holder.getRxStatementRunner()),
RxStatementRunnerHolder::getCommit,
RxStatementRunnerHolder::getRollback);
(holder, ex) -> holder.getRollback(),
RxStatementRunnerHolder::getCommit);
}
@Override
public ReactiveRunnableSpec query(String cypher) {
public RunnableSpec query(String cypher) {
return query(() -> cypher);
}
@Override
public ReactiveRunnableSpec query(Supplier<String> cypherSupplier) {
return new DefaultReactiveRunnableSpec(cypherSupplier);
public RunnableSpec query(Supplier<String> cypherSupplier) {
return new DefaultRunnableSpec(cypherSupplier);
}
@Override
public <T> OngoingReactiveDelegation<T> delegateTo(Function<RxStatementRunner, Mono<T>> callback) {
return new DefaultReactiveRunnableDelegation<>(callback);
public <T> OngoingDelegation<T> delegateTo(Function<RxStatementRunner, Mono<T>> callback) {
return new DefaultRunnableDelegation<>(callback);
}
@Override
public <T> ExecutableQuery<T> toExecutableQuery(PreparedQuery<T> preparedQuery) {
Class<T> resultType = preparedQuery.getResultType();
Neo4jClient.MappingSpec<Mono<T>, Flux<T>, T> mappingSpec = this
MappingSpec<T> mappingSpec = this
.query(preparedQuery.getCypherQuery())
.bindAll(preparedQuery.getParameters())
.fetchAs(resultType);
Neo4jClient.RecordFetchSpec<Mono<T>, Flux<T>, T> fetchSpec = preparedQuery
RecordFetchSpec<T> fetchSpec = preparedQuery
.getOptionalMappingFunction()
.map(mappingFunction -> mappingSpec.mappedBy(mappingFunction))
.orElse(mappingSpec);
@@ -133,7 +132,7 @@ class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
return new DefaultReactiveExecutableQuery<>(fetchSpec);
}
class DefaultReactiveRunnableSpec implements ReactiveRunnableSpec {
class DefaultRunnableSpec implements RunnableSpec {
private final Supplier<String> cypherSupplier;
@@ -141,18 +140,18 @@ class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
private final NamedParameters parameters = new NamedParameters();
DefaultReactiveRunnableSpec(Supplier<String> cypherSupplier) {
DefaultRunnableSpec(Supplier<String> cypherSupplier) {
this.cypherSupplier = cypherSupplier;
}
@Override
public ReactiveRunnableSpecTightToDatabase in(@SuppressWarnings("HiddenField") String targetDatabase) {
public RunnableSpecTightToDatabase in(@SuppressWarnings("HiddenField") String targetDatabase) {
this.targetDatabase = verifyDatabaseName(targetDatabase);
return this;
}
class DefaultOngoingBindSpec<T> implements OngoingBindSpec<T, ReactiveRunnableSpecTightToDatabase> {
class DefaultOngoingBindSpec<T> implements OngoingBindSpec<T, RunnableSpecTightToDatabase> {
@Nullable
private final T value;
@@ -162,14 +161,14 @@ class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
}
@Override
public ReactiveRunnableSpecTightToDatabase to(String name) {
public RunnableSpecTightToDatabase to(String name) {
DefaultReactiveRunnableSpec.this.parameters.add(name, value);
return DefaultReactiveRunnableSpec.this;
DefaultRunnableSpec.this.parameters.add(name, value);
return DefaultRunnableSpec.this;
}
@Override
public ReactiveRunnableSpecTightToDatabase with(Function<T, Map<String, Object>> binder) {
public RunnableSpecTightToDatabase with(Function<T, Map<String, Object>> binder) {
Assert.notNull(binder, "Binder is required.");
@@ -178,42 +177,41 @@ class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
}
@Override
public OngoingBindSpec<?, ReactiveRunnableSpecTightToDatabase> bind(@Nullable Object value) {
public OngoingBindSpec<?, RunnableSpecTightToDatabase> bind(@Nullable Object value) {
return new DefaultOngoingBindSpec(value);
}
@Override
public ReactiveRunnableSpecTightToDatabase bindAll(Map<String, Object> newParameters) {
public RunnableSpecTightToDatabase bindAll(Map<String, Object> newParameters) {
this.parameters.addAll(newParameters);
return this;
}
@Override
public <R> MappingSpec<Mono<R>, Flux<R>, R> fetchAs(Class<R> targetClass) {
public <R> MappingSpec<R> fetchAs(Class<R> targetClass) {
return new DefaultReactiveRecordFetchSpec<>(this.targetDatabase, this.cypherSupplier, this.parameters,
return new DefaultRecordFetchSpec<>(this.targetDatabase, this.cypherSupplier, this.parameters,
new SingleValueMappingFunction(conversionService, targetClass));
}
@Override
public RecordFetchSpec<Mono<Map<String, Object>>, Flux<Map<String, Object>>, Map<String, Object>> fetch() {
public RecordFetchSpec<Map<String, Object>> fetch() {
return new DefaultReactiveRecordFetchSpec<>(targetDatabase, cypherSupplier, parameters,
return new DefaultRecordFetchSpec<>(targetDatabase, cypherSupplier, parameters,
(t, r) -> r.asMap());
}
@Override
public Mono<ResultSummary> run() {
return new DefaultReactiveRecordFetchSpec<>(
return new DefaultRecordFetchSpec<>(
this.targetDatabase,
this.cypherSupplier,
this.parameters).run();
}
}
class DefaultReactiveRecordFetchSpec<T>
implements RecordFetchSpec<Mono<T>, Flux<T>, T>, MappingSpec<Mono<T>, Flux<T>, T> {
class DefaultRecordFetchSpec<T> implements RecordFetchSpec<T>, MappingSpec<T> {
private final String targetDatabase;
@@ -223,12 +221,12 @@ class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
private BiFunction<TypeSystem, Record, T> mappingFunction;
DefaultReactiveRecordFetchSpec(String targetDatabase, Supplier<String> cypherSupplier,
DefaultRecordFetchSpec(String targetDatabase, Supplier<String> cypherSupplier,
NamedParameters parameters) {
this(targetDatabase, cypherSupplier, parameters, null);
}
DefaultReactiveRecordFetchSpec(
DefaultRecordFetchSpec(
String targetDatabase, Supplier<String> cypherSupplier, NamedParameters parameters,
@Nullable BiFunction<TypeSystem, Record, T> mappingFunction) {
this.targetDatabase = targetDatabase;
@@ -238,7 +236,7 @@ class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
}
@Override
public RecordFetchSpec<Mono<T>, Flux<T>, T> mappedBy(BiFunction<TypeSystem, Record, T> mappingFunction) {
public RecordFetchSpec<T> mappedBy(BiFunction<TypeSystem, Record, T> mappingFunction) {
this.mappingFunction = new DelegatingMappingFunctionWithNullCheck<>(mappingFunction);
return this;
@@ -294,24 +292,24 @@ class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
}
}
class DefaultReactiveRunnableDelegation<T> implements ReactiveRunnableDelegation<T>, OngoingReactiveDelegation<T> {
class DefaultRunnableDelegation<T> implements RunnableDelegation<T>, OngoingDelegation<T> {
private final Function<RxStatementRunner, Mono<T>> callback;
private String targetDatabase;
DefaultReactiveRunnableDelegation(Function<RxStatementRunner, Mono<T>> callback) {
DefaultRunnableDelegation(Function<RxStatementRunner, Mono<T>> callback) {
this(callback, null);
}
DefaultReactiveRunnableDelegation(Function<RxStatementRunner, Mono<T>> callback,
DefaultRunnableDelegation(Function<RxStatementRunner, Mono<T>> callback,
@Nullable String targetDatabase) {
this.callback = callback;
this.targetDatabase = targetDatabase;
}
@Override
public ReactiveRunnableDelegation in(@Nullable @SuppressWarnings("HiddenField") String targetDatabase) {
public RunnableDelegation in(@Nullable @SuppressWarnings("HiddenField") String targetDatabase) {
this.targetDatabase = verifyDatabaseName(targetDatabase);
return this;
@@ -330,9 +328,9 @@ class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
final class DefaultReactiveExecutableQuery<T> implements ExecutableQuery<T> {
private final Neo4jClient.RecordFetchSpec<Mono<T>, Flux<T>, T> fetchSpec;
private final RecordFetchSpec<T> fetchSpec;
DefaultReactiveExecutableQuery(RecordFetchSpec<Mono<T>, Flux<T>, T> fetchSpec) {
DefaultReactiveExecutableQuery(RecordFetchSpec<T> fetchSpec) {
this.fetchSpec = fetchSpec;
}

20
spring-data-neo4j/src/main/java/org/neo4j/springframework/data/core/Neo4jClient.java
View File

@@ -124,14 +124,14 @@ public interface Neo4jClient {
* @param <T> The type of the class
* @return A mapping spec that allows specifying a mapping function.
*/
<T> MappingSpec<Optional<T>, Collection<T>, T> fetchAs(Class<T> targetClass);
<T> MappingSpec<T> fetchAs(Class<T> targetClass);
/**
* Fetch all records mapped into generic maps
*
* @return A fetch specification that maps into generic maps.
*/
RecordFetchSpec<Optional<Map<String, Object>>, Collection<Map<String, Object>>, Map<String, Object>> fetch();
RecordFetchSpec<Map<String, Object>> fetch();
/**
* Execute the query and discard the results. It returns the drivers result summary, including various counters
@@ -186,12 +186,10 @@ public interface Neo4jClient {
}
/**
* @param <S> The type of the class holding zero or one result element
* @param <M> The type of the class holding zero or more result elements
* @param <T> The resulting type of this mapping
* @since 1.0
*/
interface MappingSpec<S, M, T> extends RecordFetchSpec<S, M, T> {
interface MappingSpec<T> extends RecordFetchSpec<T> {
/**
* The mapping function is responsible to turn one record into one domain object. It will receive the record
@@ -200,37 +198,35 @@ public interface Neo4jClient {
* @param mappingFunction The mapping function used to create new domain objects
* @return A specification how to fetch one or more records.
*/
RecordFetchSpec<S, M, T> mappedBy(BiFunction<TypeSystem, Record, T> mappingFunction);
RecordFetchSpec<T> mappedBy(BiFunction<TypeSystem, Record, T> mappingFunction);
}
/**
* @param <S> The type of the class holding zero or one result element
* @param <M> The type of the class holding zero or more result elements
* @param <T> The type to which the fetched records are eventually mapped
* @since 1.0
*/
interface RecordFetchSpec<S, M, T> {
interface RecordFetchSpec<T> {
/**
* Fetches exactly one record and throws an exception if there are more entries.
*
* @return The one and only record.
*/
S one();
Optional<T> one();
/**
* Fetches only the first record. Returns an empty holder if there are no records.
*
* @return The first record if any.
*/
S first();
Optional<T> first();
/**
* Fetches all records.
*
* @return All records.
*/
M all();
Collection<T> all();
}
/**

71
spring-data-neo4j/src/main/java/org/neo4j/springframework/data/core/ReactiveNeo4jClient.java
View File

@@ -22,17 +22,18 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.Map;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;
import org.apache.commons.logging.LogFactory;
import org.apiguardian.api.API;
import org.neo4j.driver.Driver;
import org.neo4j.driver.Record;
import org.neo4j.driver.reactive.RxStatementRunner;
import org.neo4j.driver.summary.ResultSummary;
import org.neo4j.driver.types.TypeSystem;
import org.neo4j.springframework.data.core.Neo4jClient.BindSpec;
import org.neo4j.springframework.data.core.Neo4jClient.MappingSpec;
import org.neo4j.springframework.data.core.Neo4jClient.RecordFetchSpec;
import org.springframework.core.log.LogAccessor;
import org.springframework.dao.IncorrectResultSizeDataAccessException;
@@ -61,7 +62,7 @@ public interface ReactiveNeo4jClient {
* @param cypher The cypher code that shall be executed
* @return A new CypherSpec
*/
ReactiveRunnableSpec query(String cypher);
RunnableSpec query(String cypher);
/**
* Entrypoint for creating a new Cypher query based on a supplier. Doesn't matter at this point whether it's a match,
@@ -71,7 +72,7 @@ public interface ReactiveNeo4jClient {
* @param cypherSupplier A supplier of arbitrary Cypher code
* @return A runnable query specification.
*/
ReactiveRunnableSpec query(Supplier<String> cypherSupplier);
RunnableSpec query(Supplier<String> cypherSupplier);
/**
* Delegates interaction with the default database to the given callback.
@@ -80,7 +81,7 @@ public interface ReactiveNeo4jClient {
* @param <T> The type of the result being produced
* @return A single publisher containing none or exactly one element that will be produced by the callback
*/
<T> OngoingReactiveDelegation<T> delegateTo(Function<RxStatementRunner, Mono<T>> callback);
<T> OngoingDelegation<T> delegateTo(Function<RxStatementRunner, Mono<T>> callback);
/**
* Takes a prepared query, containing all the information about the cypher template to be used, needed parameters and
@@ -92,11 +93,55 @@ public interface ReactiveNeo4jClient {
*/
<T> ExecutableQuery<T> toExecutableQuery(PreparedQuery<T> preparedQuery);
/**
* @param <T> The resulting type of this mapping
* @since 1.0
*/
interface MappingSpec<T> extends RecordFetchSpec<T> {
/**
* The mapping function is responsible to turn one record into one domain object. It will receive the record
* itself and in addition, the type system that the Neo4j Java-Driver used while executing the query.
*
* @param mappingFunction The mapping function used to create new domain objects
* @return A specification how to fetch one or more records.
*/
RecordFetchSpec<T> mappedBy(BiFunction<TypeSystem, Record, T> mappingFunction);
}
/**
* @param <T> The type to which the fetched records are eventually mapped
* @since 1.0
*/
interface RecordFetchSpec<T> {
/**
* Fetches exactly one record and throws an exception if there are more entries.
*
* @return The one and only record.
*/
Mono<T> one();
/**
* Fetches only the first record. Returns an empty holder if there are no records.
*
* @return The first record if any.
*/
Mono<T> first();
/**
* Fetches all records.
*
* @return All records.
*/
Flux<T> all();
}
/**
* Contract for a runnable query that can be either run returning it's result, run without results or be parameterized.
* @since 1.0
*/
interface ReactiveRunnableSpec extends ReactiveRunnableSpecTightToDatabase {
interface RunnableSpec extends RunnableSpecTightToDatabase {
/**
* Pins the previously defined query to a specific database.
@@ -104,14 +149,14 @@ public interface ReactiveNeo4jClient {
* @param targetDatabase selected database to use
* @return A runnable query specification that is now tight to a given database.
*/
ReactiveRunnableSpecTightToDatabase in(String targetDatabase);
RunnableSpecTightToDatabase in(String targetDatabase);
}
/**
* Contract for a runnable query inside a dedicated database.
* @since 1.0
*/
interface ReactiveRunnableSpecTightToDatabase extends BindSpec<ReactiveRunnableSpecTightToDatabase> {
interface RunnableSpecTightToDatabase extends BindSpec<RunnableSpecTightToDatabase> {
/**
* Create a mapping for each record return to a specific type.
@@ -120,14 +165,14 @@ public interface ReactiveNeo4jClient {
* @param <T> The type of the class
* @return A mapping spec that allows specifying a mapping function
*/
<T> MappingSpec<Mono<T>, Flux<T>, T> fetchAs(Class<T> targetClass);
<T> MappingSpec<T> fetchAs(Class<T> targetClass);
/**
* Fetch all records mapped into generic maps
*
* @return A fetch specification that maps into generic maps
*/
RecordFetchSpec<Mono<Map<String, Object>>, Flux<Map<String, Object>>, Map<String, Object>> fetch();
RecordFetchSpec<Map<String, Object>> fetch();
/**
* Execute the query and discard the results. It returns the drivers result summary, including various counters
@@ -144,7 +189,7 @@ public interface ReactiveNeo4jClient {
* @param <T> The type of the returned value.
* @since 1.0
*/
interface OngoingReactiveDelegation<T> extends ReactiveRunnableDelegation<T> {
interface OngoingDelegation<T> extends RunnableDelegation<T> {
/**
* Runs the delegation in the given target database.
@@ -152,7 +197,7 @@ public interface ReactiveNeo4jClient {
* @param targetDatabase selected database to use
* @return An ongoing delegation
*/
ReactiveRunnableDelegation<T> in(String targetDatabase);
RunnableDelegation<T> in(String targetDatabase);
}
/**
@@ -161,7 +206,7 @@ public interface ReactiveNeo4jClient {
* @param <T> the type that gets returned by the query
* @since 1.0
*/
interface ReactiveRunnableDelegation<T> {
interface RunnableDelegation<T> {
/**
* Runs the stored callback.

32
spring-data-neo4j/src/main/kotlin/org/neo4j/springframework/data/core/Neo4jClientExtensions.kt
View File

@@ -42,30 +42,30 @@ fun <T : Any?> Neo4jClient.OngoingDelegation<T>.inDatabase(targetDatabase: Strin
`in`(targetDatabase)
/**
* An implementation of a fetch spec that replaces Java's Optional with a nullable.
* A fetch spec that replaces Java's Optional with a nullable.
* @author Michael J. Simons
*/
class DelegatingFetchSpec<T : Any>(private val delegate: Neo4jClient.RecordFetchSpec<Optional<T>, Collection<T>, T>) : Neo4jClient.RecordFetchSpec<T?, Collection<T>, T> {
override fun one(): T? = delegate.one().orElse(null)
class KRecordFetchSpec<T : Any> (private val delegate: Neo4jClient.RecordFetchSpec<T>) {
fun one(): T? = delegate.one().orElse(null)
override fun first(): T = delegate.first().orElse(null)
fun first(): T = delegate.first().orElse(null)
override fun all(): Collection<T> = delegate.all()
fun all(): Collection<T> = delegate.all()
}
/**
* An implementation of a mapping spec that replaces Java's Optional with a nullable.
* A mapping spec that replaces Java's Optional with a nullable.
* @author Michael J. Simons
*/
class DelegatingMappingSpec<T : Any>(private val delegate: Neo4jClient.MappingSpec<Optional<T>, Collection<T>, T>) : Neo4jClient.MappingSpec<T?, Collection<T>, T> {
override fun mappedBy(mappingFunction: BiFunction<TypeSystem, Record, T>): Neo4jClient.RecordFetchSpec<T?, Collection<T>, T> =
DelegatingFetchSpec(delegate.mappedBy(mappingFunction))
class KMappingSpec<T : Any>(private val delegate: Neo4jClient.MappingSpec<T>) {
fun mappedBy(mappingFunction: BiFunction<TypeSystem, Record, T>): KRecordFetchSpec<T> =
KRecordFetchSpec(delegate.mappedBy(mappingFunction))
override fun one(): T? = delegate.one().orElse(null)
fun one(): T? = delegate.one().orElse(null)
override fun first(): T = delegate.first().orElse(null)
fun first(): T = delegate.first().orElse(null)
override fun all(): Collection<T> = delegate.all()
fun all(): Collection<T> = delegate.all()
}
/**
@@ -73,8 +73,8 @@ class DelegatingMappingSpec<T : Any>(private val delegate: Neo4jClient.MappingSp
* @author Michael J. Simons
* @since 1.0
*/
inline fun <reified T : Any> Neo4jClient.RunnableSpecTightToDatabase.fetchAs(): Neo4jClient.MappingSpec<T?, Collection<T>, T>
= DelegatingMappingSpec(fetchAs(T::class.java))
inline fun <reified T : Any> Neo4jClient.RunnableSpecTightToDatabase.fetchAs(): KMappingSpec<T> =
KMappingSpec(fetchAs(T::class.java))
/**
* Extension for [Neo4jClient.RunnableSpecTightToDatabase.mappedBy] leveraging reified type parameters and removing
@@ -82,5 +82,5 @@ inline fun <reified T : Any> Neo4jClient.RunnableSpecTightToDatabase.fetchAs():
* @author Michael J. Simons
* @since 1.0
*/
inline fun <reified T : Any> Neo4jClient.RunnableSpecTightToDatabase.mappedBy(noinline mappingFunction: (TypeSystem, Record) -> T): Neo4jClient.RecordFetchSpec<T?, Collection<T>, T>
= DelegatingFetchSpec(fetchAs(T::class.java).mappedBy(mappingFunction))
inline fun <reified T : Any> Neo4jClient.RunnableSpecTightToDatabase.mappedBy(noinline mappingFunction: (TypeSystem, Record) -> T)
= KRecordFetchSpec(fetchAs(T::class.java).mappedBy(mappingFunction))

81
spring-data-neo4j/src/main/kotlin/org/neo4j/springframework/data/core/ReactiveNeo4jClientExtensions.kt
View File

@@ -18,27 +18,31 @@
*/
package org.neo4j.springframework.data.core
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.reactive.asFlow
import kotlinx.coroutines.reactive.awaitFirstOrNull
import kotlinx.coroutines.reactive.awaitSingle
import org.neo4j.driver.Record
import org.neo4j.driver.summary.ResultSummary
import org.neo4j.driver.types.TypeSystem
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
/**
* Extension for [ReactiveNeo4jClient.ReactiveRunnableSpec.in] providing an `inDatabase` alias since `in` is a reserved keyword in Kotlin.
* Extension for [ReactiveNeo4jClient.RunnableSpec.in] providing an `inDatabase` alias since `in` is a reserved keyword in Kotlin.
*
* @author Michael J. Simons
* @since 1.0
*/
fun ReactiveNeo4jClient.ReactiveRunnableSpec.inDatabase(targetDatabase: String): ReactiveNeo4jClient.ReactiveRunnableSpecTightToDatabase =
`in`(targetDatabase)
fun ReactiveNeo4jClient.RunnableSpec.inDatabase(targetDatabase: String): ReactiveNeo4jClient.RunnableSpecTightToDatabase
= `in`(targetDatabase)
/**
* Extension for [ReactiveNeo4jClient.OngoingReactiveDelegation.in] providing an `inDatabase` alias since `in` is a reserved keyword in Kotlin.
* Extension for [ReactiveNeo4jClient.OngoingDelegation.in] providing an `inDatabase` alias since `in` is a reserved keyword in Kotlin.
*
* @author Michael J. Simons
* @since 1.0
*/
fun <T : Any?> ReactiveNeo4jClient.OngoingReactiveDelegation<T>.inDatabase(targetDatabase: String): ReactiveNeo4jClient.ReactiveRunnableDelegation<T>
fun <T : Any?> ReactiveNeo4jClient.OngoingDelegation<T>.inDatabase(targetDatabase: String): ReactiveNeo4jClient.RunnableDelegation<T>
= `in`(targetDatabase)
/**
@@ -46,7 +50,7 @@ fun <T : Any?> ReactiveNeo4jClient.OngoingReactiveDelegation<T>.inDatabase(targe
* @author Michael J. Simons
* @since 1.0
*/
inline fun <reified T : Any> ReactiveNeo4jClient.ReactiveRunnableSpecTightToDatabase.fetchAs(): Neo4jClient.MappingSpec<Mono<T>, Flux<T>, T>
inline fun <reified T : Any> ReactiveNeo4jClient.RunnableSpecTightToDatabase.fetchAs(): ReactiveNeo4jClient.MappingSpec<T>
= fetchAs(T::class.java)
/**
@@ -55,5 +59,64 @@ inline fun <reified T : Any> ReactiveNeo4jClient.ReactiveRunnableSpecTightToData
* @author Michael J. Simons
* @since 1.0
*/
inline fun <reified T : Any> ReactiveNeo4jClient.ReactiveRunnableSpecTightToDatabase.mappedBy(noinline mappingFunction: (TypeSystem, Record) -> T): Neo4jClient.RecordFetchSpec<Mono<T>, Flux<T>, T>
inline fun <reified T : Any> ReactiveNeo4jClient.RunnableSpecTightToDatabase.mappedBy(noinline mappingFunction: (TypeSystem, Record) -> T): ReactiveNeo4jClient.RecordFetchSpec<T>
= fetchAs(T::class.java).mappedBy(mappingFunction)
/**
* Non-nullable Coroutines variant of [ReactiveNeo4jClient.RunnableSpecTightToDatabase.run].
*
* @author Michael J. Simons
* @since 1.0
*/
suspend inline fun ReactiveNeo4jClient.RunnableSpecTightToDatabase.await(): ResultSummary =
run().awaitSingle()
/**
* Nullable Coroutines variant of [ReactiveNeo4jClient.RecordFetchSpec.one].
*
* @author Michael J. Simons
* @since 1.0
*/
suspend inline fun <reified T : Any> ReactiveNeo4jClient.RecordFetchSpec<T>.awaitOneOrNull(): T? = one().awaitFirstOrNull()
/**
* Nullable Coroutines variant of [ReactiveNeo4jClient.RecordFetchSpec.first].
*
* @author Michael J. Simons
* @since 1.0
*/
suspend inline fun <reified T : Any> ReactiveNeo4jClient.RecordFetchSpec<T>.awaitFirstOrNull(): T? = first().awaitFirstOrNull()
/**
* Coroutines [Flow] variant of [ReactiveNeo4jClient.RecordFetchSpec.all].
*
* @author Michael J. Simons
* @since 1.0
*/
@ExperimentalCoroutinesApi
inline fun <reified T : Any> ReactiveNeo4jClient.RecordFetchSpec<T>.fetchAll(): Flow<T> = all().asFlow()
/**
* Coroutines [Flow] variant of [ReactiveNeo4jClient.ExecutableQuery.getResults].
*
* @author Michael J. Simons
* @since 1.0
*/
@ExperimentalCoroutinesApi
inline fun <reified T : Any> ReactiveNeo4jClient.ExecutableQuery<T>.fetchAllResults(): Flow<T> = results.asFlow()
/**
* Nullable Coroutines variant of [ReactiveNeo4jClient.ExecutableQuery.getSingleResult].
*
* @author Michael J. Simons
* @since 1.0
*/
suspend inline fun <reified T : Any> ReactiveNeo4jClient.ExecutableQuery<T>.awaitSingleResultOrNull(): T? = singleResult.awaitFirstOrNull()
/**
* Nullable Coroutines variant of [ReactiveNeo4jClient.RunnableDelegation.run].
*
* @author Michael J. Simons
* @since 1.0
*/
suspend inline fun <reified T : Any> ReactiveNeo4jClient.RunnableDelegation<T>.awaitFirstOrNull(): T? = run().awaitFirstOrNull()

4
spring-data-neo4j/src/test/kotlin/org/neo4j/springframework/data/core/Neo4jClientExtensionsTest.kt
View File

@@ -52,8 +52,8 @@ class Neo4jClientExtensionsTest {
val runnableSpec = mockk<Neo4jClient.RunnableSpecTightToDatabase>(relaxed = true)
val mappingSpec: Neo4jClient.RecordFetchSpec<String?, Collection<String>, String> =
runnableSpec.mappedBy { _, record -> "Foo" };
val mappingSpec: KRecordFetchSpec<String> =
runnableSpec.mappedBy { _, _ -> "Foo" };
verify(exactly = 1) { runnableSpec.fetchAs(String::class.java) }
}

178
spring-data-neo4j/src/test/kotlin/org/neo4j/springframework/data/core/ReactiveNeo4jClientExtensionsTest.kt
View File

@@ -18,9 +18,15 @@
*/
package org.neo4j.springframework.data.core
import io.mockk.every
import io.mockk.mockk
import io.mockk.verify
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.Nested
import org.junit.jupiter.api.Test
import org.neo4j.driver.summary.ResultSummary
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
@@ -32,7 +38,7 @@ class ReactiveNeo4jClientExtensionsTest {
@Test
fun `RunnableSpec#inDatabase(targetDatabase) extension should call its Java counterpart`() {
val runnableSpec = mockk<ReactiveNeo4jClient.ReactiveRunnableSpec>(relaxed = true)
val runnableSpec = mockk<ReactiveNeo4jClient.RunnableSpec>(relaxed = true)
runnableSpec.inDatabase("foobar");
@@ -42,7 +48,7 @@ class ReactiveNeo4jClientExtensionsTest {
@Test
fun `OngoingDelegation#inDatabase(targetDatabase) extension should call its Java counterpart`() {
val ongoingDelegation = mockk<ReactiveNeo4jClient.OngoingReactiveDelegation<Any>>(relaxed = true)
val ongoingDelegation = mockk<ReactiveNeo4jClient.OngoingDelegation<Any>>(relaxed = true)
ongoingDelegation.inDatabase("foobar");
@@ -52,11 +58,173 @@ class ReactiveNeo4jClientExtensionsTest {
@Test
fun `ReactiveRunnableDelegation#fetchAs() extension should call its Java counterpart`() {
val runnableSpec = mockk<ReactiveNeo4jClient.ReactiveRunnableSpecTightToDatabase>(relaxed = true)
val runnableSpec = mockk<ReactiveNeo4jClient.RunnableSpecTightToDatabase>(relaxed = true)
val mappingSpec : Neo4jClient.MappingSpec<Mono<String>, Flux<String>, String> =
runnableSpec.fetchAs();
val mappingSpec: ReactiveNeo4jClient.MappingSpec<String> = runnableSpec.fetchAs();
verify(exactly = 1) { runnableSpec.fetchAs(String::class.java) }
}
@Test
fun runnableSpecShouldReturnSuspendedResultSummary() {
val runnableSpec = mockk<ReactiveNeo4jClient.RunnableSpecTightToDatabase>()
val resultSummary = mockk<ResultSummary>()
every { runnableSpec.run() } returns Mono.just(resultSummary)
runBlocking {
assertThat(runnableSpec.await()).isEqualTo(resultSummary)
}
verify {
runnableSpec.run()
}
}
@Nested
inner class CoroutinesVariantsOfRunnableDelegation {
private val runnableDelegation = mockk<ReactiveNeo4jClient.RunnableDelegation<String>>()
@Test
fun `awaitFirstOrNull should return value`() {
every { runnableDelegation.run() } returns Mono.just("bazbar")
runBlocking {
assertThat(runnableDelegation.awaitFirstOrNull()).isEqualTo("bazbar")
}
verify {
runnableDelegation.run()
}
}
@Test
fun `awaitFirstOrNull should return null`() {
every { runnableDelegation.run() } returns Mono.empty()
runBlocking {
assertThat(runnableDelegation.awaitFirstOrNull()).isNull()
}
verify {
runnableDelegation.run()
}
}
}
@Nested
inner class CoroutinesVariantsOfRecordFetchSpec {
private val recordFetchSpec = mockk<ReactiveNeo4jClient.RecordFetchSpec<String>>()
@Test
fun `awaitOne should return value`() {
every { recordFetchSpec.one() } returns Mono.just("foo")
runBlocking {
assertThat(recordFetchSpec.awaitOneOrNull()).isEqualTo("foo")
}
verify {
recordFetchSpec.one()
}
}
@Test
fun `awaitOne should return null`() {
every { recordFetchSpec.one() } returns Mono.empty()
runBlocking {
assertThat(recordFetchSpec.awaitOneOrNull()).isNull()
}
verify {
recordFetchSpec.one()
}
}
@Test
fun `awaitFirstOrNull should return value`() {
every { recordFetchSpec.first() } returns Mono.just("bar")
runBlocking {
assertThat(recordFetchSpec.awaitFirstOrNull()).isEqualTo("bar")
}
verify {
recordFetchSpec.first()
}
}
@Test
fun `awaitFirstOrNull should return null`() {
every { recordFetchSpec.first() } returns Mono.empty()
runBlocking {
assertThat(recordFetchSpec.awaitFirstOrNull()).isNull()
}
verify {
recordFetchSpec.first()
}
}
@Test
fun `fetchAll should return a flow of thing`() {
every { recordFetchSpec.all() } returns Flux.just("foo", "bar")
runBlocking {
assertThat(recordFetchSpec.fetchAll().toList()).contains("foo", "bar")
}
verify {
recordFetchSpec.all()
}
}
}
@Nested
inner class CoroutinesVariantsOfExecutableQuery {
private val executableQuery = mockk<ReactiveNeo4jClient.ExecutableQuery<String>>()
@Test
fun `fetchAllResults should return a flow of thing`() {
every { executableQuery.results } returns Flux.just("foo", "bar")
runBlocking {
assertThat(executableQuery.fetchAllResults().toList()).contains("foo", "bar")
}
verify {
executableQuery.results
}
}
@Test
fun `awaitSingleResultOrNull should return value`() {
every { executableQuery.singleResult } returns Mono.just("baz")
runBlocking {
assertThat(executableQuery.awaitSingleResultOrNull()).isEqualTo("baz")
}
verify {
executableQuery.singleResult
}
}
@Test
fun `awaitFirstOrNull should return null`() {
every { executableQuery.singleResult } returns Mono.empty()
runBlocking {
assertThat(executableQuery.awaitSingleResultOrNull()).isNull()
}
verify {
executableQuery.singleResult
}
}
}
}

35
spring-data-neo4j/src/test/kotlin/org/neo4j/springframework/data/integration/reactive/ReactiveNeo4jClientKotlinInteropIT.kt
View File

@@ -18,16 +18,18 @@
*/
package org.neo4j.springframework.data.integration.reactive
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.AfterEach
import org.junit.jupiter.api.BeforeEach
import org.junit.jupiter.api.Test
import org.neo4j.driver.Driver
import org.neo4j.driver.Record
import org.neo4j.driver.Values
import org.neo4j.driver.types.TypeSystem
import org.neo4j.springframework.data.config.AbstractReactiveNeo4jConfig
import org.neo4j.springframework.data.core.ReactiveNeo4jClient
import org.neo4j.springframework.data.core.fetchAs
import org.neo4j.springframework.data.core.mappedBy
import org.neo4j.springframework.data.core.*
import org.neo4j.springframework.data.test.Neo4jExtension
import org.neo4j.springframework.data.test.Neo4jIntegrationTest
import org.springframework.beans.factory.annotation.Autowired
@@ -100,6 +102,33 @@ class ReactiveNeo4jClientKotlinInteropIT @Autowired constructor(
.verifyComplete();
}
@Test
fun `The reactive Neo4j client should be usable with Co-Routines`() {
val recordToArtist: (TypeSystem, Record) -> Artist = { _, r -> Artist(r["m"]["name"].asString()) }
runBlocking {
val artists = neo4jClient
.query("MATCH (m:Member) RETURN m ORDER BY m.name ASC")
.mappedBy(recordToArtist)
.fetchAll()
.toList()
assertThat(artists).hasSize(7)
assertThat(artists.map { it.name }).contains("Bela", "Roger")
}
runBlocking {
val freddie = neo4jClient
.query("MATCH (m:Member) WHERE m.name =~ \$needle RETURN m ORDER BY m.name ASC")
.bind("Fre.*").to("needle")
.mappedBy(recordToArtist)
.awaitOneOrNull()
assertThat(freddie).isNotNull
}
}
@Configuration
@EnableTransactionManagement
open class Config : AbstractReactiveNeo4jConfig() {