From 5a5310af4d416bb558dd23bc267409eef2522a77 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 28 Nov 2018 14:40:26 +0100 Subject: [PATCH] =?UTF-8?q?#8=20-=20Replace=20exchange()=20method=20with?= =?UTF-8?q?=20map(=E2=80=A6)=20and=20then()=20methods.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DatabaseClient now no longer exposes the exchange() method but rather provides map(…) and then(…) methods. Calling map(…) extracts values from tabular results by propagating the mapping function to Result.map(…). then() allows statement execution without returning the result propagating only completion and error signals. Original pull request: #33. --- README.adoc | 20 +- .../data/r2dbc/function/DatabaseClient.java | 106 ++++++---- .../r2dbc/function/DefaultDatabaseClient.java | 184 +++++++++++++----- .../data/r2dbc/function/DefaultSqlResult.java | 42 +++- .../data/r2dbc/function/SqlResult.java | 2 +- .../support/SimpleR2dbcRepository.java | 18 +- ...bstractDatabaseClientIntegrationTests.java | 38 +++- ...ctionalDatabaseClientIntegrationTests.java | 6 +- 8 files changed, 297 insertions(+), 119 deletions(-) diff --git a/README.adoc b/README.adoc index c03cddb..1033deb 100644 --- a/README.adoc +++ b/README.adoc @@ -8,7 +8,7 @@ The state of R2DBC is incubating to evaluate how an reactive integration could l == This is NOT an ORM -Spring Data R2DBC does not try to be an ORM. +Spring Data R2DBC does not try to be an ORM. Instead it is more of a construction kit for your personal reactive relational data access component that you can define the way you like or need it. == Maven Coordinates @@ -64,12 +64,13 @@ Mono count = databaseClient.execute() Flux> rows = databaseClient.execute() .sql("SELECT id, name, manual FROM legoset") - .fetch().all(); + .fetch() + .all(); Flux result = db.execute() .sql("SELECT txid_current();") - .exchange() - .flatMapMany(it -> it.extract((r, md) -> r.get(0, Long.class)).all()); + .map((r, md) -> r.get(0, Long.class)) + .all(); ---- === Examples selecting data @@ -100,14 +101,13 @@ Flux ids = databaseClient.insert() .value("id", 42055) .value("name", "Description") .nullValue("manual", Integer.class) - .exchange() // - .flatMapMany(it -> it.extract((r, m) -> r.get("id", Integer.class)).all()) + .map((r, m) -> r.get("id", Integer.class) + .all(); -Flux ids = databaseClient.insert() +Mono completion = databaseClient.insert() .into(LegoSet.class) .using(legoSet) - .exchange() - .flatMapMany(it -> it.extract((r, m) -> r.get("id", Integer.class)).all()) + .then(); ---- === Examples using reactive repositories @@ -162,7 +162,7 @@ Flux rowsUpdated = databaseClient.inTransaction(db -> { [source,java] ---- Flux txId = databaseClient.execute().sql("SELECT txid_current();").exchange() - .flatMapMany(it -> it.extract((r, md) -> r.get(0, Long.class)).all()); + .flatMapMany(it -> it.map((r, md) -> r.get(0, Long.class)).all()); Mono then = databaseClient.enableTransactionSynchronization(databaseClient.beginTransaction() // .thenMany(txId)) // diff --git a/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java b/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java index 4e49dcb..2ad6c5a 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java +++ b/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java @@ -156,17 +156,26 @@ public interface DatabaseClient { */ TypedExecuteSpec as(Class resultType); + /** + * Configure a result mapping {@link java.util.function.BiFunction function}. + * + * @param mappingFunction must not be {@literal null}. + * @param result type. + * @return + */ + FetchSpec map(BiFunction mappingFunction); + /** * Perform the SQL call and retrieve the result. */ FetchSpec> fetch(); /** - * Perform the SQL request and return a {@link SqlResult}. + * Perform the SQL call and return a {@link Mono} that completes without result on statement completion. * - * @return a {@code Mono} for the result + * @return a {@link Mono} ignoring its payload (actively dropping). */ - Mono>> exchange(); + Mono then(); } /** @@ -183,17 +192,26 @@ public interface DatabaseClient { */ TypedExecuteSpec as(Class resultType); + /** + * Configure a result mapping {@link java.util.function.BiFunction function}. + * + * @param mappingFunction must not be {@literal null}. + * @param result type. + * @return + */ + FetchSpec map(BiFunction mappingFunction); + /** * Perform the SQL call and retrieve the result. */ FetchSpec fetch(); /** - * Perform the SQL request and return a {@link SqlResult}. + * Perform the SQL call and return a {@link Mono} that completes without result on statement completion. * - * @return a {@code Mono} for the result + * @return a {@link Mono} ignoring its payload (actively dropping). */ - Mono> exchange(); + Mono then(); } /** @@ -229,7 +247,7 @@ public interface DatabaseClient { * @param table must not be {@literal null} or empty. * @return */ - GenericInsertSpec into(String table); + GenericInsertSpec> into(String table); /** * Specify the target table to insert to using the {@link Class entity class}. @@ -254,17 +272,19 @@ public interface DatabaseClient { */ TypedSelectSpec as(Class resultType); + /** + * Configure a result mapping {@link java.util.function.BiFunction function}. + * + * @param mappingFunction must not be {@literal null}. + * @param result type. + * @return + */ + FetchSpec map(BiFunction mappingFunction); + /** * Perform the SQL call and retrieve the result. */ FetchSpec> fetch(); - - /** - * Perform the SQL request and return a {@link SqlResult}. - * - * @return a {@code Mono} for the result - */ - Mono>> exchange(); } /** @@ -279,28 +299,21 @@ public interface DatabaseClient { * @param resultType must not be {@literal null}. * @param result type. */ - TypedSelectSpec as(Class resultType); + FetchSpec as(Class resultType); /** - * Configure a result mapping {@link java.util.function.Function}. + * Configure a result mapping {@link java.util.function.BiFunction function}. * * @param mappingFunction must not be {@literal null}. * @param result type. * @return */ - TypedSelectSpec extract(BiFunction mappingFunction); + FetchSpec map(BiFunction mappingFunction); /** * Perform the SQL call and retrieve the result. */ FetchSpec fetch(); - - /** - * Perform the SQL request and return a {@link SqlResult}. - * - * @return a {@code Mono} for the result - */ - Mono> exchange(); } /** @@ -332,8 +345,10 @@ public interface DatabaseClient { /** * Contract for specifying {@code INSERT} options leading to the exchange. + * + * @param Result type of tabular insert results. */ - interface GenericInsertSpec extends InsertSpec { + interface GenericInsertSpec extends InsertSpec { /** * Specify a field and non-{@literal null} value to insert. @@ -341,7 +356,7 @@ public interface DatabaseClient { * @param field must not be {@literal null} or empty. * @param value must not be {@literal null} */ - GenericInsertSpec value(String field, Object value); + GenericInsertSpec value(String field, Object value); /** * Specify a {@literal null} value to insert. @@ -349,7 +364,7 @@ public interface DatabaseClient { * @param field must not be {@literal null} or empty. * @param type must not be {@literal null}. */ - GenericInsertSpec nullValue(String field, Class type); + GenericInsertSpec nullValue(String field, Class type); } /** @@ -363,7 +378,7 @@ public interface DatabaseClient { * @param objectToInsert * @return */ - InsertSpec using(T objectToInsert); + InsertSpec> using(T objectToInsert); /** * Use the given {@code tableName} as insert target. @@ -374,30 +389,43 @@ public interface DatabaseClient { TypedInsertSpec table(String tableName); /** - * Insert the given {@link Publisher} to insert one or more objects. + * Insert the given {@link Publisher} to insert one or more objects. Inserts only a single object when calling + * {@link FetchSpec#one()} or {@link FetchSpec#first()}. * * @param objectToInsert * @return + * @see InsertSpec#fetch() */ - InsertSpec using(Publisher objectToInsert); + InsertSpec> using(Publisher objectToInsert); } /** * Contract for specifying {@code INSERT} options leading to the exchange. + * + * @param Result type of tabular insert results. */ - interface InsertSpec { + interface InsertSpec { /** - * Perform the SQL call. + * Configure a result mapping {@link java.util.function.BiFunction function}. + * + * @param mappingFunction must not be {@literal null}. + * @param result type. + * @return + */ + FetchSpec map(BiFunction mappingFunction); + + /** + * Perform the SQL call and retrieve the result. + */ + FetchSpec fetch(); + + /** + * Perform the SQL call and return a {@link Mono} that completes without result on statement completion. + * + * @return a {@link Mono} ignoring its payload (actively dropping). */ Mono then(); - - /** - * Perform the SQL request and return a {@link SqlResult}. - * - * @return a {@code Mono} for the result - */ - Mono>> exchange(); } /** diff --git a/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java b/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java index a6545a9..400e7eb 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java +++ b/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java @@ -41,6 +41,7 @@ import java.util.Set; import java.util.function.BiFunction; import java.util.function.Function; import java.util.function.Supplier; +import java.util.stream.Collectors; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -220,6 +221,15 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { return new DefaultTypedExecuteSpec<>(byIndex, byName, sqlSupplier, typeToRead); } + /** + * Customization hook. + */ + protected DefaultTypedExecuteSpec createTypedExecuteSpec(Map byIndex, + Map byName, Supplier sqlSupplier, + BiFunction mappingFunction) { + return new DefaultTypedExecuteSpec<>(byIndex, byName, sqlSupplier, mappingFunction); + } + /** * Customization hook. */ @@ -295,6 +305,13 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { this.sqlSupplier = sqlSupplier; } + ExecuteSpecSupport(ExecuteSpecSupport other) { + + this.byIndex = other.byIndex; + this.byName = other.byName; + this.sqlSupplier = other.sqlSupplier; + } + protected String getSql() { String sql = sqlSupplier.get(); @@ -396,14 +413,22 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { return createTypedExecuteSpec(this.byIndex, this.byName, this.sqlSupplier, resultType); } + @Override + public FetchSpec map(BiFunction mappingFunction) { + + Assert.notNull(mappingFunction, "Mapping function must not be null!"); + + return exchange(getSql(), mappingFunction); + } + @Override public FetchSpec> fetch() { return exchange(getSql(), ColumnMapRowMapper.INSTANCE); } @Override - public Mono>> exchange() { - return Mono.just(exchange(getSql(), ColumnMapRowMapper.INSTANCE)); + public Mono then() { + return fetch().rowsUpdated().then(); } @Override @@ -456,6 +481,14 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { this.mappingFunction = dataAccessStrategy.getRowMapper(typeToRead); } + DefaultTypedExecuteSpec(Map byIndex, Map byName, + Supplier sqlSupplier, BiFunction mappingFunction) { + + super(byIndex, byName, sqlSupplier); + this.typeToRead = null; + this.mappingFunction = mappingFunction; + } + @Override public TypedExecuteSpec as(Class resultType) { @@ -464,14 +497,22 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { return createTypedExecuteSpec(this.byIndex, this.byName, this.sqlSupplier, resultType); } + @Override + public FetchSpec map(BiFunction mappingFunction) { + + Assert.notNull(mappingFunction, "Mapping function must not be null!"); + + return exchange(getSql(), mappingFunction); + } + @Override public FetchSpec fetch() { return exchange(getSql(), mappingFunction); } @Override - public Mono> exchange() { - return Mono.just(exchange(getSql(), mappingFunction)); + public Mono then() { + return fetch().rowsUpdated().then(); } @Override @@ -603,10 +644,21 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { @Override public TypedSelectSpec as(Class resultType) { + + Assert.notNull(resultType, "Result type must not be null!"); + return new DefaultTypedSelectSpec<>(table, projectedFields, sort, page, resultType, dataAccessStrategy.getRowMapper(resultType)); } + @Override + public FetchSpec map(BiFunction mappingFunction) { + + Assert.notNull(mappingFunction, "Mapping function must not be null!"); + + return exchange(mappingFunction); + } + @Override public DefaultGenericSelectSpec project(String... selectedFields) { return (DefaultGenericSelectSpec) super.project(selectedFields); @@ -627,11 +679,6 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { return exchange(ColumnMapRowMapper.INSTANCE); } - @Override - public Mono>> exchange() { - return Mono.just(exchange(ColumnMapRowMapper.INSTANCE)); - } - private SqlResult exchange(BiFunction mappingFunction) { Set columns; @@ -660,7 +707,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { @SuppressWarnings("unchecked") private class DefaultTypedSelectSpec extends DefaultSelectSpecSupport implements TypedSelectSpec { - private final Class typeToRead; + private final @Nullable Class typeToRead; private final BiFunction mappingFunction; DefaultTypedSelectSpec(Class typeToRead) { @@ -671,7 +718,12 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { this.mappingFunction = dataAccessStrategy.getRowMapper(typeToRead); } - DefaultTypedSelectSpec(String table, List projectedFields, Sort sort, Pageable page, Class typeToRead, + DefaultTypedSelectSpec(String table, List projectedFields, Sort sort, Pageable page, + BiFunction mappingFunction) { + this(table, projectedFields, sort, page, null, mappingFunction); + } + + DefaultTypedSelectSpec(String table, List projectedFields, Sort sort, Pageable page, Class typeToRead, BiFunction mappingFunction) { super(table, projectedFields, sort, page); this.typeToRead = typeToRead; @@ -679,20 +731,19 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { } @Override - public TypedSelectSpec as(Class resultType) { + public FetchSpec as(Class resultType) { Assert.notNull(resultType, "Result type must not be null!"); - return new DefaultTypedSelectSpec<>(table, projectedFields, sort, page, typeToRead, - dataAccessStrategy.getRowMapper(resultType)); + return exchange(dataAccessStrategy.getRowMapper(resultType)); } @Override - public TypedSelectSpec extract(BiFunction mappingFunction) { + public FetchSpec map(BiFunction mappingFunction) { Assert.notNull(mappingFunction, "Mapping function must not be null!"); - return new DefaultTypedSelectSpec<>(table, projectedFields, sort, page, typeToRead, mappingFunction); + return exchange(mappingFunction); } @Override @@ -711,15 +762,10 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { } @Override - public FetchSpec fetch() { + public SqlResult fetch() { return exchange(mappingFunction); } - @Override - public Mono> exchange() { - return Mono.just(exchange(mappingFunction)); - } - private SqlResult exchange(BiFunction mappingFunction) { List columns; @@ -749,13 +795,13 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { class DefaultInsertIntoSpec implements InsertIntoSpec { @Override - public GenericInsertSpec into(String table) { - return new DefaultGenericInsertSpec(table, Collections.emptyMap()); + public GenericInsertSpec> into(String table) { + return new DefaultGenericInsertSpec<>(table, Collections.emptyMap(), ColumnMapRowMapper.INSTANCE); } @Override public TypedInsertSpec into(Class table) { - return new DefaultTypedInsertSpec<>(table); + return new DefaultTypedInsertSpec<>(table, ColumnMapRowMapper.INSTANCE); } } @@ -763,10 +809,11 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { * Default implementation of {@link DatabaseClient.GenericInsertSpec}. */ @RequiredArgsConstructor - class DefaultGenericInsertSpec implements GenericInsertSpec { + class DefaultGenericInsertSpec implements GenericInsertSpec { private final String table; private final Map byName; + private final BiFunction mappingFunction; @Override public GenericInsertSpec value(String field, Object value) { @@ -776,7 +823,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { Map byName = new LinkedHashMap<>(this.byName); byName.put(field, new SettableValue(field, value, null)); - return new DefaultGenericInsertSpec(this.table, byName); + return new DefaultGenericInsertSpec<>(this.table, byName, this.mappingFunction); } @Override @@ -787,20 +834,28 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { Map byName = new LinkedHashMap<>(this.byName); byName.put(field, new SettableValue(field, null, type)); - return new DefaultGenericInsertSpec(this.table, byName); + return new DefaultGenericInsertSpec<>(this.table, byName, this.mappingFunction); + } + + @Override + public FetchSpec map(BiFunction mappingFunction) { + + Assert.notNull(mappingFunction, "Mapping function must not be null!"); + + return exchange(mappingFunction); + } + + @Override + public FetchSpec fetch() { + return exchange(this.mappingFunction); } @Override public Mono then() { - return exchange((row, md) -> row).all().then(); + return fetch().rowsUpdated().then(); } - @Override - public Mono>> exchange() { - return Mono.just(exchange(ColumnMapRowMapper.INSTANCE)); - } - - private SqlResult exchange(BiFunction mappingFunction) { + private SqlResult exchange(BiFunction mappingFunction) { if (byName.isEmpty()) { throw new IllegalStateException("Insert fields is empty!"); @@ -809,7 +864,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { BindableOperation bindableInsert = dataAccessStrategy.insertAndReturnGeneratedKeys(table, byName.keySet()); String sql = bindableInsert.toQuery(); - Function insertFunction = it -> { + Function> insertFunction = it -> { if (logger.isDebugEnabled()) { logger.debug("Executing SQL statement [" + sql + "]"); @@ -837,17 +892,19 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { * Default implementation of {@link DatabaseClient.TypedInsertSpec}. */ @RequiredArgsConstructor - class DefaultTypedInsertSpec implements TypedInsertSpec, InsertSpec { + class DefaultTypedInsertSpec implements TypedInsertSpec, InsertSpec { private final Class typeToInsert; private final String table; private final Publisher objectToInsert; + private final BiFunction mappingFunction; - DefaultTypedInsertSpec(Class typeToInsert) { + DefaultTypedInsertSpec(Class typeToInsert, BiFunction mappingFunction) { this.typeToInsert = typeToInsert; this.table = dataAccessStrategy.getTableName(typeToInsert); this.objectToInsert = Mono.empty(); + this.mappingFunction = mappingFunction; } @Override @@ -855,7 +912,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { Assert.hasText(tableName, "Table name must not be null or empty!"); - return new DefaultTypedInsertSpec<>(typeToInsert, tableName, objectToInsert); + return new DefaultTypedInsertSpec<>(typeToInsert, tableName, objectToInsert, this.mappingFunction); } @Override @@ -863,7 +920,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { Assert.notNull(objectToInsert, "Object to insert must not be null!"); - return new DefaultTypedInsertSpec<>(typeToInsert, table, Mono.just(objectToInsert)); + return new DefaultTypedInsertSpec<>(typeToInsert, table, Mono.just(objectToInsert), this.mappingFunction); } @Override @@ -871,7 +928,20 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { Assert.notNull(objectToInsert, "Publisher to insert must not be null!"); - return new DefaultTypedInsertSpec<>(typeToInsert, table, objectToInsert); + return new DefaultTypedInsertSpec<>(typeToInsert, table, objectToInsert, this.mappingFunction); + } + + @Override + public FetchSpec map(BiFunction mappingFunction) { + + Assert.notNull(mappingFunction, "Mapping function must not be null!"); + + return exchange(mappingFunction); + } + + @Override + public FetchSpec fetch() { + return exchange(this.mappingFunction); } @Override @@ -879,12 +949,33 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { return Mono.from(objectToInsert).flatMapMany(toInsert -> exchange(toInsert, (row, md) -> row).all()).then(); } - @Override - public Mono>> exchange() { - return Mono.from(objectToInsert).map(toInsert -> exchange(toInsert, ColumnMapRowMapper.INSTANCE)); + private FetchSpec exchange(BiFunction mappingFunction) { + + return new FetchSpec() { + @Override + public Mono one() { + return Mono.from(objectToInsert).flatMap(toInsert -> exchange(toInsert, mappingFunction).one()); + } + + @Override + public Mono first() { + return Mono.from(objectToInsert).flatMap(toInsert -> exchange(toInsert, mappingFunction).first()); + } + + @Override + public Flux all() { + return Flux.from(objectToInsert).flatMap(toInsert -> exchange(toInsert, mappingFunction).all()); + } + + @Override + public Mono rowsUpdated() { + return Mono.from(objectToInsert).flatMapMany(toInsert -> exchange(toInsert, mappingFunction).rowsUpdated()) + .collect(Collectors.summingInt(Integer::intValue)); + } + }; } - private SqlResult exchange(Object toInsert, BiFunction mappingFunction) { + private SqlResult exchange(Object toInsert, BiFunction mappingFunction) { List insertValues = dataAccessStrategy.getValuesToInsert(toInsert); Set columns = new LinkedHashSet<>(); @@ -919,7 +1010,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { return new DefaultSqlResult<>(DefaultDatabaseClient.this, // sql, // resultFunction, // - it -> resultFunction.apply(it).flatMap(Result::getRowsUpdated).next(), // + it -> resultFunction.apply(it).flatMap(Result::getRowsUpdated) + .collect(Collectors.summingInt(Integer::intValue)), // mappingFunction); } } diff --git a/src/main/java/org/springframework/data/r2dbc/function/DefaultSqlResult.java b/src/main/java/org/springframework/data/r2dbc/function/DefaultSqlResult.java index 378b473..44f1165 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/DefaultSqlResult.java +++ b/src/main/java/org/springframework/data/r2dbc/function/DefaultSqlResult.java @@ -34,6 +34,33 @@ import org.springframework.jdbc.core.SqlProvider; */ class DefaultSqlResult implements SqlResult { + private final static SqlResult EMPTY = new SqlResult() { + @Override + public SqlResult map(BiFunction mappingFunction) { + return DefaultSqlResult.empty(); + } + + @Override + public Mono one() { + return Mono.empty(); + } + + @Override + public Mono first() { + return Mono.empty(); + } + + @Override + public Flux all() { + return Flux.empty(); + } + + @Override + public Mono rowsUpdated() { + return Mono.empty(); + } + }; + private final ConnectionAccessor connectionAccessor; private final String sql; private final Function> resultFunction; @@ -71,11 +98,22 @@ class DefaultSqlResult implements SqlResult { }); } + /** + * Returns an empty {@link SqlResult}. + * + * @param + * @return + */ + @SuppressWarnings("unchecked") + public static SqlResult empty() { + return (SqlResult) EMPTY; + } + /* (non-Javadoc) - * @see org.springframework.data.jdbc.core.function.SqlResult#extract(java.util.function.BiFunction) + * @see org.springframework.data.jdbc.core.function.SqlResult#map(java.util.function.BiFunction) */ @Override - public SqlResult extract(BiFunction mappingFunction) { + public SqlResult map(BiFunction mappingFunction) { return new DefaultSqlResult<>(connectionAccessor, sql, resultFunction, updatedRowsFunction, mappingFunction); } diff --git a/src/main/java/org/springframework/data/r2dbc/function/SqlResult.java b/src/main/java/org/springframework/data/r2dbc/function/SqlResult.java index eabf54a..a0b3a0c 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/SqlResult.java +++ b/src/main/java/org/springframework/data/r2dbc/function/SqlResult.java @@ -34,5 +34,5 @@ public interface SqlResult extends FetchSpec { * @param * @return a new {@link SqlResult} with {@link BiFunction mapping function} applied. */ - SqlResult extract(BiFunction mappingFunction); + SqlResult map(BiFunction mappingFunction); } diff --git a/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java b/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java index 2383034..fb1458b 100644 --- a/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java +++ b/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java @@ -32,7 +32,6 @@ import org.springframework.data.r2dbc.function.BindIdOperation; import org.springframework.data.r2dbc.function.BindableOperation; import org.springframework.data.r2dbc.function.DatabaseClient; import org.springframework.data.r2dbc.function.DatabaseClient.GenericExecuteSpec; -import org.springframework.data.r2dbc.function.FetchSpec; import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy; import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter; import org.springframework.data.r2dbc.function.convert.SettableValue; @@ -66,8 +65,8 @@ public class SimpleR2dbcRepository implements ReactiveCrudRepository it.extract(converter.populateIdIfNecessary(objectToSave)).one()); + .map(converter.populateIdIfNecessary(objectToSave)) // + .one(); } Object id = entity.getRequiredId(objectToSave); @@ -83,8 +82,7 @@ public class SimpleR2dbcRepository implements ReactiveCrudRepository implements ReactiveCrudRepository it.extract((r, md) -> r).first()).hasElement(); + .map((r, md) -> r) // + .first() // + .hasElement(); } /* (non-Javadoc) @@ -220,8 +219,8 @@ public class SimpleR2dbcRepository implements ReactiveCrudRepository it.extract((r, md) -> r.get(0, Long.class)).first()) // + .map((r, md) -> r.get(0, Long.class)) // + .first() // .defaultIfEmpty(0L); } @@ -312,7 +311,6 @@ public class SimpleR2dbcRepository implements ReactiveCrudRepository deleteAll() { return databaseClient.execute().sql(String.format("DELETE FROM %s", entity.getTableName())) // - .exchange() // .then(); } diff --git a/src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java index 3ebf5bc..bfe27ad 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java @@ -20,6 +20,7 @@ import static org.springframework.data.domain.Sort.Order.*; import io.r2dbc.spi.ConnectionFactory; import lombok.Data; +import reactor.core.publisher.Flux; import reactor.core.publisher.Hooks; import reactor.test.StepVerifier; @@ -108,6 +109,9 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr .expectNext(1) // .verifyComplete(); + Flux rows = databaseClient.select().from("legoset").orderBy(Sort.by(desc("id"))).as(LegoSet.class).fetch() + .all(); + assertThat(jdbc.queryForMap("SELECT id, name, manual FROM legoset")).containsEntry("id", 42055); } @@ -160,8 +164,8 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr .value("id", 42055) // .value("name", "SCHAUFELRADBAGGER") // .nullValue("manual", Integer.class) // - .exchange() // - .flatMapMany(FetchSpec::rowsUpdated) // + .fetch() // + .rowsUpdated() // .as(StepVerifier::create) // .expectNext(1).verifyComplete(); @@ -195,16 +199,18 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr DatabaseClient databaseClient = DatabaseClient.create(connectionFactory); databaseClient.insert().into(LegoSet.class)// - .using(legoSet).exchange() // - .flatMapMany(FetchSpec::rowsUpdated) // + .using(legoSet) // + .fetch() // + .rowsUpdated() // .as(StepVerifier::create) // - .expectNext(1).verifyComplete(); + .expectNext(1) // + .verifyComplete(); assertThat(jdbc.queryForMap("SELECT id, name, manual FROM legoset")).containsEntry("id", 42055); } @Test - public void select() { + public void selectAsMap() { jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)"); @@ -213,7 +219,8 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr databaseClient.select().from(LegoSet.class) // .project("id", "name", "manual") // .orderBy(Sort.by("id")) // - .fetch().all() // + .fetch() // + .all() // .as(StepVerifier::create) // .assertNext(actual -> { assertThat(actual.getId()).isEqualTo(42055); @@ -222,6 +229,23 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr }).verifyComplete(); } + @Test + public void selectExtracting() { + + jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)"); + + DatabaseClient databaseClient = DatabaseClient.create(connectionFactory); + + databaseClient.select().from("legoset") // + .project("id", "name", "manual") // + .orderBy(Sort.by("id")) // + .map((r, md) -> r.get("id", Integer.class)) // + .all() // + .as(StepVerifier::create) // + .expectNext(42055) // + .verifyComplete(); + } + @Test public void selectOrderByIdDesc() { diff --git a/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java index c50b305..a57c5f8 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java @@ -148,8 +148,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend Queue transactionIds = new ArrayBlockingQueue<>(5); TransactionalDatabaseClient databaseClient = TransactionalDatabaseClient.create(connectionFactory); - Flux txId = databaseClient.execute().sql(getCurrentTransactionIdStatement()).exchange() - .flatMapMany(it -> it.extract((r, md) -> r.get(0, Long.class)).all()); + Flux txId = databaseClient.execute().sql(getCurrentTransactionIdStatement()).map((r, md) -> r.get(0, Long.class)).all(); Mono then = databaseClient.enableTransactionSynchronization(databaseClient.beginTransaction() // .thenMany(txId.concatWith(txId).doOnNext(transactionIds::add)) // @@ -207,8 +206,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend Flux transactionIds = databaseClient.inTransaction(db -> { - Flux txId = db.execute().sql(getCurrentTransactionIdStatement()).exchange() - .flatMapMany(it -> it.extract((r, md) -> r.get(0)).all()); + Flux txId = db.execute().sql(getCurrentTransactionIdStatement()).map((r, md) -> r.get(0)).all(); return txId.concatWith(txId); });