Realm の通知を RxSwift の Observable に包み込む

Realm のクエリ結果が変化するたびに RxSwift のストリームに結果が流れる処理を考えてみました。
Realm の通知トークンが購読中は強参照になるのが肝です。

realm.objects(Dog).filter("age > 3").rx_response()
    .subscribeNext { /* UI の更新処理 */ }
    .addDisposableTo(disposeBag)

実装

RxCocoa/UIControl+Rx.swift を参考

import RxSwift
import RealmSwift

extension Results {
    
    public func rx_response() -> Observable<Results<Element>> {
        return Observable.create { observer in
            
            MainScheduler.ensureExecutingOnScheduler()
            
            let token = self.addNotificationBlock { results, error in
                
                if let error = error {
                    observer.onError(error)
                } else {
                    observer.onNext(results!)
                }
            }
            
            return AnonymousDisposable {
                token.stop()
            }
        }
    }
}