Simple extension on results object for swift realm

RxSwift+Realm.swift

extension Realm {
    public func rx_objectForPrimaryKey<T: Object>(type: T.Type, key: AnyObject) -> Observable<T?> {
        return self.objects(type)
            .filter("%K = %@", self.schema[type.className()]!.primaryKeyProperty!.name, key)
            .rx_result
    }
}

extension Results where T: Object {
    
    public var rx_result: Observable<T?> {
        return self.rx_results.map { $0.first }
    }
    
    public var rx_results: Observable<[T]> {
        
        var token: NotificationToken? = nil
        
        return Observable.create { observer in
            
            token = self.addNotificationBlock { results, error in
                if let error = error {
                    observer.onError(error)
                } else {
                    if let results = results {
                        observer.onNext(results.map { $0 })
                    }
                }
            }
            
            observer.onNext(self.map { $0 })
            
            return AnonymousDisposable {
                if let token = token {
                    self.realm!.removeNotification(token)
                }
            }
        }.shareReplayLatestWhileConnected()
    }
}