adjusted for upcoming reactive-streams#108 change

Author: ktoso

README.md

@@ -69,7 +69,7 @@ onError | (onSubscribe onNext* (onError | onComplete)?)
 
 ```java
 public interface Publisher<T> {
-    public void subscribe(Subscriber<T> s);
+    public void subscribe(Subscriber<? super T> s);
 }
 ````
 

api/src/main/java/org/reactivestreams/Publisher.java

@@ -16,5 +16,5 @@ public interface Publisher<T> {
      * 
      * @param s the {@link Subscriber} that will consume signals from this {@link Publisher}
      */
-    public void subscribe(Subscriber<T> s);
+    public void subscribe(Subscriber<? super T> s);
 }

tck/src/main/java/org/reactivestreams/tck/SubscriberBlackboxVerification.java

@@ -244,7 +244,7 @@ public void spec209_blackbox_mustBePreparedToReceiveAnOnCompleteSignalWithoutPre
       public void run(BlackboxTestStage stage) throws Throwable {
         final Publisher<T> pub = new Publisher<T>() {
           @Override
-          public void subscribe(Subscriber<T> s) {
+          public void subscribe(Subscriber<? super T> s) {
             s.onComplete();
           }
         };
@@ -399,7 +399,7 @@ public BlackboxTestStage(TestEnvironment env, boolean runDefaultInit) throws Int
       }
     }
 
-    public Subscriber<T> sub() {
+    public Subscriber<? super T> sub() {
       return subscriber.value();
     }
 

tck/src/main/java/org/reactivestreams/tck/SubscriberWhiteboxVerification.java

@@ -529,7 +529,7 @@ public WhiteboxTestStage(TestEnvironment env, boolean runDefaultInit) throws Int
       }
     }
 
-    public Subscriber<T> sub() {
+    public Subscriber<? super T> sub() {
       return subscriber.value();
     }
 
@@ -568,8 +568,8 @@ public void verifyNoAsyncErrors() {
    */
   public static class BlackboxSubscriberProxy<T> extends BlackboxProbe<T> implements Subscriber<T> {
 
-    public BlackboxSubscriberProxy(TestEnvironment env, Subscriber<T> subscriber) {
-      super(env, Promise.completed(env, subscriber));
+    public BlackboxSubscriberProxy(TestEnvironment env, Subscriber<? super T> subscriber) {
+      super(env, new Promise<Subscriber<? super T>>(env, subscriber));
     }
 
     @Override
@@ -598,12 +598,12 @@ public void onComplete() {
 
   public static class BlackboxProbe<T> implements SubscriberProbe<T> {
     protected final TestEnvironment env;
-    protected final Promise<Subscriber<T>> subscriber;
+    protected final Promise<Subscriber<? super T>> subscriber;
 
     protected final Receptacle<T> elements;
     protected final Promise<Throwable> error;
 
-    public BlackboxProbe(TestEnvironment env, Promise<Subscriber<T>> subscriber) {
+    public BlackboxProbe(TestEnvironment env, Promise<Subscriber<? super T>> subscriber) {
       this.env = env;
       this.subscriber = subscriber;
       elements = new Receptacle<T>(env);
@@ -640,7 +640,7 @@ public void expectNext(T expected, long timeoutMillis) throws InterruptedExcepti
       }
     }
 
-    public Subscriber<T> sub() {
+    public Subscriber<? super T> sub() {
       return subscriber.value();
     }
 
@@ -706,7 +706,7 @@ public void expectNone(long withinMillis) throws InterruptedException {
   public static class WhiteboxSubscriberProbe<T> extends BlackboxProbe<T> implements SubscriberPuppeteer {
     protected Promise<SubscriberPuppet> puppet;
 
-    public WhiteboxSubscriberProbe(TestEnvironment env, Promise<Subscriber<T>> subscriber) {
+    public WhiteboxSubscriberProbe(TestEnvironment env, Promise<Subscriber<? super T>> subscriber) {
       super(env, subscriber);
       puppet = new Promise<SubscriberPuppet>(env);
     }

tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java

@@ -441,7 +441,7 @@ public static class ManualPublisher<T> implements Publisher<T> {
     protected final TestEnvironment env;
 
     protected long pendingDemand = 0L;
-    protected Promise<Subscriber<T>> subscriber;
+    protected Promise<Subscriber<? super T>> subscriber;
 
     protected final Receptacle<Long> requests;
 
@@ -451,11 +451,11 @@ public ManualPublisher(TestEnvironment env) {
       this.env = env;
       requests = new Receptacle<Long>(env);
       cancelled = new Latch(env);
-      subscriber = new Promise<Subscriber<T>>(this.env);
+      subscriber = new Promise<Subscriber<? super T>>(this.env);
     }
 
     @Override
-    public void subscribe(Subscriber<T> s) {
+    public void subscribe(Subscriber<? super T> s) {
       if (!subscriber.isCompleted()) {
         subscriber.completeImmediatly(s);
 
@@ -596,14 +596,13 @@ public void expectClose(long timeoutMillis, String notClosedErrorMsg) throws Int
   public static class Promise<T> {
     private final TestEnvironment env;
 
-    public static <T> Promise<T> completed(TestEnvironment env, T value) {
-      Promise<T> promise = new Promise<T>(env);
-      promise.completeImmediatly(value);
-      return promise;
+    public Promise(TestEnvironment env) {
+      this.env = env;
     }
 
-    public Promise(TestEnvironment env) {
+    public Promise(TestEnvironment env, T value) {
       this.env = env;
+      completeImmediatly(value);
     }
 
     private ArrayBlockingQueue<T> abq = new ArrayBlockingQueue<T>(1);