<commit_before>#if USING_SQLCIPHER
    import GRDBCipher
#else
    import GRDB
#endif
import RxSwift

extension ValueObservation : ReactiveCompatible { }

/// :nodoc:
public protocol _ValueObservationProtocol: ReactiveCompatible {
    associatedtype _Reducer: ValueReducer
    var scheduling: ValueScheduling { get set }
    func start(
        in reader: DatabaseReader,
        onError: ((Error) -> Void)?,
        onChange: @escaping (_Reducer.Value) -> Void) throws -> TransactionObserver
}

/// :nodoc:
extension ValueObservation: _ValueObservationProtocol where Reducer: ValueReducer {
    public typealias _Reducer = Reducer
}

extension Reactive where Base: _ValueObservationProtocol {
    /// Returns an Observable that emits the same values as a ValueObservation.
    ///
    ///     let observation = ValueObservation.trackingAll(Player.all())
    ///     observation.rx
    ///         .fetch(in: dbQueue)
    ///         .subscribe(onNext: { players: [Player] in
    ///             print("Players have changed")
    ///         })
    ///
    /// By default, all values are emitted on the main dispatch queue. If you
    /// give a *scheduler*, values 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
    ///     observation.rx
    ///         .fetch(in: dbQueue)
    ///         .subscribe(onNext: { players: [Player] in
    ///             // on the main queue
    ///             print("Values have changed")
    ///         })
    ///     // <- here "Values have changed" has been printed
    ///
    ///     // on any queue
    ///     observation.rx
    ///         .fetch(in: dbQueue)
    ///         .subscribe(onNext: { players: [Player] in
    ///             // on the main queue
    ///             print("Values have changed")
    ///         })
    ///     // <- here "Values 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 fetch(
        in reader: DatabaseReader,
        startImmediately: Bool = true,
        scheduler: ImmediateSchedulerType? = nil)
        -> Observable<Base._Reducer.Value>
    {
        var observation = base
        
        if let scheduler = scheduler {
            // Honor the scheduler parameter
            observation.scheduling = .unsafe(startImmediately: startImmediately)
            return observation.makeObservable(in: reader).observeOn(scheduler)
            
        } else if case .mainQueue = observation.scheduling , startImmediately == false {
            // Honor the startImmediately parameter
            observation.scheduling = .async(onQueue: DispatchQueue.main, startImmediately: false)
            return observation.makeObservable(in: reader)
            
        } else {
            // Honor observation scheduling
            return observation.makeObservable(in: reader)
        }
    }
}

extension _ValueObservationProtocol {
    fileprivate func makeObservable(in reader: DatabaseReader) -> Observable<_Reducer.Value> {<commit_msg>minor<commit_after>#if USING_SQLCIPHER
    import GRDBCipher
#else
    import GRDB
#endif
import RxSwift

extension ValueObservation : ReactiveCompatible { }

/// :nodoc:
public protocol _ValueObservationProtocol: ReactiveCompatible {
    associatedtype _Reducer: ValueReducer
    var scheduling: ValueScheduling { get set }
    func start(
        in reader: DatabaseReader,
        onError: ((Error) -> Void)?,
        onChange: @escaping (_Reducer.Value) -> Void) throws -> TransactionObserver
}

/// :nodoc:
extension ValueObservation: _ValueObservationProtocol where Reducer: ValueReducer {
    public typealias _Reducer = Reducer
}

extension Reactive where Base: _ValueObservationProtocol {
    /// Returns an Observable that emits the same values as a ValueObservation.
    ///
    ///     let observation = ValueObservation.trackingAll(Player.all())
    ///     observation.rx
    ///         .fetch(in: dbQueue)
    ///         .subscribe(onNext: { players: [Player] in
    ///             print("Players have changed")
    ///         })
    ///
    /// By default, all values are emitted on the main dispatch queue. If you
    /// give a *scheduler*, values 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
    ///     observation.rx
    ///         .fetch(in: dbQueue)
    ///         .subscribe(onNext: { players: [Player] in
    ///             // on the main queue
    ///             print("Values have changed")
    ///         })
    ///     // <- here "Values have changed" has been printed
    ///
    ///     // on any queue
    ///     observation.rx
    ///         .fetch(in: dbQueue)
    ///         .subscribe(onNext: { players: [Player] in
    ///             // on the main queue
    ///             print("Values have changed")
    ///         })
    ///     // <- here "Values 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 fetch(
        in reader: DatabaseReader,
        startImmediately: Bool = true,
        scheduler: ImmediateSchedulerType? = nil)
        -> Observable<Base._Reducer.Value>
    {
        var observation = base
        
        if let scheduler = scheduler {
            // Honor the scheduler parameter
            observation.scheduling = .unsafe(startImmediately: startImmediately)
            return observation.makeObservable(in: reader).observeOn(scheduler)
            
        } else if case .mainQueue = observation.scheduling , startImmediately == false {
            // Honor the startImmediately parameter
            observation.scheduling = .async(onQueue: .main, startImmediately: false)
            return observation.makeObservable(in: reader)
            
        } else {
            // Honor observation scheduling
            return observation.makeObservable(in: reader)
        }
    }
}

extension _ValueObservationProtocol {
    fileprivate func makeObservable(in reader: DatabaseReader) -> Observable<_Reducer.Value> {
{'fraction_non_alphanumeric': 0.10414603084760153, 'fraction_numerical': 0.0, 'mean_word_length': 2.011504047720494, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3287916', 'n_tokens_mistral': 1875, 'n_tokens_neox': 1799, 'n_words': 576}