Fixed issue #595 about null in toList operator

Author: zsxwing

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

@@ -55,10 +55,9 @@ public ToObservableList(Observable<? extends T> that) {
         public Subscription onSubscribe(final Observer<? super List<T>> observer) {
 
             return that.subscribe(new Observer<T>() {
-                final ConcurrentLinkedQueue<T> list = new ConcurrentLinkedQueue<T>();
+                final List<T> list = new ArrayList<T>();
 
                 public void onNext(T value) {
-                    // onNext can be concurrently executed so list must be thread-safe
                     list.add(value);
                 }
 
@@ -68,16 +67,10 @@ public void onError(Throwable ex) {
 
                 public void onCompleted() {
                     try {
-                        // copy from LinkedQueue to List since ConcurrentLinkedQueue does not implement the List interface
-                        ArrayList<T> l = new ArrayList<T>(list.size());
-                        for (T t : list) {
-                            l.add(t);
-                        }
-
                         // benjchristensen => I want to make this list immutable but some clients are sorting this
                         // instead of using toSortedList() and this change breaks them until we migrate their code.
                         // observer.onNext(Collections.unmodifiableList(l));
-                        observer.onNext(l);
+                        observer.onNext(new ArrayList<T>(list));
                         observer.onCompleted();
                     } catch (Throwable e) {
                         onError(e);

rxjava-core/src/test/java/rx/operators/OperationToObservableListTest.java

@@ -66,4 +66,17 @@ public void testListMultipleObservers() {
         verify(o2, Mockito.never()).onError(any(Throwable.class));
         verify(o2, times(1)).onCompleted();
     }
+
+    @Test
+    public void testListWithNullValue() {
+        Observable<String> w = Observable.from("one", null, "three");
+        Observable<List<String>> observable = Observable.create(toObservableList(w));
+
+        @SuppressWarnings("unchecked")
+        Observer<List<String>> aObserver = mock(Observer.class);
+        observable.subscribe(aObserver);
+        verify(aObserver, times(1)).onNext(Arrays.asList("one", null, "three"));
+        verify(aObserver, Mockito.never()).onError(any(Throwable.class));
+        verify(aObserver, times(1)).onCompleted();
+    }
 }