Android Room с RXJava2; onNext () эмиттера не запускается должным образом

Я перехожу с асинхронных задач на rxjava2, и у меня возникли проблемы с тестами кода.

У меня есть таблица элементов комнаты, у которых есть определенная денежная сумма. В пользовательском элементе управления, который называется DisplayCurrentBudget, должна отображаться сумма всех сумм. Это число должно обновляться каждый раз при вставке нового элемента. Я справился с этим требованием двумя способами, но оба дают один и тот же результат: мой код не заботится о том, обновляется ли база данных, он обновляется только при воссоздании фрагмента (onCreateView).

Моя первая попытка была такой:

//RxJava2 Test
    Observable<ItemS> ItemObservable = Observable.create( emitter -> {
        try {
            List<ItemS> movies = oStandardModel.getItemsVanilla();
            for (ItemS movie : movies) {
                emitter.onNext(movie);
            }
            emitter.onComplete();
        } catch (Exception e) {
            emitter.onError(e);
        }
    });


    DisposableObserver<ItemS> disposable = ItemObservable.
        subscribeOn(Schedulers.io()).
        observeOn(AndroidSchedulers.mainThread()).
        subscribeWith(new DisposableObserver<ItemS>() {
            public List<ItemS> BadFeelingAboutThis = new ArrayList<ItemS>();

            @Override
            public void onNext(ItemS movie) {
                // Access your Movie object here
                BadFeelingAboutThis.add(movie);
            }
            @Override
            public void onError(Throwable e) {
                // Show the user that an error has occurred
            }
            @Override
            public void onComplete() {
                // Show the user that the operation is complete
                oBinding.DisplayCurrentBudget.setText(Manager.GetBigSum(BadFeelingAboutThis).toString());
            }
    });

Этот код меня уже не устраивал. Моя вторая попытка дает точно такой же результат:

Observable<BigDecimal> ItemObservable2 = Observable.create( emitter -> {
        try {
            BigDecimal mySum = oStandardModel.getWholeBudget();
            emitter.onNext(mySum);
            emitter.onComplete();
        } catch (Exception e) {
            emitter.onError(e);
        }
    });

    DisposableObserver<BigDecimal> disposable = ItemObservable2.
            subscribeOn(Schedulers.io()).
            observeOn(AndroidSchedulers.mainThread()).
            subscribeWith(new DisposableObserver<BigDecimal>() {
                @Override
                public void onNext(BigDecimal sum) {
                    // Access your Movie object here
                    oBinding.DisplayCurrentBudget.setText(sum.toString());
                }
                @Override
                public void onError(Throwable e) {
                    // Show the user that an error has occurred
                }
                @Override
                public void onComplete() {
                    // Show the user that the operation is complete
                }
            });

Есть ли очевидные проблемы с моим кодом?

Спасибо за чтение, очень признателен!

Редактировать: Меня спросили, что делает Manager.GetBigSum, на самом деле это не так много. Он только добавляет BigDecimal-Values ​​в список элементов.

public static BigDecimal GetBigSum(List<ItemS> ListP){

    List<BigDecimal> bigDList = ListP.stream().map(ItemS::get_dAmount).collect(Collectors.toList());

    return bigDList.stream()
            .reduce(BigDecimal.ZERO, BigDecimal::add);
}

Далее я упростил запрос. Но он по-прежнему не заботится об обновлениях БД, а только о воссоздании фрагментов:

Single.fromCallable(() -> oStandardModel.getItemsVanilla())
            .map(Manager::GetBigSum)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                    e -> oBinding.DisplayCurrentBudget.setText(e.toString())
            );