1.x: Fix the observer array becoming visible too early and causing NPE

Author: akarnokd

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

@@ -177,7 +177,10 @@ public void request(long n) {
 
     }
 
-    private static final class Zip<R> extends AtomicLong {
+    static final class Zip<R> extends AtomicLong {
+        /** */
+        private static final long serialVersionUID = 5995274816189928317L;
+        
         final Observer<? super R> child;
         private final FuncN<? extends R> zipFunction;
         private final CompositeSubscription childSubscription = new CompositeSubscription();
@@ -186,7 +189,7 @@ private static final class Zip<R> extends AtomicLong {
         int emitted = 0; // not volatile/synchronized as accessed inside COUNTER_UPDATER block
 
         /* initialized when started in `start` */
-        private Object[] observers;
+        private volatile Object[] observers;
         private AtomicLong requested;
 
         public Zip(final Subscriber<? super R> child, FuncN<? extends R> zipFunction) {
@@ -197,14 +200,16 @@ public Zip(final Subscriber<? super R> child, FuncN<? extends R> zipFunction) {
 
         @SuppressWarnings("unchecked")
         public void start(@SuppressWarnings("rawtypes") Observable[] os, AtomicLong requested) {
-            observers = new Object[os.length];
-            this.requested = requested;
+            Object[] observers = new Object[os.length];
             for (int i = 0; i < os.length; i++) {
                 InnerSubscriber io = new InnerSubscriber();
                 observers[i] = io;
                 childSubscription.add(io);
             }
-
+            
+            this.requested = requested;
+            this.observers = observers; // full memory barrier: release all above
+            
             for (int i = 0; i < os.length; i++) {
                 os[i].unsafeSubscribe((InnerSubscriber) observers[i]);
             }