src/main/java/io/reactivex/internal/operators/observable/ObservableCache.java
void add(CacheDisposable<T> consumer) {
        for (;;) {
            CacheDisposable<T>[] current = observers.get();
            if (current == TERMINATED) {
                return;
            }
            int n = current.length;

            @SuppressWarnings("unchecked")
            CacheDisposable<T>[] next = new CacheDisposable[n + 1];
            System.arraycopy(current, 0, next, 0, n);
            next[n] = consumer;

            if (observers.compareAndSet(current, next)) {
                return;
            }
        }
    }
void add(CacheDisposable<T> consumer) {
        for (;;) {
            CacheDisposable<T>[] current = observers.get();
            if (current == TERMINATED) {
                return;
            }
            int n = current.length;

            @SuppressWarnings("unchecked")
            CacheDisposable<T>[] next = new CacheDisposable[n + 1];
            System.arraycopy(current, 0, next, 0, n);
            next[n] = consumer;

            if (observers.compareAndSet(current, next)) {
                return;
            }
        }
    }
void add ( CacheDisposable < T > consumer ) { for ( ; ; ) { CacheDisposable < T > [ ] current = observers . get ( ) ; if ( current == TERMINATED ) { return ; } int n = current . length ; @ SuppressWarnings ( "unchecked" ) CacheDisposable < T > [ ] next = new CacheDisposable [ n + 1 ] ; System . arraycopy ( current , 0 , next , 0 , n ) ; next [ n ] = consumer ; if ( observers . compareAndSet ( current , next ) ) { return ; } } }
Atomically adds the consumer to the {@link #observers} copy-on-write array
if the source has not yet terminated.
@param consumer the consumer to add
Atomically adds the consumer to the {