Merge pull request ReactiveX#1926 from zsxwing/fix-merge-finally

Move the codes out of the finally block

Author: benjchristensen

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

@@ -218,15 +218,17 @@ private void handleScalarSynchronousObservable(ScalarSynchronousObservable<? ext
         private void handleScalarSynchronousObservableWithoutRequestLimits(ScalarSynchronousObservable<? extends T> t) {
             T value = t.get();
             if (getEmitLock()) {
+                boolean moreToDrain;
                 try {
                     actual.onNext(value);
-                    return;
                 } finally {
-                    if (releaseEmitLock()) {
-                        drainQueuesIfNeeded();
-                    }
-                    request(1);
+                    moreToDrain = releaseEmitLock();
                 }
+                if (moreToDrain) {
+                    drainQueuesIfNeeded();
+                }
+                request(1);
+                return;
             } else {
                 initScalarValueQueueIfNeeded();
                 try {
@@ -241,22 +243,28 @@ private void handleScalarSynchronousObservableWithoutRequestLimits(ScalarSynchro
         private void handleScalarSynchronousObservableWithRequestLimits(ScalarSynchronousObservable<? extends T> t) {
             if (getEmitLock()) {
                 boolean emitted = false;
+                boolean moreToDrain;
+                boolean isReturn = false;
                 try {
                     long r = mergeProducer.requested;
                     if (r > 0) {
                         emitted = true;
                         actual.onNext(t.get());
                         MergeProducer.REQUESTED.decrementAndGet(mergeProducer);
                         // we handle this Observable without ever incrementing the wip or touching other machinery so just return here
-                        return;
+                        isReturn = true;
                     }
                 } finally {
-                    if (releaseEmitLock()) {
-                        drainQueuesIfNeeded();
-                    }
-                    if (emitted) {
-                        request(1);
-                    }
+                    moreToDrain = releaseEmitLock();
+                }
+                if (moreToDrain) {
+                    drainQueuesIfNeeded();
+                }
+                if (emitted) {
+                    request(1);
+                }
+                if (isReturn) {
+                    return;
                 }
             }
 
@@ -301,20 +309,21 @@ private boolean drainQueuesIfNeeded() {
             while (true) {
                 if (getEmitLock()) {
                     int emitted = 0;
+                    boolean moreToDrain;
                     try {
                         emitted = drainScalarValueQueue();
                         drainChildrenQueues();
                     } finally {
-                        boolean moreToDrain = releaseEmitLock();
-                        // request outside of lock
-                        if (emitted > 0) {
-                            request(emitted);
-                        }
-                        if (!moreToDrain) {
-                            return true;
-                        }
-                        // otherwise we'll loop and get whatever was added 
+                        moreToDrain = releaseEmitLock();
+                    }
+                    // request outside of lock
+                    if (emitted > 0) {
+                        request(emitted);
+                    }
+                    if (!moreToDrain) {
+                        return true;
                     }
+                    // otherwise we'll loop and get whatever was added
                 } else {
                     return false;
                 }