Synchronization of Merge operator (fixes)

- return AtomicSubscription not MergeSubscription which I was accidentally still returning
- try/finally in unit test so threads are released even if assertion is thrown

Author: benjchristensen

rxjava-core/src/main/java/rx/operators/OperationMerge.java

@@ -118,7 +118,7 @@ private MergeObservable(Observable<Observable<T>> sequences) {
             this.sequences = sequences;
         }
 
-        public MergeSubscription call(Observer<T> actualObserver) {
+        public Subscription call(Observer<T> actualObserver) {
 
             /**
              * We must synchronize a merge because we subscribe to multiple sequences in parallel that will each be emitting.
@@ -127,15 +127,16 @@ public MergeSubscription call(Observer<T> actualObserver) {
              * <p>
              * Bug report: https://github.com/Netflix/RxJava/issues/200
              */
-            SynchronizedObserver<T> synchronizedObserver = new SynchronizedObserver<T>(actualObserver, new AtomicObservableSubscription(ourSubscription));
+            AtomicObservableSubscription subscription = new AtomicObservableSubscription(ourSubscription);
+            SynchronizedObserver<T> synchronizedObserver = new SynchronizedObserver<T>(actualObserver, subscription);
 
             /**
              * Subscribe to the parent Observable to get to the children Observables
              */
             sequences.subscribe(new ParentObserver(synchronizedObserver));
 
             /* return our subscription to allow unsubscribing */
-            return ourSubscription;
+            return subscription;
         }
 
         /**
@@ -439,11 +440,13 @@ public void onNext(String v) {
             // wait for both observables to send (one should be blocked)
             o1.onNextBeingSent.await();
             o2.onNextBeingSent.await();
-            
-            assertEquals(1, concurrentCounter.get());
 
-            // release so it can finish
-            endLatch.countDown();
+            try { // in try/finally so threads are released via latch countDown even if assertion fails
+                assertEquals(1, concurrentCounter.get());
+            } finally {
+                // release so it can finish
+                endLatch.countDown();
+            }
 
             try {
                 o1.t.join();