Update to Reactive Streams 1.0.0.M1

Author: ldaley

rxjava-reactive-streams/build.gradle

@@ -2,8 +2,8 @@ description = "Adapter between RxJava and ReactiveStreams"
 
 dependencies {
     compile 'io.reactivex:rxjava:1.0.+'
-    compile 'org.reactivestreams:reactive-streams:0.4.0'
-    testCompile 'org.reactivestreams:reactive-streams-tck:0.4.0'
+    compile 'org.reactivestreams:reactive-streams:1.0.0.M1'
+    testCompile 'org.reactivestreams:reactive-streams-tck:1.0.0.M1'
 }
 
 test {

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

@@ -23,6 +23,7 @@
 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> {
 
@@ -39,9 +40,17 @@ 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) {
-                    request(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);
+                        }
+                    }
                 }
 
                 @Override
@@ -70,27 +79,34 @@ public void cancel() {
                     }
                 }
 
-                private void fireDone() {
-                    if (done.compareAndSet(false, true)) {
+                private boolean fireDone() {
+                    boolean first = done.compareAndSet(false, true);
+                    if (first) {
                         subscribers.remove(s);
                     }
+                    return first;
                 }
 
                 @Override
                 public void onCompleted() {
-                    s.onComplete();
+                    if (fireDone()) {
+                        s.onComplete();
+                    }
                 }
 
                 @Override
                 public void onError(Throwable e) {
-                    s.onError(e);
-                    fireDone();
+                    if (fireDone()) {
+                        s.onError(e);
+                    }
                 }
 
                 @Override
                 public void onNext(T t) {
-                    s.onNext(t);
-                    fireDone();
+                    if (!done.get()) {
+                        pending.decrementAndGet();
+                        s.onNext(t);
+                    }
                 }
             });
         } else {

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

@@ -23,6 +23,7 @@
 import rx.Observable;
 import rx.RxReactiveStreams;
 import rx.reactivestreams.test.CountdownIterable;
+import rx.schedulers.Schedulers;
 
 @Test
 public class TckPublisherTest extends PublisherVerification<Long> {
@@ -34,14 +35,12 @@ public TckPublisherTest() {
         super(new TestEnvironment(DEFAULT_TIMEOUT_MILLIS), PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS);
     }
 
-    @Override
-    public long maxElementsFromPublisher() {
-        return Long.MAX_VALUE;
-    }
-
     @Override
     public Publisher<Long> createPublisher(long elements) {
-        return RxReactiveStreams.toPublisher(Observable.from(new CountdownIterable(elements)));
+        return RxReactiveStreams.toPublisher(
+            Observable.from(new CountdownIterable(elements))
+                .observeOn(Schedulers.computation())
+        );
     }
 
     @Override

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

@@ -66,5 +66,4 @@ public Publisher<Long> createHelperPublisher(long elements) {
         return RxReactiveStreams.toPublisher(Observable.from(new CountdownIterable(elements)));
     }
 
-
 }

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

@@ -113,17 +113,4 @@ public Publisher<Long> createHelperPublisher(long elements) {
         return RxReactiveStreams.toPublisher(Observable.from(new CountdownIterable(elements)));
     }
 
-    public void spec309_callingRequestWithNegativeNumberMustThrow() throws Throwable {
-        notVerified(); // nonsense test, subscriber doesn't create any Subscription implementations
-    }
-
-    @Override
-    public void spec309_callingRequestZeroMustThrow() throws Throwable {
-        notVerified(); // nonsense test, subscriber doesn't create any Subscription implementations
-    }
-
-    @Override
-    public void spec317_mustSignalOnErrorWhenPendingAboveLongMaxValue() throws Throwable {
-        notVerified(); // nonsense test, the publisher should implement this
-    }
 }