ConnectableObservables (part 2)

Introduction

In the previous post, I've shown how one can write a "simple" ConnectableObservable that uses a Subject to dispatch events to subscribers once it has been connected.

The shortcoming of the solution is that there is no request coordination and everything runs in unbounded mode: the developers have to apply onBackpressureXXX strategies per subscriber, however, that leads to either dropping data or buffer bloat.

If the underlying Observable is cold, there should be a way to make sure it emits only as much elements as the child subscribers can process. To achieve this, we need request coordination.

Request coordination

So far, the operators we were implementing had to deal with a single child subscriber and its request at a time. One had to either pass it through, rebatch it or accumulate it, based on the business logic of said operator.

When there are multiple child Subscribers, the problem space suddenly receives a new dimension. What are the new problems?

Every bit counts

First, different child subscribers may request different amounts. Some may request small amounts, some may request larger amounts and others may want to run in unbounded mode (i.e., request(Long.MAX_VALUE)). In addition, the request calls may happen any time and with any amount.

Given such heterogeneous request pattern, what should be the request amount sent to the upstream Observable source?

There are two main options:

1. request as much that the smallest child Subscriber requested and
2. request as much as the largest child Subscriber requested.

Option 1) is essentially the lockstep approach. Its benefit is that there is no no need for request re-batching and buffering since once the upstream emits, everybody can receive it immediately. (Rebatching and buffering is an option in case the request amounts are really 1s or 10s at a time.) The drawback is that the whole setup slows down to the slowest child Subscriber, which if "forgets" to request, nobody gets anything.

Option 2) gives more room to individual child Subscribers and allows them to run on their own pace. However, this solution requires unbounded buffering capability (which may be shared or per each Subscriber). This means if there is an unbounded child Subscriber, the operator has to request Long.MAX_VALUE and fill the buffers for everyone. This, depending on the operator, may be of no problem though.

Subscribers may come and go at will

The second problem is that the the number of Subscribers may not be constant: new subscribers arrive, old ones leave. This poses another set of problems:

1. A child Subscriber may request Long.MAX_VALUE then leave after a few (or no) elements.
2. A child Subscriber may arrive but not request anything, stopping everyone else.
3. A child Subscriber may leave at any time and thus its request amount "pressure" has to be released.
4. All child Subscribers leave before the upstream Observable completes. What should happen in this case?

Unfortunately, problems 1) and 2) require mutually exclusive approaches explained above (lockstep vs. unbounded buffering). Problem 3) requires unsubscription action.

Problem 4) depends on the approach taken in respect to 1) and 2).

Within the lockstep approach, two sub-options arise. Either one has to introduce some bounded buffers that will hold onto the requested amounts, which now has to be re-batched to fit in, and simply await the new Subscribers. Otherwise, one has to slowly "drip" away the source values until a child Subscriber arrives. 

Within the unbounded buffering approach, one can simply keep buffering or again, start dropping values.

Approaches taken in RxJava

RxJava has two operators that return a ConnectableObservable: publish() and replay(). For a long time, these were ignoring backpressure completely and behaved just like the MulticastSupplier in the previous part.

These operators were rewritten to support backpressure (in 1.0.13 and 1.0.14 respectively) and had to take the problems mentioned before into account. The solutions were as follows

Operator publish() does lockstepping with a fixed prefetch buffer: the buffer is only drained (and then replenished) if all known child Subscribers can take a value. If there are no child Subscribers, it "slowly drips away" it source, which means it starts to request 1 by 1 and drops these values.

Operator replay() does unbounded buffering. The reason for this is that both the bounded and unbounded version of replay() has to buffer and replay all values from the upstream anyway. You may think, why buffer everything when the replay is time and/or size bound. The answer is that these operators, similar to Subjects, have to deliver events continuously and without skips; if there is an child Subscriber that arrived at some time, requested 1 then went to "sleep", the next time it requests the bounded replay has to present the next value, no matter how far ahead the other Subscribers went in the meantime.

The effect of disconnection

There is a problem that isn't dealt with in the RxJava operators but has to be mentioned. If one unsubscribes the Subscription returned by the connect() method, the upstream will stop sending further events.

The problem is that this may leave the child Subscribers hanging: they won't receive any further events (beyond those that are already in some buffer of the respective operator). We have similar problems with CompletableFutures in Java 8. One can cancel a Future but what happens to those that were awaiting its result?

The solution in Java 8 is to emit a CancellationException as the result in this case so that the dependent computations can terminate. However, this isn't the case with RxJava (in both 1.x and 2.x branches). The current implementation will just hang the child Subscribers.

This problem may appear outside of a ConnectableObservable as well. For some time, the RxAndroid 0.x library contained an operator that were applied to all sequences and unsubscribed them if the lifecycle required cleanup. The problem was that this left child Subscribers without termination events. I suggested emitting an onError and onCompleted event for this case. There was no resolution of the problem and the operator was removed before 1.0.

On a personal note, I don't remember anyone from the community complaining about this problem and it seems nobody is really affected by this behavior. As with many obscure and corner cases, if I don't mention them, nobody else seems to discover them.

The effect of termination

Upstream Observables may terminate normally, in which case the ConnectableObservable will emit the terminal event to child Subscribers.

At this point, a new Subscriber may subscribe to the terminated ConnectableObservable. What should happen in this case? Does the termination also mean disconnection? Should the child Subscriber get terminated instantly, similar to PublishSubject?

Again the solution requires business decision. RxJava chose the approach that a terminal event sent to a ConnectableObservables is considered a disconnect event and late coming Subscribers won't receive any terminal event but will be remembered until another call to connect() happens.

This has the benefit that the developers can "prepare" child Subscribers before the upstream Observable gets run and thus avoid losing events. The drawback is that one has to remember to call connect() again, otherwise nothing runs and the Subscribers are left hanging.

Family of collectors and emitters

Before we jump into some code, I'd like to sketch out a pattern that is the foundation of almost all operators that deal with either multiple sources or multiple child Subscribers.

I've written dozens of such operators and I've noticed they all use the same set of components and methods:

1. They all need to track Subscribers, either the child Subscribers or the Subscribers that are subscribed to the source Observables. The tracking structure uses the copy-on-write approach of array-based resource containers.
2. They all use an emitter loop (synchronized) or drain loop (atomics) which has to be triggered from many places: when an event is emitted from upstream(s), when a new child Subscriber arrives, when a request comes from child Subscribers and sometimes when a child unsubscribes.
3. The loop has some preprocessing step: figuring out where the Subscribers are at the moment, selecting which source to drain or combining available values from sources in some fashion
4. Finally, the events are delivered to Subscriber(s) and replenishments are requested from source Observable(s).

Which operator?

Now that we are aware of the problems, let's implement a ConnectableObservable which does request coordination.

I've been thinking what operator to implement. My first thought was to show how to implement the operator pair of an AsyncSubject or BehaviorSubject (similar to how publish() is the pair of PublishSubject), however, the former can be implemented using plain composition plus replay():

public ConnectableObservable<T> async() {
    return takeLast(1).replay();
}

Implementing the pair of BehaviorSubject is a bit more involved. The naive implementation would use composition such as this:

public ConnectableObservable<T> behave() {
    return replay(1);
}

However, this doesn't properly capture the behavior of a terminated BehaviorSubject: child Subscribers get nothing but a terminal event whereas replay will always replay 1 value and 1 terminal event after it completed.

To minimize brain melting, I'm not going to show how to implement a variant of the least complex of the operators: publish().

Publish (or die)

First, let's sketch out all the requirements we want to achieve:

