java - 使用 Rxjava 作为 EventBus 时如何从事件中获取参数

Question

我在服务中接受来自网络的 json。

它通知 RxBus 事件：

      try {
            String m = msg.getData().getString("message");
            Log.i("handleMessage", m);
            JSONObject message1 = (JSONObject) new JSONTokener(m).nextValue();
            if (_rxBus.hasObservers()) {
                _rxBus.send(new Events.IncomingMsg(message1));
            }

在订阅方面，我如何使用“message1”参数，这是我需要操作的 json。如何从事件中提取和使用 json：

@Override
public void onStart() {
    super.onStart();
    subscriptions = new CompositeSubscription();
    subscriptions//
            .add(bindActivity(this, rxBus.toObserverable())//
                    .subscribe(new Action1<Object>() {
                        @Override
                        public void call(Object event) {
                            if (event instanceof Events.IncomingMsg) {
                                Log.v(TAG,"RXBUS!!!!");
                            }
                        }
                    }));
}

Answer 1

您可以将其过滤为 JSONObject 流，如下所示：

（Java 8 lambda 风格）

rxBus.toObservable()
    .ofType(Events.IncomingMsg.class)
    // I'm making a big assumption that getMessage() is a thing here.
    .map((event) -> event.getMessage())
    .subscribe((message) -> {
        // Do thing with message here!
    });

（Java 7“经典”风格）

rxBus.toObservable()
    .ofType(Events.IncomingMsg.class)
    // I'm making a big assumption that getMessage() is a thing here.
    .map(new Func1<Events.IncomingMsg, JSONObject>() {

        @Override
        public JSONObject call(final Events.IncomingMsg event) {
            return event.getMessage();
        }

    })
    .subscribe(new Action1<JSONObject>() {

        @Override
        public void call(final JSONObject message) {
            // Do something with message here.
        }

    });

（Java 7“经典”风格，过滤“位置”字符串）

rxBus.toObservable()
    .ofType(Events.IncomingMsg.class)
    // I'm making a big assumption that getMessage() is a thing here.
    .map(new Func1<Events.IncomingMsg, String>() {

        @Override
        public String call(final Events.IncomingMsg event) {
            return event.getMessage().getString("location");
        }

    })
    .subscribe(new Action1<String>() {

        @Override
        public void call(final String warehouse) {
            // Do something with warehouse here.
        }

    });