just() 在订阅和 flatmap() 中不起作用

Question

想法是订阅

订阅 sbscribe 订阅，当他们获得一个值时，他们会自动取消订阅。

但不知何故，onComplete() 没有被调用，订阅者继续接收消息。

测试示例：

public class MyActivity extends Activity {
    private final Subject<String, String> mStringSubject = new      
            SerializedSubject<>(PublishSubject.create());


    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        Handler h = new Handler();

        h.postDelayed(() -> mStringSubject.onNext("Hello1"), 500);
        h.postDelayed(() -> mStringSubject.onNext("Hello2"), 1000);

        mStringSubject.flatMap(Observable::just)
            .subscribe(s -> Log.e("MainActivity", s),
                    Throwable::printStackTrace,
                    () -> Log.e("MainActivity", "Completed"));
    }
}

我在日志中看到

Hello 1
Hello 2

但我预计：

Hello 1
Completed

Answer 1

没有任何东西可以从 mStringSubject.flatMap(Observable::just) 取消订阅，因此订阅者会收到 mStringSubject 发出的所有内容。如果您想确保每个订阅者只获得一个值，请将 take(1) 应用于源：

Observable<String> o = mStringSubject.flatMap(Observable::just).take(1);

o.subscribe(...)
o.subscribe(...)

顺便说一句，mStringSubject.flatMap(Observable::just)只是增加了开销，相当于mStringSubject。

just() 在订阅和 flatmap() 中不起作用

just() doesn't work in subscription and flatmap()

rx-java