Single Observable with Multiple Subscribers

Viewed 14632

I have an Observable<<List<Foo>> getFoo() that is created from a Retrofit Service and after calling the .getFoo() method, I need to share it with Multiple Subscribers. Calling the .share() method though, it causes the Network Call to be re-executed. Replay Operator does not work either. I know that a potential solution might be .cache(), but I do not know why this behaviour is caused.

// Create an instance of our GitHub API interface.
Retrofit retrofit = new Retrofit.Builder()
            .baseUrl(API_URL)
            .addConverterFactory(GsonConverterFactory.create())
            .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
            .build();

// Create a call instance for looking up Retrofit contributors.
Observable<List<Contributor>> testObservable = retrofit
        .create(GitHub.class)
        .contributors("square", "retrofit")
        .share();

Subscription subscription1 = testObservable
       .subscribe(new Subscriber<List<Contributor>>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onNext(List<Contributor> contributors) {
                System.out.println(contributors);
            }
         });

Subscription subscription2 = testObservable
        .subscribe(new Subscriber<List<Contributor>>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onNext(List<Contributor> contributors) {
                System.out.println(contributors + " -> 2");
            }
         });

subscription1.unsubscribe();
subscription2.unsubscribe();

The code above can reproduce the aforementioned behaviour. You can debug it and see that the Lists received belong to a different MemoryAddress.

I have also looked at ConnectableObservables as a potential solution, but this requires me carrying the original observable around, and calling .connect() each time I want to add a new Subscriber.

This kind of behaviour with the .share() was working fine till Retrofit 1.9. It stopped working on Retrofit 2 - beta. I have not yet tested it with the Retrofit 2 Release Version, that was released some hours ago.

EDIT: 01/02/2017

For future readers, I have written an article here explaining more about the case!

2 Answers
Related