more WIP

Author: ktoso

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

@@ -64,7 +64,7 @@ public static <T> Flow.Publisher<T> toFlow(
         }
         return new FlowPublisherFromReactive<T>(reactiveStreamsPublisher);
     }
-    
+
     /**
      * Converts a Flow Processor into a Reactive Streams Processor.
      * @param <T> the input value type
@@ -117,7 +117,7 @@ public static <T, U> Flow.Processor<T, U> toFlow(
      * @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) {
+    public static <T> Flow.Subscriber<T> toFlow(org.reactivestreams.Subscriber<T> reactiveStreamsSubscriber) {
         if (reactiveStreamsSubscriber == null) {
             throw new NullPointerException("reactiveStreamsSubscriber");
         }
@@ -130,7 +130,7 @@ 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) {
+    public static <T> org.reactivestreams.Subscriber<T> toReactiveStreams(Flow.Subscriber<T> flowSubscriber) {
         if (flowSubscriber == null) {
             throw new NullPointerException("flowSubscriber");
         }
@@ -142,7 +142,7 @@ public static <T> org.reactivestreams.Subscriber<T> toReactiveStreamsSubscriber(
      */
     static final class FlowToReactiveSubscription implements Flow.Subscription {
         private final org.reactivestreams.Subscription reactiveStreams;
-        
+
         public FlowToReactiveSubscription(org.reactivestreams.Subscription reactive) {
             this.reactiveStreams = reactive;
         }
@@ -156,15 +156,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;
-        
+
         public ReactiveToFlowSubscription(Flow.Subscription flow) {
             this.flow = flow;
         }
@@ -178,18 +178,18 @@ 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;
-        
+
         public FlowToReactiveSubscriber(org.reactivestreams.Subscriber<? super T> reactive) {
             this.reactiveStreams = reactive;
         }
@@ -213,17 +213,17 @@ 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;
-        
+
         public ReactiveToFlowSubscriber(Flow.Subscriber<? super T> flow) {
             this.flow = flow;
         }
@@ -247,9 +247,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,7 +258,7 @@ 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;
         }
@@ -292,7 +292,7 @@ public void subscribe(org.reactivestreams.Subscriber<? super U> s) {
             flow.subscribe(new FlowToReactiveSubscriber<U>(s));
         }
     }
-    
+
     /**
      * Wraps a Reactive Streams Processor and forwards methods of the Flow Processor to it.
      * @param <T> the input type
@@ -301,7 +301,7 @@ 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;
         }
@@ -380,4 +380,4 @@ public void subscribe(Flow.Subscriber<? super T> flow) {
         }
     }
 
-}
+}

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

@@ -116,7 +116,7 @@ public void execute(Runnable command) {
     public void reactiveStreamsToFlowSubscriber() {
         TestEitherConsumer<Integer> tc = new TestEitherConsumer<Integer>();
 
-        Flow.Subscriber<Integer> fs = ReactiveStreamsFlowBridge.toFlowSubscriber(tc);
+        Flow.Subscriber<Integer> fs = ReactiveStreamsFlowBridge.toFlow(tc);
 
         final Object[] state = { null, null };
 
@@ -148,7 +148,7 @@ public void cancel() {
     public void flowToReactiveStreamsSubscriber() {
         TestEitherConsumer<Integer> tc = new TestEitherConsumer<Integer>();
 
-        org.reactivestreams.Subscriber<Integer> fs = ReactiveStreamsFlowBridge.toReactiveStreamsSubscriber(tc);
+        org.reactivestreams.Subscriber<Integer> fs = ReactiveStreamsFlowBridge.toReactiveStreams(tc);
 
         final Object[] state = { null, null };
 

tck-flow/src/main/java/org/reactivestreams/tck/flow/FlowPublisherVerification.java

@@ -54,7 +54,7 @@ final public Publisher<T> createFailedPublisher() {
   /**
    * By implementing this method, additional TCK tests concerning a "failed" publishers will be run.
    *
-   * The expected behaviour of the {@link Publisher} returned by this method is hand out a subscription,
+   * The expected behaviour of the {@link Flow.Publisher} returned by this method is hand out a subscription,
    * followed by signalling {@code onError} on it, as specified by Rule 1.9.
    *
    * If you ignore these additional tests, return {@code null} from this method.

tck-flow/src/main/java/org/reactivestreams/tck/flow/FlowSubscriberBlackboxVerification.java

@@ -37,28 +37,28 @@ protected FlowSubscriberBlackboxVerification(TestEnvironment env) {
     super(env);
   }
 
+  @Override
+  public final void triggerRequest(Subscriber<? super T> subscriber) {
+    triggerFlowRequest(ReactiveStreamsFlowBridge.toFlow(subscriber));
+  }
   /**
    * Override this method if the {@link java.util.concurrent.Flow.Subscriber} implementation you are verifying
    * needs an external signal before it signals demand to its Publisher.
    *
    * By default this method does nothing.
    */
