         scheduler: ImmediateSchedulerType? = nil)
         -> Observable<[Base.RowDecoder]>
     {
+        // TODO: test
         return ValueObservation.trackingAll(base).rx.fetch(
             in: reader,
             startImmediately:

    }
}

// MARK: - DatabaseValueConvertible

extension Reactive where Base: FetchRequest, Base.RowDecoder: DatabaseValueConvertible {

        scheduler: ImmediateSchedulerType? = nil)
        -> Observable<Base.RowDecoder?>
    {
        // TODO: test
        return ValueObservation.trackingOne(base).rx.fetch(
            in: reader,
            startImmediately: startImmediately,
            scheduler: scheduler)
    }
}

/// :nodoc:
public protocol _QueryInterfaceRequestProtocol: FetchRequest { }
extension QueryInterfaceRequest: _QueryInterfaceRequestProtocol { }
extension Reactive where Base: _QueryInterfaceRequestProtocol, Base.RowDecoder: FetchableRecord {
    /// Returns an Observable that emits after each committed database
    /// transaction that has modified the tables and columns fetched by
    /// the request.
    ///
    ///     let dbQueue = DatabaseQueue()
    ///     let request = Player.all()
    ///     request.rx
    ///         .fetchAll(in: dbQueue)
    ///         .subscribe(onNext: { players: [Player] in
    ///             print("Players have changed")
    ///         })
    ///
    /// By default, all records are emitted on the main dispatch queue. If you
    /// give a *scheduler*, records are emitted on that scheduler.
    ///
    /// If you set *startImmediately* to true (the default value), the first
    /// element is emitted right upon subscription. It is *synchronously*
    /// emitted if and only if the observable is subscribed on the main queue,
    /// and is given a nil *scheduler* argument:
    ///
    ///     // on the main queue
    ///     request.rx
    ///         .fetchAll(in: dbQueue)
    ///         .subscribe(onNext: { players: [Player] in
    ///             // on the main queue
    ///             print("Players have changed")
    ///         })
    ///     // <- here "Players have changed" has been printed
    ///
    ///     // on any queue
    ///     request.rx
    ///         .fetchAll(in: dbQueue, scheduler: MainScheduler.instance)
    ///         .subscribe(onNext: { players: [Player] in
    ///             // on the main queue
    ///             print("Players have changed")
    ///         })
    ///     // <- here "Players have changed" may not be printed yet
    ///
    /// - parameter reader: A DatabaseReader (DatabaseQueue or DatabasePool).
    /// - parameter startImmediately: When true (the default), the first
    ///   element is emitted right upon subscription.
    /// - parameter scheduler: The eventual scheduler on which elements
    ///   are emitted.
    public func fetchAll(
        in reader: DatabaseReader,
        startImmediately: Bool = true,
        scheduler: ImmediateSchedulerType? = nil)
        -> Observable<[Base.RowDecoder]>
    {
        let request = base as! QueryInterfaceRequest<Base.RowDecoder>
        return ValueObservation.trackingAll(request).rx.fetch(
            in: reader,
            startImmediately:
            startImmediately,
            scheduler: scheduler)
    }
    
    /// Returns an Observable that emits after each committed database
    /// transaction that has modified the tables and columns fetched by
    /// the request.
    ///
    ///     let dbQueue = DatabaseQueue()
    ///     let request = Player.filter(key: 1)
    ///     request.rx
    ///         .fetchOne(in: dbQueue)
    ///         .subscribe(onNext: { player: Player? in
    ///             print("Player 1 has changed")
    ///         })
    ///
    /// By default, all records are emitted on the main dispatch queue. If you
    /// give a *scheduler*, records are emitted on that scheduler.
    ///
    /// If you set *startImmediately* to true (the default value), the first
    /// element is emitted right upon subscription. It is *synchronously*
    /// emitted if and only if the observable is subscribed on the main queue,
    /// and is given a nil *scheduler* argument:
    ///
    ///     // on the main queue
    ///     request.rx
    ///         .fetchOne(in: dbQueue)
    ///         .subscribe(onNext: { player: Player? in
    ///             // on the main queue
    ///             print("Player 1 has changed")
    ///         })
    ///     // <- here "Player 1 has changed" has been printed
    ///
    ///     // on any queue
    ///     request.rx
    ///         .fetchOne(in: dbQueue, scheduler: MainScheduler.instance)
    ///         .subscribe(onNext: { player: Player? in
    ///             // on the main queue
    ///             print("Player 1 has changed")
    ///         })
    ///     // <- here "Player 1 has changed" may not be printed yet
    ///
    /// - parameter reader: A DatabaseReader (DatabaseQueue or DatabasePool).
    /// - parameter startImmediately: When true (the default), the first
    ///   element is emitted right upon subscription.
    /// - parameter scheduler: The eventual scheduler on which elements
    ///   are emitted.
    public func fetchOne(
        in reader: DatabaseReader,
        startImmediately: Bool = true,
        scheduler: ImmediateSchedulerType? = nil)
        -> Observable<Base.RowDecoder?>
    {
        let request = base as! QueryInterfaceRequest<Base.RowDecoder>
        return ValueObservation.trackingOne(request).rx.fetch(
            in: reader,
            startImmediately: startImmediately,
            scheduler: scheduler)
