ReactiveFlowBridge.java

package hu.akarnokd.reactiveflowbridge;

import java.util.Objects;
import java.util.concurrent.Flow;
import java.util.function.Function;

/**
 * Bridge between Reactive-Streams API and the Java 9 Flow API.
 */
public final class ReactiveFlowBridge {
    /** Utility class. */
    private ReactiveFlowBridge() {
        throw new IllegalStateException("No instances!");
    }

    /**
     * Converts a Flow Publisher into a Reactive Publisher.
     * @param <T> the value type
     * @param flowPublisher the source Flow Publisher to convert
     * @return the equivalent Reactive Publisher
     */
    @SuppressWarnings("unchecked")
    public static <T> org.reactivestreams.Publisher<T> toReactive(
            Flow.Publisher<? extends T> flowPublisher) {
        Objects.requireNonNull(flowPublisher, "flowPublisher");
        if (flowPublisher instanceof org.reactivestreams.Publisher) {
            return (org.reactivestreams.Publisher<T>)flowPublisher;
        }
        if (flowPublisher instanceof FlowPublisherFromReactive) {
            return (org.reactivestreams.Publisher<T>)(((FlowPublisherFromReactive<T>)flowPublisher).reactive);
        }
        return new ReactivePublisherFromFlow<>(flowPublisher);
    }

    /**
     * Converts a Reactive Publisher into a Flow Publisher.
     * @param <T> the value type
     * @param reactivePublisher the source Reactive Publisher to convert
     * @return the equivalent Flow Publisher
     */
    @SuppressWarnings("unchecked")
    public static <T> Flow.Publisher<T> toFlow(
            org.reactivestreams.Publisher<? extends T> reactivePublisher
    ) {
        Objects.requireNonNull(reactivePublisher, "reactivePublisher");
        if (reactivePublisher instanceof org.reactivestreams.Publisher) {
            return (Flow.Publisher<T>)reactivePublisher;
        }
        if (reactivePublisher instanceof ReactivePublisherFromFlow) {
            return (Flow.Publisher<T>)(((ReactivePublisherFromFlow<T>)reactivePublisher).flow);
        }
        return new FlowPublisherFromReactive<>(reactivePublisher);
    }
    
    /**
     * Converts a Flow Processor into a Reactive Processor.
     * @param <T> the input value type
     * @param <U> the output value type
     * @param flowProcessor the source Flow Processor to convert
     * @return the equivalent Reactive Processor
     */
    @SuppressWarnings("unchecked")
    public static <T, U> org.reactivestreams.Processor<T, U> toReactive(
            Flow.Processor<? super T, ? extends U> flowProcessor
    ) {
        Objects.requireNonNull(flowProcessor, "flowProcessor");
        if (flowProcessor instanceof org.reactivestreams.Processor) {
            return (org.reactivestreams.Processor<T, U>)flowProcessor;
        }
        if (flowProcessor instanceof FlowToReactiveProcessor) {
            return (org.reactivestreams.Processor<T, U>)(((FlowToReactiveProcessor<T, U>)flowProcessor).reactive);
        }
        return new ReactiveToFlowProcessor<>(flowProcessor);
    }

