#189 - Accept StatementFilterFunction in DatabaseClient.

We now accept StatementFilterFunction and ExecuteFunction via DatabaseClient to filter Statement execution. StatementFilterFunctions can be used to pre-process the statement or post-process Result objects.

databaseClient.execute(…)
		.filter((s, next) -> next.execute(s.returnGeneratedValues("my_id")))
		.filter((s, next) -> next.execute(s.fetchSize(25)))

databaseClient.execute(…)
		.filter(s -> s.returnGeneratedValues("my_id"))
		.filter(s -> s.fetchSize(25))

Original pull request: #308.

Author: mp911de

Diff for: src/main/asciidoc/new-features.adoc

@@ -5,7 +5,8 @@
 == What's New in Spring Data R2DBC 1.1.0 RELEASE
 
 * Introduction of `R2dbcEntityTemplate` for entity-oriented operations.
-* Support interface projections with `DatabaseClient.as(…)`
+* Support interface projections with `DatabaseClient.as(…)`.
+* <<r2dbc.datbaseclient.filter,Support for `ExecuteFunction` and `StatementFilterFunction` via `DatabaseClient.filter(…)`>>.
 
 [[new-features.1-0-0-RELEASE]]
 == What's New in Spring Data R2DBC 1.0.0 RELEASE

Diff for: src/main/asciidoc/reference/r2dbc-sql.adoc

@@ -134,7 +134,7 @@ In JDBC, the actual drivers translate `?` bind markers to database-native marker
 
 Spring Data R2DBC lets you use native bind markers or named bind markers with the `:name` syntax.
 
-Named parameter support leverages a `R2dbcDialect` instance  to expand named parameters to native bind markers at the time of query execution, which gives you a certain degree of query portability across various database vendors.
+Named parameter support leverages a `R2dbcDialect` instance to expand named parameters to native bind markers at the time of query execution, which gives you a certain degree of query portability across various database vendors.
 ****
 
 The query-preprocessor unrolls named `Collection` parameters into a series of bind markers to remove the need of dynamic query creation based on the number of arguments.
@@ -159,7 +159,7 @@ tuples.add(new Object[] {"John", 35});
 tuples.add(new Object[] {"Ann",  50});
 
 db.execute("SELECT id, name, state FROM table WHERE (name, age) IN (:tuples)")
-    .bind("tuples", tuples);
+    .bind("tuples", tuples)
 ----
 ====
 
@@ -171,6 +171,38 @@ The following example shows a simpler variant using `IN` predicates:
 [source,java]
 ----
 db.execute("SELECT id, name, state FROM table WHERE age IN (:ages)")
-    .bind("ages", Arrays.asList(35, 50));
+    .bind("ages", Arrays.asList(35, 50))
 ----
 ====
+
+[[r2dbc.datbaseclient.filter]]
+== Statement Filters
+
+You can register a `Statement` filter (`StatementFilterFunction`) through `DatabaseClient` to intercept and modify statements in their execution, as the following example shows:
+
+====
+[source,java]
+----
+db.execute("INSERT INTO table (name, state) VALUES(:name, :state)")
+    .filter((s, next) -> next.execute(s.returnGeneratedValues("id")))
+    .bind("name", …)
+    .bind("state", …)
+----
+====
+
+`DatabaseClient` exposes also simplified `filter(…)` overload accepting `UnaryOperator<Statement>`:
+
+====
+[source,java]
+----
+db.execute("INSERT INTO table (name, state) VALUES(:name, :state)")
+    .filter(s -> s.returnGeneratedValues("id"))
+    .bind("name", …)
+    .bind("state", …)
+
+db.execute("SELECT id, name, state FROM table")
+    .filter(s -> s.fetchSize(25))
+----
+====
+
+`StatementFilterFunction` allow filtering of the executed `Statement` and filtering of `Result` objects.

Diff for: src/main/java/org/springframework/data/r2dbc/core/DatabaseClient.java

@@ -18,6 +18,7 @@
 import io.r2dbc.spi.ConnectionFactory;
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
+import io.r2dbc.spi.Statement;
 import reactor.core.publisher.Mono;
 
 import java.util.Arrays;
@@ -26,6 +27,7 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.function.UnaryOperator;
 
 import org.reactivestreams.Publisher;
 
@@ -37,6 +39,7 @@
 import org.springframework.data.r2dbc.query.Update;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.util.Assert;
 
 /**
  * A non-blocking, reactive client for performing database calls requests with Reactive Streams back pressure. Provides
@@ -142,6 +145,16 @@ interface Builder {
 		 */
 		Builder exceptionTranslator(R2dbcExceptionTranslator exceptionTranslator);
 
+		/**
+		 * Configures a {@link ExecuteFunction} to execute {@link Statement} objects.
+		 *
+		 * @param executeFunction must not be {@literal null}.
+		 * @return {@code this} {@link Builder}.
+		 * @since 1.1
+		 * @see Statement#execute()
+		 */
+		Builder executeFunction(ExecuteFunction executeFunction);
+
 		/**
 		 * Configures a {@link ReactiveDataAccessStrategy}.
 		 *
@@ -186,7 +199,7 @@ interface Builder {
 	/**
 	 * Contract for specifying a SQL call along with options leading to the exchange.
 	 */
-	interface GenericExecuteSpec extends BindSpec<GenericExecuteSpec> {
+	interface GenericExecuteSpec extends BindSpec<GenericExecuteSpec>, StatementFilterSpec<GenericExecuteSpec> {
 
 		/**
 		 * Define the target type the result should be mapped to. <br />
@@ -231,7 +244,7 @@ interface GenericExecuteSpec extends BindSpec<GenericExecuteSpec> {
 	/**
 	 * Contract for specifying a SQL call along with options leading to the exchange.
 	 */
-	interface TypedExecuteSpec<T> extends BindSpec<TypedExecuteSpec<T>> {
+	interface TypedExecuteSpec<T> extends BindSpec<TypedExecuteSpec<T>>, StatementFilterSpec<TypedExecuteSpec<T>> {
 
 		/**
 		 * Define the target type the result should be mapped to. <br />
@@ -866,4 +879,31 @@ interface BindSpec<S extends BindSpec<S>> {
 		 */
 		S bindNull(String name, Class<?> type);
 	}
+
+	/**
+	 * Contract for applying a {@link StatementFilterFunction}.
+	 *
+	 * @since 1.1
+	 */
+	interface StatementFilterSpec<S extends StatementFilterSpec<S>> {
+
+		/**
+		 * Add the given filter to the end of the filter chain.
+		 *
+		 * @param filter the filter to be added to the chain.
+		 */
+		default S filter(UnaryOperator<Statement> filter) {
+
+			Assert.notNull(filter, "Statement FilterFunction must not be null!");
+
+			return filter((statement, next) -> next.execute(filter.apply(statement)));
+		}
+
+		/**
+		 * Add the given filter to the end of the filter chain.
+		 *
+		 * @param filter the filter to be added to the chain.
+		 */
+		S filter(StatementFilterFunction filter);
+	}
 }