Rxjava - emit on PublishSubject before subscribe

Question

Consider a scenario where we have a stream emitting strings, and we want to save strings in files.

I am using PublishSubject and this works fine:

Subject<String> stream = PublishSubject.create();
stream.subscribe(str -> saveFile(str));
mReverseGeocoderStream.onNext("some-string1")
mReverseGeocoderStream.onNext("some-string2")

However, this doesn't work (only some-string2 gets delivered)

Subject<String> stream = PublishSubject.create();
mReverseGeocoderStream.onNext("some-string1")
stream.subscribe(str -> saveFile(str));
mReverseGeocoderStream.onNext("some-string2")

Is there a way to make the second scenario work as well?

i.e., Can we change PublishSubject to make sure it buffers the events until a subscriber consumes them?

Note that BehaviorSubject is not an option because re-subscribing causes another file save. It doesn't have the concept of "consuming events".

I found UnicastSubject which is pretty much what I want, except, it fails with IllegalStateException when I unsubscribe and later re-subscribe with a different subscriber.

Use case:

Assume we have an android app. It makes a network request, at the back of the network request, it needs to show a dialog. While the request is being made, the user backgrounds the app. At this point, we unsubscribe our observers that are listening for the signal to show the dialog.

Network request comes back and signal to show the dialog gets fired to the stream. No one is listening at this point. User foregrounds the app. A new subscriber gets attached to the network request manager (ViewModel). At this point, I'd like the "unconsumed" signal to get delivered to the subscriber.

Note: I cannot use the behavior subject. If I do that, every time user backgrounds and foregrounds the app the dialog will show up. I want the event to be consumed and finished once the dialog is shown.

The first part of your question made me think that for sure you needed a ReplaySubject, but then you mention that you want to unsubscribe and subscribe again in different points of your program and get the emissions but ignore them (not save them in the file). That's quite confusing. I feel the problem is not related to the type of subject. Could you explain further what is what you want to achieve? — dglozano
I've added a use case. Let me know if it's clear or not. I've also added a potential solution. Please let me know if you have a better solution. — Kasra
Hmm not sure if this will help or if it is a better solution or not, but I came across a similar use case while using LiveData in my Android app. The suggested approach I found is to create a wrapper classed called for example Event<T> like the following: gist.github.com/dglozano/dd02efa15d070c5517c9e62e212ecd24 . Then, you can use a Behaviour subject that will emit the Event everytime, but you will only process it if it hasn't been handled yet. — dglozano
No, this is unfortunately antipattern as per RxJava standard. Objects should be immutable. Once they are emitted, they are not supposed to be altered. — Kasra

Kasra Kasra · Accepted Answer · 2019-05-02T19:15:53

Did some more researching and found this:

https://gist.github.com/xsveda/8c556516079fde97d04b4b7e14a18463

from : Queue like Subject in RxJava

Note that it uses Relay, but you can easily replace Relay with Subject if you don't want to bring in another dependency.

I am open to other solutions and perhaps criticising why this solution is not a good one.

/**
 * Relay that buffers values when no Observer subscribed and replays them to Observer as requested. Such values are not replayed
 * to any other Observer.
 * <p>
 * This relay holds an unbounded internal buffer.
 * <p>
 * This relay allows only a single Observer at a time to be subscribed to it.
 * <p>
 * If more than one Observer attempts to subscribe to this Relay at the same time, they
 * will receive an IllegalStateException.
 *
 * @param <T> the value type received and emitted by this Relay subclass
 */
public final class CacheRelay<T> extends Relay<T> {

    private final ConcurrentLinkedQueue<T> queue = new ConcurrentLinkedQueue<>();
    private final PublishRelay<T> relay = PublishRelay.create();

    private CacheRelay() {
    }

    public static <T> CacheRelay<T> create() {
        return new CacheRelay<>();
    }

    @Override
    public void accept(T value) {
        if (relay.hasObservers()) {
            relay.accept(value);
        } else {
            queue.add(value);
        }
    }

    @Override
    public boolean hasObservers() {
        return relay.hasObservers();
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        if (hasObservers()) {
            EmptyDisposable.error(new IllegalStateException("Only a single observer at a time allowed."), observer);
        } else {
            for (T element; (element = queue.poll()) != null; ) {
                observer.onNext(element);
            }
            relay.subscribeActual(observer);
        }
    }
}

Rxjava - emit on PublishSubject before subscribe

1 Answers