Fixing wrapping and unwrapping of the wrappers themselves.

Author: viktorklang

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

@@ -34,11 +34,11 @@ public static <T> org.reactivestreams.Publisher<T> toReactiveStreams(
         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);
     }
@@ -56,12 +56,12 @@ public static <T> Flow.Publisher<T> toFlow(
         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);
     }
 
@@ -79,12 +79,12 @@ public static <T, U> org.reactivestreams.Processor<T, U> toReactiveStreams(
         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);
     }
 
@@ -102,12 +102,12 @@ public static <T, U> Flow.Processor<T, U> toFlow(
         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> toFlow(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,18 +137,25 @@ public static <T> Flow.Subscriber<T> toFlow(org.reactivestreams.Subscriber<T> re
      * @param flowSubscriber the Flow Subscriber instance to convert
      * @return the equivalent Reactive Streams Subscriber
      */
+    @SuppressWarnings("unchecked")
     public static <T> org.reactivestreams.Subscriber<T> toReactiveStreams(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;
@@ -163,7 +177,7 @@ public void 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;
@@ -188,16 +202,15 @@ public void cancel() {
      */
     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) {
-            if (reactive == null) throw null;
             this.reactiveStreams = reactive;
         }
 
         @Override
         public void onSubscribe(Flow.Subscription subscription) {
-            reactiveStreams.onSubscribe(new ReactiveToFlowSubscription(subscription));
+            reactiveStreams.onSubscribe((subscription == null) ? null : new ReactiveToFlowSubscription(subscription));
         }
 
         @Override
@@ -223,15 +236,15 @@ public void onComplete() {
      */
     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
@@ -261,13 +274,12 @@ static final class ReactiveToFlowProcessor<T, U>
         final Flow.Processor<? super T, ? extends U> flow;
 
         public ReactiveToFlowProcessor(Flow.Processor<? super T, ? extends U> flow) {
-            if (flow == null) throw null;
             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
@@ -287,11 +299,7 @@ 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));
         }
     }
 
@@ -309,8 +317,8 @@ public FlowToReactiveProcessor(org.reactivestreams.Processor<? super T, ? extend
         }
 
         @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
@@ -330,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));
         }
     }
 
@@ -352,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));
         }
     }
 
@@ -374,11 +374,7 @@ 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));
         }
     }
 

tck/src/test/java/org/reactivestreams/tck/BrokenExampleBlackboxTest.java

@@ -48,33 +48,33 @@ public Integer createElement(int element) {
   public Subscriber<Integer> createSubscriber() {
     return new MySubscriber<Integer>();
   }
-
-  public class MySubscriber<T> implements Subscriber<T> {
-    private Subscription subscription;
-
-    @Override
+  
+  public class MySubscriber<T> implements Subscriber<T> {  
+    private Subscription subscription;  
+    
+    @Override  
     public void onSubscribe(Subscription subscription) {
       if (this.subscription == null) {
         this.subscription = subscription;
-        subscription.request(1); //a value of Long.MAX_VALUE may be considered as effectively unbounded
+        subscription.request(1); //a value of  Long.MAX_VALUE may be considered as effectively unbounded
       } else subscription.cancel();
-    }
-
-    @Override
+    }  
+    
+    @Override  
     public void onNext(T item) {
       if (item == null) throw null;
-      System.out.println("Got : " + item);
-      subscription.request(1); //a value of Long.MAX_VALUE may be considered as effectively unbounded
-    }
-
-    @Override
-    public void onError(Throwable t) {
-      t.printStackTrace();
-    }
-
-    @Override
-    public void onComplete() {
-      System.out.println("Done");
-    }
-  }
+      System.out.println("Got : " + item);  
+      subscription.request(1); //a value of  Long.MAX_VALUE may be considered as effectively unbounded
+    }  
+    
+    @Override  
+    public void onError(Throwable t) {  
+      t.printStackTrace();  
+    }  
+    
+    @Override  
+    public void onComplete() {  
+      System.out.println("Done");  
+    }  
+  }  
 }