在RXJava2中是否有等效的doAfterSubscribe？

Question

    PublishSubject<String> publishSubject = PublishSubject.create();
    Observable<String> fooObservable = publishSubject.filter(value -> value.startsWith("foo"))
            .doOnSubscribe(disposable -> {
                publishSubject.onNext("foobar");
            })
            .replay(1)
            .refCount();

    fooObservable.subscribe(val -> log.info("A val : <{}>", val));

A val : <foobar>

请注意，我使用以下代码得到了相同的结果：

PublishSubject<String> publishSubject = PublishSubject.create();
Observable<String> fooObservable = publishSubject.filter(value -> value.startsWith("foo"))
        .replay(1)
        .refCount();

publishSubject.onNext("foobar");
fooObservable.subscribe(val -> log.info("A val : <{}>", val));

所以问题是，fooObservable直到订阅了PublishSubject之后才订阅PublishSubject,

是否有一种方法可以在第一次订阅FooObservable之后立即运行代码？

PublishSubject<String> publishSubject = PublishSubject.create();
BehaviorSubject<String> fooObservable = BehaviorSubject.create();
publishSubject.filter(value -> value.startsWith("foo")).subscribe(fooObservable);
publishSubject.onNext("foobar");
fooObservable.subscribe(val -> log.info("A val : <{}>", val));

public Observable<Response> getObservable(Request req);

如果请求与已经订阅的请求匹配，那么observable应该在订阅时立即提供最新的匹配值。

当没有订阅者时，我需要取消我包装的服务的订阅。

Answer 1

我认为您需要publish()和connect()。在这里你可以读到更多关于它的信息。

在你的情况下，它会是这样的：

PublishSubject<String> publishSubject = PublishSubject.create();
Observable<String> fooObservable = publishSubject
    .filter(value -> value.startsWith("foo"))
    .publish();

fooObservable.subscribe(val -> log.info("A val : <{}>", val));
Observable<Object> o2 = fooObservable.map { new Object() }
Observable<Object> o3 = fooObservable.map { /* Something here*/ }

Disposable disposable = fooObservable.connect()

但请记住执行disposable.dispose()以避免泄漏

BehaviorSubject<String> publishSubject = BehaviorSubject.create();
Observable<String> fooObservable = publishSubject.filter(value ->         
value.startsWith("foo"));
fooObservable.subscribe(val -> log.info("A val : <{}>", val));
publishSubject.onNext("foobar");