-  abstract public void triggerFlowRequest(Flow.Subscriber<? super T> subscriber);
+  public void triggerFlowRequest(Flow.Subscriber<? super T> subscriber) {
+    // this method is intentionally left blank
+  }
 
   @Override
-  public final void triggerRequest(Subscriber<? super T> subscriber) {
-    triggerFlowRequest(ReactiveStreamsFlowBridge.toFlow(subscriber));
+  public final Subscriber<T> createSubscriber() {
+    return ReactiveStreamsFlowBridge.<T>toReactiveStreams(createFlowSubscriber());
   }
-
   /**
    * This is the main method you must implement in your test incarnation.
-   * It must create a new {@link java.util.concurrent.Flow.Subscriber} instance to be subjected to the testing logic.
+   * It must create a new {@link Flow.Subscriber} instance to be subjected to the testing logic.
    */
   abstract public Flow.Subscriber<T> createFlowSubscriber();
 
-  @Override
-  public final Subscriber<T> createSubscriber() {
-    return ReactiveStreamsFlowBridge.<T>toReactiveStreams(createFlowSubscriber());
-  }
-
 }

tck-flow/src/main/java/org/reactivestreams/tck/flow/FlowSubscriberWhiteboxVerification.java

@@ -11,16 +11,20 @@
 
 package org.reactivestreams.tck.flow;
 
+import org.reactivestreams.ReactiveStreamsFlowBridge;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.tck.SubscriberWhiteboxVerification;
 import org.reactivestreams.tck.TestEnvironment;
 import org.reactivestreams.tck.flow.support.SubscriberWhiteboxVerificationRules;
 
+import java.util.concurrent.Flow;
+
 /**
- * Provides tests for verifying {@link org.reactivestreams.Subscriber} and {@link org.reactivestreams.Subscription} specification rules.
+ * Provides whitebox style tests for verifying {@link java.util.concurrent.Flow.Subscriber}
+ * and {@link java.util.concurrent.Flow.Subscription} specification rules.
  *
- * @see org.reactivestreams.Subscriber
- * @see org.reactivestreams.Subscription
+ * @see java.util.concurrent.Flow.Subscriber
+ * @see java.util.concurrent.Flow.Subscription
  */
 public abstract class FlowSubscriberWhiteboxVerification<T> extends SubscriberWhiteboxVerification<T>
   implements SubscriberWhiteboxVerificationRules {
@@ -31,8 +35,14 @@ protected FlowSubscriberWhiteboxVerification(TestEnvironment env) {
 
   @Override
   final public Subscriber<T> createSubscriber(WhiteboxSubscriberProbe<T> probe) {
-    return createFlowSubscriber(probe);
+    return ReactiveStreamsFlowBridge.toReactiveStreams(createFlowSubscriber(probe));
   }
-
-  protected abstract Subscriber<T> createFlowSubscriber(WhiteboxSubscriberProbe<T> probe);
+  /**
+   * This is the main method you must implement in your test incarnation.
+   * It must create a new {@link org.reactivestreams.Subscriber} instance to be subjected to the testing logic.
+   *
+   * In order to be meaningfully testable your Subscriber must inform the given
+   * `WhiteboxSubscriberProbe` of the respective events having been received.
+   */
+  protected abstract Flow.Subscriber<T> createFlowSubscriber(WhiteboxSubscriberProbe<T> probe);
 }