         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)

    }
}

// MARK: - DatabaseValueConvertible

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

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