Merge pull request ReactiveX#206 from thegeez/list-subscribe

Observable.toList breaks with multiple subscribers

Author: benjchristensen

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

@@ -40,7 +40,6 @@ public static <T> Func1<Observer<List<T>>, Subscription> toObservableList(Observ
     private static class ToObservableList<T> implements Func1<Observer<List<T>>, Subscription> {
 
         private final Observable<T> that;
-        final ConcurrentLinkedQueue<T> list = new ConcurrentLinkedQueue<T>();
 
         public ToObservableList(Observable<T> that) {
             this.that = that;
@@ -49,6 +48,7 @@ public ToObservableList(Observable<T> that) {
         public Subscription call(final Observer<List<T>> observer) {
 
             return that.subscribe(new Observer<T>() {
+                final ConcurrentLinkedQueue<T> list = new ConcurrentLinkedQueue<T>();
                 public void onNext(T value) {
                     // onNext can be concurrently executed so list must be thread-safe
                     list.add(value);
@@ -94,5 +94,28 @@ public void testList() {
             verify(aObserver, Mockito.never()).onError(any(Exception.class));
             verify(aObserver, times(1)).onCompleted();
         }
+
+        @Test
+        public void testListMultipleObservers() {
+            Observable<String> w = Observable.toObservable("one", "two", "three");
+            Observable<List<String>> observable = Observable.create(toObservableList(w));
+
+            @SuppressWarnings("unchecked")
+            Observer<List<String>> o1 = mock(Observer.class);
+            observable.subscribe(o1);
+
+            Observer<List<String>> o2 = mock(Observer.class);
+            observable.subscribe(o2);
+
+            List<String> expected = Arrays.asList("one", "two", "three");
+
+            verify(o1, times(1)).onNext(expected);
+            verify(o1, Mockito.never()).onError(any(Exception.class));
+            verify(o1, times(1)).onCompleted();
+
+            verify(o2, times(1)).onNext(expected);
+            verify(o2, Mockito.never()).onError(any(Exception.class));
+            verify(o2, times(1)).onCompleted();
+        }
     }
 }