client.java

//
// Rest client for Foo service
//
public interface FooClient {
    /**
     * @return a {@link Mono} upon subscription it sends request to retrieve Foo and once response
     * is received it emits the retrieved Foo.
     */
    Mono<Foo> getFoo();

    /**
     * Extract a response flux from the given {@link RestResponse}'s headers and it's content.
     *
     * @param receiver extracting receiver function, receives a {@link Mono} upon subscription it sends
     *                 request to retrieve Foo and once response is received it emits the {@link RestResponse}
     *                 holding Foo and response headers.
     *                 The recevier can: 
     *                   1. Completely process the response within it's scope and forward a completion event.
     *                   2. Simply forward the response without processing it.
     *                   3. Process the response, transform it and forward the transformed response.
     *                 The forwarded event will piped through {@link Flux} returned by getFoo.  
     * @param <T> the extracted flux type
     * @return a {@link Flux} forwarding the returned {@link Publisher} sequence
     */
    <T> Flux<T> getFoo(Function<Mono<RestResponse<Foo>>, ? extends Publisher<T>> receiver);

    /**
     * @return a {@link Mono} upon subscription it submits Foo and once response is received it emits
     * {@link Subscriber#onComplete()} event.
     */
    Mono<Void> putFoo();

    /**
     * Extract a response flux from the given {@link RestResponse}'s headers.
     *
     * @param receiver extracting receiver, receives a {@link Mono} upon subscription it submits Foo
     *                 and once response is received it emits {@link RestResponse} containing response
     *                 headers.
     *                 The recevier can: 
     *                   1. Completely process the response within it's scope and forward a completion event.
     *                   2. Simply forward the response without processing it.
     *                   3. Process the response, transform it and forward the transformed response.
     *                 The forwarded event will piped through {@link Flux} returned by putFoo. 
     * @param <T> the extracted flux type
     * @return a {@link Flux} forwarding the returned {@link Publisher} sequence
     */
    <T> Flux<T> putFoo(Function<Mono<RestResponse<Void>>, ? extends Publisher<T>> receiver);

    /**
     * @return a {@link Flux} upon subscription it sends series of paged-list requests to list Foo instances
     * and stream the returned Foo instances.
     */
    Flux<Foo> listFoo();

    /**
     * Extract a response flux from the given {@link PageResponse}'s headers and it's content.
     *
     * @param receiver extracting receiver, receives a {@link Flux} upon subscription send paged-list
     *                 requests to list Foo instances and stream {@link PageResponse} containing each page
     *                 and response headers.
     *                 The recevier can: 
     *                   1. Completely process the response within it's scope and forward a completion event.
     *                   2. Simply forward the response without processing it.
     *                   3. Process the response, transform it and forward the transformed response.
     *                 The forwarded event will piped through {@link Flux} returned by listFoo. 
     * @param <T> the extracted flux type
     * @return  a {@link Flux} forwarding the returned {@link Publisher} sequence
     */
    <T> Flux<T> listFoo(Function<Flux<PageResponse<Foo>>, ? extends Publisher<T>> receiver);

    /**
     * @return a {@link Mono} upon subscription sends request to create Foo and emits Foo once
     * it is created.
     */
    Mono<Foo> createFoo();

    /**
     * Extract a response flux from the given {@link LroResponse}'s headers, it's content and polling state.
     *
     * @param receiver extracting receiver, receives a {@link Flux} upon subscription its send a create
     *                 request and multiple polling requests. Responses for these requests will be streamed as LroResponse
     *                 containing LRO operation state & headers. Stream ends with a LroResponse containing the final
     *                 result T (created resource) and headers.
     *                 The recevier can: 
     *                   1. Completely process the response within it's scope and forward a completion event.
     *                   2. Simply forward the response without processing it.
     *                   3. Process the response, transform it and forward the transformed response.
     *                 The forwarded event will piped through {@link Flux} returned by beginCreateFoo.
     * @param <T> the extracted flux type
     * @return a {@link Flux} forwarding the returned {@link Publisher} sequence
     */
    <T> Flux<T> beginCreateFoo(Function<Flux<LroResponse<Foo>>, ? extends Publisher<T>> receiver);
}

//
// Types representing Extractor input.
//
/**
 * The general response of a rest request.
 *
 * @param <T> result type
 */
interface RestResponse<T> {
    T content();
    Map<String, List<String>> headers();
}

/**
 * Response type holding a items in single page and it's headers.
 *
 * @param <T> type of page items
 */
interface PageResponse<T> {
    List<T> items();
    Map<String, List<String>> headers();
}

/**
 * Response type holding result of LRO polling operation.
 *
 * @param <T> the type of the final result of LRO
 */
interface LroResponse<T> {
    PollingState state();
    T content();
    Map<String, List<String>> headers();
}

/**
 * Data class holding polling state.
 */
class PollingState {
}

/**
 * result type
 */
class Foo {
}

// 
// FooClient usage
//
FooClient client = createClient();
//
Flux<Foo> getFooContentFlux = client.getFoo(responseMono ->
        responseMono.flatMapMany(response -> Flux.just(response.content())));
//
Flux<Map<String, List<String>>> getFooHeadersFlux =
        client.getFoo(responseMono -> responseMono
                .flatMapMany(response -> Flux.just(response.headers())));
//
Flux<Map<String, List<String>>> putFooHeadersFlux = client.putFoo(responseMono ->
        responseMono.flatMapMany(response -> Flux.just(response.headers())));
//
Flux<Foo> listFooFlux = client.listFoo(pageResponseFlux ->
        pageResponseFlux.map(r -> r.items()).flatMapIterable(i -> i));