Как проверить наличие записи в Rxjava2?

Question

У меня есть реализация с реактивным программированием, и я использую Springboot Framework, Rxjava2, среду реактивных данных Spring.У меня есть сценарий сохранения записи в mongodb, когда запись не существует.

Но когда я проверяю запись в ответ, я обнаружил, что излучатель остановлен, чтобы продолжить работу.

Ниже приведен мой пример кода.У меня есть 4 данных и 2 из которых не было в базе данных.Я обнаружил, что эмитент продолжает работу с данными, которые существуют только в базе данных.

val movies = mutableListOf("Secret Mother (Mainland) - 秘密媽媽","Life For Life - 命情真","Before Dawn - 愛在暴風的日子","The Threat Of Love 2 - Loving ou 我愛你2")

        Observable
            .fromIterable(movies)
            .flatMapMaybe {
                videoInfoService
                    .findVideoByTitle(it)
                    .switchIfEmpty(Maybe.empty())
            }
            .subscribe(object: Observer<VideoInfo>{
                override fun onComplete() {
                    println("on complete ")
                }

                override fun onSubscribe(d: Disposable) {
                    println("on subscribe ")
                }

                override fun onNext(t: VideoInfo) {
                    println("on next: ${t.title}")
                }

                override fun onError(e: Throwable) {
                    e.printStackTrace()
                }

            })

Благодарим вас за руководство.Я знаю, что Rxjava2 не может продолжить работу с нулевым значением из другого поста, который я написал этим утром, я думаю, что это должен быть какой-то способ справиться с этим сценарием.

Спасибо,

Answer 1

Это вывод, который я хочу.

Спасибо, @Gustavo предоставил ответ, и я получил подсказку от ответа @ Gustavo, чтобы получить этот ответ.

val movies = mutableListOf("Secret Mother (Mainland) - 秘密媽媽","Life For Life - 命情真","Before Dawn - 愛在暴風的日子","The Threat Of Love 2 - Loving ou 我愛你2")

        Observable
            .fromIterable(movies)
            .flatMap {
                videoInfoService
                    .findVideoByTitle(it)
                    .toObservable()
                    .switchIfEmpty{
                        // Save the video
                        save(it)
                    }
            }
            .subscribe(object: Observer<VideoInfo>{
                override fun onComplete() {
                    println("on complete ")
                }

                override fun onSubscribe(d: Disposable) {
                    println("on subscribe ")
                }

                override fun onNext(t: VideoInfo) {
                    println("on next: ${t.title}")
                }

                override fun onError(e: Throwable) {
                    e.printStackTrace()
                }

            })

Answer 2

при условии, что ваш метод сохранения записи в mongodb возвращает Completable:

fun store(movie: Movie): Completable

и ваш метод videoInfoService.findVideoByTitle возвращает экземпляр Movie, вы можете изменить свой код на:

    Observable
        .fromIterable(movies)
        .flatMapSingle { title ->
            videoInfoService
                .findVideoByTitle(title)
                .switchIfEmpty {
                    Single.defer {
                        val movie = Movie(title)
                        store(movie)
                            .andThen(Single.just(movie))
                    }
                }
        }