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

#8 - Polishing.

Browse Source 
Formatting, JavaDoc, issue comments on `@Test` annotations.
Removed some dead code.

Original pull request: #33.
This commit is contained in:
Jens Schauder
2018-12-04 16:37:12 +01:00
parent 5a5310af4d
commit 7db8e64393
7 changed files with 143 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java
View File

@@ -64,7 +64,9 @@ public interface DatabaseClient {
// Static, factory methods
/**
* A variant of {@link #create()} that accepts a {@link io.r2dbc.spi.ConnectionFactory}
* Creates a {@code DatabaseClient} that will use the provided {@link io.r2dbc.spi.ConnectionFactory}.
* @param factory The {@code ConnectionFactory} to use for obtaining connections.
* @return a new {@code DatabaseClient}. Guaranteed to be not {@code null}.
*/
static DatabaseClient create(ConnectionFactory factory) {
return new DefaultDatabaseClientBuilder().connectionFactory(factory).build();
@@ -161,7 +163,7 @@ public interface DatabaseClient {
*
* @param mappingFunction must not be {@literal null}.
* @param <R> result type.
* @return
* @return a {@link FetchSpec} for configuration what to fetch. Guaranteed to be not {@code null}.
*/
<R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);
@@ -197,7 +199,7 @@ public interface DatabaseClient {
*
* @param mappingFunction must not be {@literal null}.
* @param <R> result type.
* @return
* @return a {@link FetchSpec} for configuration what to fetch. Guaranteed to be not {@code null}.
*/
<R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);
@@ -223,7 +225,7 @@ public interface DatabaseClient {
* Specify the source {@literal table} to select from.
*
* @param table must not be {@literal null} or empty.
* @return
* @return a {@link GenericSelectSpec} for further configuration of the select. Guaranteed to be not {@code null}.
*/
GenericSelectSpec from(String table);
@@ -231,7 +233,7 @@ public interface DatabaseClient {
* Specify the source table to select from to using the {@link Class entity class}.
*
* @param table must not be {@literal null}.
* @return
* @return a {@link TypedSelectSpec} for further configuration of the select. Guaranteed to be not {@code null}.
*/
<T> TypedSelectSpec<T> from(Class<T> table);
}
@@ -245,7 +247,7 @@ public interface DatabaseClient {
* Specify the target {@literal table} to insert into.
*
* @param table must not be {@literal null} or empty.
* @return
* @return a {@link GenericInsertSpec} for further configuration of the insert. Guaranteed to be not {@code null}.
*/
GenericInsertSpec<Map<String, Object>> into(String table);
@@ -253,7 +255,7 @@ public interface DatabaseClient {
* Specify the target table to insert to using the {@link Class entity class}.
*
* @param table must not be {@literal null}.
* @return
* @return a {@link TypedInsertSpec} for further configuration of the insert. Guaranteed to be not {@code null}.
*/
<T> TypedInsertSpec<T> into(Class<T> table);
}
@@ -277,7 +279,7 @@ public interface DatabaseClient {
*
* @param mappingFunction must not be {@literal null}.
* @param <R> result type.
* @return
* @return a {@link FetchSpec} for configuration what to fetch. Guaranteed to be not {@code null}.
*/
<R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);
@@ -306,7 +308,7 @@ public interface DatabaseClient {
*
* @param mappingFunction must not be {@literal null}.
* @param <R> result type.
* @return
* @return a {@link FetchSpec} for configuration what to fetch. Guaranteed to be not {@code null}.
*/
<R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);
@@ -375,8 +377,8 @@ public interface DatabaseClient {
/**
* Insert the given {@code objectToInsert}.
*
* @param objectToInsert
* @return
* @param objectToInsert the object of which the attributes will provide the values for the insert. Must not be {@code null}.
* @return a {@link InsertSpec} for further configuration of the insert. Guaranteed to be not {@code null}.
*/
InsertSpec<Map<String, Object>> using(T objectToInsert);
@@ -384,7 +386,7 @@ public interface DatabaseClient {
* Use the given {@code tableName} as insert target.
*
* @param tableName must not be {@literal null} or empty.
* @return
* @return a {@link TypedInsertSpec} for further configuration of the insert. Guaranteed to be not {@code null}.
*/
TypedInsertSpec<T> table(String tableName);
@@ -392,8 +394,8 @@ public interface DatabaseClient {
* 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
* @param objectToInsert a publisher providing the objects of which the attributes will provide the values for the insert. Must not be {@code null}.
* @return a {@link InsertSpec} for further configuration of the insert. Guaranteed to be not {@code null}.
* @see InsertSpec#fetch()
*/
InsertSpec<Map<String, Object>> using(Publisher<T> objectToInsert);
@@ -411,7 +413,7 @@ public interface DatabaseClient {
*
* @param mappingFunction must not be {@literal null}.
* @param <R> result type.
* @return
* @return a {@link FetchSpec} for configuration what to fetch. Guaranteed to be not {@code null}.
*/
<R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);
@@ -436,15 +438,15 @@ public interface DatabaseClient {
/**
* Bind a non-{@literal null} value to a parameter identified by its {@code index}.
*
* @param index
* @param value must not be {@literal null}.
* @param index zero based index to bind the parameter to.
* @param value to bind. Must not be {@literal null}.
*/
S bind(int index, Object value);
/**
* Bind a {@literal null} value to a parameter identified by its {@code index}.
*
* @param index
* @param index zero based index to bind the parameter to.
* @param type must not be {@literal null}.
*/
S bindNull(int index, Class<?> type);

115
src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
View File

@@ -23,6 +23,20 @@ import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import io.r2dbc.spi.Statement;
import lombok.RequiredArgsConstructor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.r2dbc.UncategorizedR2dbcException;
import org.springframework.data.r2dbc.function.connectionfactory.ConnectionProxy;
import org.springframework.data.r2dbc.function.convert.ColumnMapRowMapper;
import org.springframework.data.r2dbc.function.convert.SettableValue;
import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
import org.springframework.jdbc.core.SqlProvider;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -43,21 +57,6 @@ 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;
import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.r2dbc.UncategorizedR2dbcException;
import org.springframework.data.r2dbc.function.connectionfactory.ConnectionProxy;
import org.springframework.data.r2dbc.function.convert.ColumnMapRowMapper;
import org.springframework.data.r2dbc.function.convert.SettableValue;
import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
import org.springframework.jdbc.core.SqlProvider;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* Default implementation of {@link DatabaseClient}.
*
@@ -65,7 +64,9 @@ import org.springframework.util.Assert;
*/
class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
/** Logger available to subclasses */
/**
* Logger available to subclasses
*/
private final Log logger = LogFactory.getLog(getClass());
private final ConnectionFactory connector;
@@ -77,7 +78,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
private final DefaultDatabaseClientBuilder builder;
DefaultDatabaseClient(ConnectionFactory connector, R2dbcExceptionTranslator exceptionTranslator,
ReactiveDataAccessStrategy dataAccessStrategy, DefaultDatabaseClientBuilder builder) {
ReactiveDataAccessStrategy dataAccessStrategy, DefaultDatabaseClientBuilder builder) {
this.connector = connector;
this.exceptionTranslator = exceptionTranslator;
@@ -113,7 +114,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
*
* @param action must not be {@literal null}.
* @return the resulting {@link Mono}.
* @throws DataAccessException
* @throws DataAccessException when during construction of the {@link Mono} a problem occurs.
*/
@Override
public <T> Mono<T> inConnection(Function<Connection, Mono<T>> action) throws DataAccessException {
@@ -140,7 +141,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
*
* @param action must not be {@literal null}.
* @return the resulting {@link Flux}.
* @throws DataAccessException
* @throws DataAccessException when during construction of the {@link Mono} a problem occurs.
*/
@Override
public <T> Flux<T> inConnectionMany(Function<Connection, Flux<T>> action) throws DataAccessException {
@@ -162,7 +163,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
/**
* Obtain a {@link Connection}.
*
* @return
* @return a {@link Mono} able to emit a {@link Connection}.
*/
protected Mono<Connection> getConnection() {
return Mono.from(obtainConnectionFactory().create());
@@ -171,8 +172,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
/**
* Release the {@link Connection}.
*
* @param connection
* @return
* @param connection to close.
* @return a {@link Publisher} that completes successfully when the connection is closed.
*/
protected Publisher<Void> closeConnection(Connection connection) {
return connection.close();
@@ -196,15 +197,15 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
*/
protected Connection createConnectionProxy(Connection con) {
return (Connection) Proxy.newProxyInstance(ConnectionProxy.class.getClassLoader(),
new Class<?>[] { ConnectionProxy.class }, new CloseSuppressingInvocationHandler(con));
new Class<?>[]{ConnectionProxy.class}, new CloseSuppressingInvocationHandler(con));
}
/**
* Translate the given {@link R2dbcException} into a generic {@link DataAccessException}.
*
* @param task readable text describing the task being attempted.
* @param sql SQL query or update that caused the problem (may be {@literal null}).
* @param ex the offending {@link R2dbcException}.
* @param sql SQL query or update that caused the problem (may be {@literal null}).
* @param ex the offending {@link R2dbcException}.
* @return a DataAccessException wrapping the {@link R2dbcException} (never {@literal null}).
*/
protected DataAccessException translateException(String task, @Nullable String sql, R2dbcException ex) {
@@ -217,7 +218,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
* Customization hook.
*/
protected <T> DefaultTypedExecuteSpec<T> createTypedExecuteSpec(Map<Integer, SettableValue> byIndex,
Map<String, SettableValue> byName, Supplier<String> sqlSupplier, Class<T> typeToRead) {
Map<String, SettableValue> byName, Supplier<String> sqlSupplier, Class<T> typeToRead) {
return new DefaultTypedExecuteSpec<>(byIndex, byName, sqlSupplier, typeToRead);
}
@@ -225,8 +226,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
* Customization hook.
*/
protected <T> DefaultTypedExecuteSpec<T> createTypedExecuteSpec(Map<Integer, SettableValue> byIndex,
Map<String, SettableValue> byName, Supplier<String> sqlSupplier,
BiFunction<Row, RowMetadata, T> mappingFunction) {
Map<String, SettableValue> byName, Supplier<String> sqlSupplier,
BiFunction<Row, RowMetadata, T> mappingFunction) {
return new DefaultTypedExecuteSpec<>(byIndex, byName, sqlSupplier, mappingFunction);
}
@@ -234,7 +235,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
* Customization hook.
*/
protected ExecuteSpecSupport createGenericExecuteSpec(Map<Integer, SettableValue> byIndex,
Map<String, SettableValue> byName, Supplier<String> sqlSupplier) {
Map<String, SettableValue> byName, Supplier<String> sqlSupplier) {
return new DefaultGenericExecuteSpec(byIndex, byName, sqlSupplier);
}
@@ -246,7 +247,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
}
private static void doBind(Statement<?> statement, Map<String, SettableValue> byName,
Map<Integer, SettableValue> byIndex) {
Map<Integer, SettableValue> byIndex) {
byIndex.forEach((i, o) -> {
@@ -305,13 +306,6 @@ 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();
@@ -379,7 +373,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
}
protected ExecuteSpecSupport createInstance(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
Supplier<String> sqlSupplier) {
Supplier<String> sqlSupplier) {
return new ExecuteSpecSupport(byIndex, byName, sqlSupplier);
}
@@ -397,7 +391,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
protected class DefaultGenericExecuteSpec extends ExecuteSpecSupport implements GenericExecuteSpec {
DefaultGenericExecuteSpec(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
Supplier<String> sqlSupplier) {
Supplier<String> sqlSupplier) {
super(byIndex, byName, sqlSupplier);
}
@@ -458,7 +452,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
@Override
protected ExecuteSpecSupport createInstance(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
Supplier<String> sqlSupplier) {
Supplier<String> sqlSupplier) {
return createGenericExecuteSpec(byIndex, byName, sqlSupplier);
}
}
@@ -473,7 +467,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
private final BiFunction<Row, RowMetadata, T> mappingFunction;
DefaultTypedExecuteSpec(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
Supplier<String> sqlSupplier, Class<T> typeToRead) {
Supplier<String> sqlSupplier, Class<T> typeToRead) {
super(byIndex, byName, sqlSupplier);
@@ -482,9 +476,10 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
}
DefaultTypedExecuteSpec(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
Supplier<String> sqlSupplier, BiFunction<Row, RowMetadata, T> mappingFunction) {
Supplier<String> sqlSupplier, BiFunction<Row, RowMetadata, T> mappingFunction) {
super(byIndex, byName, sqlSupplier);
this.typeToRead = null;
this.mappingFunction = mappingFunction;
}
@@ -542,7 +537,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
@Override
protected DefaultTypedExecuteSpec<T> createInstance(Map<Integer, SettableValue> byIndex,
Map<String, SettableValue> byName, Supplier<String> sqlSupplier) {
Map<String, SettableValue> byName, Supplier<String> sqlSupplier) {
return createTypedExecuteSpec(byIndex, byName, sqlSupplier, typeToRead);
}
}
@@ -629,12 +624,12 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
}
protected abstract DefaultSelectSpecSupport createInstance(String table, List<String> projectedFields, Sort sort,
Pageable page);
Pageable page);
}
private class DefaultGenericSelectSpec extends DefaultSelectSpecSupport implements GenericSelectSpec {
public DefaultGenericSelectSpec(String table, List<String> projectedFields, Sort sort, Pageable page) {
DefaultGenericSelectSpec(String table, List<String> projectedFields, Sort sort, Pageable page) {
super(table, projectedFields, sort, page);
}
@@ -696,7 +691,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
@Override
protected DefaultGenericSelectSpec createInstance(String table, List<String> projectedFields, Sort sort,
Pageable page) {
Pageable page) {
return new DefaultGenericSelectSpec(table, projectedFields, sort, page);
}
}
@@ -707,7 +702,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
@SuppressWarnings("unchecked")
private class DefaultTypedSelectSpec<T> extends DefaultSelectSpecSupport implements TypedSelectSpec<T> {
private final @Nullable Class<T> typeToRead;
private final @Nullable
Class<T> typeToRead;
private final BiFunction<Row, RowMetadata, T> mappingFunction;
DefaultTypedSelectSpec(Class<T> typeToRead) {
@@ -719,12 +715,12 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
}
DefaultTypedSelectSpec(String table, List<String> projectedFields, Sort sort, Pageable page,
BiFunction<Row, RowMetadata, T> mappingFunction) {
BiFunction<Row, RowMetadata, T> mappingFunction) {
this(table, projectedFields, sort, page, null, mappingFunction);
}
DefaultTypedSelectSpec(String table, List<String> projectedFields, Sort sort, Pageable page, Class<T> typeToRead,
BiFunction<Row, RowMetadata, T> mappingFunction) {
BiFunction<Row, RowMetadata, T> mappingFunction) {
super(table, projectedFields, sort, page);
this.typeToRead = typeToRead;
this.mappingFunction = mappingFunction;
@@ -784,7 +780,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
@Override
protected DefaultTypedSelectSpec<T> createInstance(String table, List<String> projectedFields, Sort sort,
Pageable page) {
Pageable page) {
return new DefaultTypedSelectSpec<>(table, projectedFields, sort, page, typeToRead, mappingFunction);
}
}
@@ -872,9 +868,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
Statement<?> statement = it.createStatement(sql);
byName.forEach((k, v) -> {
bindableInsert.bind(statement, v);
});
byName.forEach((k, v) -> bindableInsert.bind(statement, v));
return statement;
};
@@ -1003,14 +998,14 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
return statement;
};
Function<Connection, Flux<Result>> resultFunction = it -> {
return Flux.from(insertFunction.apply(it).execute());
};
Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(insertFunction.apply(it).execute());
return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
sql, //
resultFunction, //
it -> resultFunction.apply(it).flatMap(Result::getRowsUpdated)
it -> resultFunction //
.apply(it) //
.flatMap(Result::getRowsUpdated) //
.collect(Collectors.summingInt(Integer::intValue)), //
mappingFunction);
}
@@ -1023,7 +1018,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
} catch (R2dbcException e) {
String sql = getSql(action);
return Flux.error(new UncategorizedR2dbcException("doInConnectionMany", sql, e) {});
return Flux.error(new UncategorizedR2dbcException("doInConnectionMany", sql, e) {
});
}
}
@@ -1034,7 +1030,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
} catch (R2dbcException e) {
String sql = getSql(action);
return Mono.error(new UncategorizedR2dbcException("doInConnection", sql, e) {});
return Mono.error(new UncategorizedR2dbcException("doInConnection", sql, e) {
});
}
}

4
src/main/java/org/springframework/data/r2dbc/function/DefaultSqlResult.java
View File

@@ -101,8 +101,8 @@ class DefaultSqlResult<T> implements SqlResult<T> {
/**
* Returns an empty {@link SqlResult}.
*
* @param <R>
* @return
* @param <R> value type of the {@code SqlResult}.
* @return a {@code SqlResult}.
*/
@SuppressWarnings("unchecked")
public static <R> SqlResult<R> empty() {

2
src/main/java/org/springframework/data/r2dbc/function/SqlResult.java
View File

@@ -31,7 +31,7 @@ public interface SqlResult<T> extends FetchSpec<T> {
* Apply a {@link BiFunction mapping function} to the result that emits {@link Row}s.
*
* @param mappingFunction must not be {@literal null}.
* @param <R>
* @param <R> the value type of the {@code SqlResult}.
* @return a new {@link SqlResult} with {@link BiFunction mapping function} applied.
*/
<R> SqlResult<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);

10
src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java
View File

@@ -315,14 +315,16 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
}
private String getIdColumnName() {
return converter.getMappingContext().getRequiredPersistentEntity(entity.getJavaType()).getRequiredIdProperty()
return converter //
.getMappingContext() //
.getRequiredPersistentEntity(entity.getJavaType()) //
.getRequiredIdProperty() //
.getColumnName();
}
private BiConsumer<String, SettableValue> bind(BindableOperation operation, Statement<?> statement) {
return (k, v) -> {
operation.bind(statement, v);
};
return (k, v) -> operation.bind(statement, v);
}
}

53
src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java
View File

@@ -15,17 +15,8 @@
*/
package org.springframework.data.r2dbc.function;
import static org.assertj.core.api.Assertions.*;
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;
import javax.sql.DataSource;
import org.junit.Before;
import org.junit.Test;
import org.springframework.dao.DataAccessException;
@@ -35,6 +26,14 @@ import org.springframework.data.domain.Sort;
import org.springframework.data.r2dbc.testing.R2dbcIntegrationTestSupport;
import org.springframework.data.relational.core.mapping.Table;
import org.springframework.jdbc.core.JdbcTemplate;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Hooks;
import reactor.test.StepVerifier;
import javax.sql.DataSource;
import static org.assertj.core.api.Assertions.*;
import static org.springframework.data.domain.Sort.Order.*;
/**
* Integration tests for {@link DatabaseClient}.
@@ -58,7 +57,8 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
try {
jdbc.execute("DROP TABLE legoset");
} catch (DataAccessException e) {}
} catch (DataAccessException e) {
}
jdbc.execute(getCreateTableStatement());
}
@@ -90,12 +90,10 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
/**
* Get a parameterized {@code INSERT INTO legoset} statement setting id, name, and manual values.
*
* @return
*/
protected abstract String getInsertIntoLegosetStatement();
@Test
@Test // gh-2
public void executeInsert() {
DatabaseClient databaseClient = DatabaseClient.create(connectionFactory);
@@ -115,7 +113,7 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
assertThat(jdbc.queryForMap("SELECT id, name, manual FROM legoset")).containsEntry("id", 42055);
}
@Test
@Test // gh-2
public void shouldTranslateDuplicateKeyException() {
DatabaseClient databaseClient = DatabaseClient.create(connectionFactory);
@@ -128,15 +126,13 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
.bindNull(2, Integer.class) //
.fetch().rowsUpdated() //
.as(StepVerifier::create) //
.expectErrorSatisfies(exception -> {
assertThat(exception).isInstanceOf(DuplicateKeyException.class)
.hasMessageContaining("execute; SQL [INSERT INTO legoset");
}) //
.expectErrorSatisfies(exception -> assertThat(exception) //
.isInstanceOf(DuplicateKeyException.class) //
.hasMessageContaining("execute; SQL [INSERT INTO legoset")) //
.verify();
}
@Test
@Test // gh-2
public void executeSelect() {
jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
@@ -155,7 +151,7 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
}).verifyComplete();
}
@Test
@Test // gh-2
public void insert() {
DatabaseClient databaseClient = DatabaseClient.create(connectionFactory);
@@ -172,7 +168,7 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
assertThat(jdbc.queryForMap("SELECT id, name, manual FROM legoset")).containsEntry("id", 42055);
}
@Test
@Test // gh-2
public void insertWithoutResult() {
DatabaseClient databaseClient = DatabaseClient.create(connectionFactory);
@@ -188,7 +184,7 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
assertThat(jdbc.queryForMap("SELECT id, name, manual FROM legoset")).containsEntry("id", 42055);
}
@Test
@Test // gh-2
public void insertTypedObject() {
LegoSet legoSet = new LegoSet();
@@ -209,7 +205,7 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
assertThat(jdbc.queryForMap("SELECT id, name, manual FROM legoset")).containsEntry("id", 42055);
}
@Test
@Test // gh-2
public void selectAsMap() {
jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
@@ -229,7 +225,7 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
}).verifyComplete();
}
@Test
@Test // gh-8
public void selectExtracting() {
jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
@@ -246,7 +242,7 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
.verifyComplete();
}
@Test
@Test // gh-2
public void selectOrderByIdDesc() {
jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
@@ -264,7 +260,7 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
.verifyComplete();
}
@Test
@Test // gh-2
public void selectOrderPaged() {
jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
@@ -282,7 +278,7 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
.verifyComplete();
}
@Test
@Test // gh-2
public void selectTypedLater() {
jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
@@ -304,6 +300,7 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
@Data
@Table("legoset")
static class LegoSet {
int id;
String name;
Integer manual;

67
src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java
View File

@@ -15,27 +15,25 @@
*/
package org.springframework.data.r2dbc.function;
import static org.assertj.core.api.Assertions.*;
import io.r2dbc.spi.ConnectionFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Hooks;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import javax.sql.DataSource;
import org.junit.Before;
import org.junit.Test;
import org.springframework.dao.DataAccessException;
import org.springframework.data.r2dbc.testing.R2dbcIntegrationTestSupport;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.NoTransactionException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Hooks;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import static org.assertj.core.api.Assertions.*;
/**
* Abstract base class for integration tests for {@link TransactionalDatabaseClient}.
@@ -58,7 +56,8 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
jdbc = createJdbcTemplate(createDataSource());
try {
jdbc.execute("DROP TABLE legoset");
} catch (DataAccessException e) {}
} catch (DataAccessException e) {
}
jdbc.execute(getCreateTableStatement());
jdbc.execute("DELETE FROM legoset");
}
@@ -91,31 +90,27 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
/**
* Get a parameterized {@code INSERT INTO legoset} statement setting id, name, and manual values.
*
* @return
*/
protected abstract String getInsertIntoLegosetStatement();
/**
* Get a statement that returns the current transactionId.
*
* @return
*/
protected abstract String getCurrentTransactionIdStatement();
@Test
@Test // gh-2
public void executeInsertInManagedTransaction() {
TransactionalDatabaseClient databaseClient = TransactionalDatabaseClient.create(connectionFactory);
Flux<Integer> integerFlux = databaseClient.inTransaction(db -> {
return db.execute().sql(getInsertIntoLegosetStatement()) //
.bind(0, 42055) //
.bind(1, "SCHAUFELRADBAGGER") //
.bindNull(2, Integer.class) //
.fetch().rowsUpdated();
});
Flux<Integer> integerFlux = databaseClient.inTransaction(db -> db //
.execute() //
.sql(getInsertIntoLegosetStatement()) //
.bind(0, 42055) //
.bind(1, "SCHAUFELRADBAGGER") //
.bindNull(2, Integer.class) //
.fetch().rowsUpdated() //
);
integerFlux.as(StepVerifier::create) //
.expectNext(1) //
@@ -124,7 +119,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
assertThat(jdbc.queryForMap("SELECT id, name, manual FROM legoset")).containsEntry("id", 42055);
}
@Test
@Test // gh-2
public void executeInsertInAutoCommitTransaction() {
TransactionalDatabaseClient databaseClient = TransactionalDatabaseClient.create(connectionFactory);
@@ -142,13 +137,17 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
assertThat(jdbc.queryForMap("SELECT id, name, manual FROM legoset")).containsEntry("id", 42055);
}
@Test
@Test // gh-2
public void shouldManageUserTransaction() {
Queue<Long> transactionIds = new ArrayBlockingQueue<>(5);
TransactionalDatabaseClient databaseClient = TransactionalDatabaseClient.create(connectionFactory);
Flux<Long> txId = databaseClient.execute().sql(getCurrentTransactionIdStatement()).map((r, md) -> r.get(0, Long.class)).all();
Flux<Long> txId = databaseClient //
.execute() //
.sql(getCurrentTransactionIdStatement()) //
.map((r, md) -> r.get(0, Long.class)) //
.all();
Mono<Void> then = databaseClient.enableTransactionSynchronization(databaseClient.beginTransaction() //
.thenMany(txId.concatWith(txId).doOnNext(transactionIds::add)) //
@@ -162,7 +161,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
assertThat(listOfTxIds).containsExactly(listOfTxIds.get(1), listOfTxIds.get(0));
}
@Test
@Test // gh-2
public void userTransactionManagementShouldFailWithoutSynchronizer() {
TransactionalDatabaseClient databaseClient = TransactionalDatabaseClient.create(connectionFactory);
@@ -177,7 +176,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
}).verify();
}
@Test
@Test // gh-2
public void shouldRollbackTransaction() {
TransactionalDatabaseClient databaseClient = TransactionalDatabaseClient.create(connectionFactory);
@@ -199,7 +198,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
assertThat(count).isEqualTo(0);
}
@Test
@Test // gh-2
public void emitTransactionIds() {
TransactionalDatabaseClient databaseClient = TransactionalDatabaseClient.create(connectionFactory);