    /**
     * Converts a Reactive Processor into a Flow Processor.
     * @param <T> the input value type
     * @param <U> the output value type
     * @param reactiveProcessor the source Reactive Processor to convert
     * @return the equivalent Flow Processor
     */
    @SuppressWarnings("unchecked")
    public static <T, U> Flow.Processor<T, U> toFlow(
            org.reactivestreams.Processor<? super T, ? extends U> reactiveProcessor
        ) {
        Objects.requireNonNull(reactiveProcessor, "reactiveProcessor");
        if (reactiveProcessor instanceof org.reactivestreams.Processor) {
            return (Flow.Processor<T, U>)reactiveProcessor;
        }
        if (reactiveProcessor instanceof ReactiveToFlowProcessor) {
            return (Flow.Processor<T, U>)(((ReactiveToFlowProcessor<T, U>)reactiveProcessor).flow);
        }
        return new FlowToReactiveProcessor<>(reactiveProcessor);
    }

    
    /**
     * Converts a Reactive Publisher into a Flow Publisher via a function then
     * applies the returned Flow Publisher to a transformer function that returns a value.
     * <p>
     * This allows a more inlined conversion from a Reactive Publisher to a
     * fluent Flow-based API:
     * <pre><code>
     * org.reactivestreams.Publisher&lt;T> source = ...
     * 
     * ReactiveFlowBridge.to(
     *     source, 
     *     ReactiveFlowBridge::toFlow, 
     *     FluentFlow::from)
     * .map(v -> v)
     * .subscribe(...);
     * </code></pre>
     * @param <T> the element type
     * @param <U> the output type
     * @param reactivePublisher the source Reactive Publisher
     * @param toFlow the function that converts from Reactive Publisher into Flow Publisher
     * @param transformer the transformer that takes a Flow Publisher and returns some transformation of it
     * @return the transformation result
     */
    public static <T, U> U to(
            org.reactivestreams.Publisher<? extends T> reactivePublisher, 
            Function<? super org.reactivestreams.Publisher<? extends T>, ? extends Flow.Publisher<? extends T>> toFlow,
            Function<Flow.Publisher<? extends T>, U> transformer) {
        return transformer.apply(toFlow.apply(reactivePublisher));
    }

    /**
     * Converts a Flow Publisher into a Reactive Publisher via a function then
     * applies the returned Reactive Publisher to a transformer function that returns a value.
     * <p>
     * This allows a more inlined conversion from a Flow Publisher to a
     * fluent Reactive-based API:
     * <pre><code>
     * Flow.Publisher&lt;T> source = ...
     * 
     * ReactiveFlowBridge.to(
     *     source, 
     *     ReactiveFlowBridge::toFlow, 
     *     Observable::fromPublisher)
     * .map(v -> v)
     * .subscribe(...);
     * </code></pre>
     * @param <T> the element type
     * @param <U> the output type
     * @param flowPublisher the source Reactive Publisher
     * @param toReactive the function that converts from Reactive Publisher into Flow Publisher
     * @param transformer the transformer that takes a Flow Publisher and returns some transformation of it
     * @return the transformation result
     */
    public static <T, U> U to(
            Flow.Publisher<? extends T> flowPublisher, 
            Function<? super Flow.Publisher<? extends T>, ? extends org.reactivestreams.Publisher<? extends T>> toReactive,
            Function<org.reactivestreams.Publisher<? extends T>, U> transformer) {
        return transformer.apply(toReactive.apply(flowPublisher));
    }
    
    /**
     * Wraps a Reactive Subscription and converts the calls to a Flow Subscription.
     */
    static final class FlowToReactiveSubscription implements Flow.Subscription {
        private final org.reactivestreams.Subscription reactive;
        
        public FlowToReactiveSubscription(org.reactivestreams.Subscription reactive) {
            this.reactive = reactive;
        }

        @Override
        public void request(long n) {
            reactive.request(n);
        }

        @Override
        public void cancel() {
            reactive.cancel();
        }
        
    }
    
    /**
     * Wraps a Flow Subscription and converts the calls to a Reactive Subscription.
     */
    static final class ReactiveToFlowSubscription implements org.reactivestreams.Subscription {
        private final Flow.Subscription flow;
        
        public ReactiveToFlowSubscription(Flow.Subscription flow) {
            this.flow = flow;
        }

        @Override
        public void request(long n) {
            flow.request(n);
        }

        @Override
        public void cancel() {
            flow.cancel();
        }
        
        
    }
    
    /**
     * Wraps a Reactive Subscriber and forwards methods of the Flow Subscriber to it.
     * @param <T> the value type
     */
    static final class FlowToReactiveSubscriber<T> 
            implements Flow.Subscriber<T> {
        private final org.reactivestreams.Subscriber<? super T> reactive;
        
        public FlowToReactiveSubscriber(org.reactivestreams.Subscriber<? super T> reactive) {
            this.reactive = reactive;
        }

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            reactive.onSubscribe(new ReactiveToFlowSubscription(subscription));
        }

        @Override
        public void onNext(T item) {
            reactive.onNext(item);
        }

        @Override
        public void onError(Throwable throwable) {
            reactive.onError(throwable);
        }

