Proper usage of Observable.create() in RxJava 2 (Best Practices)

Question

I'm currently building a small Social Media style App which leverages RxJava 2 and Firebase. I'm using MVP style architecture, and I've abstracted out my AuthService with an interface called AuthSource.

For simplicity's sake, I'll work with a Single method in my Service:

public class FirebaseAuthService implements AuthSource {

private FirebaseAuth auth;
private FirebaseAuth.AuthStateListener listener;

//initialization code

@Override
public Maybe<User> getUser() {
    return Maybe.create(new MaybeOnSubscribe<User>() {
                            @Override
                            public void subscribe(final MaybeEmitter<User> e) throws Exception {
                                if (auth == null) {
                                    auth = FirebaseAuth.getInstance();
                                }

                                if (listener != null) {
                                    auth.removeAuthStateListener(listener);
                                }

                                listener = new FirebaseAuth.AuthStateListener() {
                                    @Override
                                    public void onAuthStateChanged(@NonNull FirebaseAuth firebaseAuth) {
                                        FirebaseUser firebaseUser = firebaseAuth.getCurrentUser();
                                        auth.removeAuthStateListener(listener);
                                        if (firebaseUser != null) {
                                            User user = new User(
                                                    firebaseUser.getDisplayName(),
                                                    firebaseUser.getEmail());

                                            user.setUserId(firebaseUser.getUid());


                                            Uri photoUrl = firebaseUser.getPhotoUrl();
                                            if (photoUrl != null){
                                                user.setProfilePhotoUrl(photoUrl.toString());
                                            }
                                            e.onSuccess(user);
                                        } else {
                                            e.onComplete();
                                        }
                                    }
                                };

                                auth.addAuthStateListener(listener);
                            }
                        }
    );

}

}

interface AuthSource {
    Maybe<User> getUser();
//Other methods etc.
}

Finally, I'll show my Presenter method which handles the call:

//from with a Presenter:
@Override
private void getUserData() {
    disposableSubscriptions.add(
            auth.getUser().subscribeOn(schedulerProvider.io())
                    .observeOn(schedulerProvider.ui())
                    .subscribeWith(
                            new DisposableMaybeObserver<User>() {

                                @Override
                                public void onError(Throwable e) {
                                    view.makeToast(R.string.error_retrieving_data);
                                    view.startDispatchActivity();
                                }

                                @Override
                                public void onComplete() {

                                }

                                @Override
                                public void onSuccess(User user) {
                                    ProfilePagePresenter.this.currentUser = user;
                                    view.setName(user.getName());
                                    view.setEmail(user.getEmail());
                                    if (user.getProfilePhotoUrl().equals("")) {
                                        view.setDefaultProfilePhoto();
                                    } else {
                                        view.setProfilePhotoURI(user.getProfilePhotoUrl());
                                    }

                                    getUserProfileFromDatabase();

                                }
                            }
                    )
    );
}

I realize the topic of the question is a bit general, so I'll try to narrow things down from here. The code I've posted above works insofar as I'm succesfully getting Data from Firebase's API using Create(). The problem is, I'm quite new to using RxJava 2, and I'm not certain what's going on under the hood here for garbage collection and memory leaks. I chose to use Observable.create() as per the RxJava 2 Docs:

"Provides an API (via a cold Observable) that bridges the reactive world with the callback-style world."

RxJava 2 Docs Finally, the only proactive thing I'm doing at the moment to dispose of these Observables, is to call CompositeDisposable.clear() in my Presenter when events take the user to a new Activity.

Questions:

-Is it safe to assume that simply calling CompositeDisposable.clear() when the Presenter finishes, will handle my Garbage collection? (assuming I haven't created memory leaks in the rest of the code).

-If my understanding is correct, create() is a better option to use than fromCallable() in this case, as fromCallable() should be used for Synchronous events (i.e. not something like Firebase API callbacks)?

-Is it really as simple as just throwing my Asynchronous callbacks in Observable.create()? I'm terrified at how easy that is to do...

Answer 1

Is it safe to assume that simply calling CompositeDisposable.clear() when the Presenter finishes, will handle my Garbage collection? (assuming I haven't created memory leaks in the rest of the code).

It's a little trickier than this. Non-disposed Observable won't create memory leak if everything referenced by the Observable belong to the Activity scope. Both the producer and the consumer will be garbage collected alongside Activity. Memory leak may occur if you referenced resources that will survive the Activity, a provider instantiated at Application level for example. So if you want to use CompositeDisposable.clear() make sure to implement emitter.setCancellable() inside Observable.create() to dispose those leaky resources.

If my understanding is correct, create() is a better option to use than fromCallable() in this case, as fromCallable() should be used for Synchronous events (i.e. not something like Firebase API callbacks)?

create() use to be named fromAsync(). Use fromCallable() to wrap a synchronous method call, create() when wrapping callback code.

Is it really as simple as just throwing my Asynchronous callbacks in Observable.create()? I'm terrified at how easy that is to do...

It is as easy ... if you take care of those pesky references outside of scope as mentioned at the first point.

Usually on Android, a memory leak involve the Context, which is big. Be sure to test your code. leakcanary is a great help for this matter.

Last, you could avoid doing the wrapping yourself by using an existing Firebase RxJava binding. Or take inspiration from them:

• https://github.com/kunny/RxFirebase
• https://github.com/ashdavies/rx-firebase
• https://github.com/DariusL/RxFirebaseAndroid
• https://github.com/ezhome/Android-RxFirebase
• https://github.com/nmoskalenko/RxFirebase
• https://github.com/VictorAlbertos/RxFcm