1. The operator should do a lockstep-based request coordination with prefetching (for efficiency)
2. The effect of disconnection on the child Subscribers should be parametrizable: no event, signal error or signal completion.
3. The operator should be considered terminated and new subscribers will wait for the next connect().
4. The operator will allow errors to cut ahead. (Implementing error-delay is an excercise left to the reader).
5. The operator will use a power-of-2 prefetch buffer.

With these requirements, we start with the skeleton of the class as usual:

public class PublishConnectableObservable<T> 
extends ConnectableObservable<T> {

    public enum DisconnectStrategy {                           // (1)
        NO_EVENT,
        SEND_ERROR,
        SEND_COMPLETED
    }
    
    public static <T> PublishConnectableObservable<T> 
    createWith(                                               // (2)
            Observable<T> source, 
            DisconnectStrategy strategy) {
        State<T> state = new State<>(strategy, source);
        return new PublishConnectableObservable<>(state);
    }
    
    final State<T> state;
    
    protected PublishConnectableObservable(State<T> state) {  // (3)
        super(state);
        this.state = state;
    }
    
    @Override
    public void connect(
            Action1<? super Subscription> connection) {       // (4)
        state.connect(connection);
    }
}

Nothing extraordinary so far:

1. We create an enum for the disconnection strategy
2. We have to use a factory method because the internal state has to be accessible from OnSubscribe and from instance methods of this class.
3. We construct the object where State doubles as an OnSubscribe to save on allocation.
4. Finally, we delegate the connection attempt to the state object. This gives us a less verbose source code.

Next comes the state object with some familiar structure (see last post of this series):

static final class State<T> implements OnSubscribe<T> {
    final DisconnectStrategy strategy;
    final Observable<T> source;
    
    final AtomicReference<Connection<T>> connection;      // (1)
      
    public State(DisconnectStrategy strategy, 
            Observable<T> source) {                       // (2)
        this.strategy = strategy;
        this.source = source;
        this.connection = new AtomicReference<>(
            new Connection<>(this)
        );
    }
        
    @Override
    public void call(Subscriber<? super T> s) {           // (3)
        // implement
    }
        
    public void connect(
        Action1<? super Subscription> disconnect) {       // (4)
        // implement
    }
        
    public void replaceConnection(Connection<T> conn) {   // (5)
        Connection<T> next = new Connection<>(this);
        connection.compareAndSet(conn, next);
    }
}

The state object will handle the connection, subscription and reconnection cases:

1. Because we have to reconnect, we store the current connection in an AtomicReference.
2. We initialize the source and strategy fields and set up an initial unconnected connection.
3. The method call() from OnSubscribe will handle the subscribers; I'll show the implementation further down.
4. The connect method will handle the connection attempts; I'll show the implementation further down.
5. Finally, once a connection has been terminated on its own or via unsubscribe, we have to replace the old connection with a fresh connection atomically and not overwriting somebody else's fresh connection due to races.

Before going deep into the complicated logic, two more simplistic classes remain. The first is the Subscriber that will be subscribed to the source Observable:

static final class SourceSubscriber<T> 
extends Subscriber<T> {
    final Connection<T> connection;
    public SourceSubscriber(
            Connection<T> connection) {    // (1)
        this.connection = connection;
    }
    @Override
    public void onStart() {
        request(RxRingBuffer.SIZE);        // (2)
    }

    @Override
    public void onNext(T t) {
        connection.onNext(t);              // (3)
    }

    @Override
    public void onError(Throwable e) {
        connection.onError(e);
    }

    @Override
    public void onCompleted() {
        connection.onCompleted();
    }
    
    public void requestMore(long n) {      // (4)
        request(n);
    }
}

The class, again is full of delegations:

1. We store the connection object and we will delegate events to it.
2. If this Subscriber is subscribed to the source Observable, we request only a limited number of elements upfront. (Parametrizing this is left to the reader).
3. Again, for class simplicity, we delegate the events to the connection object, which happens to implement the Observer interface for convenience
4. We will have to replenish all consumed values but request() is a protected method: it is exposed through the requestMore() method.

Next comes a Publisher and Subscriber instance that will handle the unsubscription and request accounting for the child Subscribers of our operator.

static final class PublishProducer<T> 
implements Producer, Subscription {
    final Subscriber<? super T> actual;
    final AtomicLong requested;
    final AtomicBoolean once;
    volatile Connection<T> connection;             // (1)
    
    public PublishProducer(
            Subscriber<? super T> actual) {
        this.actual = actual;
        this.requested = new AtomicLong();
        this.once = new AtomicBoolean();
    }
    
    @Override
    public void request(long n) {
        if (n < 0) {
            throw new IllegalArgumentException();
        }
        if (n > 0) {
            BackpressureUtils
                .getAndAddRequest(requested, n);
            Connection<T> conn = connection;       // (2)
            if (conn != null) {
                conn.drain();
            }
        }
    }
    
    @Override
    public boolean isUnsubscribed() {
        return once.get();
    }
    
    @Override
    public void unsubscribe() {
        if (once.compareAndSet(false, true)) {
            Connection<T> conn = connection;       // (3)
            if (conn != null) {
                conn.remove(this);
                conn.drain();
            }
        }
    }
}

This is a bit more interesting.

1. We need to know about what connection this class has to deal with for two reasons: 1) it has to notify the connection the underlying Subscriber can receive values, 2) if the subscriber unsubscribes, it may mean the other Subscribers can now receive further values.
2. Since request() runs asynchronously, the connection might not be available yet. We have to remember to call drain() once this connection becomes available (shown later on).
3. Since unsubscribe() runs asynchronously as well, it has check for non-null and only remove itself from the array of subscribers (shown later on). Note also the idempotence provided by once.

The final class, in skeleton form is the Connection itself:

@SuppressWarnings({ "unchecked", "rawtypes" })
static final class Connection<T>
 implements Observer<T> {                             // (1)

    final AtomicReference<PublishProducer<T>[]>
        subscribers;
    final State<T> state;
    final AtomicBoolean connected;
    
    final Queue<T> queue;
    final AtomicReference<Throwable> error;
    volatile boolean done;

    volatile boolean disconnected;
    
    final AtomicInteger wip;
    
    final SourceSubscriber parent;
    
    
    static final PublishProducer[] EMPTY = 
        new PublishProducer[0];

    static final PublishProducer[] TERMINATED = 
        new PublishProducer[0];
    
    public Connection(State<T> state) {               // (2)
        this.state = state;
        this.subscribers = new AtomicReference<>(EMPTY);
        this.connected = new AtomicBoolean();
        this.queue = new SpscArrayQueue(
            RxRingBuffer.SIZE);
        this.error = new AtomicReference<>();
        this.wip = new AtomicInteger();
        this.parent = createParent();
    }
    
    SourceSubscriber createParent() {                 // (3)
        // implement
    }
    
    boolean add(PublishProducer<T> producer) {        // (4)
        // implement
    }
    
    void remove(PublishProducer<T> producer) {
        // implement
    }
    
    void onConnect(
         Action1<? super Subscription> disconnect) {  // (5)
        // implement
    }
    
    @Override
    public void onNext(T t) {                         // (6)
        // implement
    }

    @Override
    public void onError(Throwable e) {
        // implement
    }

    @Override
    public void onCompleted() {
        // implement
    }
    
    void drain() {                                    // (7)
        // implement
    }
    
    boolean checkTerminated(boolean d, 
        boolean empty) {
        // implement
    }
}

The method names and fields should look familiar by now:

