2.x: Cleanup, coverage and related component fixes (#5889)

Author: akarnokd

src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMapEager.java

@@ -132,10 +132,6 @@ public void onNext(T t) {
 
             subscribers.offer(inner);
 
-            if (cancelled) {
-                return;
-            }
-
             p.subscribe(inner);
 
             if (cancelled) {

src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishMulticast.java

@@ -153,6 +153,8 @@ static final class MulticastProcessor<T> extends Flowable<T> implements Flowable
 
         int consumed;
 
+        long emitted;
+
         @SuppressWarnings("unchecked")
         MulticastProcessor(int prefetch, boolean delayError) {
             this.prefetch = prefetch;
@@ -261,10 +263,10 @@ boolean add(MulticastSubscription<T> s) {
         void remove(MulticastSubscription<T> s) {
             for (;;) {
                 MulticastSubscription<T>[] current = subscribers.get();
-                if (current == TERMINATED || current == EMPTY) {
+                int n = current.length;
+                if (n == 0) {
                     return;
                 }
-                int n = current.length;
                 int j = -1;
 
                 for (int i = 0; i < n; i++) {
@@ -323,6 +325,7 @@ void drain() {
             int upstreamConsumed = consumed;
             int localLimit = limit;
             boolean canRequest = sourceMode != QueueSubscription.SYNC;
+            long e = emitted;
 
             for (;;) {
                 MulticastSubscription<T>[] array = subscribers.get();
@@ -338,10 +341,15 @@ void drain() {
                             if (r > u) {
                                 r = u;
                             }
+                        } else {
+                            n--;
                         }
                     }
 
-                    long e = 0L;
+                    if (n == 0) {
+                        r = e;
+                    }
+
                     while (e != r) {
                         if (isDisposed()) {
                             q.clear();
@@ -425,12 +433,9 @@ void drain() {
                             return;
                         }
                     }
-
-                    for (MulticastSubscription<T> ms : array) {
-                        BackpressureHelper.produced(ms, e);
-                    }
                 }
 
+                emitted = e;
                 consumed = upstreamConsumed;
                 missed = wip.addAndGet(-missed);
                 if (missed == 0) {
@@ -465,7 +470,6 @@ static final class MulticastSubscription<T>
     extends AtomicLong
     implements Subscription {
 
-
         private static final long serialVersionUID = 8664815189257569791L;
 
         final Subscriber<? super T> actual;

src/main/java/io/reactivex/processors/UnicastProcessor.java

@@ -180,8 +180,8 @@ public static <T> UnicastProcessor<T> create(int capacityHint, Runnable onCancel
     }
 
     void doTerminate() {
-        Runnable r = onTerminate.get();
-        if (r != null && onTerminate.compareAndSet(r, null)) {
+        Runnable r = onTerminate.getAndSet(null);
+        if (r != null) {
             r.run();
         }
     }

src/main/java/io/reactivex/subjects/BehaviorSubject.java

@@ -442,26 +442,20 @@ void remove(BehaviorDisposable<T> rs) {
     @SuppressWarnings("unchecked")
     BehaviorDisposable<T>[] terminate(Object terminalValue) {
 
-        BehaviorDisposable<T>[] a = subscribers.get();
+        BehaviorDisposable<T>[] a = subscribers.getAndSet(TERMINATED);
         if (a != TERMINATED) {
-            a = subscribers.getAndSet(TERMINATED);
-            if (a != TERMINATED) {
-                // either this or atomics with lots of allocation
-                setCurrent(terminalValue);
-            }
+            // either this or atomics with lots of allocation
+            setCurrent(terminalValue);
         }
 
         return a;
     }
 
     void setCurrent(Object o) {
         writeLock.lock();
-        try {
-            index++;
-            value.lazySet(o);
-        } finally {
-            writeLock.unlock();
-        }
+        index++;
+        value.lazySet(o);
+        writeLock.unlock();
     }
 
     static final class BehaviorDisposable<T> implements Disposable, NonThrowingPredicate<Object> {