1.x: remove remaining field updaters (#3979)

This PR removes the remaining AtomicXFieldUpdaters from the library because reflection is somewhat problematic on Android.

This also deprecates the internal BackpressureUtils.getAndAddRequest to indicate requested field updater should not be used anymore.

Author: akarnokd

src/main/java/rx/internal/operators/BackpressureUtils.java

@@ -45,7 +45,9 @@ private BackpressureUtils() {
      * @param n
      *            the number of requests to add to the requested count
      * @return requested value just prior to successful addition
+     * @deprecated Android has issues with reflection-based atomics
      */
+    @Deprecated
     public static <T> long getAndAddRequest(AtomicLongFieldUpdater<T> requested, T object, long n) {
         // add n to field but check for overflow
         while (true) {

src/main/java/rx/internal/operators/CompletableOnSubscribeConcat.java

@@ -52,9 +52,7 @@ static final class CompletableConcatSubscriber
 
         volatile boolean done;
 
-        volatile int once;
-        static final AtomicIntegerFieldUpdater<CompletableConcatSubscriber> ONCE =
-                AtomicIntegerFieldUpdater.newUpdater(CompletableConcatSubscriber.class, "once");
+        final AtomicBoolean once;
 
         final ConcatInnerSubscriber inner;
 
@@ -67,6 +65,7 @@ public CompletableConcatSubscriber(CompletableSubscriber actual, int prefetch) {
             this.sr = new SerialSubscription();
             this.inner = new ConcatInnerSubscriber();
             this.wip = new AtomicInteger();
+            this.once = new AtomicBoolean();
             add(sr);
             request(prefetch);
         }
@@ -84,7 +83,7 @@ public void onNext(Completable t) {
 
         @Override
         public void onError(Throwable t) {
-            if (ONCE.compareAndSet(this, 0, 1)) {
+            if (once.compareAndSet(false, true)) {
                 actual.onError(t);
                 return;
             }
@@ -121,7 +120,7 @@ void next() {
             Completable c = queue.poll();
             if (c == null) {
                 if (d) {
-                    if (ONCE.compareAndSet(this, 0, 1)) {
+                    if (once.compareAndSet(false, true)) {
                         actual.onCompleted();
                     }
                     return;

src/main/java/rx/internal/operators/CompletableOnSubscribeMerge.java

@@ -55,14 +55,9 @@ static final class CompletableMergeSubscriber
 
         volatile boolean done;
 
-        volatile Queue<Throwable> errors;
-        @SuppressWarnings("rawtypes")
-        static final AtomicReferenceFieldUpdater<CompletableMergeSubscriber, Queue> ERRORS =
-                AtomicReferenceFieldUpdater.newUpdater(CompletableMergeSubscriber.class, Queue.class, "errors");
+        final AtomicReference<Queue<Throwable>> errors;
 
-        volatile int once;
-        static final AtomicIntegerFieldUpdater<CompletableMergeSubscriber> ONCE =
-                AtomicIntegerFieldUpdater.newUpdater(CompletableMergeSubscriber.class, "once");
+        final AtomicBoolean once;
 
         final AtomicInteger wip;
 
@@ -72,6 +67,8 @@ public CompletableMergeSubscriber(CompletableSubscriber actual, int maxConcurren
             this.delayErrors = delayErrors;
             this.set = new CompositeSubscription();
             this.wip = new AtomicInteger(1);
+            this.once = new AtomicBoolean();
+            this.errors = new AtomicReference<Queue<Throwable>>();
             if (maxConcurrency == Integer.MAX_VALUE) {
                 request(Long.MAX_VALUE);
             } else {
@@ -80,17 +77,17 @@ public CompletableMergeSubscriber(CompletableSubscriber actual, int maxConcurren
         }
 
         Queue<Throwable> getOrCreateErrors() {
-            Queue<Throwable> q = errors;
+            Queue<Throwable> q = errors.get();
 
             if (q != null) {
                 return q;
             }
 
             q = new ConcurrentLinkedQueue<Throwable>();
-            if (ERRORS.compareAndSet(this, null, q)) {
+            if (errors.compareAndSet(null, q)) {
                 return q;
             }
-            return errors;
+            return errors.get();
         }
 
         @Override
@@ -167,23 +164,23 @@ public void onCompleted() {
 
         void terminate() {
             if (wip.decrementAndGet() == 0) {
-                Queue<Throwable> q = errors;
+                Queue<Throwable> q = errors.get();
                 if (q == null || q.isEmpty()) {
                     actual.onCompleted();
                 } else {
                     Throwable e = collectErrors(q);
-                    if (ONCE.compareAndSet(this, 0, 1)) {
+                    if (once.compareAndSet(false, true)) {
                         actual.onError(e);
                     } else {
                         RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
                     }
                 }
             } else
             if (!delayErrors) {
-                Queue<Throwable> q = errors;
+                Queue<Throwable> q = errors.get();
                 if (q != null && !q.isEmpty()) {
                     Throwable e = collectErrors(q);
-                    if (ONCE.compareAndSet(this, 0, 1)) {
+                    if (once.compareAndSet(false, true)) {
                         actual.onError(e);
                     } else {
                         RxJavaPlugins.getInstance().getErrorHandler().handleError(e);

src/main/java/rx/internal/operators/OperatorGroupBy.java

@@ -106,28 +106,16 @@ public static final class GroupBySubscriber<T, K, V>
 
         final ProducerArbiter s;
 
-        volatile int cancelled;
-        @SuppressWarnings("rawtypes")
-        static final AtomicIntegerFieldUpdater<GroupBySubscriber> CANCELLED =
-                AtomicIntegerFieldUpdater.newUpdater(GroupBySubscriber.class, "cancelled");
+        final AtomicBoolean cancelled;
 
-        volatile long requested;
-        @SuppressWarnings("rawtypes")
-        static final AtomicLongFieldUpdater<GroupBySubscriber> REQUESTED =
-                AtomicLongFieldUpdater.newUpdater(GroupBySubscriber.class, "requested");
-        
-        volatile int groupCount;
-        @SuppressWarnings("rawtypes")
-        static final AtomicIntegerFieldUpdater<GroupBySubscriber> GROUP_COUNT =
-                AtomicIntegerFieldUpdater.newUpdater(GroupBySubscriber.class, "groupCount");
+        final AtomicLong requested;
+
+        final AtomicInteger groupCount;
 
         Throwable error;
         volatile boolean done;
 
-        volatile int wip;
-        @SuppressWarnings("rawtypes")
-        static final AtomicIntegerFieldUpdater<GroupBySubscriber> WIP =
-                AtomicIntegerFieldUpdater.newUpdater(GroupBySubscriber.class, "wip");
+        final AtomicInteger wip;
 
         public GroupBySubscriber(Subscriber<? super GroupedObservable<K, V>> actual, Func1<? super T, ? extends K> keySelector, Func1<? super T, ? extends V> valueSelector, int bufferSize, boolean delayError) {
             this.actual = actual;
@@ -137,10 +125,13 @@ public GroupBySubscriber(Subscriber<? super GroupedObservable<K, V>> actual, Fun
             this.delayError = delayError;
             this.groups = new ConcurrentHashMap<Object, GroupedUnicast<K, V>>();
             this.queue = new ConcurrentLinkedQueue<GroupedObservable<K, V>>();
-            GROUP_COUNT.lazySet(this, 1);
             this.s = new ProducerArbiter();
             this.s.request(bufferSize);
             this.producer = new GroupByProducer(this);
+            this.cancelled = new AtomicBoolean();
+            this.requested = new AtomicLong();
+            this.groupCount = new AtomicInteger(1);
+            this.wip = new AtomicInteger();
         }
 
         @Override
@@ -172,11 +163,11 @@ public void onNext(T t) {
             if (group == null) {
                 // if the main has been cancelled, stop creating groups
                 // and skip this value
-                if (cancelled == 0) {
+                if (!cancelled.get()) {
                     group = GroupedUnicast.createWith(key, bufferSize, this, delayError);
                     groups.put(mapKey, group);
 
-                    GROUP_COUNT.getAndIncrement(this);
+                    groupCount.getAndIncrement();
 
                     notNew = false;
                     q.offer(group);
@@ -210,7 +201,7 @@ public void onError(Throwable t) {
             }
             error = t;
             done = true;
-            GROUP_COUNT.decrementAndGet(this);
+            groupCount.decrementAndGet();
             drain();
         }
 
@@ -226,7 +217,7 @@ public void onCompleted() {
             groups.clear();
 
             done = true;
-            GROUP_COUNT.decrementAndGet(this);
+            groupCount.decrementAndGet();
             drain();
         }
 
@@ -235,15 +226,15 @@ public void requestMore(long n) {
                 throw new IllegalArgumentException("n >= 0 required but it was " + n);
             }
 
-            BackpressureUtils.getAndAddRequest(REQUESTED, this, n);
+            BackpressureUtils.getAndAddRequest(requested, n);
             drain();
         }
 
         public void cancel() {
             // cancelling the main source means we don't want any more groups
             // but running groups still require new values
-            if (CANCELLED.compareAndSet(this, 0, 1)) {
-                if (GROUP_COUNT.decrementAndGet(this) == 0) {
+            if (cancelled.compareAndSet(false, true)) {
+                if (groupCount.decrementAndGet() == 0) {
                     unsubscribe();
                 }
             }
@@ -252,14 +243,14 @@ public void cancel() {
         public void cancel(K key) {
             Object mapKey = key != null ? key : NULL_KEY;
             if (groups.remove(mapKey) != null) {
-                if (GROUP_COUNT.decrementAndGet(this) == 0) {
+                if (groupCount.decrementAndGet() == 0) {
                     unsubscribe();
                 }
             }
         }
 
         void drain() {
-            if (WIP.getAndIncrement(this) != 0) {
+            if (wip.getAndIncrement() != 0) {
                 return;
             }
 
@@ -274,7 +265,7 @@ void drain() {
                     return;
                 }
 
-                long r = requested;
+                long r = requested.get();
                 boolean unbounded = r == Long.MAX_VALUE;
                 long e = 0L;
 
@@ -301,12 +292,12 @@ void drain() {
 
                 if (e != 0L) {
                     if (!unbounded) {
-                        REQUESTED.addAndGet(this, e);
+                        requested.addAndGet(e);
                     }
                     s.request(-e);
                 }
 
-                missed = WIP.addAndGet(this, -missed);
+                missed = wip.addAndGet(-missed);
                 if (missed == 0) {
                     break;
                 }
@@ -378,35 +369,27 @@ static final class State<T, K> extends AtomicInteger implements Producer, Subscr
         final GroupBySubscriber<?, K, T> parent;
         final boolean delayError;
 
-        volatile long requested;
-        @SuppressWarnings("rawtypes")
-        static final AtomicLongFieldUpdater<State> REQUESTED =
-                AtomicLongFieldUpdater.newUpdater(State.class, "requested");
+        final AtomicLong requested;
 
         volatile boolean done;
         Throwable error;
 
-        volatile int cancelled;
-        @SuppressWarnings("rawtypes")
-        static final AtomicIntegerFieldUpdater<State> CANCELLED =
-                AtomicIntegerFieldUpdater.newUpdater(State.class, "cancelled");
-        
-        volatile Subscriber<? super T> actual;
-        @SuppressWarnings("rawtypes")
-        static final AtomicReferenceFieldUpdater<State, Subscriber> ACTUAL =
-                AtomicReferenceFieldUpdater.newUpdater(State.class, Subscriber.class, "actual");
+        final AtomicBoolean cancelled;
+
+        final AtomicReference<Subscriber<? super T>> actual;
 
-        volatile int once;
-        @SuppressWarnings("rawtypes")
-        static final AtomicIntegerFieldUpdater<State> ONCE =
-                AtomicIntegerFieldUpdater.newUpdater(State.class, "once");
+        final AtomicBoolean once;
 
 
         public State(int bufferSize, GroupBySubscriber<?, K, T> parent, K key, boolean delayError) {
             this.queue = new ConcurrentLinkedQueue<Object>();
             this.parent = parent;
             this.key = key;
             this.delayError = delayError;
+            this.cancelled = new AtomicBoolean();
+            this.actual = new AtomicReference<Subscriber<? super T>>();
+            this.once = new AtomicBoolean();
+            this.requested = new AtomicLong();
         }
 
         @Override
@@ -415,19 +398,19 @@ public void request(long n) {
                 throw new IllegalArgumentException("n >= required but it was " + n);
             }
             if (n != 0L) {
-                BackpressureUtils.getAndAddRequest(REQUESTED, this, n);
+                BackpressureUtils.getAndAddRequest(requested, n);
                 drain();
             }
         }
 
         @Override
         public boolean isUnsubscribed() {
-            return cancelled != 0;
+            return cancelled.get();
         }
 
         @Override
         public void unsubscribe() {
-            if (CANCELLED.compareAndSet(this, 0, 1)) {
+            if (cancelled.compareAndSet(false, true)) {
                 if (getAndIncrement() == 0) {
                     parent.cancel(key);
                 }
@@ -436,10 +419,10 @@ public void unsubscribe() {
 
         @Override
         public void call(Subscriber<? super T> s) {
-            if (ONCE.compareAndSet(this, 0, 1)) {
+            if (once.compareAndSet(false, true)) {
                 s.add(this);
                 s.setProducer(this);
-                ACTUAL.lazySet(this, s);
+                actual.lazySet(s);
                 drain();
             } else {
                 s.onError(new IllegalStateException("Only one Subscriber allowed!"));
@@ -475,15 +458,15 @@ void drain() {
 
             final Queue<Object> q = queue;
             final boolean delayError = this.delayError;
-            Subscriber<? super T> a = actual;
+            Subscriber<? super T> a = actual.get();
             NotificationLite<T> nl = NotificationLite.instance();
             for (;;) {
                 if (a != null) {
                     if (checkTerminated(done, q.isEmpty(), a, delayError)) {
                         return;
                     }
 
-                    long r = requested;
+                    long r = requested.get();
                     boolean unbounded = r == Long.MAX_VALUE;
                     long e = 0;
 
@@ -508,7 +491,7 @@ void drain() {
 
                     if (e != 0L) {
                         if (!unbounded) {
-                            REQUESTED.addAndGet(this, e);
+                            requested.addAndGet(e);
                         }
                         parent.s.request(-e);
                     }
@@ -519,13 +502,13 @@ void drain() {
                     break;
                 }
                 if (a == null) {
-                    a = actual;
+                    a = actual.get();
                 }
             }
         }
 
         boolean checkTerminated(boolean d, boolean empty, Subscriber<? super T> a, boolean delayError) {
-            if (cancelled != 0) {
+            if (cancelled.get()) {
                 queue.clear();
                 parent.cancel(key);
                 return true;