将Firebase实时数据侦听器与RxJava相结合 [英] Combining Firebase realtime data listener with RxJava

问题描述

我在我的应用中使用了 Firebase 以及RxJava。
只要后端数据发生变化（添加，删除，更改...），Firebase就能够通知您的应用程序。
我试图将Firebase与RxJava的功能结合在一起。

我正在听的数据叫做 Leisure $ c $，并且 Observable 发出 LeisureUpdate ，其中包含 Leisure 和更新类型（添加，删除，移动，更改）。



这是我的方法，它允许订阅这个事件。

  private Observable< LeisureUpdate> leisureUpdatesObservable; 
私人ChildEventListener leisureUpdatesListener; 
 private int leisureUpdatesSubscriptionsCount; 
 
 @NonNull 
 public Observable< LeisureUpdate> subscribeToLeisuresUpdates（）{
 if（leisureUpdatesObservable == null）{
 leisureUpdatesObservable = Observable.create（new Observable.OnSubscribe< LeisureUpdate>（）{
 $ b $ @Override 
 public void call（final订阅者< ;?超级LeisureUpdate>订阅者）{
 leisureUpdatesListener = firebase.child（FirebaseStructure.LEISURES）.addChildEventListener（new ChildEventListener（）{
 @Override 
 public void onChildAdded DataSnapshot dataSnapshot，String s）{
 final Leisure Leisure = convertMapToLeisure（（Map< String，Object>）dataSnapshot.getValue（））; 
 subscriber.onNext（new LeisureUpdate（leisure，LeisureUpdate.ADDED）） ; 
} 
 
 @Override 
 public void onChildChanged（DataSnapshot dataSnapshot，String s）{
 final Lei （休闲）= convertMapToLeisure（（Map< String，Object>）dataSnapshot.getValue（））; 
 subscriber.onNext（新的LeisureUpdate（休闲，LeisureUpdate.CHANGED））; （Map< String，Object>）dataSnapshot.getValue（）返回一个数组，其中包含一个数据快照（DataSnapshot dataSnapshot） ）; 
 subscriber.onNext（新的LeisureUpdate（leisure，LeisureUpdate.REMOVED））; 
 
 $ b @Override 
 public void onChildMoved（DataSnapshot dataSnapshot，String s）{
 Leisure Leisure = convertMapToLeisure（（Map< String，Object>）dataSnapshot。的getValue（））; 
 subscriber.onNext（新的LeisureUpdate（休闲，LeisureUpdate.MOVED））; 
 
 $ b @Override 
 public void onCancelled（FirebaseError firebaseError）{
 subscriber.onError（new Error（firebaseError.getMessage（）））; 
} 
}）; 
} 
}）; 
} 
 leisureUpdatesSubscriptionsCount ++; 
返回leisureUpdatesObservable; 
 
 
 
 $ b首先，我想使用 Observable。 fromCallable（）方法，以创建 Observable ，但我猜这是不可能的，因为Firebase使用回调，对吗？ b 
 $ b 
为了始终有一个 Observable  Observable  >其中多个订阅者可以订阅。
 
 
 
当大家取消订阅，我需要停止监听事件在Firebase中。 
如果还有任何订阅，我还是没有找到使 Observable 理解。所以我一直在计算多少个电话到 subscribeToLeisuresUpdates（），用 leisureUpdatesSubscriptionsCount 。
 
 
 
然后每次有人想取消订阅，都必须打电话给b 
 $ b 
  @Override 
 public void unsubscribeFromLeisuresUpdates（）{
 if（leisureUpdatesObservable == null）{
 return; 
} 
 leisureUpdatesSubscriptionsCount--; 
 if（leisureUpdatesSubscriptionsCount == 0）{
 firebase.child（FirebaseStructure.LEISURES）.removeEventListener（leisureUpdatesListener）; 
 leisureUpdatesObservable = null; 
 
 
 
 
 $ b $ p 
 $ b $ p $这是我发现的唯一方法使 Observable 会在有订阅者的时候发射条目，但是我觉得有一个更简单的方法，特别是当没有更多的订阅者正在监听observable的时候。 b 
 $ b 
任何遇到类似问题或有不同做法的人？另外，你可以使用我的库 rxFirebase  
 
I am using Firebase in my app, along with RxJava.
Firebase is capable of notify your app whenever something changed in the backend data (addition, removals, changes, ...).
I am trying to combine the feature of Firebase with RxJava.

The data I am listening for is called Leisure, and the Observable emits LeisureUpdate which contains a Leisure and the type of update (add, remove, moved, changed).

Here is my method which allows to subscribe to this events.
private Observable<LeisureUpdate> leisureUpdatesObservable;
private ChildEventListener leisureUpdatesListener;
private int leisureUpdatesSubscriptionsCount;

@NonNull
public Observable<LeisureUpdate> subscribeToLeisuresUpdates() {
    if (leisureUpdatesObservable == null) {
        leisureUpdatesObservable = Observable.create(new Observable.OnSubscribe<LeisureUpdate>() {

            @Override
            public void call(final Subscriber<? super LeisureUpdate> subscriber) {
                leisureUpdatesListener = firebase.child(FirebaseStructure.LEISURES).addChildEventListener(new ChildEventListener() {
                    @Override
                    public void onChildAdded(DataSnapshot dataSnapshot, String s) {
                        final Leisure leisure = convertMapToLeisure((Map<String, Object>) dataSnapshot.getValue());
                        subscriber.onNext(new LeisureUpdate(leisure, LeisureUpdate.ADDED));
                    }

                    @Override
                    public void onChildChanged(DataSnapshot dataSnapshot, String s) {
                        final Leisure leisure = convertMapToLeisure((Map<String, Object>) dataSnapshot.getValue());
                        subscriber.onNext(new LeisureUpdate(leisure, LeisureUpdate.CHANGED));
                    }

                    @Override
                    public void onChildRemoved(DataSnapshot dataSnapshot) {
                        final Leisure leisure = convertMapToLeisure((Map<String, Object>) dataSnapshot.getValue());
                        subscriber.onNext(new LeisureUpdate(leisure, LeisureUpdate.REMOVED));
                    }

                    @Override
                    public void onChildMoved(DataSnapshot dataSnapshot, String s) {
                        final Leisure leisure = convertMapToLeisure((Map<String, Object>) dataSnapshot.getValue());
                        subscriber.onNext(new LeisureUpdate(leisure, LeisureUpdate.MOVED));
                    }

                    @Override
                    public void onCancelled(FirebaseError firebaseError) {
                        subscriber.onError(new Error(firebaseError.getMessage()));
                    }
                });
            }
        });
    }
    leisureUpdatesSubscriptionsCount++;
    return leisureUpdatesObservable;
}
First off, I would like to use Observable.fromCallable() method in order to create the Observable, but I guess it is impossible, since Firebase uses callbacks, right?

I keep a single instance of the Observable in order to always have one Observable where multiple Subscriber can subscribe.

The problem comes when everyone unsubscribe and I need to stop listening for the events in Firebase.
I didn't find anyway to make the Observable understand if there is any subscription still. So I keep counting how many calls I got to subscribeToLeisuresUpdates(), with leisureUpdatesSubscriptionsCount.

Then every time someone wants to unsubscribe it has to call
@Override
public void unsubscribeFromLeisuresUpdates() {
    if (leisureUpdatesObservable == null) {
        return;
    }
    leisureUpdatesSubscriptionsCount--;
    if (leisureUpdatesSubscriptionsCount == 0) {
        firebase.child(FirebaseStructure.LEISURES).removeEventListener(leisureUpdatesListener);
        leisureUpdatesObservable = null;
    }
}
This is the only way I found to make the Observable emits items when there is a subscriber, but I feel like there must be an easier way, specially understanding when there is no more subscribers listening to the observable.

Anyone who encountered a similar problem or have a different approach? 
解决方案
Also, you can use my library rxFirebase

                        
这篇关于将Firebase实时数据侦听器与RxJava相结合的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！