Restore merge changes: not related to the bug

Author: akarnokd

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

@@ -194,7 +194,7 @@ private void handleNewSource(Observable<? extends T> t) {
             }
             MergeProducer<T> producerIfNeeded = null;
             // if we have received a request then we need to respect it, otherwise we fast-path
-            if (mergeProducer.requested() != Long.MAX_VALUE) {
+            if (mergeProducer.requested != Long.MAX_VALUE) {
                 /**
                  * <pre> {@code
                  * With this optimization:
@@ -237,7 +237,7 @@ private void handleScalarSynchronousObservable(ScalarSynchronousObservable<? ext
              * } </pre>
              * 
              */
-            if (mergeProducer.requested() == Long.MAX_VALUE) {
+            if (mergeProducer.requested == Long.MAX_VALUE) {
                 handleScalarSynchronousObservableWithoutRequestLimits(t);
             } else {
                 handleScalarSynchronousObservableWithRequestLimits(t);
@@ -274,11 +274,11 @@ private void handleScalarSynchronousObservableWithRequestLimits(ScalarSynchronou
                 boolean moreToDrain;
                 boolean isReturn = false;
                 try {
-                    long r = mergeProducer.requested();
+                    long r = mergeProducer.requested;
                     if (r > 0) {
                         emitted = true;
                         actual.onNext(t.get());
-                        mergeProducer.getAndAdd(-1);
+                        MergeProducer.REQUESTED.decrementAndGet(mergeProducer);
                         // we handle this Observable without ever incrementing the wip or touching other machinery so just return here
                         isReturn = true;
                     }
@@ -376,7 +376,7 @@ private void drainChildrenQueues() {
         private int drainScalarValueQueue() {
             RxRingBuffer svq = scalarValueQueue;
             if (svq != null) {
-                long r = mergeProducer.requested();
+                long r = mergeProducer.requested;
                 int emittedWhileDraining = 0;
                 if (r < 0) {
                     // drain it all
@@ -398,7 +398,7 @@ private int drainScalarValueQueue() {
                         }
                     }
                     // decrement the number we emitted from outstanding requests
-                    mergeProducer.getAndAdd(-emittedWhileDraining);
+                    MergeProducer.REQUESTED.getAndAdd(mergeProducer, -emittedWhileDraining);
                 }
                 return emittedWhileDraining;
             }
@@ -410,7 +410,7 @@ private int drainScalarValueQueue() {
             @Override
             public Boolean call(InnerSubscriber<T> s) {
                 if (s.q != null) {
-                    long r = mergeProducer.requested();
+                    long r = mergeProducer.requested;
                     int emitted = s.drainQueue();
                     if (emitted > 0) {
                         s.requestMore(emitted);
@@ -533,26 +533,19 @@ public MergeProducer(MergeSubscriber<T> ms) {
             this.ms = ms;
         }
 
-        private volatile long rq = 0;
+        private volatile long requested = 0;
         @SuppressWarnings("rawtypes")
-        static final AtomicLongFieldUpdater<MergeProducer> RQ = AtomicLongFieldUpdater.newUpdater(MergeProducer.class, "rq");
+        static final AtomicLongFieldUpdater<MergeProducer> REQUESTED = AtomicLongFieldUpdater.newUpdater(MergeProducer.class, "requested");
 
-        public long requested() {
-            return rq;
-        }
-        public long getAndAdd(long n) {
-            return RQ.getAndAdd(this, n);
-        }
-        
         @Override
         public void request(long n) {
-            if (rq == Long.MAX_VALUE) {
+            if (requested == Long.MAX_VALUE) {
                 return;
             }
             if (n == Long.MAX_VALUE) {
-                rq = Long.MAX_VALUE;
+                requested = Long.MAX_VALUE;
             } else {
-                BackpressureUtils.getAndAddRequest(RQ, this, n);
+                BackpressureUtils.getAndAddRequest(REQUESTED, this, n);
                 if (ms.drainQueuesIfNeeded()) {
                     boolean sendComplete = false;
                     synchronized (ms) {
@@ -675,7 +668,7 @@ private void emit(T t, boolean complete) {
                     } else {
                         // this needs to check q.count() as draining above may not have drained the full queue
                         // perf tests show this to be okay, though different queue implementations could perform poorly with this
-                        if (producer.requested() > 0 && q.count() == 0) {
+                        if (producer.requested > 0 && q.count() == 0) {
                             if (complete) {
                                 parentSubscriber.completeInner(this);
                             } else {
@@ -686,7 +679,7 @@ private void emit(T t, boolean complete) {
                                     onError(OnErrorThrowable.addValueAsLastCause(e, t));
                                 }
                                 emitted++;
-                                producer.getAndAdd(-1);
+                                MergeProducer.REQUESTED.decrementAndGet(producer);
                             }
                         } else {
                             // no requests available, so enqueue it
@@ -735,7 +728,7 @@ private void enqueue(T t, boolean complete) {
         private int drainRequested() {
             int emitted = 0;
             // drain what was requested
-            long toEmit = producer.requested();
+            long toEmit = producer.requested;
             Object o;
             for (int i = 0; i < toEmit; i++) {
                 o = q.poll();
@@ -757,7 +750,7 @@ private int drainRequested() {
             }
 
             // decrement the number we emitted from outstanding requests
-            producer.getAndAdd(-emitted);
+            MergeProducer.REQUESTED.getAndAdd(producer, -emitted);
             return emitted;
         }