在RxJava2中正确使用Observable.create()(最佳实践)

我目前正在构建一个利用RxJava 2和Firebase的小型社交媒体风格的应用程序.我正在使用MVP样式架构,并且我使用名为AuthSource的接口抽象出了我的AuthService.

为简单起见,我将在我的服务中使用Single方法:

public class FirebaseAuthService implements AuthSource {

private FirebaseAuth auth;
private FirebaseAuth.AuthStateListener listener;

//initialization code

@Override
public Maybe getUser() {
    return Maybe.create(new MaybeOnSubscribe() {
                            @Override
                            public void subscribe(final MaybeEmitter 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 getUser();
//Other methods etc.
}

最后,我将展示处理调用的Presenter方法:

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

                                @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();

                                }
                            }
                    )
    );
}

我意识到这个问题的主题有点笼统,所以我会尝试从这里缩小范围.我已经张贴上面的代码工作,只要我是从火力地堡的API使用创建成功地获得数据().问题是,我对使用RxJava 2很新,而且我不确定这里有什么东西可以用于垃圾收集和内存泄漏.我根据RxJava 2 Docs选择使用Observable.create():

"提供一个API(通过冷Observable),将反应世界与回调式世界联系起来."

RxJava 2 Docs 最后,我正在处理这些Observable的唯一主动是,当事件将用户带到一个新的Activity时,在我的Presenter中调用CompositeDisposable.clear().

问题:

- 如果我的理解是正确的,在这种情况下,create()是比使用fromCallable()更好的选择,因为fromCallable()应该用于同步事件(即不像Firebase API回调)？

- 它就像在Observable.create()中抛出我的异步回调一样简单吗？我很害怕这很容易做到......

假设只是CompositeDisposable.clear() 在Presenter完成时调用,是否可以安全地处理我的垃圾收集？(假设我没有在其余代码中创建内存泄漏).

这比这更棘手.Observable如果Observable属于Activity范围所引用的所有内容,则non-dispos 将不会创建内存泄漏.生产者和消费者都将被垃圾收集在一起Activity.如果您引用的资源将Activity在Application例如级别实例化的提供程序中存在,则可能会发生内存泄漏.所以如果你想使用CompositeDisposable.clear()make确保在emitter.setCancellable()里面Observable.create()实现配置那些漏洞的资源.

如果我的理解是正确的,create()那么使用比fromCallable()在这种情况下更好的选择,fromCallable()应该用于同步事件(即不像Firebase API回调)？

create()用来命名fromAsync().用于在包装回调代码时包装fromCallable()同步方法调用create().

它真的就像扔掉我的异步回调一样简单 Observable.create()吗？我很害怕这很容易做到......

这很容易......如果你在第一点提到的范围外处理那些讨厌的引用.

通常在Android上,内存泄漏涉及到Context,这很大.一定要测试你的代码.泄漏对这件事有很大的帮助.

最后,您可以通过使用现有的Firebase RxJava绑定来避免自己进行包装.或者从他们那里获取灵感:

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