Tried to adapt according to Ben's review comments.

Author: jmhofer

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

@@ -34,6 +34,8 @@
 import rx.Observer;
 import rx.Subscription;
 import rx.subscriptions.Subscriptions;
+import rx.util.AtomicObservableSubscription;
+import rx.util.SynchronizedObserver;
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
 import rx.util.functions.Func3;
@@ -124,7 +126,7 @@ public void onNext(T args) {
      */
     private static class Aggregator<R> implements Func1<Observer<? super R>, Subscription> {
 
-        private Observer<? super R> observer;
+        private volatile Observer<? super R> observer;
 
         private final FuncN<? extends R> combineLatestFunction;
         private final AtomicBoolean running = new AtomicBoolean(true);
@@ -228,19 +230,21 @@ public Subscription call(Observer<? super R> observer) {
             if (this.observer != null) {
                 throw new IllegalStateException("Only one Observer can subscribe to this Observable.");
             }
-            this.observer = observer;
+            
+            AtomicObservableSubscription subscription = new AtomicObservableSubscription(new Subscription() {
+                @Override
+                public void unsubscribe() {
+                    stop();
+                }
+            });
+            this.observer = new SynchronizedObserver<R>(observer, subscription);
 
             /* start the observers */
             for (CombineObserver<R, ?> rw : observers) {
                 rw.startWatching();
             }
 
-            return new Subscription() {
-                @Override
-                public void unsubscribe() {
-                    stop();
-                }
-            };
+            return subscription;
         }
 
         private void stop() {

rxjava-core/src/main/java/rx/util/SynchronizedObserver.java

@@ -64,12 +64,12 @@ public final class SynchronizedObserver<T> implements Observer<T> {
      * // TODO composing of this class should rarely happen now with updated design so this decision should be revisited
      */
 
-    private final Observer<T> observer;
+    private final Observer<? super T> observer;
     private final AtomicObservableSubscription subscription;
     private volatile boolean finishRequested = false;
     private volatile boolean finished = false;
 
-    public SynchronizedObserver(Observer<T> Observer, AtomicObservableSubscription subscription) {
+    public SynchronizedObserver(Observer<? super T> Observer, AtomicObservableSubscription subscription) {
         this.observer = Observer;
         this.subscription = subscription;
     }