1. The class has to manage a set of state variables: the current array of Subscribers, the value queue plus the terminal event holders, the connection and disconnection indicators, the work counter for the queue-drain approach, the Subscriber that is subscribed to the Observable and finally the EMPTY and TERMINATED array indicators.
2. The constructor initializes the various fields.
3. The subscriber needs some preparations besides creating a new SourceSubscriber, therefore, I factored it out into a separate method.
4. The copy-on-write handling of the known subscribers is done via add and remove, similar to how we did this with Subjects and with the array-backed Subscription container.
5. We will handle the source events with these onXXX methods.
6. Finally, the drain and termination check methods for the queue-drain approach.

The meltdown

So far, the classes and those methods implemented were nothing special. However, the real complexity starts from here on. I'll show the missing implementations one by one and mention the concurrency considerations with them as well..

I suggest you take a small break, drink some power-up, clear your head at this point.

Done? All right, let'd do this.

State.call

This method is responsible for handling the incoming child Subscribers. The method has to consider that the connection may terminate on its own or get disconnected concurrently:

@Override
public void call(Subscriber<? super T> s) {
    PublishProducer<T> pp 
        = new PublishProducer<>(s);
    
    s.add(pp);
    s.setProducer(pp);                                // (1)

    for (;;) {
        Connection<T> curr = connection.get();
        
        pp.connection = curr;                         // (2)
        if (curr.add(pp)) {                           // (3)
            if (pp.isUnsubscribed()) {                // (4)
                curr.remove(pp);
            } else {
                curr.drain();                         // (5)
            }
            break;
        }
    }
}

1. First, we create a PublishProducer and set it on the subscriber to react to requests and unsubscription.
2. Next, we retrieve the current known connection and set it on the PublishProducer so it can call the drain() method if it wishes.
3. We attempt to add the PublishProducer to the internal tracking array. If this fails, it means the current connection has terminated and we have to try the next connection (once becomes available) by looping a bit.
4. Even if the add succeeded, the child might have just unsubscribed and thus the remove might not have found it. By calling it here again, we can make it sure the PublishProducer doesn't stay in the array unnecessarily.
5. Once the add succeeded, we have to call drain since a concurrent call in PublishProducer might have not seen a non-null connection and couldn't notify the connection for more values (or about unsubscription). The call will make sure this PublishProducer is handled as necessary.

State.connect

This method is responsible for triggering a single connection on an unconnected Connection instance and/or return the Subscription that let's an active Connection get unsubscribed.

public void connect(Action1<? super Subscription> disconnect) {
    for (;;) {
        Connection<T> curr = this.connection.get();
        
        if (!curr.connected.get() && 
                curr.connected.compareAndSet(false, true)) {  // (1)
            curr.doConnect(disconnect);
            return;
        }
        if (!curr.parent.isUnsubscribed()) {                  // (2)
            disconnect.call(curr.parent);
            return;
        }
        
        replaceConnection(curr);                              // (3)
    }
}

This method is also racing with a termination/disconnection and as such, it has to take them into account when attempting to establish a fresh connection.

1. It works by first retrieving the current connection and if the current thread is the first, switch it into a connected state. If successful, the doConnect method is called which will do the necessary subscription work.
2. Otherwise, check if the current connection is unsubscribed. If not return it to the callback. Note that there is a small window here where the current connection is determined active but may become disconnected/terminated when the method is called. Resolving this issue requires either blocking synchronization between termination and connection or other serialization approach. In practice, however, this is rarely an issue and can be ignored.
3. Finally, if the current connection is disconnected, let's replace it with a fresh, not-yet connected Connection and try the loop again.

Connection.createParent

The method constructs a SourceSubscriber and sets it up to behave according to the disconnection strategy:

SourceSubscriber createParent() {
    SourceSubscriber parent = new SourceSubscriber<>(this);
    
    parent.add(Subscriptions.create(() -> {
        switch (state.strategy) {
        case SEND_COMPLETED:
            onCompleted();
            break;
        case SEND_ERROR:
            onError(new CancellationException("Disconnected"));
            break;
        default:
            disconnected = true;
            drain();
        }
    }));
    
    return parent;
}

The method will instantiate a SourceSubscriber and add a Subscription to it. This subscription, depending on the disconnection strategy, will either call onCompleted, onError with a CancellationException or set the disconnect flag followed by a call to drain (the onXXX methods call drain()).

We need the disconnected flag because we can't use an isUnsubscribed check: it would always skip the terminal event and appear as if we'd have the NO_EVENT strategy.

Connection.add, Connection.remove

The algorithms for adding and removing resources to an array-based container with copy-on-write semantics should be quite familiar by now. For completeness, here are the methods anyway:

boolean add(PublishProducer<T> producer) {
    for (;;) {
        PublishProducer<T>[] curr = subscribers.get();
        if (curr == TERMINATED) {
            return false;
        }
        
        int n = curr.length;
        
        PublishProducer<T>[] next = new PublishProducer[n + 1];
        System.arraycopy(curr, 0, next, 0, n);
        next[n] = producer;
        if (subscribers.compareAndSet(curr, next)) {
            return true;
        }
    }
}

void remove(PublishProducer<T> producer) {
    for (;;) {
        PublishProducer<T>[] curr = subscribers.get();
        if (curr == TERMINATED || curr == EMPTY) {
            return;
        }
        
        int n = curr.length;
        
        int j = -1;
        for (int i = 0; i < n; i++) {
            if (curr[i] == producer) {
                j = i;
                break;
            }
        }
        
        if (j < 0) {
            break;
        }
        PublishProducer<T>[] next;
        if (n == 1) {
            next = EMPTY;
        } else {
            next = new PublishProducer[n - 1];
            System.arraycopy(curr, 0, next, 0, j);
            System.arraycopy(curr, j + 1, next, j, n - j - 1);
        }
        if (subscribers.compareAndSet(curr, next)) {
            return;
        }
    }
}

Connection.onXXX

The four onXXX methods on the class are quite sort, therefore, I'll show them togheter in this subsection:

void onConnect(
         Action1<? super Subscription> disconnect) {        // (1)
    disconnect.call(this.parent);
      
    state.source.unsafeSubscribe(parent);
}
    
@Override
public void onNext(T t) {                                   // (2)
    if (queue.offer(t)) {
        drain();
    } else {
        onError(new MissingBackpressureException());
        parent.unsubscribe();
    }
}

@Override
public void onError(Throwable e) {
    if (!error.compareAndSet(null, e)) {                    // (3)
        e.printStackTrace();
    } else {
        done = true;
        drain();
    }
}

@Override
public void onCompleted() {                                 // (4)
    done = true;
    drain();
}

Let's see them:

1. The reason we have to drag the Action1 all the way here instead of calling it State.connect at (2) is that the call must happen before the actual subscription to the underlying Observable to allow synchronous cancellation.
2. The next method offers the value and calls drain to make sure it is delivered if possible. Note that if the queue is full, we reward it with a MissingBackpressureException and unsubscription; it means the upstream doesn't handle backpressure well or at all.
3. Since we may receive an error as part of the upstream event as well as a disconnection event, we heed an AtomicReference and set only one of them as the terminal event. In this example, the first one wins, the other gets printed to the console. If the CAS succeded, we set the done flag and call drain to handle things.
4. It is true onCompleted can also be called from two places, but since it just sets the done flag to true, there is no need for any CAS-ing here. It is also true that due to the disconnection strategy, the onError and onCompleted can race with each other. However, since the difference of handling them is just that error contains null or not, it is't really a problem. Note also that since we used unsafeSubscribe in onConnect, there shouldn't be any call to the SourceSubscriber.unsubscribe coming from upstream and causing trouble if the source terminated normally and the disconnection strategy happen to be SEND_ERROR.

