RxJava/Android Observable 不响应状态变化

Question

当我为 Subject 调用 onNext 时，observable 没有接收到任何变化并且没有做出适当的反应。

所以我们的产品使用 canBus 将数据从我们的控制器发送到我们的硬件，我们正在使用 rxJava 来处理数据更改。我们已经让 rxJava 与 UI 完美配合，但对于我们的新 Alexa 功能，出于某种原因，当我直接调用模型函数来更改值（然后调用 subject.onNext）时，没有任何反应。我似乎找不到错误。流程如下：

Alexa 回复：

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

我不明白为什么没有处理更改。正在命中日志消息 "change shower state value"。当它击中 "Shower can observer log" 时，观察者也在观察。我还认为可能是这个观察者也调用了 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() {

            }
    });

但这似乎主要是 UI 变化观察者。重要说明：我没有写观察者，但是我确实理解发生了什么减去这个问题。任何帮助将不胜感激。

Answer 1

所以我忘了将它添加到 alexa 响应中 class 并且在我添加之后它起作用了：

TouchSteamApplication.getApplicationComponent().inject(this);

RxJava/Android Observable 不响应状态变化

RxJava/Android Observable not reacting to state changes

android

rx-java

rx-android