Fix: request and subscriber tracking, nonpositive request error

reporting.

Author: akarnokd

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

@@ -15,33 +15,37 @@
  */
 package rx.internal.reactivestreams;
 
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
-import rx.Observable;
 
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicBoolean;
+import rx.Observable;
+import rx.internal.operators.BackpressureUtils;
 
 public class PublisherAdapter<T> implements Publisher<T> {
 
     private final Observable<T> observable;
 
-    private final Set<Subscriber<?>> subscribers = new HashSet<Subscriber<?>>();
+    private final ConcurrentMap<Subscriber<?>, Object> subscribers = new ConcurrentHashMap<Subscriber<?>, Object>();
 
     public PublisherAdapter(final Observable<T> observable) {
         this.observable = observable.serialize();
     }
 
     @Override
     public void subscribe(final Subscriber<? super T> s) {
-        if (subscribers.add(s)) {
+        if (subscribers.putIfAbsent(s, s) == null) {
             observable.subscribe(new rx.Subscriber<T>() {
                 private final AtomicBoolean done = new AtomicBoolean();
-
+                private final AtomicLong childRequested = new AtomicLong();
                 private void doRequest(long n) {
                     if (!done.get()) {
+                        BackpressureUtils.getAndAddRequest(childRequested, n);
                         request(n);
                     }
                 }
@@ -55,6 +59,7 @@ public void request(long n) {
                             if (n < 1) {
                                 unsubscribe();
                                 onError(new IllegalArgumentException("3.9 While the Subscription is not cancelled, Subscription.request(long n) MUST throw a java.lang.IllegalArgumentException if the argument is <= 0."));
+                                return;
                             }
 
                             requested.set(true);
@@ -98,7 +103,16 @@ public void onError(Throwable e) {
                 @Override
                 public void onNext(T t) {
                     if (!done.get()) {
-                        s.onNext(t);
+                        if (childRequested.get() > 0) {
+                            s.onNext(t);
+                            childRequested.decrementAndGet();
+                        } else {
+                            try {
+                                onError(new IllegalStateException("1.1 source doesn't respect backpressure"));
+                            } finally {
+                                unsubscribe();
+                            }
+                        }
                     }
                 }
             });

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

@@ -15,22 +15,24 @@
  */
 package rx.reactivestreams;
 
+import static org.testng.Assert.*;
+import static org.testng.AssertJUnit.assertEquals;
+import static org.testng.AssertJUnit.assertNotNull;
+import static rx.RxReactiveStreams.toObservable;
+import static rx.RxReactiveStreams.toPublisher;
+
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+
 import org.reactivestreams.Publisher;
 import org.testng.annotations.Test;
+
 import rx.Observable;
 import rx.Subscriber;
 import rx.reactivestreams.test.IterablePublisher;
 import rx.reactivestreams.test.RsSubscriber;
 import rx.reactivestreams.test.RxSubscriber;
-
-import java.util.Arrays;
-import java.util.concurrent.TimeUnit;
-
-import static org.testng.Assert.*;
-import static org.testng.AssertJUnit.assertEquals;
-import static org.testng.AssertJUnit.assertNotNull;
-import static rx.RxReactiveStreams.toObservable;
-import static rx.RxReactiveStreams.toPublisher;
+import rx.subjects.PublishSubject;
 
 public class NonTckTest {
 
@@ -145,4 +147,27 @@ void subscribingToHotObservableWithBackpressureStrategy() throws InterruptedExce
         subscriber.subscription.cancel();
     }
 
+    @Test
+    public void subscribeToPublishSubjectEmitsRuleViolationException() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        RsSubscriber<Integer> rxs = subscribe(source);
+        
+        source.onNext(1);
+        
+        assertEquals(0, rxs.received.size());
+        assertNotNull(rxs.error);
+    }
+    @Test
+    public void subscribeToPublishSubjectEmitsRuleViolationExceptionAfterOneRequest() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        RsSubscriber<Integer> rxs = subscribe(source);
+        
+        rxs.subscription.request(1);
+        source.onNext(1);
+
+        source.onNext(2);
+
+        assertEquals(1, rxs.received.size());
+        assertNotNull(rxs.error);
+    }
 }