rx-java - 如何在响应式(Reactive)编程中抑制onComplete？ (RxJava)

Question

我的感冒 Observable 超出了我的控制范围。订阅时它总是产生(如果没有错误):

onNext()

onCompleted()

我有一个像这样的助手类:

public class Helper {

    private Subscription subscription;
    private ReplaySubject<MyData> subject;
    private Observable<MyData> coldObservable;

    public HelperClass(Observable<MyData> aColdObservable) {
        coldObservable = aColdObservable;
        subject = ReplaySubject.create(1);
        triggerRefresh();
    }

    public Observable<MyData> getObservable() {
        return subject.asObservable();
    }

    public void triggerRefresh() {
        if (subscription != null) {
            subscription.unsubscribe();
            subscription = null;
        }
        subscription = coldObservable
                              .subscribeOn(Schedulers.io())
                              .observeOn(Schedulers.computation())
                              .subscribe(subject);
    }
}

我对该主题有多个订阅，请为客户端获取此伪代码:

public class Client {

    private final Helper helper;
    private Observable<MyData> proxiedObservable;

    public Client(Observable<MyData> coldObservable) {
        helper = new Helper(coldObservable);
    }

    public void init() {
        proxiedObservable = helper.getObservable()
                                                        .subscribeOn(Schedulers.io)
                                                        .observeOn(Schedulers.computation());

        Subscription a = newSubscription("A");
        Subscription b = newSubscription("B");
        Subscription c = newSubscription("C");
    }

    public Subscription newSubscription(String name) {
        return proxiedObservable.subscribe(data -> log("next " + name),
                                          throwable -> log("error " + name),
                                          () -> log("complete " + name));
    }

    public void refresh() {
        helper.triggerRefresh();
    }
}

在初始化时，这会打印在日志中:

next A
complete A
next B
complete B
next C
complete C

在调用 refresh() 之后的某个时刻，我希望重复完全相同的日志，而不是打印任何内容。

显然，当冷观察者触发完整事件时，它会自动取消对我的proxiedObservable的所有订阅。

我不需要完整的事件，但我确实需要新的 MyData 来到达所有订阅。

有没有办法抑制 onComplete 事件？或者还有其他方法可以实现我在这里所需要的吗？

Answer 1

这个问题是什么RxRelay着手解决 - Subject 可以接收 onComplete 导致其关闭。

如果您不习惯添加此依赖项，那么您也可以使用 materialize() + filter() 删除 onComplete从序列中。