        @Override
        public void onComplete() {
            reactive.onComplete();
        }
        
    }

    /**
     * Wraps a Reactive Subscriber and forwards methods of the Flow Subscriber to it.
     * @param <T> the value type
     */
    static final class ReactiveToFlowSubscriber<T> 
            implements org.reactivestreams.Subscriber<T> {
        private final Flow.Subscriber<? super T> flow;
        
        public ReactiveToFlowSubscriber(Flow.Subscriber<? super T> flow) {
            this.flow = flow;
        }

        @Override
        public void onSubscribe(org.reactivestreams.Subscription subscription) {
            flow.onSubscribe(new FlowToReactiveSubscription(subscription));
        }

        @Override
        public void onNext(T item) {
            flow.onNext(item);
        }

        @Override
        public void onError(Throwable throwable) {
            flow.onError(throwable);
        }

        @Override
        public void onComplete() {
            flow.onComplete();
        }
        
    }
    
    /**
     * Wraps a Flow Processor and forwards methods of the Reactive Processor to it.
     * @param <T> the input type
     * @param <U> the output type
     */
    static final class ReactiveToFlowProcessor<T, U>
            implements org.reactivestreams.Processor<T, U> {
        private final Flow.Processor<? super T, ? extends U> flow;
        
        public ReactiveToFlowProcessor(Flow.Processor<? super T, ? extends U> flow) {
            this.flow = flow;
        }

        @Override
        public void onSubscribe(org.reactivestreams.Subscription s) {
            flow.onSubscribe(new FlowToReactiveSubscription(s));
        }

        @Override
        public void onNext(T t) {
            flow.onNext(t);
        }

        @Override
        public void onError(Throwable t) {
            flow.onError(t);
        }

        @Override
        public void onComplete() {
            flow.onComplete();
        }

        @Override
        public void subscribe(org.reactivestreams.Subscriber<? super U> s) {
            flow.subscribe(new FlowToReactiveSubscriber<>(s));
        }
    }
    
    /**
     * Wraps a Reactive Processor and forwards methods of the Flow Processor to it.
     * @param <T> the input type
     * @param <U> the output type
     */
    static final class FlowToReactiveProcessor<T, U>
            implements Flow.Processor<T, U> {
        private final org.reactivestreams.Processor<? super T, ? extends U> reactive;
        
        public FlowToReactiveProcessor(org.reactivestreams.Processor<? super T, ? extends U> reactive) {
            this.reactive = reactive;
        }

        @Override
        public void onSubscribe(Flow.Subscription s) {
            reactive.onSubscribe(new ReactiveToFlowSubscription(s));
        }

        @Override
        public void onNext(T t) {
            reactive.onNext(t);
        }

        @Override
        public void onError(Throwable t) {
            reactive.onError(t);
        }

        @Override
        public void onComplete() {
            reactive.onComplete();
        }

        @Override
        public void subscribe(Flow.Subscriber<? super U> s) {
            reactive.subscribe(new ReactiveToFlowSubscriber<>(s));
        }
    }

    /**
     * Reactive Publisher that wraps a Flow Publisher.
     * @param <T> the value type
     */
    static final class ReactivePublisherFromFlow<T> implements org.reactivestreams.Publisher<T> {

        final Flow.Publisher<? extends T> flow;

        public ReactivePublisherFromFlow(Flow.Publisher<? extends T> flowPublisher) {
            this.flow = flowPublisher;
        }

        @Override
        public void subscribe(org.reactivestreams.Subscriber<? super T> reactive) {
            flow.subscribe(new FlowToReactiveSubscriber<>(reactive));
        }
    }

    /**
     * Flow Publisher that wraps a Reactive Publisher.
     * @param <T> the value type
     */
    static final class FlowPublisherFromReactive<T> implements Flow.Publisher<T> {

        final org.reactivestreams.Publisher<? extends T> reactive;

        public FlowPublisherFromReactive(org.reactivestreams.Publisher<? extends T> reactivePublisher) {
            this.reactive = reactivePublisher;
        }

        @Override
        public void subscribe(Flow.Subscriber<? super T> flow) {
            reactive.subscribe(new ReactiveToFlowSubscriber<>(flow));
        }
    }

}