RxFirebaseUtils

RxFirebaseUtils.java

public class RxFirebaseUtils {

  public static <T> Observable<T> observeTask(final Task<T> task) {
    return Observable.<T>create(subscriber -> RxHandler.assignOnTask(subscriber, task)).observeOn(
        Schedulers.io());
  }

  public static Observable<List<DataSnapshot>> join(final DatabaseReference keysRef,
                                                    final DatabaseReference objectsRef) {
    return RxFirebaseDatabase.observeValueEvent(keysRef)
        .observeOn(Schedulers.io())
        .map(DataSnapshot::getChildren)
        .flatMap(keySnapshots -> Observable.from(keySnapshots)
            .observeOn(Schedulers.io())
            .map(DataSnapshot::getKey)
            .map(key -> RxFirebaseDatabase.observeValueEvent(objectsRef.child(key)))
            .toList()
            .flatMap(observables -> Observable.combineLatest(observables, combiner())));
  }

  private static FuncN<List<DataSnapshot>> combiner() {
    return array -> Arrays.asList(Arrays.copyOf(array, array.length, DataSnapshot[].class));
  }
}