#244 - Add Reactor Checkpoint operator for SQL execution.

We now integrate with Reactor's checkpoint operator to include more information for debugging.

Author: mp911de

src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java

@@ -386,7 +386,7 @@ <T> FetchSpec<T> exchange(Supplier<String> sqlSupplier, BiFunction<Row, RowMetad
 				return statement;
 			};
 
-			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(executeFunction.apply(it).execute());
+			Function<Connection, Flux<Result>> resultFunction = toExecuteFunction(sql, executeFunction);
 
 			return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
 					sql, //
@@ -707,7 +707,7 @@ <R> FetchSpec<R> execute(PreparedOperation<?> preparedOperation, BiFunction<Row,
 
 			String sql = getRequiredSql(preparedOperation);
 			Function<Connection, Statement> selectFunction = wrapPreparedOperation(sql, preparedOperation);
-			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(selectFunction.apply(it).execute());
+			Function<Connection, Flux<Result>> resultFunction = DefaultDatabaseClient.toExecuteFunction(sql, selectFunction);
 
 			return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
 					sql, //
@@ -1377,7 +1377,7 @@ private <R> FetchSpec<R> exchangeInsert(BiFunction<Row, RowMetadata, R> mappingF
 		String sql = getRequiredSql(operation);
 		Function<Connection, Statement> insertFunction = wrapPreparedOperation(sql, operation)
 				.andThen(statement -> statement.returnGeneratedValues());
-		Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(insertFunction.apply(it).execute());
+		Function<Connection, Flux<Result>> resultFunction = toExecuteFunction(sql, insertFunction);
 
 		return new DefaultSqlResult<>(this, //
 				sql, //
@@ -1390,7 +1390,7 @@ private UpdatedRowsFetchSpec exchangeUpdate(PreparedOperation<?> operation) {
 
 		String sql = getRequiredSql(operation);
 		Function<Connection, Statement> executeFunction = wrapPreparedOperation(sql, operation);
-		Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(executeFunction.apply(it).execute());
+		Function<Connection, Flux<Result>> resultFunction = toExecuteFunction(sql, executeFunction);
 
 		return new DefaultSqlResult<>(this, //
 				sql, //
@@ -1421,6 +1421,16 @@ private Function<Connection, Statement> wrapPreparedOperation(String sql, Prepar
 		};
 	}
 
+	private static Function<Connection, Flux<Result>> toExecuteFunction(String sql,
+			Function<Connection, Statement> executeFunction) {
+
+		return it -> {
+
+			Flux<Result> from = Flux.defer(() -> executeFunction.apply(it).execute()).cast(Result.class);
+			return from.checkpoint("SQL \"" + sql + "\" [DatabaseClient]");
+		};
+	}
+
 	private static <T> Flux<T> doInConnectionMany(Connection connection, Function<Connection, Flux<T>> action) {
 
 		try {