如何在反应式编程中抑制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()，我希望重复此完全相同的日志，而不打印任何内容。

显然，当冷事件观察者触发complete事件时，它会自动取消订阅所有订阅my proxiedObservable。

我不需要complete事件，但是我确实需要新的MyData才能访问所有订阅。

有没有办法抑制onComplete事件？还是有另一种方法可以在这里实现我的需求？

Answer 1

这个问题是RxRelay着手解决的-一个Subject可以接收的信号onComplete导致其关闭。

如果您不喜欢添加此依赖项，则可以使用use materialize()+ 从序列中filter()删除onComplete。