Connection.drain

This is unquestionably the heart of the operator and the most complicated logic due to the effects of concurrently changing values it has to rely on. I'll explain it in piece by piece:

First, it contains a familiar drain loop with wip counter and missed count:

void drain() {
    if (wip.getAndIncrement() != 0) {
        return;
    }
    
    int missed = 1;
    
    for (;;) {

        if (checkTerminated(done, queue.isEmpty())) {
            return;
        }

        // implement rest
       
        missed = wip.addAndGet(-missed);
        if (missed == 0) {
            break;
        }
    }
}

Nothing fancy yet. The wip counter doubles as the serialization entry point on a 0 - 1 transition and a missed counter above that.

If inside the loop, the first thing to do is to check for a terminal condition via checkTerminated (explained later). It checks for the terminal events and disconnected state and acts accordingly. This is done before the upcoming request coordination since terminal events are not subject to backpressure management and can be emitted before any child Subscriber requests anything.

The next step is to perform request coordination. Since we set out to do a lockstep coordination, we have to ask all known child subscribers for their current requested amount and figure out the minimum amount everybody can receive. Note that this can be zero.

        //... checkTerminated call

        PublishProducer<T>[] a = subscribers.get();
        
        int n = a.length;
        long minRequested = Long.MAX_VALUE;
        
        for (PublishProducer<T> pp : a) {
            if (!pp.isUnsubscribed()) {
                minRequested = Math.min(minRequested, pp.requested.get());
            }
        }

        // ... missed decrementing

At this point, it is possible n is zero. If there are no subscribers, we set out to "slowly drip away" the available values:

        // ... minRequested calculation

        if (n == 0) {
            if (queue.poll() != null) {
                parent.requestMore(1);
            }
        } else {
            // implement rest           
        }

        // ... missed decrementing

We have to check if the queue is non empty and consume a value with a single poll() then we ask for replenishment. Note that the "slowness" depends on the speed of the upstream Observable. If one decides to do nothing if there are no subscribers, the if statement can be simplified to if (n != 0) { } but should not be removed!

If we know there are any subscribers and we know the minimum requested amount, we can try draining our queue and emit that amount to everybody.

            // if n != 0 branch

            if (checkTerminated(done, queue.isEmpty())) {   // (1)
                return;
            }

            long e = 0L;
            while (minRequested != 0) {

                boolean d = done;
                T v = queue.poll();
                
                if (checkTerminated(d, v == null)) {        // (2)
                    return;
                }
                
                if (v == null) {
                    break;
                }

                // final detail to implement
                
                minRequested--;                             // (3)
                e++;
            }
            
            if (e != 0L) {                                  // (4)
                parent.requestMore(e);
            }
        
        // end of n != branch

This should also look familiar. We check the  terminal conditions again (1) (optional if you want to be eager). Next, we loop until the minRequested is zero or the queue becomes empty. Inside the loop we do the usual termination checks (2) and emission accounting (3). After the loop, if there were emissions, we ask for replenishment from the SourceSubscriber instance (4).

Lastly, the final piece of the drain method is the publication of each value to all subscribers:

                // ... v == null check

                for (PublishProducer<T> pp : a) {
                    pp.actual.onNext(v);
                    if (pp.requested.get() != Long.MAX_VALUE) {
                        pp.requested.decrementAndGet();
                    }
                }

                // ... minRequested--

For each of the PublishProducer (i.e., child Subscriber), we emit the value and decrement the requested amount if not Long.MAX_VALUE (i.e., unbounded child Subscriber).

Wasn't that painful, was it?

Connection.checkTerminated

The checkTerminated method has more things to do since it has to deliver the terminal events to all Subscribers while making sure new Subscribers don't succeed within the add method.

boolean checkTerminated(boolean done, boolean empty) {    // (1)
    if (disconnected) {                                   // (2)
        subscribers.set(TERMINATED);
        queue.clear();
        return true;
    }
    if (done) {
        Throwable e = error.get();
        if (e != null) {
            state.replaceConnection(this);                // (3)
            queue.clear();

            PublishProducer<T>[] a = 
                subscribers.getAndSet(TERMINATED);        // (4)
            
            for (PublishProducer<T> pp : a) {             // (5)
                if (!pp.isUnsubscribed()) {
                    pp.actual.onError(e);
                }
            }
            
            
            return true;
        } else
        if (empty) {
            state.replaceConnection(this);                // (6)

            PublishProducer<T>[] a = 
                subscribers.getAndSet(TERMINATED);
            
            for (PublishProducer<T> pp : a) {
                if (!pp.isUnsubscribed()) {
                    pp.actual.onCompleted();
                }
            }
            
            return true;
        }
    }
    return false;
}

It works as follows:

1. The method takes only a done and an empty indicator but not any individual Subscriber or the array of known subscribers.
2. Since the disconnected flag is set only if the disconnection strategy was NO_EVENT, we can't do much but just set in the TERMINATED indicator array. Anybody unlucky enough still subscribed won't get any further events.
3. If the done flag is true and there is an error we first replace the current connection with a fresh one (within the state) so newcommers won't try to subscribe to a terminated connection. 
4. After clearing the queue for any normal values, we swap in the TERMINATED indicator array so ...
5. ... anybody who got in can now receive its terminal event and the drain loop will quit.
6. The same logic applies in the case when the upstream has completed normally and the queue has become empty.

Testing it out

Finally, we reached the end of one of the most complicated operators in history of RxJava. Now let's reward us via a small unit test to see if the backpressure and the disconnection stategy really works:

Observable<Integer> source = Observable.range(1, 10);

TestSubscriber<Integer> ts = TestSubscriber.create(5);

PublishConnectableObservable<Integer> o = createWith(
    source, DisconnectStrategy.SEND_ERROR);

o.subscribe(ts);

Subscription s = o.connect();

s.unsubscribe();

System.out.println(ts.getOnNextEvents());
ts.assertValues(1, 2, 3, 4, 5);
ts.assertNotCompleted();
ts.assertError(CancellationException.class);

It should print [1, 2, 3, 4, 5] to the console and quit without any AssertionErrors. Neat, isn't it?

Conclusion

In this lenghtly and brain-stretching blog post, I've explained the requirements and problems around ConnectableObservables that want to do request coordination between its child Subscribers and its upstream Observable. I then showed an implementation of a publish() like ConnectableObservable which features disconnection strategy to avoid hanging its child Subscribers.

This is, however, not the most complicated operator in RxJava. It isn't replay(), even though the bounded version is a bit more complicated than the PublishConnectableObservable (but only due to the boundary management). It is not the most commonly used operator either and in fact, that is simpler due to fewer state-clashing. No, the most complicated operator to day has so intertwined request coordination that even I'm not sure it is possible to write a buffer-bounded version of it.

But enough of mysterious foreshadowing! In the next part, I'm going to detail what it takes to implement a replay()-like ConnectableObservable.

Operator internals: AutoConnect

Introduction

The operator autoConnect is a member of the ConnectableObservable class and allows triggering the connection to the underlying ConnectableObservable once the specified amount of subscribers have arrived. The operator returns a plain Observable and as such can be more easily included in a chain of operators.

There are two reasons why this operator exists. First, many wanted to connect to a ConnectableObservable only if a given number of subscribers have subscribed to it which, before that, was difficult to achieve due to the lack of confinement. The second reason was that another operator, cache, didn't support advanced retention policies such as size and/or time bounds and it was somewhat tedious to achieve the same first-subscriber triggered connection as cache does.

