java - RxJava 和长时间运行的监听器

Question

好吧，我是 RxJava2 的新手(嗯，我也不了解 RxJava)，并且正在尝试使用 RxJava2 和 MVP 结构开发 Android 应用程序。

在该应用程序中，我正在对使用监听器的库进行异步调用。我使用“标准”setListener/registerListener 方法设置监听器。

其中一个方法是返回“实时”值 -> 我调用库的 start() 方法，然后在每次修改列表时(当添加/删除列表时)都会在监听器上收到通知项目)。

我真的不明白如何使用 RxJava 实现这种行为，因为监听器是在发射器/订阅者的定义中订阅的？我应该在哪里声明监听器？我应该在哪里取消订阅？我应该使用什么对象？

我使用 Nucleus 启动了开发，但可以切换到另一个样板或自己做一个。

这是一些伪代码来说明我的问题:

之前

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

    mMyLib.setListener(this);
    mMyLib.startDiscovery();
}

@Override
public void itemListChanged(List<Dummy> items) {
    // update the UI with the list items
}

@Override
protected void onDestroy() {
    super.onDestroy();
    mMyLib.setListener(null);
}

在我的演示器中使用 Nucleus

如果我想在我的 Activity/演示者还活着的情况下接收列表的修改，我应该在哪里取消订阅？我是否使用了正确的语法/对象？

private static final int REQUEST_ITEMS = 1;
private PublishSubject<Integer> pageRequests = PublishSubject.create();

...

@Override
public void onCreate(Bundle savedState) {
    super.onCreate(savedState);

    restartableReplay(REQUEST_ITEMS,
            () -> Observable.create(e ->
                    {
                        mMyLib.setListener(new MyLib.Listener() {
                            @Override
                            public void itemListChanged(List<Dummy> items) {
                                Log.d(TAG, "meh itemListChanged");
                                e.onNext(items);
                                e.onComplete();
                            }
                        });
                        mMyLib.startDiscovery();
                    }
            )
            ,
            MainFragment::onItems,
            MainFragment::onNetworkError);
}

void request() {
    start(REQUEST_ITEMS);
}

Answer 1

你的方向是正确的，这是用 RxJava 包装异步回调的有效方法，一些评论:

• 您在 itemListChanged 上调用 e.onComplete()，这是错误的，因为它会结束您的 Observable 序列，您可能根本不需要调用 onComplete() ，因为它永远不会结束来自外部源的通知，或者在通知真正结束时调用它一次(外部源完成生成项目)，不要将其与取消订阅源。
• 对于要执行某些操作的取消订阅逻辑，您应该在 create() 中定义它，并使用您的取消逻辑 (mMyLib.setListener (null) 或任何其他清理资源代码)
• 在这种情况下，您似乎只有 1 个订阅者，但在其他方面考虑使用 share()，以获得一个可以多播到多个的“热门”Observable订阅者。
• 至于 Nucleus 库，据我记得 restartableReplay，将缓存您的 Observable 并重播发出的项目，这可能是这种“热”事件流是错误的(除非您需要重播上次发射或其​​他什么)，此外，缓存也会出现问题，因为您将失去取消订阅的能力，所以请务必在此处使用 Nucleus。<

Where should I unsubscribe if I want to receive modifications of the list as long as my activity / presenter are alive? Am I even using the right syntax/ objects?

您只需在想要停止接收通知的地方取消订阅即可，无论是在 onStop() 还是 onDestory() 取决于您。