#2 - Introduce ConnectionAccessor.

Declare interface exposing inConnection(…) methods to encapsulate types implementing functionality that is applies within a connection scope. Move FetchSpec and SqlResult implementations to top-level types.

Author: mp911de

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.
+ * <p/>
+ * 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
+	 */
+	<T> Mono<T> inConnection(Function<Connection, Mono<T>> 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
+	 */
+	<T> Flux<T> inConnectionMany(Function<Connection, Flux<T>> action) throws DataAccessException;
+
+}

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.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 @@ interface BindSpec<S extends BindSpec<S>> {
 		 */
 		S bind(Object bean);
 	}
-
-	/**
-	 * Contract for fetching results.
-	 */
-	interface FetchSpec<T> {
-
-		/**
-		 * 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<T> one();
-
-		/**
-		 * Get the first or no result.
-		 *
-		 * @return {@link Mono#empty()} if no match found. Never {@literal null}.
-		 */
-		Mono<T> first();
-
-		/**
-		 * Get all matching elements.
-		 *
-		 * @return never {@literal null}.
-		 */
-		Flux<T> all();
-
-		/**
-		 * Get the number of updated rows.
-		 *
-		 * @return {@link Mono} emitting the number of updated rows. Never {@literal null}.
-		 */
-		Mono<Integer> rowsUpdated();
-	}
-
 }

src/main/java/org/springframework/data/jdbc/core/function/DefaultDatabaseClient.java

@@ -46,7 +46,6 @@
 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 @@
  *
  * @author Mark Paluch
  */
-class DefaultDatabaseClient implements DatabaseClient {
+class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 	/** Logger available to subclasses */
 	protected final Log logger = LogFactory.getLog(getClass());
@@ -104,54 +103,56 @@ public InsertIntoSpec insert() {
 
 	/**
 	 * 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).
+	 * {@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 Flux}.
+	 * @return the resulting {@link Mono}.
 	 * @throws DataAccessException
 	 */
-	public <T> Flux<T> executeMany(Function<Connection, Flux<T>> action) throws DataAccessException {
+	@Override
+	public <T> Mono<T> inConnection(Function<Connection, Mono<T>> action) throws DataAccessException {
 
 		Assert.notNull(action, "Callback object must not be null");
 
 		Mono<Connection> connectionMono = getConnection();
 		// Create close-suppressing Connection proxy, also preparing returned Statements.
 
-		return Flux.usingWhen(connectionMono, it -> {
+		return Mono.usingWhen(connectionMono, it -> {
 
 			Connection connectionToUse = createConnectionProxy(it);
 
-			return doInConnectionMany(connectionToUse, action);
+			return doInConnection(connectionToUse, action);
 		}, this::closeConnection, this::closeConnection, this::closeConnection) //
-				.onErrorMap(SQLException.class, ex -> translateException("executeMany", getSql(action), ex));
+				.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 Mono}. The connection is released after the {@link Mono} terminates (or the subscription is cancelled).
+	 * {@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 Mono}.
+	 * @return the resulting {@link Flux}.
 	 * @throws DataAccessException
 	 */
-	public <T> Mono<T> execute(Function<Connection, Mono<T>> action) throws DataAccessException {
+	@Override
+	public <T> Flux<T> inConnectionMany(Function<Connection, Flux<T>> action) throws DataAccessException {
 
 		Assert.notNull(action, "Callback object must not be null");
 
 		Mono<Connection> connectionMono = getConnection();
 		// Create close-suppressing Connection proxy, also preparing returned Statements.
 
-		return Mono.usingWhen(connectionMono, it -> {
+		return Flux.usingWhen(connectionMono, it -> {
 
 			Connection connectionToUse = createConnectionProxy(it);
 
-			return doInConnection(connectionToUse, action);
+			return doInConnectionMany(connectionToUse, action);
 		}, this::closeConnection, this::closeConnection, this::closeConnection) //
-				.onErrorMap(SQLException.class, ex -> translateException("execute", getSql(action), ex));
+				.onErrorMap(SQLException.class, ex -> translateException("executeMany", getSql(action), ex));
 	}
 
 	/**
@@ -292,7 +293,8 @@ protected <T> SqlResult<T> exchange(String sql, BiFunction<Row, RowMetadata, T>
 
 			Function<Connection, Flux<Result>> 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 @@ private <T> SqlResult<T> exchange(BiFunction<Row, RowMetadata, T> mappingFunctio
 			Function<Connection, Flux<Result>> 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 @@ private <R> SqlResult<R> exchange(Object toInsert, BiFunction<Row, RowMetadata,
 			Function<Connection, Flux<Result>> 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<T> implements SqlResult<T> {
-
-		private final String sql;
-		private final Function<Connection, Flux<Result>> resultFunction;
-		private final Function<Connection, Mono<Integer>> updatedRowsFunction;
-		private final FetchSpec<T> fetchSpec;
-
-		DefaultSqlResultFunctions(String sql, Function<Connection, Flux<Result>> resultFunction,
-				Function<Connection, Mono<Integer>> updatedRowsFunction, BiFunction<Row, RowMetadata, T> 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 <R> SqlResult<R> extract(BiFunction<Row, RowMetadata, R> mappingFunction) {
-			return new DefaultSqlResultFunctions<>(sql, resultFunction, updatedRowsFunction, mappingFunction);
-		}
-
-		@Override
-		public Mono<T> one() {
-			return fetchSpec.one();
-		}
-
-		@Override
-		public Mono<T> first() {
-			return fetchSpec.first();
-		}
-
-		@Override
-		public Flux<T> all() {
-			return fetchSpec.all();
-		}
-
-		@Override
-		public Mono<Integer> rowsUpdated() {
-			return fetchSpec.rowsUpdated();
-		}
-	}
-
-	@RequiredArgsConstructor
-	class DefaultFetchFunctions<T> implements FetchSpec<T> {
-
-		private final String sql;
-		private final Function<Connection, Flux<T>> resultFunction;
-		private final Function<Connection, Mono<Integer>> updatedRowsFunction;
-
-		@Override
-		public Mono<T> 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<T> first() {
-			return all().next();
-		}
-
-		@Override
-		public Flux<T> all() {
-			return executeMany(resultFunction);
-		}
-
-		@Override
-		public Mono<Integer> rowsUpdated() {
-			return execute(updatedRowsFunction);
-		}
-	}
-
 	private static <T> Flux<T> doInConnectionMany(Connection connection, Function<Connection, Flux<T>> action) {
 
 		try {