NullPointerException in the 'toList' operator

[zsxwing]

The following code will throw a NullPointerException.

Observable.from((String) null).toList().toBlockingObservable().single();

Exception in thread "main" java.lang.NullPointerException
at java.util.concurrent.ConcurrentLinkedQueue.checkNotNull(ConcurrentLinkedQueue.java:807)
at java.util.concurrent.ConcurrentLinkedQueue.offer(ConcurrentLinkedQueue.java:326)
at java.util.concurrent.ConcurrentLinkedQueue.add(ConcurrentLinkedQueue.java:296)
at rx.operators.OperationToObservableList$ToObservableList$1.onNext(OperationToObservableList.java:62)
at rx.operators.OperationToObservableIterable$ToObservableIterable.onSubscribe(OperationToObservableIterable.java:46)
at rx.Observable.subscribe(Observable.java:236)
at rx.operators.OperationToObservableList$ToObservableList.onSubscribe(OperationToObservableList.java:57)
at rx.Observable.subscribe(Observable.java:236)
at rx.operators.OperationMaterialize$MaterializeObservable.onSubscribe(OperationMaterialize.java:57)
at rx.Observable.subscribe(Observable.java:236)
at rx.operators.OperationToIterator.toIterator(OperationToIterator.java:47)
at rx.observables.BlockingObservable.getIterator(BlockingObservable.java:172)
at rx.observables.BlockingObservable$2.iterator(BlockingObservable.java:353)
at rx.observables.BlockingObservable._singleOrDefault(BlockingObservable.java:66)
at rx.observables.BlockingObservable.single(BlockingObservable.java:278)
at rx.Test.main(Test.java:25)

The cause is rx.operators.OperationToObservableList https://github.com/Netflix/RxJava/blob/069ae42843822b846d450b38a4aa5039182e121f/rxjava-core/src/main/java/rx/operators/OperationToObservableList.java#L58 uses ConcurrentLinkedQueue to cache the values but ConcurrentLinkedQueue rejects null value.

Is ArrayList or LinkedList enough here?

[akarnokd]

The comment says below that onNext is expected to run concurrently. The solution would be to use a null sentinel:

final ConcurrentLinkedQueue<Object> list = new ConcurrentLinkedQueue<Object>();
final Object NULL_SENTINEL = new Object();
public void onNext(T value) {
    // onNext can be concurrently executed so list must be thread-safe
    list.add(value != null ? value : NULL_SENTINEL);
}

public void onError(Throwable ex) {
    observer.onError(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) {
            @SuppressWarnings("unchecked") T t0 = t != NULL_SENTINEL ? (T)t : (T)null;
            l.add(t0);
        }

        // 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.onCompleted();
    } catch (Throwable e) {
        onError(e);
    }
}

[zsxwing]

According to Rx contract, toList does not need to handle concurrency, doesn't it?

[zsxwing]

Closed as it's fixed in #633