[WIP] TCK for j.u.c.Flow types "directly" (#398)

* Add JDK9 TCK, using adapters

* Fixing wrapping and unwrapping of the wrappers themselves.

* Renames the converters to "toX" for RS and "toFlowX" for Flow.

Fixes so that the dist url for gradle is http iso https (TravisCI bug?)

Adds regression test for bridge converters.

* fix formatting

* cleanup

Author: ktoso

.gitignore

@@ -26,3 +26,4 @@ test-results
 test-tmp
 *.class
 gradle.properties
+*.orig

.java-version

@@ -0,0 +1 @@
+9

build.gradle

@@ -46,7 +46,11 @@ subprojects {
         }
     }
 
-    if (name in ["reactive-streams", "reactive-streams-tck", "reactive-streams-examples", "reactive-streams-flow-bridge"]) {
+    if (name in ["reactive-streams", 
+                 "reactive-streams-tck", 
+                 "reactive-streams-tck-flow",
+                 "reactive-streams-examples",
+                 "reactive-streams-flow-bridge"]) {
         apply plugin: "maven"
         apply plugin: "signing"
 

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

@@ -29,17 +29,17 @@ private ReactiveStreamsFlowBridge() {
      * @return the equivalent Reactive Streams Publisher
      */
     @SuppressWarnings("unchecked")
-    public static <T> org.reactivestreams.Publisher<T> toReactiveStreams(
+    public static <T> org.reactivestreams.Publisher<T> toPublisher(
             Flow.Publisher<? extends T> flowPublisher) {
         if (flowPublisher == null) {
             throw new NullPointerException("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).reactiveStreams);
         }
+        if (flowPublisher instanceof org.reactivestreams.Publisher) {
+            return (org.reactivestreams.Publisher<T>)flowPublisher;
+        }
         return new ReactivePublisherFromFlow<T>(flowPublisher);
     }
 
@@ -50,21 +50,21 @@ public static <T> org.reactivestreams.Publisher<T> toReactiveStreams(
      * @return the equivalent Flow Publisher
      */
     @SuppressWarnings("unchecked")
-    public static <T> Flow.Publisher<T> toFlow(
+    public static <T> Flow.Publisher<T> toFlowPublisher(
             org.reactivestreams.Publisher<? extends T> reactiveStreamsPublisher
     ) {
         if (reactiveStreamsPublisher == null) {
             throw new NullPointerException("reactiveStreamsPublisher");
         }
-        if (reactiveStreamsPublisher instanceof Flow.Publisher) {
-            return (Flow.Publisher<T>)reactiveStreamsPublisher;
-        }
         if (reactiveStreamsPublisher instanceof ReactivePublisherFromFlow) {
             return (Flow.Publisher<T>)(((ReactivePublisherFromFlow<T>)reactiveStreamsPublisher).flow);
         }
+        if (reactiveStreamsPublisher instanceof Flow.Publisher) {
+            return (Flow.Publisher<T>)reactiveStreamsPublisher;
+        }
         return new FlowPublisherFromReactive<T>(reactiveStreamsPublisher);
     }
-    
+
     /**
      * Converts a Flow Processor into a Reactive Streams Processor.
      * @param <T> the input value type
@@ -73,18 +73,18 @@ public static <T> Flow.Publisher<T> toFlow(
      * @return the equivalent Reactive Streams Processor
      */
     @SuppressWarnings("unchecked")
-    public static <T, U> org.reactivestreams.Processor<T, U> toReactiveStreams(
+    public static <T, U> org.reactivestreams.Processor<T, U> toProcessor(
             Flow.Processor<? super T, ? extends U> flowProcessor
     ) {
         if (flowProcessor == null) {
             throw new NullPointerException("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).reactiveStreams);
         }
+        if (flowProcessor instanceof org.reactivestreams.Processor) {
+            return (org.reactivestreams.Processor<T, U>)flowProcessor;
+        }
         return new ReactiveToFlowProcessor<T, U>(flowProcessor);
     }
 
@@ -96,18 +96,18 @@ public static <T, U> org.reactivestreams.Processor<T, U> toReactiveStreams(
      * @return the equivalent Flow Processor
      */
     @SuppressWarnings("unchecked")
-    public static <T, U> Flow.Processor<T, U> toFlow(
+    public static <T, U> Flow.Processor<T, U> toFlowProcessor(
             org.reactivestreams.Processor<? super T, ? extends U> reactiveStreamsProcessor
         ) {
         if (reactiveStreamsProcessor == null) {
             throw new NullPointerException("reactiveStreamsProcessor");
         }
-        if (reactiveStreamsProcessor instanceof Flow.Processor) {
-            return (Flow.Processor<T, U>)reactiveStreamsProcessor;
-        }
         if (reactiveStreamsProcessor instanceof ReactiveToFlowProcessor) {
             return (Flow.Processor<T, U>)(((ReactiveToFlowProcessor<T, U>)reactiveStreamsProcessor).flow);
         }
+        if (reactiveStreamsProcessor instanceof Flow.Processor) {
+            return (Flow.Processor<T, U>)reactiveStreamsProcessor;
+        }
         return new FlowToReactiveProcessor<T, U>(reactiveStreamsProcessor);
     }
 
@@ -117,10 +117,17 @@ public static <T, U> Flow.Processor<T, U> toFlow(
      * @param reactiveStreamsSubscriber the Reactive Streams Subscriber instance to convert
      * @return the equivalent Flow Subscriber
      */
+    @SuppressWarnings("unchecked")
     public static <T> Flow.Subscriber<T> toFlowSubscriber(org.reactivestreams.Subscriber<T> reactiveStreamsSubscriber) {
         if (reactiveStreamsSubscriber == null) {
             throw new NullPointerException("reactiveStreamsSubscriber");
         }
+        if (reactiveStreamsSubscriber instanceof ReactiveToFlowSubscriber) {
+            return (Flow.Subscriber<T>)((ReactiveToFlowSubscriber<T>)reactiveStreamsSubscriber).flow;
+        }
+        if (reactiveStreamsSubscriber instanceof Flow.Subscriber) {
+            return (Flow.Subscriber<T>)reactiveStreamsSubscriber;
+        }
         return new FlowToReactiveSubscriber<T>(reactiveStreamsSubscriber);
     }
 
@@ -130,19 +137,26 @@ public static <T> Flow.Subscriber<T> toFlowSubscriber(org.reactivestreams.Subscr
      * @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) {
+    @SuppressWarnings("unchecked")
+    public static <T> org.reactivestreams.Subscriber<T> toSubscriber(Flow.Subscriber<T> flowSubscriber) {
         if (flowSubscriber == null) {
             throw new NullPointerException("flowSubscriber");
         }
+        if (flowSubscriber instanceof FlowToReactiveSubscriber) {
+            return (org.reactivestreams.Subscriber<T>)((FlowToReactiveSubscriber<T>)flowSubscriber).reactiveStreams;
+        }
+        if (flowSubscriber instanceof org.reactivestreams.Subscriber) {
+            return (org.reactivestreams.Subscriber<T>)flowSubscriber;
+        }
         return new ReactiveToFlowSubscriber<T>(flowSubscriber);
     }
 
     /**
      * Wraps a Reactive Streams Subscription and converts the calls to a Flow Subscription.
      */
     static final class FlowToReactiveSubscription implements Flow.Subscription {
-        private final org.reactivestreams.Subscription reactiveStreams;
-        
+        final org.reactivestreams.Subscription reactiveStreams;
+
         public FlowToReactiveSubscription(org.reactivestreams.Subscription reactive) {
             this.reactiveStreams = reactive;
         }
@@ -156,15 +170,15 @@ public void request(long n) {
         public void cancel() {
             reactiveStreams.cancel();
         }
-        
+
     }
-    
+
     /**
      * Wraps a Flow Subscription and converts the calls to a Reactive Streams Subscription.
      */
     static final class ReactiveToFlowSubscription implements org.reactivestreams.Subscription {
-        private final Flow.Subscription flow;
-        
+        final Flow.Subscription flow;
+
         public ReactiveToFlowSubscription(Flow.Subscription flow) {
             this.flow = flow;
         }
@@ -178,25 +192,25 @@ public void request(long n) {
         public void cancel() {
             flow.cancel();
         }
-        
-        
+
+
     }
-    
+
     /**
      * Wraps a Reactive Streams Subscriber and forwards methods of the Flow Subscriber to it.
      * @param <T> the element type
      */
-    static final class FlowToReactiveSubscriber<T> 
+    static final class FlowToReactiveSubscriber<T>
             implements Flow.Subscriber<T> {
-        private final org.reactivestreams.Subscriber<? super T> reactiveStreams;
-        
+        final org.reactivestreams.Subscriber<? super T> reactiveStreams;
+
         public FlowToReactiveSubscriber(org.reactivestreams.Subscriber<? super T> reactive) {
             this.reactiveStreams = reactive;
         }
 
         @Override
         public void onSubscribe(Flow.Subscription subscription) {
-            reactiveStreams.onSubscribe(new ReactiveToFlowSubscription(subscription));
+            reactiveStreams.onSubscribe((subscription == null) ? null : new ReactiveToFlowSubscription(subscription));
         }
 
         @Override
@@ -213,24 +227,24 @@ public void onError(Throwable throwable) {
         public void onComplete() {
             reactiveStreams.onComplete();
         }
-        
+
     }
 
     /**
      * Wraps a Reactive Streams Subscriber and forwards methods of the Flow Subscriber to it.
      * @param <T> the element type
      */
-    static final class ReactiveToFlowSubscriber<T> 
+    static final class ReactiveToFlowSubscriber<T>
             implements org.reactivestreams.Subscriber<T> {
-        private final Flow.Subscriber<? super T> flow;
-        
+        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));
+            flow.onSubscribe((subscription == null) ? null : new FlowToReactiveSubscription(subscription));
         }
 
         @Override
@@ -247,9 +261,9 @@ public void onError(Throwable throwable) {
         public void onComplete() {
             flow.onComplete();
         }
-        
+
     }
-    
+
     /**
      * Wraps a Flow Processor and forwards methods of the Reactive Streams Processor to it.
      * @param <T> the input type
@@ -258,14 +272,14 @@ public void onComplete() {
     static final class ReactiveToFlowProcessor<T, U>
             implements org.reactivestreams.Processor<T, U> {
         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));
+        public void onSubscribe(org.reactivestreams.Subscription subscription) {
+            flow.onSubscribe((subscription == null) ? null : new FlowToReactiveSubscription(subscription));
         }
 
         @Override
@@ -285,14 +299,10 @@ public void onComplete() {
 
         @Override
         public void subscribe(org.reactivestreams.Subscriber<? super U> s) {
-            if (s == null) {
-                flow.subscribe(null);
-                return;
-            }
-            flow.subscribe(new FlowToReactiveSubscriber<U>(s));
+            flow.subscribe((s == null) ? null : new FlowToReactiveSubscriber<U>(s));
         }
     }
-    
+
     /**
      * Wraps a Reactive Streams Processor and forwards methods of the Flow Processor to it.
      * @param <T> the input type
@@ -301,14 +311,14 @@ public void subscribe(org.reactivestreams.Subscriber<? super U> s) {
     static final class FlowToReactiveProcessor<T, U>
             implements Flow.Processor<T, U> {
         final org.reactivestreams.Processor<? super T, ? extends U> reactiveStreams;
-        
+
         public FlowToReactiveProcessor(org.reactivestreams.Processor<? super T, ? extends U> reactive) {
             this.reactiveStreams = reactive;
         }
 
         @Override
-        public void onSubscribe(Flow.Subscription s) {
-            reactiveStreams.onSubscribe(new ReactiveToFlowSubscription(s));
+        public void onSubscribe(Flow.Subscription subscription) {
+            reactiveStreams.onSubscribe((subscription == null) ? null : new ReactiveToFlowSubscription(subscription));
         }
 
         @Override
@@ -328,11 +338,7 @@ public void onComplete() {
 
         @Override
         public void subscribe(Flow.Subscriber<? super U> s) {
-            if (s == null) {
-                reactiveStreams.subscribe(null);
-                return;
-            }
-            reactiveStreams.subscribe(new ReactiveToFlowSubscriber<U>(s));
+            reactiveStreams.subscribe((s == null) ? null : new ReactiveToFlowSubscriber<U>(s));
         }
     }
 
@@ -350,11 +356,7 @@ public ReactivePublisherFromFlow(Flow.Publisher<? extends T> flowPublisher) {
 
         @Override
         public void subscribe(org.reactivestreams.Subscriber<? super T> reactive) {
-            if (reactive == null) {
-                flow.subscribe(null);
-                return;
-            }
-            flow.subscribe(new FlowToReactiveSubscriber<T>(reactive));
+            flow.subscribe((reactive == null) ? null : new FlowToReactiveSubscriber<T>(reactive));
         }
     }
 
@@ -372,12 +374,8 @@ public FlowPublisherFromReactive(org.reactivestreams.Publisher<? extends T> reac
 
         @Override
         public void subscribe(Flow.Subscriber<? super T> flow) {
-            if (flow == null) {
-                reactiveStreams.subscribe(null);
-                return;
-            }
-            reactiveStreams.subscribe(new ReactiveToFlowSubscriber<T>(flow));
+            reactiveStreams.subscribe((flow == null) ? null : new ReactiveToFlowSubscriber<T>(flow));
         }
     }
 
-}
+}