How pass data between activities using RxJava in android?

Question

I need to pass some data between two activities MainActivity and ChildActivity. Button click on MainActivity should open ChildActivity and send event with data. I have singleton:

Subject<Object, Object> subject = new SerializedSubject<>(PublishSubject.create());

and in MainActivity I have the following button click handler:

    public void onClick(){
        startActivity(new Intent(MainActivity.this, ChildActivity.class));
        subject.onNext(new SomeEvent(data));
    }

and event listener subscription in ChildActivity :

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        addEventListeners();
    }

    private void addEventListeners() {
        subject.ofType(SomeEvent.class)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.io()).subscribe(
                event -> {
                    loadData(event.getData());
                });
    }

When I send event after starting activity and call addEventListeners in ChildActivity onCreate is still not subscribed to this event and loadData() is not called.

What is proper way to pass data between activities using RxJava (if it's possible)?

Answer 1

if anybody needs a complete solution to send data between activities using RxJava2

1- Create the bus:

public final class RxBus {

private static final BehaviorSubject<Object> behaviorSubject
        = BehaviorSubject.create();


public static BehaviorSubject<Object> getSubject() {
    return behaviorSubject;
}

}

2- the sender activity

                       //the data to be based
                    MyData  data =getMyData();
                    RxBus.getSubject().onNext(data) ;
                    startActivity(new Intent(MainActivity.this, AnotherAct.class));

3-the receiver activity

    disposable = RxBus.getSubject().
            subscribeWith(new DisposableObserver<Object>() {


                @Override
                public void onNext(Object o) {
                    if (o instanceof MyData) {
                   Log.d("tag", (MyData)o.getData();
                    }
                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onComplete() {

                }
            });
    });

4-unSubscribe to avoid memory leacks:

 @Override
protected void onDestroy() {
    super.onDestroy();
    disposable.dispose();
}

Answer 2

Reason:

Problem is that you are using PublishSubject. As per documentation of PublishSubject emits all the subsequent items of the source Observable at the time of the subscription. So in your case it will emit event only if it is subscribed.

Fix for your problem

Instead of using PublishSubject use BehaviorSubject which emits the most recently emitted item and all the subsequent items of the source Observable when a observer subscribe to it.

Browse following link for more details.