How to get the parameter from the event when using Rxjava as an EventBus

Question

I'm accepting a json from the network, in a service.

It notifies an RxBus of the event:

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

In the subscription side, how do I use that "message1" parameter which is the json i need to manipulate. How do I extract and use the json from the event:

@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

You can filter it into a stream of JSONObject like so:

(Java 8 lambda style)

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 "classic" style)

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 "classic" style, filtering "location" string)

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

    });