To sketch its implementation, one needs an AtomicInteger to count each subscriber in an OnSubscribe callback and once the count reaches the desired amount, the call to connect() can happen.

There is, however, a small complication: the synchronous unsubscription support in ConnectableObservable. By applying plain autoConnect, one loses the means to unsubscribe an ongoing stream, similar to how cache behaves. The resolution is to take a callback and hand it to the connect() method.

Implementation details

The operator is not involved in request management at all, therefore, the implementation in 1.x and the two implementation in 2.x (the other is on the NbpConnectableObservable) looks essentially the same.

In fact, it is so short I'm going to repeat it here:

public Observable<T> autoConnect(int numConnections,
        Action1<Subscription> connection) {               // (1)
    if (numConnection == 0) {
        connect(connection);                              // (2)
        return this;
    }
    AtomicInteger count = new AtomicInteger();
    return create(s -> {
        unsafeSubscribe(s);                               // (3)
        if (count.incrementAndGet() == numConnections) {  // (4)
            connect(connection);
        }
    });
}

Let's discuss the interesting points:

1. RxJava has two extra overloads of this method, one that defaults to 1 required connection and the other asks for the number of connections. Both ignore the connection callback.
2. If the number of connection is zero, we interpret it as an immediate connection. In this case, we don't have to do any kind of wrapping and just return the ConnectableObservable instance as is.
3. If the number of connection is non-zero, we have to capture the subscription attempts and do extra work once the number of subscribers reached the required amount. Before we even test for that, we subscribe the incoming Subscriber to the underlying ConnectableObservable. This is necessary to happen first because if the numConnections is 1, the connection may drain the underlying sequence and the Subscriber may not receive any values at all.
4. Once the subscriber count reached the required amount, we trigger a connection (that will call back the connection callback supplied originally.

At this point, you might think what happens if numConnections is 2, a Subscriber subscribes then unsubscribes immediately. Should the next Subscriber really trigger the connection? It depends on your requirements. The autoConnect operator, clearly, doesn't do this (a decrementAndGet() somewhere would indicate this). 

One reason for this is that originally, the operator was meant to be a simple replacement for using refCount() or share() in certain situations.

Conclusion

The operator autoConnect is among the simplest operators there are, 1 / 10, and thus has clear and simple feature set.

ConnectableObservables (part 1)

Introduction

We learned about constructing cold (i.e., range) and hot observables (i.e., UnicastSubject) but nothing specific so far about how to convert between the two.

Clearly, since subjects are also Observers, one only has to subscribe them to a cold source and let all the child Subscribers subscribe to the subject only.

But why would one do that in the first place? The conversion has one major benefit, namely it makes side-effects in the cold source happen once (per Subject subscribed to it). From a usage perspective, it means that you can reuse the same stream for multiple purposes and not having multiple and likely independent sequences.

For example, if you wanted to work on subsequent elements of the same stream, you could publish it and observe different parts of it through different subscriptions and combine the results:

Observable<Integer> source = Observable.range(1, 10);

ConnectableObservable<Integer> published = source.publish();

Observable<Integer> first = published;
Observable<Integer> second = published.skip(1);

Observable<String> both = first.zipWith(second, 
    (a, b) -> a + "+" + b);

both.subscribe(System.out::println);

published.connect();

Now let's see what ConnectableObservables should do.

ConnectableObservable requirements

ConnectableObservable is an abstract class that extends Observable and requires one extra method to be implemented.

By extending an Observable, it is subject to the same construction difficulties as are Subjects: namely their constructor requires an OnSubscribe callback which can't really access the outer class' methods at construction time so one needs a factory and an intermediate state object.

The second requirement, also coming from Observable, is that subscription should be thread safe and the implementation should allow it to happen any time, before, during and after the ConnectableObservable "runs".

It may come as a surprise that the extra abstract method isn't connect() but connect(Action1<Subscription> s) instead. The reason for this is due to the synchronous unsubscription possibility with a ConnectableObservable. But when does this come into play?

There are two cases when this feature is essential, one is more public and one is hidden away in certain operators.

The problem with connect() is that if it connects to an underlying cold and synchronous observable, that could run to completion (or never terminate) thus the method never returns. If you didn't have subscribers subscribed to it then those values may be gone forever. In addition, given an infinite synchronous stream, you may attempt to unsubscribe it via the Subscription returned by connect() after a while but then again, connect() never returns.

This comes up quite often with the second case mostly with multicasting operator overloads such as publish(Func1) and replay(Func1). These operators create a ConnectableObservable behind the scenes, run it through the Func1 provided and return a plain Observable which when subscribed to will connect the ConnectableObservable. Now if the source is synchronous and you want to take only a few elements of the returned Observable, the child subscription would never return a Subscription and the whole stream would just keep running.

The solution is to have the second, callback version of connect implemented which calls the callback with a subscription before it connects and thus allows it to be unsubscribed in sequence.

Finally, connection and disconnection has to be idempotent. It means that calling connect twice on a running stream should do nothing as well as calling unsubscribe on such stream twice should unsubscribe a running stream once. One extra thing to be careful with unsubscription is that if one unsubscribes a stream the connects again, the Subscription from the first connection should not affect the state of the second connection.

To summarize, ConnectableObservable has to

• be thread safe when subscribing to it at any time and from any thread,
• allow synchronous unsubscription at any time and from any thread and
• be idempotent in respect of connect and disconnect (unsubscribe).

A basic implementation

Given what we know about ConnectableObservables and Subjects so far, it may come trivial to implement the former with the help of the latter. Let's implement a ConnectableObservable which takes a subject of your chosing and "publishes" a source Observable's values through it.

public final class Multicast<T>
extends ConnectableObservable<T> {
    
    final Observable<T> source;
    final Subject<T, T> subject;
    
    final AtomicReference<Subscription> subscription;         // (1)
    
    public Multicast(Observable<T> source, 
            Subject<T, T> subject) {
        super(s -> {
            subject.subscribe(s);                             // (2)
        });
        this.source = source;
        this.subject = subject;
        this.subscription = new AtomicReference<>();
    }
    
    @Override
    public void connect(
        Action1<? super Subscription> connection) {
        // implement
    }
}

So far, nothing special. We take a source observable, a subject and we will hold the current connection in an AtomicReference instance (1). The OnSubscribe logic is this case is simple and there is no need for the factory approach unlike UnicastSubject: for each incoming subscriber, we subscribe them to the subject directly (2).

The body of the connect() method is a bit more involved but not too complicated:

@Override
public void connect(Action1<? super Subscription> connection) {
    for (;;) {
        Subscription s = subscription.get();                   // (1)
        if (s != null) {
            connection.call(s);                                // (2)
            return;
        }
        
        Subscriber<T> subscriber = new Subscriber<T>() {       // (3)
            @Override
            public void onNext(T t) {
                subject.onNext(t);
            }
            
            @Override
            public void onError(Throwable e) {
                subject.onError(e);
            }
            
            @Override
            public void onCompleted() {
                subject.onCompleted();
            }
        };
        
        subscriber.add(Subscriptions.create(() -> {            // (4)
            subscription.set(null);
        }));
        
        if (subscription.compareAndSet(null, subscriber)) {    // (5)
            connection.call(subscriber);                       // (6)
            
            source.subscribe(subscriber);                      // (7)
            
            return;
        }
    }
}

The implementation is basically a CAS loop:

1. We keep the current connection's Subscription in the subscription field and if it is not null, it means there is an active connection.
2. Given an active connection, we simply call the action with it.
3. Otherwise, there seems to be no active connection and we have to establish one. You may think, why not subscribe the subject directly to the source? The reason is the requirement of synchronous unsubscription: the call to subscribe() returns a Subscription too late, thus we need a Subscription before that. The Subscriber we create will forward events and also present this unsubscription possibility (remember, Subscriber extends Subscription).
4. When the connection, our Subscriber is unsubscribed, we have to set the subscription field back to null, allowing the next connect() to happen.
5. To achieve idempotence with a connect, we CAS in a subscriber in place of a null value. If it fails, due to a concurrent call to connect(), the loop resumes at (1).
6. If the CAS succeeded, we first call the callback with our Subscriber which will allow synchronous cancellation of the connection.
7. Finally, we subscribe our Subscriber to the source and quit.

Limitations of the basic implementation

The basic implementation seem to work but has some limitations.

Side note: With this blog, I hope to teach the reader how to detect bugs and shortcomings in operators; this is why some examples are not prepared for everything up front.

The first limitation is that if the source terminates, the clearing of the subscription may happen sometime in the future (through a SafeSubscriber) or not at all. The solution is to clear the subscription in the onError and onCompleted methods of our Subscriber, but we can't use set(null) there. We have to conditionally clear it there and in the regular unsubscription path because it is possible that subscription is cleared by the other party (termination vs. unsubscription race). In short, the methods should be changed like this:

    // ...
    @Override
    public void onError(Throwable e) {
        subject.onError(e);
        subscription.compareAndSet(this, null);
    }

    @Override
    public void onCompleted() {
        subject.onCompleted();
        subscription.compareAndSet(this, null);
    }
    // ...

subscriber.add(Subscriptions.create(() -> {
    subscription.compareAndSet(subscriber, null);
}));

In all three places, the clearing only happens if the current connection is still the known subscriber. This way, if there is a termination by any means followed by a reconnection, an unsubscribe() call to an old connection won't affect the new connection.

The second limitation is that once the source runs to termination, the Subject will come to its terminal state as well. New connection attempts will disconnect immediately and child Subscribers will only receive a terminal event (with the standards Subjects of RxJava).

Most likely this isn't what the business logic dictates, therefore, we have to change the parametrization of the Multicast so we can get a fresh Subject for any new connection. I'll show an implementation of this in the next subsection but before that, let's see the final limitation of the basic implementation.

The final limitation is that there is no request coordination: our Subscriber and the Subject itself will run in unbounded mode and ignore all backpressure requests. Since the standard RxJava 1.x Subjects don't support backpressure, we may run into MissingBackpressureExceptions somewhere in the downstream. Although 2.x Subjects are backpressure-aware, 2.x PublishSubject will still throw MissingBackpressureException if the child subscriber can't keep up and 2.x ReplaySubject does effectively unbounded buffering (similar to onBackpressureBuffer)

The resolution is a larger step up on the complexity ladder and will be detailed in the next part of this series about ConnectableObservables.

Fresh Subject on connect

The solution to the lack of reusability with the basic implementation can be solved by using a supplier function instead of a Subject instance and call it just before the connection happens.

This, however, creates another problem. Because the subject doesn't exist the time the constructor sets the OnSubscribe callback, we somehow have to remember the Subscribers that have attempted to subscribe when there was no connection yet but then subscribe to the actual Subject when there is a connection.

First, we now have to manage a more complex state. I'll create a Connection class that represents the state of a connection:

static final class Connection<T> {
    Subject<T, T> subject;                           // (1)
    List<Subscriber<? super T>> subscribers;         // (2)
    boolean connect;                                 // (3)
    final SerialSubscription parent;                 // (4)
    
    public Connection() {
        this.subscribers = new ArrayList<>();
        this.parent = new SerialSubscription();
    }
    
    public void setSubject(Subject<T, T> subject) {  // (5)
        // implement
        
    }
    
    public void subscribe(Subscriber<? super T> s) { // (6)
        // implement
    }
    
    public boolean tryConnect() {                    // (7)
        // implement
    }
}

Let's see its parts:

1. We have to store a Subject so subscribers can be subscribed to it any time.
2. Since the subject doesn't exist until connect() is called, we have to store the early birds in a list and subscribe them all once the subject becomes available.
3. The connection has to happen once per Connection object (termination or unsubscription then has to create an entirely new Connection object, see later).
4. We have to keep reference to the subscription to the source Observable. However, we can't just store a Subscriber because the connection process may longer at which a concurrent connection might found that reference to be still null (unlike the basic example where the Subscription was atomically established). The container is non null and ensures proper unsubscription on arrival if necessary.
5. We have to set a Subject once available and subscribe all early bird Subscribers to it.
6. We also have to provide a way for the OnSubscribe in the constructor to add new subscribers properly, depending on the current state of the connection.
7. Finally, connection has to happen once per Connection object which is managed by the tryConnect() method.

The implementation of the methods (5-7) are relatively simple but need some short explanation:

public void setSubject(Subject<T, T> subject) {
    List<Subscriber<? super T>> list;
    synchronized (this) {
        this.subject = subject;
        list = subscribers;
        subscribers = null;
    }
    for (Subscriber<? super T> s : list) {
        subject.subscribe(s);
    }
}

In this method, the subject is set while holding a lock on this. The reason for it is that to prevent concurrent subscribe() calls (see below) to happen while the Subject is set. This way, early bird Subscribers will be subscribed to the subject in this method whereas late subscribers will be directly subscribed to the subject once the unlock happens, skipping the list entirely. Subscribing the early birds outside the lock reduces the likelihood of deadlock and also doesn't block the concurrent subscribers while the loop is running.

Next comes the subscribe() method.

public void subscribe(Subscriber<? super T> s) {
    Subject<T, T> subject;
    synchronized (this) {
        subject = this.subject;
        if (subject == null) {
            subscribers.add(s);
            return;
        }
    }
    subject.subscribe(s);
}

What happens here is that, atomically, if the subject is still null (i.e., connect() hasn't been called yet), we add the subscriber to the inner list. If, however, the subject is non-null, we subscribe the Subscriber directly to it. One optimization here would be to have the subject field volatile and do a double-checked locking (since subject will be only set once).

Finally, the tryConnect() method is pretty simple:

public boolean tryConnect() {
   synchronized (this) {
        if (!connect) {
            connect = true;
            return true;
        }
        return false;
    }
}

Atomically check if the connection is false and switch it to true. If this switch happened, return true, otherwise return false. The former will trigger the connection logic while the latter will simply "return" the parent SerialSubscription in connect() (detailed later).

I'll call the new ConnectableObservable MulticastSupplier and it will have the following skeleton:

public final class MulticastSupplier<T> 
extends ConnectableObservable<T> {
    public static <T> MulticastSupplier<T> create(        // (1)
            Observable<T> source, 
            Supplier<Subject<T, T>> subjectSupplier) {
        AtomicReference<Connection<T>> conn = 
            new AtomicReference<>(new Connection<>());    // (2)
        
        return new MulticastSupplier<>(
            source, subjectSupplier, conn);
    }
    

    
    final Observable<T> source;
    final Supplier<Subject<T, T>> subjectSupplier;
    final AtomicReference<Connection<T>> connection;      // (3)
    
    
    protected MulticastSupplier(Observable<T> source, 
            Supplier<Subject<T, T>> subjectSupplier,
            AtomicReference<Connection<T>> connection) {
        super(s -> {
            Connection<T> conn = connection.get();        // (4)
            conn.subscribe(s);
        });
        this.source = source;
        this.subjectSupplier = subjectSupplier;
        this.connection = connection; 
    }

    void replaceConnection(Connection<T> conn) {          // (5)
        Connection<T> next = new Connection<>();
        connection.compareAndSet(conn, next);
    }
    
    @Override
    public void connect(Action1<? super Subscription> connection) {
        // implement
    }
}

Let's see why this looks like as it is:

1. Since the constructor doesn't allow access to instance field before super is called, we have to create the Connection state before the instantiation of MulticastSupplier so both its body and the OnSubscribe callback can access it. 
2. Since the connection is not constant (can be reconnected any number of times), we have to use a holder for the connection, an AtomicReference in this case. It will come in handy when the state changes have to happen atomically. In addition, since Subscribers have to be remembered before a connection is established, we can't use a null state anymore.
3. The same AtomicReference has to be accessible from connect() later on.
4. The OnSubscribe callback is slightly modified: we retrieve the current Connection instance and call subscribe() on it. If it is connected, it will go straight and subscribe to the underlying Subject, otherwise the Subscriber will be remembered.
5. Finally, if either a disconnect or source termination happens, we have to replace the old connection with a fresh one so connect() can start again. The implementation first creates a new (empty) Connection and tries to CAS it in, replacing the known Connection when the last connect() has established it. This will prevent old Subscriptions to disconnect newer connections.

Finally, let's see the connect() implementation:

@Override
public void connect(
        Action1<? super Subscription> connection) {

    Connection<T> conn = this.connection.get();          // (1)
    
    if (conn.tryConnect()) {                             // (2)
        Subject<T, T> subject = subjectSupplier.get();
        
        Subscriber<T> parent = new Subscriber<T>() {     // (3)
            @Override
            public void onNext(T t) {
                subject.onNext(t);
            }
            
            @Override
            public void onError(Throwable e) {           // (4)
                subject.onError(e);
                replaceConnection(conn);
            }
            
            @Override
            public void onCompleted() {
                subject.onCompleted();
                replaceConnection(conn);
            }
        };
        
        conn.parent.set(parent);                         // (5)
        
        parent.add(Subscriptions.create(() -> {          // (6)
            replaceConnection(conn);
        }));
        
        conn.setSubject(subject);                        // (7)
        
        connection.call(conn.parent);                    // (8)
        
        source.subscribe(parent);                        // (9)
    } else {
        connection.call(conn.parent);                    // (10)
    }
}

The implementation no longer has a CAS loop because the atomic connection requirement is handled a bit differently:

1. First, we retrieve the current Connection object from the AtomicReference.
2. If not connected, set the state to connected and perform the connection logic, otherwise, go to (10).
3. Once we got a Subject, we create our wrapper Subscriber as before that references the Subject.
4. We'd like to disconnect eagerly once a terminal event has been received, therefore, we call replaceConnection() with the known connection object. Depending on what kind of races one wish to tolerate, you can swap the call on the subject with the replacement: this way, subscribers racing with the termination event will be added to the next connection instead of the current.
5. We then set the parent onto the connection. If there was a concurrent disconnect, this will unsubscribe the parent Subscriber immediately. Depending on how a connect-disconnect race should be handled, one can quit if isUnsubscribed() is true, don't even try to subscribe to the source but return a unsubscribed Subscription or retry the connection attempt. The latter requires a similar loop as in the basic example.
6. We set the unsubscribe action to replace the connection.
7. We set the subject on the current connection, which may trigger the early birds' subscription to the Subject.
8. Before the parent Subscriber is connected, we call the callback with the SerialSubscription (not the Subscriber!) to allow synchronous cancellation.
9. Then we subscribe the parent Subscriber to the source Observable.
10. If the tryConnect() returned false, we simply call the callback with the SerialSubscription of the current connection. Note here that this has to be non-null and thus the need of indirection around the parent Subscriber of (3).

Conclusion

In this blog post, I've detailed the requirements of ConnectableObservables and showed two simpler variants of implementing one.

However, one would expect request coordination from a ConnectableObservable which neither of the Multicast or MulticastSelector supports.

Looking at them wasn't in vain, because they feature construction approaches that will come in handy with the next part of this mini-series.

So far, operators and classes were low to medium complexity due to the fact that the event and method call "streams" were not really stepping on each other. Next, however, we step up on the complexity ladder and look at how one can coordinate requests within a ConnectableObservable.

This is, in my opinion, a master-level implementation task and if understood, it opens the door to the most complex operator implementations in RxJava. Stay tuned!

Operator internals: All, Any, Exists

Introduction

The operator all checks if a given condition (predicate) holds for all elements of the upstream, emitting a single true value at the end, or emits false immediately if the predicate returns false. The operator any is its logical inverse and looks quite like all, except it returns immediately if the predicate returns true and returns false for empty upstreams. They can and do support backpressure.

We need to consider the following properties/requirements with this operator:

• Since the output is a single value, one doesn't need to play around with request accounting and can let the operator request Long.MAX_VALUE from upstream. This gives the added benefit that it may trigger a fast-path and thus run with reduced overhead.
• Since the output is a single value, emitted even if the upstream is empty, one has to prepare for handling request amounts from downstream and only then emit the result.

Implementation: 1.x

The 1.x implementation is straightforward. The Subscriber requests Long.MAX_VALUE and uses the SingleDelayedProducer to delay the emission of the resulting boolean until the downstream actually requests.

Since backpressure handling is optional in 1.x, one can't emit false in case the predicate returns false in onNext because only the SingleDelayedProducer knows if there was actually an request call or not.

Implementation: 2.x

The 2.x implementation is a bit longer because I chose to inline the behavior of the SingleDelayedProducer and thus saving on allocation costs.

The backpressure requirement still holds but with one exception: since the call to onSubscribe is mandatory, onNext is only ever called if there was a request to it. Therefore, the operator has to insert itself between the upstream and the downstream request-wise.

Failing the predicate in onNext no longer requires buffering of the value but can be simply emit directly (because we know there was at least a request(1) beforehand). An empty upstream, however, still requires "buffering" the result until a request comes along. The related state machine is quite similar to the one described in an earlier post. The notable difference is that we know the delayed emission will always emit true thus no need for an instance variable holding it until needed.

It is worth looking at the onNext() method in AllSubscriber:

@Override
public void onNext(T t) {
    if (done) {                             // (1)
        return;
    }
    boolean b;
    try {
         b = predicate.test(t);
    } catch (Throwable e) {                 // (2)
         lazySet(HAS_REQUEST_HAS_VALUE);
         done = true;
         s.cancel();
         actual.onError(e);
         return;
    }
    if (!b) {
        lazySet(HAS_REQUEST_HAS_VALUE);     // (3)
        done = true;
        s.cancel();
        actual.onNext(false);
        actual.onComplete();
    }
}

1. Cancellation is best effort in both Reactive-Streams and 1.x Observables and one can't rely upon the cancellation alone. The done flag drops all events after the termination/cancellation of the operator.
2. Callbacks can crash, in which case we set the state-machine to its terminal value HAS_REQUEST_HAS_VALUE which should prevent any value emission in a request call. In addition we set the done flag and call cancel on the Subscription.
3. If the predicate returned false, we can shortcut the stream by cancelling it and emitting the constant false as the result. Here, the state machine is also brought to its terminal state.

Conclusion

The all and any operators is among the simpler operators, 2/10 maybe, but one needs to recognize an empty upstream would overflow the downstream in a naive implementation and thus there is a need for the SingleDelayedProducer to bridge the gap.

Operator internals: Amb, AmbWith

Introduction

The amb operator, shorthand for ambiguous, subscribes to a set of source Observables and keeps relaying events from the first Observable that signaled any event while unsubscribing and ignoring the rest. The operator can and does support backpressure.

From the operator building's perspective, we need to consider the following properties/requirements:

• The number of source Observables is known when the downstream subscribes.
• We need to track all subscriptions in a collection other than CompositeSubscription because there is no way to cherry pick one and unsubscribe the rest.
• One has to relay the downstream request to all sources.
• Unsubscription, requesting and even the choosing of a "winner" may happen while subscribing to the source Observables.

The two major versions are implemented slightly differently.

Implementation: 1.x

The 1.x implementation is slightly more verbose. It uses a simple ConcurrentLinkedQueue to keep track of the subscribers, AmbSubscriber. In addition, the winner AmbSubscriber is kept in an AtomicReference.

In case an unsubscription happens, we need to attach a callback to the child subscriber which when called, will loop through the collection of AmbSubscribers and unsubscribes them one by one.

When the subscription happens, a loop goes through all available sources, instantiated an AmbSubscriber and subscribes. Since unsubscription can happen at any time or any previous source may have already won, the loop has to check for both condition and quit early.

Once all sources have been subscribed, the child subscriber receives its Producer. The task of this producer is to dispatch all requests to every AmbSubscriber, or in case of a winner, dispatch the request only to that particular AmbSubscriber.

It may seem odd, but if there is a winner before the Producer is set, there is no need to set the Producer because the winner has obviously ignored any backpressure and started emitting anyways.

In the AmbSubscriber, any event fired will check if the current AmbSubscriber is the winner or not. If so, the event is relayed. Otherwise, an atomic dance happens where the AmbSubscriber tries to CAS itself into the winning position and if successful, it unsubscribes the others. If the CAS failed, the AmbSubscriber unsubscribes itself.

Implementation: 2.x

The 2.x implementation is less verbose and exploits the fact that the number of source Observables is known. Therefore, an array of AmbInnerSubscriber is used and the "winner" indicator is now an volatile integer field backed by a field updater. 

When the child subscribes, a loop first creates every AmbInnerSubscriber, sets a custom Subscription on the child (which is the coordinator class itself) and then subscribes to each source Observable. This second loop also checks for a winner in the process.

The winner field has multiple meanings depending on the state of the operator. Minus one indicates the child cancelled, zero means there is no winner yet and any positive number indicates the index plus 1 of the winner AmbInnerSubscriber.

In the reactive-streams world, there is an increased likelihood a Subscription arrives later than any request or cancellation attempt, therefore, one has to be prepared for it. Therefore, AmbInnerSubscriber has to keep its Subscription in a volatile field plus it has to track all the missed requests in another. This pattern is so common with 2.x, it is worth detailing it here:

class AmbSubscriber<T> 
extends AtomicReference<Subscription>
implements Subscriber<T>, Subscription {
    volatile long missedRequested;
    static final AtomicLongFieldUpdater MISSED_REQUESTED = ...;

    static final Subscription CANCELLED = ...;
}

The class implements Subscriber, naturally, and Subscription for convenience (so we have request() and cancel() to implement). The class also has a static final field holding an empty implementation of the Subscription interface. We will use this instance to indicate a cancelled state and also notify any late-coming request() or onSubscribe() to do nothing. By extending AtomicReference directly, we will keep the incoming Subscription in a (hidden) instance field and access it via atomic methods of this.

Let's see the implementation of the cancel() method first:

@Override
public void cancel() {
    Subscription s = get();
    if (s != CANCELLED) {
        s = getAndSet(CANCELLED);
        if (s != CANCELLED && s != null) {
            s.cancel();
        }
    }
}

This atomic getAndSet() should look familiar by now. When called, if the current subscription is not the constant CANCELLED, we getAndSet it to cancelled. The atomicity guarantees that there will be only one thread that experiences a non-CANCELLED previous state in which case we call cancel on it. Note that cancel() may be called before onSubscribe thus the current subscription may be null.

Next, let's see the onSubscribe() method:

@Override
public void onSubscribe(Subscription s) {
    if (!compareAndSet(null, s)) {                         // (1)
        s.cancel();                                        // (2)
        if (get() != CANCELLED) {                          // (3)
            SubscriptionHelper.reportSubscriptionSet();
        }
        return;
    }
            
    long r = MISSED_REQUESTED.getAndSet(this, 0L);         // (4)
    if (r != 0L) {                                         // (5)
        s.request(r);
    }
}

1. First, we try to CAS in the incoming Subscription and replace a null value.
2. If there is already a Subscription, we cancel the incoming one in any case.
3. It is possible, although unlikely, multiple calls to onSubscribe happens due to bogous source. If the current value isn't the cancelled indicator, we have to report the incident in some way and just quit.
4. If the CAS succeeded, we now have to take all missed requested amount via getAndSet.
5. If there were in fact missed requests, we do request that amount from the Subscription at hand.

Finally, let's look at the request() method:

@Override
public void request(long n) {
    Subscription s = get();
    if (s != null) {                                       // (1)
        s.request(n);
    } else {
        BackpressureHelper.add(MISSED_REQUESTED, this, n); // (2)
        s = get();
        if (s != null && s != CANCELLED) {                 // (3)
            long r = MISSED_REQUESTED.getAndSet(this, 0L); // (4)
            if (r != 0L) {                                 // (5)
                s.request(r);
            }
        }
    }
}

1. When the request is called, first we check if the current Subscription isn't null. If so, we request the amount directly. The current Subscription might be the CANCELLED instance in which case this call is a no-op.
2. We use the backpressure-helper routine to safely add the number to the missedRequested field (which caps at Long.MAX_VALUE). 2.x Bug: validation of n is missing here.
3. Once the missed amount has been added, we need to check the Subscription again since it might have been set asynchronously. 
4. If not null and not cancelled, we call getAndSet the missed amount. This makes sure the missed amount is either retrieved by this method or by the onSubscribe method atomically.
5. If the missed amount is non-zero, we request it from the Subscription. Otherwise, the onSubscribe has already taken any missed value for us.

The other onXXX methods of the AmbInnerSubscriber work similarly to the 1.x version. There is a local won field (no need for volatile) that if set, serves as a fast-path for delivering events. If it is false, there is an attempt to win the race and if won, the won field is set to true. Otherwise the AmbInnerSubscriber cancels the Subscription at hand (which shouldn't be null at this point as RS requires calling onSubscribe before any other onXXX methods).

2.x Bug: If the AmbSubscriber wins, it doesn't cancel the other AmbSubscribers and thus they remain subscribed indefinitely.

Conclusion

The operator amb isn't a complicated operator, 5/10 maybe, but it requires some custom logic to deal with unsubscription/cancellation and request dispatching.

While reviewing the 2.x implementation, I found two oversights that can be easily addressed via a PR.

Operator internals: introduction

Developing an operator is usually a non-trivial task. In the Advanced RxJava blog, I've tried to convey many of the foundational elements and experience I've got from building them.

However, RxJava has around 150 unique operators and many of them required some custom logic or "unconventional" idea. Such knowledge is hard or next to impossible to explain in any meaningful generalization.

Therefore, I'll start a parallel series where I'm going to dive into each operator, except perhaps the most trivial ones. I'll call all of them operators, regardless of whether they implement Operator or OnSubscribe for convenience. I'll go by name ascending and cover aliases of an operator into the same post.

I'll look into both 1.x and 2.x implementations which has the added benefit of doing an effective review of them while also pointing out what it takes to make said operator Reactive-Streams compliant.