Java 9 Flow bridge: add Subscriber converters (#399)

* Java 9 Flow bridge: add Subscriber converters

* Fix return type javadoc

Author: akarnokd

flow-bridge/src/main/java/org/reactivestreams/ReactiveStreamsFlowBridge.java

@@ -110,7 +110,33 @@ public static <T, U> Flow.Processor<T, U> toFlow(
         }
         return new FlowToReactiveProcessor<T, U>(reactiveStreamsProcessor);
     }
-    
+
+    /**
+     * Converts a Reactive Streams Subscriber into a Flow Subscriber.
+     * @param <T> the input and output value type
+     * @param reactiveStreamsSubscriber the Reactive Streams Subscriber instance to convert
+     * @return the equivalent Flow Subscriber
+     */
+    public static <T> Flow.Subscriber<T> toFlowSubscriber(org.reactivestreams.Subscriber<T> reactiveStreamsSubscriber) {
+        if (reactiveStreamsSubscriber == null) {
+            throw new NullPointerException("reactiveStreamsSubscriber");
+        }
+        return new FlowToReactiveSubscriber<T>(reactiveStreamsSubscriber);
+    }
+
+    /**
+     * Converts a Flow Subscriber into a Reactive Streams Subscriber.
+     * @param <T> the input and output value type
+     * @param flowSubscriber the Flow Subscriber instance to convert
+     * @return the equivalent Reactive Streams Subscriber
+     */
+    public static <T> org.reactivestreams.Subscriber<T> toReactiveStreamsSubscriber(Flow.Subscriber<T> flowSubscriber) {
+        if (flowSubscriber == null) {
+            throw new NullPointerException("flowSubscriber");
+        }
+        return new ReactiveToFlowSubscriber<T>(flowSubscriber);
+    }
+
     /**
      * Wraps a Reactive Streams Subscription and converts the calls to a Flow Subscription.
      */

flow-bridge/src/test/java/org/reactivestreams/ReactiveStreamsFlowBridgeTest.java

@@ -11,6 +11,7 @@
 
 package org.reactivestreams;
 
+import org.testng.Assert;
 import org.testng.annotations.Test;
 
 import java.io.IOException;
@@ -110,4 +111,68 @@ public void execute(Runnable command) {
 
         tc.assertFailure(IOException.class, 1, 2, 3, 4, 5);
     }
+
+    @Test
+    public void reactiveStreamsToFlowSubscriber() {
+        TestEitherConsumer<Integer> tc = new TestEitherConsumer<Integer>();
+
+        Flow.Subscriber<Integer> fs = ReactiveStreamsFlowBridge.toFlowSubscriber(tc);
+
+        final Object[] state = { null, null };
+
+        fs.onSubscribe(new Flow.Subscription() {
+            @Override
+            public void request(long n) {
+                state[0] = n;
+            }
+
+            @Override
+            public void cancel() {
+                state[1] = true;
+            }
+        });
+
+        Assert.assertEquals(state[0], Long.MAX_VALUE);
+
+        fs.onNext(1);
+        fs.onNext(2);
+        fs.onNext(3);
+        fs.onComplete();
+
+        tc.assertResult(1, 2, 3);
+
+        Assert.assertNull(state[1]);
+    }
+
+    @Test
+    public void flowToReactiveStreamsSubscriber() {
+        TestEitherConsumer<Integer> tc = new TestEitherConsumer<Integer>();
+
+        org.reactivestreams.Subscriber<Integer> fs = ReactiveStreamsFlowBridge.toReactiveStreamsSubscriber(tc);
+
+        final Object[] state = { null, null };
+
+        fs.onSubscribe(new org.reactivestreams.Subscription() {
+            @Override
+            public void request(long n) {
+                state[0] = n;
+            }
+
+            @Override
+            public void cancel() {
+                state[1] = true;
+            }
+        });
+
+        Assert.assertEquals(state[0], Long.MAX_VALUE);
+
+        fs.onNext(1);
+        fs.onNext(2);
+        fs.onNext(3);
+        fs.onComplete();
+
+        tc.assertResult(1, 2, 3);
+
+        Assert.assertNull(state[1]);
+    }
 }