Firebase リスナーを RxJava / RxKotlin に変換する Extension集

今回はみんな大好きFirebaseを使用する際に、
簡単にRxに変換する方法をご紹介したいと思います。

Firebaseといえど、認証系のAuth、データベースのFirestore、ストレージのStorageなど様々な機能があります。
そのすべてのリスナーをRxで流すような実装はよくあります。

/**
 * FirebaseのTaskをFlowableに変換
 *
 * @param T オブジェクト
 * @param strategy BackpressureStrategy
 * @return Flowable
 */
@CheckResult
inline fun <reified T> Task<T?>.toFlowable(strategy: BackpressureStrategy = BackpressureStrategy.LATEST): Flowable<T> {
    return toObservable().toFlowable(strategy)
}

/**
 * FirebaseのTaskをObservableに変換
 *
 * @param T オブジェクト
 * @return Observable
 */
@CheckResult
inline fun <reified T> Task<T?>.toObservable(): Observable<T> {
    return Observable.create { emitter ->
        addOnSuccessListener {
            if (emitter.isDisposed) return@addOnSuccessListener
            if (it != null) emitter.onNext(it) else emitter.onComplete()
        }
        addOnFailureListener {
            if (emitter.isDisposed) return@addOnFailureListener
            emitter.onError(it)
        }
        addOnCompleteListener {
            if (emitter.isDisposed) return@addOnCompleteListener
            emitter.onComplete()
        }
    }
}

/**
 * FirebaseのTaskをSingleに変換
 *
 * @param T オブジェクト
 * @return Single
 */
@CheckResult
inline fun <reified T> Task<T?>.toSingle(): Single<T> {
    return Single.create { emitter ->
        addOnSuccessListener {
            if (emitter.isDisposed) return@addOnSuccessListener
            if (it != null) emitter.onSuccess(it) else emitter.onError(Exception("T is null."))
        }
        addOnFailureListener {
            if (emitter.isDisposed) return@addOnFailureListener
            emitter.onError(it)
        }
    }
}

/**
 * FirebaseのTaskをMaybeに変換
 *
 * @param T オブジェクト
 * @return Maybe
 */
@CheckResult
inline fun <reified T> Task<T?>.toMaybe(): Maybe<T> {
    return Maybe.create { emitter ->
        addOnSuccessListener {
            if (emitter.isDisposed) return@addOnSuccessListener
            if (it != null) emitter.onSuccess(it) else emitter.onComplete()
        }
        addOnFailureListener {
            if (emitter.isDisposed) return@addOnFailureListener
            emitter.onError(it)
        }
        addOnCanceledListener {
            if (emitter.isDisposed) return@addOnCanceledListener
            emitter.onComplete()
        }
    }
}

/**
 * FirebaseのTaskをFlowableに変換
 *
 * @param strategy BackpressureStrategy
 * @return Flowable
 */
@CheckResult
inline fun <reified T> Task<T?>.toFlowableUnit(strategy: BackpressureStrategy = BackpressureStrategy.LATEST): Flowable<Unit> {
    return toObservableUnit().toFlowable(strategy)
}

/**
 * FirebaseのTaskをObservableに変換
 *
 * @return Observable
 */
@CheckResult
inline fun <reified T> Task<T?>.toObservableUnit(): Observable<Unit> {
    return Observable.create { emitter ->
        addOnSuccessListener {
            if (emitter.isDisposed) return@addOnSuccessListener
            emitter.onNext(Unit)
        }
        addOnFailureListener {
            if (emitter.isDisposed) return@addOnFailureListener
            emitter.onError(it)
        }
        addOnCompleteListener {
            if (emitter.isDisposed) return@addOnCompleteListener
            emitter.onComplete()
        }
    }
}

/**
 * FirebaseのTaskをSingleに変換
 *
 * @return Single
 */
@CheckResult
inline fun <reified T> Task<T?>.toSingleUnit(): Single<Unit> {
    return Single.create { emitter ->
        addOnSuccessListener {
            if (emitter.isDisposed) return@addOnSuccessListener
            emitter.onSuccess(Unit)
        }
        addOnFailureListener {
            if (emitter.isDisposed) return@addOnFailureListener
            emitter.onError(it)
        }
    }
}

/**
 * FirebaseのTaskをMaybeに変換
 *
 * @return Maybe
 */
@CheckResult
inline fun <reified T> Task<T?>.toMaybeUnit(): Maybe<Unit> {
    return Maybe.create { emitter ->
        addOnSuccessListener {
            if (emitter.isDisposed) return@addOnSuccessListener
            emitter.onSuccess(Unit)
        }
        addOnFailureListener {
            if (emitter.isDisposed) return@addOnFailureListener
            emitter.onError(it)
        }
        addOnCanceledListener {
            if (emitter.isDisposed) return@addOnCanceledListener
            emitter.onComplete()
        }
    }
}

ポイントは３つ

• FirebaseのリスナーはRxがすでに購読解除されている場合は流さないようにしています
• toSingleはオブジェクトがnullだった場合はエラーとして扱うのに対し、toSingleUnitはオブジェクトがnullであってもエラーとして扱わないようにしています。理由として、Firebaseの通信が成功してaddOnSuccessListenerが呼ばれるけど、通知するオブジェクトがない場合はVoidが通知されるからです。JavaまたはKotlinではVoidを生成できないため、代わりにUnitを返してあげています。

たった１行でFirebaseからRxに変換できてしまう

Auth

/**
 * Emailでサインアップ
 *
 * @param email メールアドレス
 * @param password パスワード
 * @return Firebaseユーザー
 */
fun signUpWithEmail(email: String, password: String): Maybe<AuthResult> {
    return FirebaseAuth.getInstance()
        .createUserWithEmailAndPassword(email, password)
        .toMaybe()
}

Firestore

/**
 * すべてのユーザーを取得
 *
 * @return すべてのユーザー
 */
fun findAll(): Single<QuerySnapshot> {
    return FirebaseFirestore.getInstance()
        .collection("users")
        .get()
        .toSingle()
}

/**
 * IDでユーザーを取得
 *
 * @param id ユーザーID
 * @return ユーザー
 */
fun findById(id: String): Single<DocumentSnapshot> {
    return FirebaseFirestore.getInstance()
        .collection("users")
        .document(id)
        .get()
        .toSingle()
}

Storage

/**
 * プロフィール画像をStorageから取得
 *
 * @param id ユーザーID
 * @param fileName プロフィール画像
 * @return Uri
 */
fun fetchImage(id: String, fileName: String): Single<Uri> {
    return FirebaseStorage.getInstance()
        .reference
        .child("users")
        .child(id)
        .child(fileName)
        .downloadUrl
        .toSingle()
}