Upgrade to 1.0 RC4 of RS.

Closes #142.

Author: ldaley

rxjava-reactive-streams/build.gradle

@@ -4,8 +4,8 @@ apply plugin: 'java'
 
 dependencies {
     compile 'io.reactivex:rxjava:1.0.+'
-    compile 'org.reactivestreams:reactive-streams:1.0.0.RC1'
-    testCompile 'org.reactivestreams:reactive-streams-tck:1.0.0.RC1'
+    compile 'org.reactivestreams:reactive-streams:1.0.0.RC4'
+    testCompile 'org.reactivestreams:reactive-streams-tck:1.0.0.RC4'
 }
 
 test {

rxjava-reactive-streams/src/main/java/rx/internal/reactivestreams/PublisherAdapter.java

@@ -23,7 +23,6 @@
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
 
 public class PublisherAdapter<T> implements Publisher<T> {
 
@@ -40,16 +39,10 @@ public void subscribe(final Subscriber<? super T> s) {
         if (subscribers.add(s)) {
             observable.subscribe(new rx.Subscriber<T>() {
                 private final AtomicBoolean done = new AtomicBoolean();
-                private final AtomicLong pending = new AtomicLong(Long.MIN_VALUE);
 
                 private void doRequest(long n) {
                     if (!done.get()) {
-                        if (pending.addAndGet(n) >= 0) {
-                            unsubscribe();
-                            onError(new IllegalStateException("Violation of rule 3.17 - more than Long.MAX_VALUE elements requested"));
-                        } else {
-                            request(n);
-                        }
+                        request(n);
                     }
                 }
 
@@ -105,7 +98,6 @@ public void onError(Throwable e) {
                 @Override
                 public void onNext(T t) {
                     if (!done.get()) {
-                        pending.decrementAndGet();
                         s.onNext(t);
                     }
                 }

rxjava-reactive-streams/src/main/java/rx/internal/reactivestreams/SubscriberAdapter.java

@@ -34,6 +34,10 @@ public SubscriberAdapter(rx.Subscriber<? super T> rxSubscriber) {
 
     @Override
     public void onSubscribe(final Subscription rsSubscription) {
+        if (rsSubscription == null) {
+            throw new NullPointerException("onSubscribe(null)");
+        }
+
         if (started.compareAndSet(false, true)) {
             rxSubscriber.add(Subscriptions.create(new Action0() {
                 @Override
@@ -57,11 +61,17 @@ public void request(long n) {
 
     @Override
     public void onNext(T t) {
+        if (t == null) {
+            throw new NullPointerException("onNext(null)");
+        }
         rxSubscriber.onNext(t);
     }
 
     @Override
     public void onError(Throwable t) {
+        if (t == null) {
+            throw new NullPointerException("onError(null)");
+        }
         rxSubscriber.onError(t);
     }
 

rxjava-reactive-streams/src/test/java/rx/reactivestreams/TckPublisherTest.java

@@ -44,7 +44,7 @@ public Publisher<Long> createPublisher(long elements) {
     }
 
     @Override
-    public Publisher<Long> createErrorStatePublisher() {
+    public Publisher<Long> createFailedPublisher() {
         // Null because we always successfully subscribe.
         // If the observable is in error state, it will subscribe and then emit the error as the first item
         // This is not an “error state” publisher as defined by RS

rxjava-reactive-streams/src/test/java/rx/reactivestreams/TckSubscriberBlackboxTest.java

@@ -61,6 +61,11 @@ public void onNext(Long aLong) {
         });
     }
 
+    @Override
+    public Long createElement(int element) {
+        return Long.valueOf(Integer.toString(element));
+    }
+
     @Override
     public Publisher<Long> createHelperPublisher(long elements) {
         return RxReactiveStreams.toPublisher(Observable.from(new CountdownIterable(elements)));

rxjava-reactive-streams/src/test/java/rx/reactivestreams/TckSubscriberWhiteboxTest.java

@@ -108,6 +108,11 @@ public void onComplete() {
         };
     }
 
+    @Override
+    public Long createElement(int element) {
+        return Long.valueOf(Integer.toString(element));
+    }
+
     @Override
     public Publisher<Long> createHelperPublisher(long elements) {
         return RxReactiveStreams.toPublisher(Observable.from(new CountdownIterable(elements)));

rxjava-reactive-streams/src/test/java/rx/reactivestreams/test/RsSubscriber.java

@@ -45,11 +45,18 @@ public Wait(long count, CountDownLatch latch) {
 
     @Override
     public void onSubscribe(Subscription s) {
+        if (s == null) {
+            throw new NullPointerException("onSubscribe(null)");
+        }
+
         subscription = s;
     }
 
     @Override
     public void onNext(T t) {
+        if (t == null) {
+            throw new NullPointerException("onNext(null)");
+        }
         received.add(t);
         for (Wait wait : waits) {
             if (received.size() >= wait.count) {
@@ -60,6 +67,9 @@ public void onNext(T t) {
 
     @Override
     public void onError(Throwable t) {
+        if (t == null) {
+            throw new NullPointerException("onError(null)");
+        }
         error = t;
         unwaitAll();
     }