Механизм ворот RxJava

Question

Я хочу реализовать что-то вроде механизма ворот. Мне нужен один PublishSubject и пара подписчиков. Когда PublishSubject отправляет данные через onNext, только один подписчик получит их.

Например: У меня 3 равно фрагментам внутри вкладок. У них есть подписка на глобальную публикацию под названием onLoginPublisher . Когда onResume или onPause , вызываемый gate становится открытым или закрытым. Когда вызывается onLogin и ворота не открываются из-за отсутствия на экране ни одного из этих фрагментов, onNext будет ожидать onResume фрагмента

Посмотрите на картинку:

Answer 1

Вы можете использовать filter с состоянием ворот. Например, вы можете обернуть всю логику в класс:

public final class GatedSubject<T> {
    final PublishSubject<T> subject = PublishSubject.create();
    final AtomicReferenceArray<Boolean> gates;

    public GatedSubject(int numGates) {
        gates = new AtomicReferenceArray<>(numGates);
    }

    public boolean getGateStatus(int gateIndex) {
        return gates.get(gateIndex) != null;
    }

    public void setGateStatus(int gateIndex, boolean status) {
        gates.set(gateIndex, status ? Boolean.TRUE : null);
    }

    public void Observable<T> getGate(int gateIndex) {
        return subject.filter(v -> getGateStatus(gateIndex));
    }

    public void onNext(T item) {
        subject.onNext(item);
    }

    public void onError(Throwable error) {
        subject.onError(error);
    }

    public void onComplete() {
        subject.onComplete();
    }
}