diff --git a/src/main/java/org/springframework/data/jdbc/core/function/ConnectionAccessor.java b/src/main/java/org/springframework/data/jdbc/core/function/ConnectionAccessor.java new file mode 100644 index 0000000..aad998a --- /dev/null +++ b/src/main/java/org/springframework/data/jdbc/core/function/ConnectionAccessor.java @@ -0,0 +1,62 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.jdbc.core.function; + +import io.r2dbc.spi.Connection; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.util.function.Function; + +import org.springframework.dao.DataAccessException; + +/** + * Interface declaring methods that accept callback {@link Function} to operate within the scope of a + * {@link Connection}. Callback functions operate on a provided connection and must not close the connection as the + * connections may be pooled or be subject to other kinds of resource management. + *

+ * Callback functions are responsible for creating a {@link org.reactivestreams.Publisher} that defines the scope of how + * long the allocated {@link Connection} is valid. Connections are released after the publisher terminates. + * + * @author Mark Paluch + */ +public interface ConnectionAccessor { + + /** + * Execute a callback {@link Function} within a {@link Connection} scope. The function is responsible for creating a + * {@link Mono}. The connection is released after the {@link Mono} terminates (or the subscription is cancelled). + * Connection resources must not be passed outside of the {@link Function} closure, otherwise resources may get + * defunct. + * + * @param action must not be {@literal null}. + * @return the resulting {@link Mono}. + * @throws DataAccessException + */ + Mono inConnection(Function> action) throws DataAccessException; + + /** + * Execute a callback {@link Function} within a {@link Connection} scope. The function is responsible for creating a + * {@link Flux}. The connection is released after the {@link Flux} terminates (or the subscription is cancelled). + * Connection resources must not be passed outside of the {@link Function} closure, otherwise resources may get + * defunct. + * + * @param action must not be {@literal null}. + * @return the resulting {@link Flux}. + * @throws DataAccessException + */ + Flux inConnectionMany(Function> action) throws DataAccessException; + +} diff --git a/src/main/java/org/springframework/data/jdbc/core/function/DatabaseClient.java b/src/main/java/org/springframework/data/jdbc/core/function/DatabaseClient.java index 5926592..a17ad27 100644 --- a/src/main/java/org/springframework/data/jdbc/core/function/DatabaseClient.java +++ b/src/main/java/org/springframework/data/jdbc/core/function/DatabaseClient.java @@ -15,7 +15,9 @@ */ package org.springframework.data.jdbc.core.function; -import reactor.core.publisher.Flux; +import io.r2dbc.spi.ConnectionFactory; +import io.r2dbc.spi.Row; +import io.r2dbc.spi.RowMetadata; import reactor.core.publisher.Mono; import java.util.Map; @@ -28,10 +30,6 @@ import org.springframework.data.domain.Pageable; import org.springframework.data.domain.Sort; import org.springframework.jdbc.support.SQLExceptionTranslator; -import io.r2dbc.spi.ConnectionFactory; -import io.r2dbc.spi.Row; -import io.r2dbc.spi.RowMetadata; - /** * A non-blocking, reactive client for performing database calls requests with Reactive Streams back pressure. Provides * a higher level, common API over R2DBC client libraries. @@ -412,40 +410,4 @@ public interface DatabaseClient { */ S bind(Object bean); } - - /** - * Contract for fetching results. - */ - interface FetchSpec { - - /** - * Get exactly zero or one result. - * - * @return {@link Mono#empty()} if no match found. Never {@literal null}. - * @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found. - */ - Mono one(); - - /** - * Get the first or no result. - * - * @return {@link Mono#empty()} if no match found. Never {@literal null}. - */ - Mono first(); - - /** - * Get all matching elements. - * - * @return never {@literal null}. - */ - Flux all(); - - /** - * Get the number of updated rows. - * - * @return {@link Mono} emitting the number of updated rows. Never {@literal null}. - */ - Mono rowsUpdated(); - } - } diff --git a/src/main/java/org/springframework/data/jdbc/core/function/DefaultDatabaseClient.java b/src/main/java/org/springframework/data/jdbc/core/function/DefaultDatabaseClient.java index d478676..e254735 100644 --- a/src/main/java/org/springframework/data/jdbc/core/function/DefaultDatabaseClient.java +++ b/src/main/java/org/springframework/data/jdbc/core/function/DefaultDatabaseClient.java @@ -46,7 +46,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.reactivestreams.Publisher; import org.springframework.dao.DataAccessException; -import org.springframework.dao.IncorrectResultSizeDataAccessException; import org.springframework.dao.UncategorizedDataAccessException; import org.springframework.data.jdbc.core.function.connectionfactory.ConnectionProxy; import org.springframework.data.util.Pair; @@ -60,7 +59,7 @@ import org.springframework.util.Assert; * * @author Mark Paluch */ -class DefaultDatabaseClient implements DatabaseClient { +class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { /** Logger available to subclasses */ protected final Log logger = LogFactory.getLog(getClass()); @@ -102,32 +101,6 @@ class DefaultDatabaseClient implements DatabaseClient { return new DefaultInsertIntoSpec(); } - /** - * Execute a callback {@link Function} within a {@link Connection} scope. The function is responsible for creating a - * {@link Flux}. The connection is released after the {@link Flux} terminates (or the subscription is cancelled). - * Connection resources must not be passed outside of the {@link Function} closure, otherwise resources may get - * defunct. - * - * @param action must not be {@literal null}. - * @return the resulting {@link Flux}. - * @throws DataAccessException - */ - public Flux executeMany(Function> action) throws DataAccessException { - - Assert.notNull(action, "Callback object must not be null"); - - Mono connectionMono = getConnection(); - // Create close-suppressing Connection proxy, also preparing returned Statements. - - return Flux.usingWhen(connectionMono, it -> { - - Connection connectionToUse = createConnectionProxy(it); - - return doInConnectionMany(connectionToUse, action); - }, this::closeConnection, this::closeConnection, this::closeConnection) // - .onErrorMap(SQLException.class, ex -> translateException("executeMany", getSql(action), ex)); - } - /** * Execute a callback {@link Function} within a {@link Connection} scope. The function is responsible for creating a * {@link Mono}. The connection is released after the {@link Mono} terminates (or the subscription is cancelled). @@ -138,7 +111,8 @@ class DefaultDatabaseClient implements DatabaseClient { * @return the resulting {@link Mono}. * @throws DataAccessException */ - public Mono execute(Function> action) throws DataAccessException { + @Override + public Mono inConnection(Function> action) throws DataAccessException { Assert.notNull(action, "Callback object must not be null"); @@ -154,6 +128,33 @@ class DefaultDatabaseClient implements DatabaseClient { .onErrorMap(SQLException.class, ex -> translateException("execute", getSql(action), ex)); } + /** + * Execute a callback {@link Function} within a {@link Connection} scope. The function is responsible for creating a + * {@link Flux}. The connection is released after the {@link Flux} terminates (or the subscription is cancelled). + * Connection resources must not be passed outside of the {@link Function} closure, otherwise resources may get + * defunct. + * + * @param action must not be {@literal null}. + * @return the resulting {@link Flux}. + * @throws DataAccessException + */ + @Override + public Flux inConnectionMany(Function> action) throws DataAccessException { + + Assert.notNull(action, "Callback object must not be null"); + + Mono connectionMono = getConnection(); + // Create close-suppressing Connection proxy, also preparing returned Statements. + + return Flux.usingWhen(connectionMono, it -> { + + Connection connectionToUse = createConnectionProxy(it); + + return doInConnectionMany(connectionToUse, action); + }, this::closeConnection, this::closeConnection, this::closeConnection) // + .onErrorMap(SQLException.class, ex -> translateException("executeMany", getSql(action), ex)); + } + /** * Obtain a {@link Connection}. * @@ -292,7 +293,8 @@ class DefaultDatabaseClient implements DatabaseClient { Function> resultFunction = it -> Flux.from(executeFunction.apply(it).execute()); - return new DefaultSqlResultFunctions<>(sql, // + return new DefaultSqlResult<>(DefaultDatabaseClient.this, // + sql, // resultFunction, // it -> resultFunction.apply(it).flatMap(Result::getRowsUpdated).next(), // mappingFunction); @@ -564,7 +566,8 @@ class DefaultDatabaseClient implements DatabaseClient { Function> resultFunction = it -> Flux .from(insertFunction.apply(it).executeReturningGeneratedKeys()); - return new DefaultSqlResultFunctions<>(sql, // + return new DefaultSqlResult<>(DefaultDatabaseClient.this, // + sql, // resultFunction, // it -> resultFunction.apply(it).flatMap(Result::getRowsUpdated).next(), // mappingFunction); @@ -680,102 +683,14 @@ class DefaultDatabaseClient implements DatabaseClient { Function> resultFunction = it -> Flux .from(insertFunction.apply(it).executeReturningGeneratedKeys()); - return new DefaultSqlResultFunctions<>(sql, // + return new DefaultSqlResult<>(DefaultDatabaseClient.this, // + sql, // resultFunction, // it -> resultFunction.apply(it).flatMap(Result::getRowsUpdated).next(), // mappingFunction); } } - /** - * Default {@link org.springframework.data.jdbc.core.function.SqlResult} implementation. - */ - class DefaultSqlResultFunctions implements SqlResult { - - private final String sql; - private final Function> resultFunction; - private final Function> updatedRowsFunction; - private final FetchSpec fetchSpec; - - DefaultSqlResultFunctions(String sql, Function> resultFunction, - Function> updatedRowsFunction, BiFunction mappingFunction) { - - this.sql = sql; - this.resultFunction = resultFunction; - this.updatedRowsFunction = updatedRowsFunction; - - this.fetchSpec = new DefaultFetchFunctions<>(sql, - it -> resultFunction.apply(it).flatMap(result -> result.map(mappingFunction)), updatedRowsFunction); - } - - @Override - public SqlResult extract(BiFunction mappingFunction) { - return new DefaultSqlResultFunctions<>(sql, resultFunction, updatedRowsFunction, mappingFunction); - } - - @Override - public Mono one() { - return fetchSpec.one(); - } - - @Override - public Mono first() { - return fetchSpec.first(); - } - - @Override - public Flux all() { - return fetchSpec.all(); - } - - @Override - public Mono rowsUpdated() { - return fetchSpec.rowsUpdated(); - } - } - - @RequiredArgsConstructor - class DefaultFetchFunctions implements FetchSpec { - - private final String sql; - private final Function> resultFunction; - private final Function> updatedRowsFunction; - - @Override - public Mono one() { - - return all().buffer(2) // - .flatMap(it -> { - - if (it.isEmpty()) { - return Mono.empty(); - } - - if (it.size() > 1) { - return Mono.error(new IncorrectResultSizeDataAccessException( - String.format("Query [%s] returned non unique result.", this.sql), 1)); - } - - return Mono.just(it.get(0)); - }).next(); - } - - @Override - public Mono first() { - return all().next(); - } - - @Override - public Flux all() { - return executeMany(resultFunction); - } - - @Override - public Mono rowsUpdated() { - return execute(updatedRowsFunction); - } - } - private static Flux doInConnectionMany(Connection connection, Function> action) { try { diff --git a/src/main/java/org/springframework/data/jdbc/core/function/DefaultFetchSpec.java b/src/main/java/org/springframework/data/jdbc/core/function/DefaultFetchSpec.java new file mode 100644 index 0000000..5db2a68 --- /dev/null +++ b/src/main/java/org/springframework/data/jdbc/core/function/DefaultFetchSpec.java @@ -0,0 +1,85 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.jdbc.core.function; + +import io.r2dbc.spi.Connection; +import lombok.RequiredArgsConstructor; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.util.function.Function; + +import org.springframework.dao.IncorrectResultSizeDataAccessException; + +/** + * Default implementation of {@link FetchSpec}. + * + * @author Mark Paluch + */ +@RequiredArgsConstructor +class DefaultFetchSpec implements FetchSpec { + + private final ConnectionAccessor connectionAccessor; + private final String sql; + private final Function> resultFunction; + private final Function> updatedRowsFunction; + + /* (non-Javadoc) + * @see org.springframework.data.jdbc.core.function.FetchSpec#one() + */ + @Override + public Mono one() { + + return all().buffer(2) // + .flatMap(it -> { + + if (it.isEmpty()) { + return Mono.empty(); + } + + if (it.size() > 1) { + return Mono.error(new IncorrectResultSizeDataAccessException( + String.format("Query [%s] returned non unique result.", this.sql), 1)); + } + + return Mono.just(it.get(0)); + }).next(); + } + + /* (non-Javadoc) + * @see org.springframework.data.jdbc.core.function.FetchSpec#first() + */ + @Override + public Mono first() { + return all().next(); + } + + /* (non-Javadoc) + * @see org.springframework.data.jdbc.core.function.FetchSpec#all() + */ + @Override + public Flux all() { + return connectionAccessor.inConnectionMany(resultFunction); + } + + /* (non-Javadoc) + * @see org.springframework.data.jdbc.core.function.FetchSpec#rowsUpdated() + */ + @Override + public Mono rowsUpdated() { + return connectionAccessor.inConnection(updatedRowsFunction); + } +} diff --git a/src/main/java/org/springframework/data/jdbc/core/function/DefaultSqlResult.java b/src/main/java/org/springframework/data/jdbc/core/function/DefaultSqlResult.java new file mode 100644 index 0000000..3cafbb3 --- /dev/null +++ b/src/main/java/org/springframework/data/jdbc/core/function/DefaultSqlResult.java @@ -0,0 +1,92 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.jdbc.core.function; + +import io.r2dbc.spi.Connection; +import io.r2dbc.spi.Result; +import io.r2dbc.spi.Row; +import io.r2dbc.spi.RowMetadata; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.util.function.BiFunction; +import java.util.function.Function; + +/** + * Default {@link SqlResult} implementation. + * + * @author Mark Paluch + */ +class DefaultSqlResult implements SqlResult { + + private final ConnectionAccessor connectionAccessor; + private final String sql; + private final Function> resultFunction; + private final Function> updatedRowsFunction; + private final FetchSpec fetchSpec; + + DefaultSqlResult(ConnectionAccessor connectionAccessor, String sql, Function> resultFunction, + Function> updatedRowsFunction, BiFunction mappingFunction) { + + this.sql = sql; + this.connectionAccessor = connectionAccessor; + this.resultFunction = resultFunction; + this.updatedRowsFunction = updatedRowsFunction; + + this.fetchSpec = new DefaultFetchSpec<>(connectionAccessor, sql, + it -> resultFunction.apply(it).flatMap(result -> result.map(mappingFunction)), updatedRowsFunction); + } + + /* (non-Javadoc) + * @see org.springframework.data.jdbc.core.function.SqlResult#extract(java.util.function.BiFunction) + */ + @Override + public SqlResult extract(BiFunction mappingFunction) { + return new DefaultSqlResult<>(connectionAccessor, sql, resultFunction, updatedRowsFunction, mappingFunction); + } + + /* (non-Javadoc) + * @see org.springframework.data.jdbc.core.function.FetchSpec#one() + */ + @Override + public Mono one() { + return fetchSpec.one(); + } + + /* (non-Javadoc) + * @see org.springframework.data.jdbc.core.function.FetchSpec#first() + */ + @Override + public Mono first() { + return fetchSpec.first(); + } + + /* (non-Javadoc) + * @see org.springframework.data.jdbc.core.function.FetchSpec#all() + */ + @Override + public Flux all() { + return fetchSpec.all(); + } + + /* (non-Javadoc) + * @see org.springframework.data.jdbc.core.function.FetchSpec#rowsUpdated() + */ + @Override + public Mono rowsUpdated() { + return fetchSpec.rowsUpdated(); + } +} diff --git a/src/main/java/org/springframework/data/jdbc/core/function/FetchSpec.java b/src/main/java/org/springframework/data/jdbc/core/function/FetchSpec.java new file mode 100644 index 0000000..74fca95 --- /dev/null +++ b/src/main/java/org/springframework/data/jdbc/core/function/FetchSpec.java @@ -0,0 +1,56 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.jdbc.core.function; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * Contract for fetching results. + * + * @author Mark Paluch + */ +public interface FetchSpec { + + /** + * Get exactly zero or one result. + * + * @return {@link Mono#empty()} if no match found. Never {@literal null}. + * @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found. + */ + Mono one(); + + /** + * Get the first or no result. + * + * @return {@link Mono#empty()} if no match found. Never {@literal null}. + */ + Mono first(); + + /** + * Get all matching elements. + * + * @return never {@literal null}. + */ + Flux all(); + + /** + * Get the number of updated rows. + * + * @return {@link Mono} emitting the number of updated rows. Never {@literal null}. + */ + Mono rowsUpdated(); +} diff --git a/src/main/java/org/springframework/data/jdbc/core/function/SqlResult.java b/src/main/java/org/springframework/data/jdbc/core/function/SqlResult.java index 70e746f..b6d3d66 100644 --- a/src/main/java/org/springframework/data/jdbc/core/function/SqlResult.java +++ b/src/main/java/org/springframework/data/jdbc/core/function/SqlResult.java @@ -15,17 +15,24 @@ */ package org.springframework.data.jdbc.core.function; -import java.util.function.BiFunction; - -import org.springframework.data.jdbc.core.function.DatabaseClient.FetchSpec; - import io.r2dbc.spi.Row; import io.r2dbc.spi.RowMetadata; +import java.util.function.BiFunction; + /** + * Mappable {@link FetchSpec} that accepts a {@link BiFunction mapping function} to map SQL {@link Row}s. + * * @author Mark Paluch */ public interface SqlResult extends FetchSpec { + /** + * Apply a {@link BiFunction mapping function} to the result that emits {@link Row}s. + * + * @param mappingFunction must not be {@literal null}. + * @param + * @return a new {@link SqlResult} with {@link BiFunction mapping function} applied. + */ SqlResult extract(BiFunction mappingFunction); } diff --git a/src/main/java/org/springframework/data/jdbc/core/function/connectionfactory/ConnectionProxy.java b/src/main/java/org/springframework/data/jdbc/core/function/connectionfactory/ConnectionProxy.java index a473df7..6f5891f 100644 --- a/src/main/java/org/springframework/data/jdbc/core/function/connectionfactory/ConnectionProxy.java +++ b/src/main/java/org/springframework/data/jdbc/core/function/connectionfactory/ConnectionProxy.java @@ -22,9 +22,8 @@ import java.sql.Wrapper; /** * Subinterface of {@link Connection} to be implemented by Connection proxies. Allows access to the underlying target * Connection. - *

- * This interface can be checked when there is a need to cast to a native R2DBC {@link Connection}. Alternatively, all - * such connections also support JDBC 4.0's {@link Connection#unwrap}. + *

+ * This interface can be checked when there is a need to cast to a native R2DBC {@link Connection}. * * @author Mark Paluch */ @@ -32,7 +31,7 @@ public interface ConnectionProxy extends Connection, Wrapper { /** * Return the target Connection of this proxy. - *

+ *

* This will typically be the native driver Connection or a wrapper from a connection pool. * * @return the underlying Connection (never {@code null})