R2DBC Stored Procedures Proposal

Parameter.java

/**
 * OUT and IN/OUT parameter wrapper for stored procedures (callable functions).
 *
 * @param <T>
 */
class Parameter<T> {

    @Nullable
    private final T value;

    private final Class<T> javaType;

    @Nullable
    private final String declaration;

    private final boolean inOut;

    private Parameter(@Nullable T value, Class<T> javaType, @Nullable String declaration, boolean inOut) {
        this.value = value;
        this.javaType = javaType;
        this.declaration = declaration;
        this.inOut = inOut;
    }

    /**
     * Creates a new {@code OUT} parameter registration given a {@link Class Java type}.
     * Maps to its appropriate vendor-specific type.
     *
     * @param valueType Java type the out parameter maps to.
     * @return
     */
    public static <T> Parameter<T> out(Class<T> valueType) {
        return new Parameter<>(null, valueType, null, false);
    }

    /**
     * Creates a new {@code OUT} parameter registration given a {@link Class Java type} and type {@code declaration}.
     *
     * @param valueType   Java type the out parameter maps to.
     * @param declaration vendor-specific parameter type declaration.
     * @return
     */
    public static <T> Parameter<T> out(Class<T> valueType, String declaration) {
        return new Parameter<>(null, valueType, declaration, false);
    }

    /**
     * Creates a new {@code INOUT} parameter registration given a {@link Class Java type}.
     * Maps to its appropriate vendor-specific type.
     *
     * @param valueType Java type the out parameter maps to.
     * @return
     */
    public static <T> Parameter<T> inOut(Class<T> valueType) {
        return new Parameter<>(null, valueType, null, true);
    }

    /**
     * Creates a new {@code INOUT} parameter registration given a {@link Class Java type} and type {@code declaration}.
     *
     * @param valueType   Java type the out parameter maps to.
     * @param declaration vendor-specific parameter type declaration.
     * @return
     */
    public static <T> Parameter<T> inOut(Class<T> valueType, String declaration) {
        return new Parameter<>(null, valueType, declaration, true);
    }

    /**
     * Creates a new {@code INOUT} parameter registration a {@code value}.
     * Maps to its appropriate vendor-specific type using the {@code value} type.
     *
     * @param value parameter value.
     * @return
     */
    public static <T> Parameter<T> inOut(T value) {
        return new Parameter<>(value, (Class<T>) value.getClass(), null, true);
    }

    /**
     * Creates a new {@code INOUT} parameter registration given a {@code value}, {@link Class Java type} and type {@code declaration}.
     *
     * @param value       parameter value.
     * @param declaration vendor-specific parameter type declaration.
     * @return
     */
    public static <T> Parameter<T> inOut(T value, String declaration) {
        return new Parameter<>(value, (Class<T>) value.getClass(), declaration, true);
    }

    /**
     * Creates a new {@code INOUT} parameter registration given a {@code value}, {@link Class Java type} and type {@code declaration}.
     *
     * @param value       parameter value.
     * @param valueType   Java type the out parameter maps to.
     * @param declaration vendor-specific parameter type declaration.
     * @return
     */
    public static <T> Parameter<T> inOut(T value, Class<T> valueType, String declaration) {
        return new Parameter<>(value, valueType, declaration, true);
    }

    public boolean hasValue() {
        return this.value != null;
    }

    @Nullable
    public T getValue() {
        return value;
    }

    @Nullable
    public Class<T> getJavaType() {
        return javaType;
    }

    @Nullable
    public String getDeclaration() {
        return this.declaration;
    }

    public boolean isInOut() {
        return this.inOut;
    }

    public boolean isOut() {
        return !this.inOut;
    }
}

Responses.java

/**
 * Generic marker interface for response elements.
 * A response element can be a:
 *  - Row
 *  - Notice (Typically INFO or ERROR along with SQLState and a numeric code)
 *  - OutParameter
 *  - RowCount
 *  - Optional: Any vendor-specific type
 */
interface Response {
}

/**
 * Out-parameter resulting from a stored procedure/trigger.
 *
 * Impl note: Parameters are subject to vendor-specific ordering.
 * Drivers could provide an option to emit out parameters sorted by {@link #getIndex()}.
 */
interface OutParameter extends Response {

    /**
     * @return the parameter index.
     */
    int getIndex();

    /**
     * @return parameter name, optional.
     */
    @Nullable
    String getName();

    /**
     * Returns the value for this parameter.
     *
     * @param type       the type of item to return. This type must be assignable to, and allows for variance.
     * @param <T>        the type of the item being returned
     * @return the value for a column in this row.  Value can be {@code null}.
     * @throws IllegalArgumentException if {@code identifier} or {@code type} is {@code null}
     */
    @Nullable
    <T> T get(Class<T> type);

    /**
     * Returns the value for this parameter using the default type mapping.  The default implementation of this method calls {@link #get(Class)} passing {@link Object} as the type in
     * order to allow the implementation to make the loosest possible match.
     *
     * @return the value for a column in this row.  Value can be {@code null}.
     * @throws IllegalArgumentException if {@code identifier} or {@code type} is {@code null}
     */
    @Nullable
    default Object get() {
        return get(Object.class);
    }
}

/**
 * Change to Row: implement Response interface.
 */
interface Row extends Response {
}

/**
 * Encapsumates info or error messages. Allows categorization and
 * consumption as a server response does not terminate if an error happens.
 * Subsequent errors or other elements might be emitted.
 */
interface Notice extends Response {

    /**
     * @return {@literal true} if this is an information.
     */
    boolean isInfo();

    /**
     * @return {@literal true} if this is a warning.
     */
    boolean isWarning();

    /**
     * @return {@literal true} if this is an error.
     */
    boolean isError();

    /**
     * @return creates the appropriate {@link R2dbcException}.
     */
    R2dbcException toException();

    /**
     * @return the message associated with this notice.
     */
    String getMessage();
}

/**
 * Encapsulates a row count message, typically used with INSERT/UPDATE/DELETE.
 *
 * Impl notice: Drivers can emit multiple row counts if they operate in a chunked mode.
 * Clients can summarize these to a total count (that's what happens in {@link Result#getRowsUpdated()}).
 */
interface RowCount extends Response {

    /**
     * @return the number of affected rows.
     */
    long getCount();
}

Result.java

public interface Result {

    // Methods to add

    /**
     * Returns a mapping of the {@link Response responses} on a {@link Predicate filtered stream of results} that is the result of a query against a database. Response elements can be out 
     * parameters, row counts, rows or notifications.
     *
     * @param filter
     * @param f
     * @param <T>
     * @return
     */
    <T> Publisher<T> map(Predicate<? extends Response> filter, Function<? extends Response, ? extends T> f);

    /**
     * Returns a mapping of the {@link Response responses} that are the results of a query against a database. Response elements can be out parameters, row counts, rows or notifications.
     *
     * @param f
     * @param <T>
     * @return
     */
    <T> Publisher<T> flatMap(Function<? extends Response, ? extends Publisher<? extends T>> f);

}

Usage.java

Connection connection = …;

Publisher<String> functionResult = connection.createStatement("CALL my_function($1, $2, $3)")
    .bind(0, Parameter.out(String.class))
    .bind(1, Parameter.inOut("foo", "VARCHAR(400)"))
    .bind(2, Parameter.out(String.class, "NVARCHAR(8000)"))
    .execute()
    .flatMap(result -> result. map(OutParameter.class::isInstance, response -> {
           OutParameter param = (OutParameter) response;
           return param.get(String.class).
        }));