Merge pull request #108 from rkuhn/wip-contravariant-subscriber

allow contravariant use of Subscriber

Author: smaldini

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/examples/java/org/reactivestreams/example/multicast/StockPricePublisher.java

@@ -17,7 +17,7 @@
 public class StockPricePublisher implements Publisher<Stock> {
 
     @Override
-    public void subscribe(final Subscriber<Stock> s) {
+    public void subscribe(final Subscriber<? super Stock> s) {
         s.onSubscribe(new Subscription() {
 
             AtomicLong capacity = new AtomicLong();
@@ -46,10 +46,10 @@ public void startConsuming() {
     }
 
     private static final class EventHandler implements Handler {
-        private final Subscriber<Stock> s;
+        private final Subscriber<? super Stock> s;
         private final AtomicLong capacity;
 
-        private EventHandler(Subscriber<Stock> s, AtomicLong capacity) {
+        private EventHandler(Subscriber<? super Stock> s, AtomicLong capacity) {
             this.s = s;
             this.capacity = capacity;
         }

api/src/examples/java/org/reactivestreams/example/unicast/InfiniteIncrementNumberPublisher.java

@@ -1,6 +1,7 @@
 package org.reactivestreams.example.unicast;
 
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.reactivestreams.Subscription;
 import org.reactivestreams.Subscriber;
@@ -9,13 +10,13 @@
 class InfiniteIncrementNumberPublisher implements Publisher<Integer> {
 
     @Override
-    public void subscribe(final Subscriber<Integer> s) {
+    public void subscribe(final Subscriber<? super Integer> s) {
 
         final AtomicInteger i = new AtomicInteger();
 
         Subscription subscription = new Subscription() {
 
-            AtomicInteger capacity = new AtomicInteger();
+            AtomicLong capacity = new AtomicLong();
 
             @Override
             public void request(long n) {

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);
 }