RxJava и долгоиграющий слушатель

Итак, я новичок в RxJava2 (ну, я тоже не знаю RxJava) и пытаюсь разработать приложение для Android, используя структуру RxJava2 и MVP.

В этом приложении я делаю асинхронные вызовы к библиотеке, которая использует прослушиватели. Я устанавливаю слушателя, используя «стандартный» метод 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 в моем презентере

Где я должен отказаться от подписки, если я хочу получать изменения в списке, пока мои действия/ведущие активны? Использую ли я правильный синтаксис/объекты?

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);
}