RxJava门机制

Question

我想实现一些类似大门机制的东西。我需要一个PublishSubject和几个订阅者。当PublishSubject通过onNext发送数据时，只有一个订阅服务器将接收数据。

例如：我有3个等于片断在标签。他们订阅了全局发布的OnLoginPublisher。当ONREME或onPause时，CALL gate变为打开或关闭。当onLogin被调用并且由于屏幕上没有这些片段而没有打开任何门时，onNext将等待fragment的onremume

Answer 1

您可以将过滤器与门的状态一起使用。例如，您可以将所有逻辑封装到一个类中：

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();
    }
}