Remove CHM check to allow reusing the Subscriber instance (#159)

ReactiveX · Nov 15, 2016 · 1c6a64d · 1c6a64d
1 parent a3c326c
commit 1c6a64d
Show file tree

Hide file tree

Showing 2 changed files with 60 additions and 75 deletions.
diff --git a/rxjava-reactive-streams/build.gradle b/rxjava-reactive-streams/build.gradle
@@ -4,7 +4,7 @@ apply plugin: 'java'
 apply plugin: 'jacoco'
 
 dependencies {
-    compile 'io.reactivex:rxjava:1.2.1'
+    compile 'io.reactivex:rxjava:1.2.2'
     compile 'org.reactivestreams:reactive-streams:1.0.0'
     testCompile 'org.reactivestreams:reactive-streams-tck:1.0.0'
     testCompile group: 'org.testng', name: 'testng', version: '6.9.10'

diff --git a/rxjava-reactive-streams/src/main/java/rx/internal/reactivestreams/PublisherAdapter.java b/rxjava-reactive-streams/src/main/java/rx/internal/reactivestreams/PublisherAdapter.java
@@ -15,14 +15,9 @@
  */
 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 java.util.concurrent.atomic.*;
 
-import org.reactivestreams.Publisher;
-import org.reactivestreams.Subscriber;
-import org.reactivestreams.Subscription;
+import org.reactivestreams.*;
 
 import rx.Observable;
 import rx.internal.operators.BackpressureUtils;
@@ -31,94 +26,84 @@ public class PublisherAdapter<T> implements Publisher<T> {
 
     private final Observable<T> observable;
 
-    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.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);
-                    }
+        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);
                 }
-
-                @Override
-                public void onStart() {
-                    final AtomicBoolean requested = new AtomicBoolean();
-                    s.onSubscribe(new Subscription() {
-                        @Override
-                        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);
-                            doRequest(n);
-                        }
-
-                        @Override
-                        public void cancel() {
+            }
+
+            @Override
+            public void onStart() {
+                final AtomicBoolean requested = new AtomicBoolean();
+                s.onSubscribe(new Subscription() {
+                    @Override
+                    public void request(long n) {
+                        if (n < 1) {
                             unsubscribe();
-                            fireDone();
+                            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;
                         }
-                    });
 
-                    if (!requested.get()) {
-                        request(0);
+                        requested.set(true);
+                        doRequest(n);
                     }
-                }
 
-                private boolean fireDone() {
-                    boolean first = done.compareAndSet(false, true);
-                    if (first) {
-                        subscribers.remove(s);
+                    @Override
+                    public void cancel() {
+                        unsubscribe();
+                        fireDone();
                     }
-                    return first;
-                }
+                });
 
-                @Override
-                public void onCompleted() {
-                    if (fireDone()) {
-                        s.onComplete();
-                    }
+                if (!requested.get()) {
+                    request(0);
                 }
+            }
 
-                @Override
-                public void onError(Throwable e) {
-                    if (fireDone()) {
-                        s.onError(e);
-                    }
+            boolean fireDone() {
+                return done.compareAndSet(false, true);
+            }
+
+            @Override
+            public void onCompleted() {
+                if (fireDone()) {
+                    s.onComplete();
                 }
+            }
 
-                @Override
-                public void onNext(T t) {
-                    if (!done.get()) {
-                        if (childRequested.get() > 0) {
-                            s.onNext(t);
-                            childRequested.decrementAndGet();
-                        } else {
-                            try {
-                                onError(new IllegalStateException("1.1 source doesn't respect backpressure"));
-                            } finally {
-                                unsubscribe();
-                            }
+            @Override
+            public void onError(Throwable e) {
+                if (fireDone()) {
+                    s.onError(e);
+                }
+            }
+
+            @Override
+            public void onNext(T t) {
+                if (!done.get()) {
+                    if (childRequested.get() > 0) {
+                        s.onNext(t);
+                        childRequested.decrementAndGet();
+                    } else {
+                        try {
+                            onError(new IllegalStateException("1.1 source doesn't respect backpressure"));
+                        } finally {
+                            unsubscribe();
                         }
                     }
                 }
-            });
-        } else {
-            s.onError(new IllegalArgumentException("1.10 Subscriber cannot subscribe more than once"));
-        }
+            }
+        });
     }
 
 }