RxJava。订阅开始时的初始下一步？

Question

我正在尝试实现一个使用可观察的发出更改的类。当对此可观察对象完成订阅时，我想发送一个启动/初始化事件。然后我想发送通常的事件。

例如假设我有两个不同的订户A和B。A和B在不同的时间开始订阅。如果是MyClass。getChanges（）发出事件1、2、3、4和5。

如果A在事件1、2之间启动it订阅，则应接收以下事件：InitialEvent、2、3、4、5。

如果B在事件4和5之间启动it订阅，则B应接收以下事件：InitialEvent，5。

如何使用RxJava做到这一点？

谢谢

编辑1

我想我需要解释一下，“InitialEvent”每次发出时都是不同的。每次新订阅者开始从getChanged（）订阅时，它都由MyClass计算。

我的场景是MyClass包含一个列表。“initialEvent”包含订阅完成时的列表。然后，对该列表的每个更改都从getChanges（）发出。

Answer 1

可能不是很优雅的方式仅仅使用一面旗帜怎么样？看起来您只想替换第一个发出的事件。

e、 g.对于一个订阅，以下逻辑：

boolean firstTimeA = true;

myCustomObservable.subscribe(s -> {
   System.out.println(firstTimeA ? "initEvent" : s.toString());
   if(firstTimeA) firstTimeA = false;
});

由于您想拥有第二个订阅，只需创建firstTimeB并将其更新为您的B订阅。

Answer 2

你要找的是PublishSubject。受试者是热门的可观察对象，因为他们不会等到观察者订阅后才开始发出他们的项目。这里有一些关于主题的信息。

这是您的用例的快速演示

    PublishSubject<String> subject = PublishSubject.create();
    Observable<String> InitEvent = Observable.just("init");
    Observable<String> A = subject.asObservable();
    Observable<String> B = subject.asObservable();

    subject.onNext("1");

    A.startWith(InitEvent)
            .subscribe(s -> System.out.println("A: " + s));

    subject.onNext("2");
    subject.onNext("3");
    subject.onNext("4");

    B.startWith(InitEvent)
            .subscribe(s -> System.out.println("B: " + s));

    subject.onNext("5");

Answer 3

很抱歉在两年后发布此消息，但我也有同样的需要，发现这个问题没有答案。

我所做的是：

public Observable<Event> observe() {
    return Observable.defer(() -> 
        subject.startWith(createInitialEvent())
    );
}

想法如下：

• deer（）在观察者订阅由观察（）方法返回的可观察对象时执行传入的lambda表达式。所以基本上，它执行subject.start（...），它返回一个可观察对象，它是订阅者的实际事件源。
• subject.startwith（...）发出一个初始事件（由start To（...）指定），后跟主题发出的事件。

因此，如果我回到原来的帖子：如果观察者在事件1,2之间启动它的订阅，那么它应该接收以下事件：InitialEvent，2，3，4，5。