RxJava / Android Observable не реагирует на изменения состояния

Question

Когда я вызываю onNext для субъекта, наблюдаемое не воспринимает никаких изменений и не реагирует соответствующим образом.

Поэтому наш продукт использует canBus для отправки данных с нашего контроллера на наше оборудование, и мы используемrxJava для обработки изменений данных.Уже у нас rxJava отлично работает с пользовательским интерфейсом, но для нашей новой функции Alexa, по какой-то причине, когда я вызываю функции модели напрямую для изменения значений (и, в свою очередь, вызываю subject.onNext), ничего не происходит.Я не могу найти ошибку.Поток ниже:

Ответ Алексы:

case "ShowerState":
            //TODO figure out shower state alexa case
            Log.d(TAG, "Intent: ShowerIntent");
            if(action.equals("on")){
                Log.d(TAG, "Action: On");
                digitalShowerModel.changeShowerStateValue(DigitalShowerModel.DigitalShowerState.ON);
            }else if(action.equals("off")){
                Log.d(TAG, "Action: Off");
                digitalShowerModel.changeShowerStateValue(DigitalShowerModel.DigitalShowerState.OFF);
            }
            break;

Модель цифрового душа:

public void changeShowerStateValue(DigitalShowerState digitalShowerState) {
    Log.d(TAG, "Change shower state value");
    currentShowerState = digitalShowerState;
    currentShowerStateSubject.onNext(currentShowerState);
}

Поток наблюдателя:

private void initDigitalShowerCanMsgs() {
    Observable.combineLatest(
            digitalShowerRepository.getShowerStateChanges(),
            digitalShowerRepository.getWaterTempChanges(),
            digitalShowerRepository.getWaterOutputChanges(),
            digitalShowerRepository.getWaterFlowChanges(),
            Observable.interval(TIME_BETWEEN_DIGITAL_SHOWER_MSGS, TimeUnit.SECONDS),
            new Function5<DigitalShowerModel.DigitalShowerState,
                                    Integer, DigitalShowerModel.WaterOutput,
                                    DigitalShowerModel.WaterFlow, Long, ArrayList<Integer>>() {
                @Override
                public ArrayList<Integer> apply(DigitalShowerModel.DigitalShowerState digitalShowerState, Integer temp, DigitalShowerModel.WaterOutput waterOutput, DigitalShowerModel.WaterFlow waterFlow, Long aLong) throws Exception {
                    ArrayList<Integer> dataBytes = new ArrayList<>();
                    dataBytes.add(digitalShowerState.ordinal());
                    dataBytes.add(temp);
                    dataBytes.add(waterOutput.ordinal());
                    dataBytes.add(waterFlow.ordinal());
                    return dataBytes;
                }
            })
            .observeOn(Schedulers.io())
            .subscribe(new Observer<ArrayList<Integer>>() {
                @Override
                public void onSubscribe(Disposable d) {

                }

                @Override
                public void onNext(ArrayList<Integer> dataBytes) {
                    Log.d(TAG, "SHOWER CAN MESSAGE OBSERVER");
                    FC_Cmd[FlexDefs.CTLESVSTATE].fflxout.execute(dataBytes.get(0), dataBytes.get(1),
                            dataBytes.get(2), dataBytes.get(3), 0, 0);
                }

                @Override
                public void onError(Throwable e) {
                    e.printStackTrace();
                }

                @Override
                public void onComplete() {

                }
            });

Цифровой душРепозиторий:

public Observable<DigitalShowerModel.DigitalShowerState> getShowerStateChanges() {
    Log.d(TAG, "Get shower state changes DigitalShowerRepository");
    return digitalShowerModel.getCurrentShowerStateSubject();
}

Модель цифрового душа (наблюдатель):

    public Subject<DigitalShowerState> getCurrentShowerStateSubject() {
    Log.d(TAG, "Get shower state value");
    return currentShowerStateSubject;
}

Я не понимаю, почему изменения не обрабатываются.Журнал сообщения "изменить значение состояния душа" находится в процессе.Также наблюдатель наблюдает за тем, как он нажимает на «Журнал наблюдателя в душе».Я также подумал, что это может быть наблюдатель, который также вызывает getShowerStateChanges, так или иначе влияя на него:

Фрагмент цифрового экрана:

 Observable.combineLatest(
        digitalShowerRepository.getShowerStateChanges(),
        digitalShowerRepository.getWaterOutputChanges(), new BiFunction<DigitalShowerModel.DigitalShowerState, DigitalShowerModel.WaterOutput, Pair>() {
            @Override
            public Pair<DigitalShowerModel.DigitalShowerState, DigitalShowerModel.WaterOutput>
                    apply(DigitalShowerModel.DigitalShowerState digitalShowerState, DigitalShowerModel.WaterOutput waterOutput) throws Exception {
                return new Pair<>(digitalShowerState, waterOutput);
            }
        })
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<Pair>() {
            @Override
            public void onSubscribe(Disposable d) {
                digitalShowerDisposables.add(d);
            }

            @Override
            public void onNext(Pair pair) {
                if(!wasInitiatedByUser) {
                    rainHeadRdoBtn.setOnTouchListener(null);
                    showerHeadRdoBtn.setOnTouchListener(null);
                    wandRdoBtn.setOnTouchListener(null);
                    showerOptions.setOnCheckedChangeListener(null);

                    if (pair.first == DigitalShowerModel.DigitalShowerState.ON) {
                        switch ((DigitalShowerModel.WaterOutput) pair.second) {
                            case RAINHEAD:
                                rainHeadRdoBtn.setChecked(true);
                                break;
                            case SHOWERHEAD:
                                showerHeadRdoBtn.setChecked(true);
                                break;
                            case WAND:
                                wandRdoBtn.setChecked(true);
                                break;
                        }
                    } else if (pair.first == DigitalShowerModel.DigitalShowerState.OFF) {
                        showerOptions.clearCheck();
                    }

                    rainHeadRdoBtn.setOnTouchListener(DigitalShowerScreenFragment.this);
                    showerHeadRdoBtn.setOnTouchListener(DigitalShowerScreenFragment.this);
                    wandRdoBtn.setOnTouchListener(DigitalShowerScreenFragment.this);
                    showerOptions.setOnCheckedChangeListener(DigitalShowerScreenFragment.this);
                }
                wasInitiatedByUser = false;
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onComplete() {

            }
    });

Но это, по-видимому, в первую очередь наблюдатель за изменением пользовательского интерфейса.Большое примечание: я не писал наблюдателю, однако я понимаю, что происходит, минус эта проблема.Любая помощь будет оценена.