#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.

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.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<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.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 <T> Flux<T> executeMany(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 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 <T> Mono<T> execute(Function<Connection, Mono<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");
 
@@ -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 <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 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<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 @@ class DefaultDatabaseClient implements DatabaseClient {
 			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 @@ class DefaultDatabaseClient implements DatabaseClient {
 			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 {

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<T> implements FetchSpec<T> {
+
+	private final ConnectionAccessor connectionAccessor;
+	private final String sql;
+	private final Function<Connection, Flux<T>> resultFunction;
+	private final Function<Connection, Mono<Integer>> updatedRowsFunction;
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#one()
+	 */
+	@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();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#first()
+	 */
+	@Override
+	public Mono<T> first() {
+		return all().next();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#all()
+	 */
+	@Override
+	public Flux<T> all() {
+		return connectionAccessor.inConnectionMany(resultFunction);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#rowsUpdated()
+	 */
+	@Override
+	public Mono<Integer> rowsUpdated() {
+		return connectionAccessor.inConnection(updatedRowsFunction);
+	}
+}

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<T> implements SqlResult<T> {
+
+	private final ConnectionAccessor connectionAccessor;
+	private final String sql;
+	private final Function<Connection, Flux<Result>> resultFunction;
+	private final Function<Connection, Mono<Integer>> updatedRowsFunction;
+	private final FetchSpec<T> fetchSpec;
+
+	DefaultSqlResult(ConnectionAccessor connectionAccessor, String sql, Function<Connection, Flux<Result>> resultFunction,
+			Function<Connection, Mono<Integer>> updatedRowsFunction, BiFunction<Row, RowMetadata, T> 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 <R> SqlResult<R> extract(BiFunction<Row, RowMetadata, R> mappingFunction) {
+		return new DefaultSqlResult<>(connectionAccessor, sql, resultFunction, updatedRowsFunction, mappingFunction);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#one()
+	 */
+	@Override
+	public Mono<T> one() {
+		return fetchSpec.one();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#first()
+	 */
+	@Override
+	public Mono<T> first() {
+		return fetchSpec.first();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#all()
+	 */
+	@Override
+	public Flux<T> all() {
+		return fetchSpec.all();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#rowsUpdated()
+	 */
+	@Override
+	public Mono<Integer> rowsUpdated() {
+		return fetchSpec.rowsUpdated();
+	}
+}

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<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/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<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>
+	 * @return a new {@link SqlResult} with {@link BiFunction mapping function} applied.
+	 */
 	<R> SqlResult<R> extract(BiFunction<Row, RowMetadata, R> mappingFunction);
 }

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.
- * <p>
- * 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}.
+ * <p/>
+ * 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.
-	 * <p>
+	 * <p/>
 	 * This will typically be the native driver Connection or a wrapper from a connection pool.
 	 *
 	 * @return the underlying Connection (never {@code null})