<commit_before>import GRDB
import RxSwift

/// We want the `rx` joiner on DatabaseWriter.
/// Normally we'd use ReactiveCompatible. But ReactiveCompatible is unable to
/// define `rx` on existentials as well:
///
///     let writer: DatabaseWriter
///     writer.rx...
///
/// :nodoc:
extension DatabaseWriter {
    public var rx: Reactive<AnyDatabaseWriter> {
        return Reactive(AnyDatabaseWriter(self))
    }
}

extension Reactive where Base: DatabaseWriter {
    /// Returns an Observable that asynchronously writes into the database.
    ///
    /// For example:
    ///
    ///     // Observable<Int>
    ///     let newPlayerCount = dbQueue.rx.flatMapWrite { db in
    ///         try Player(...).insert(db)
    ///         let newPlayerCount = try Player.fetchAll(db)
    ///         return Observable.just(newPlayerCount)
    ///     }
    ///
    /// The database updates are executed inside a database transaction. If the
    /// transaction completes successfully, the observable returned from the
    /// closure is subscribed, from a database protected dispatch queue,
    /// immediately after the transaction has been committed.
    ///
    /// - parameter scheduler: The scheduler on which the observable completes.
    ///   Defaults to MainScheduler.instance.
    /// - parameter updates: A closure which writes in the database.
    func flatMapWrite<T>(
        observeOn scheduler: ImmediateSchedulerType = MainScheduler.instance,
        updates: @escaping (Database) throws -> Observable<T>)
        -> Observable<T>
    {
        let writer = base
        return Observable
            .create { observer in
                var disposable: Disposable?
                writer.asyncWriteWithoutTransaction { db in
                    var observable: Observable<T>?
                    do {
                        try db.inTransaction {
                            observable = try updates(db)
                            return .commit
                        }
                        // Subscribe after transaction
                        disposable = observable!.subscribe(observer)
                    } catch {
                        observer.onError(error)
                    }
                }
                return Disposables.create {
                    disposable?.dispose()
                }
            }
            .observeOn(scheduler)
    }
    
    /// Returns a Completable that asynchronously writes into the database.
    ///
    ///     let dbQueue = DatabaseQueue()
    ///     let completable: Completable = dbQueue.rx.write { db in
    ///         try Player(...).insert(db)
    ///     }
    ///
    /// By default, the completable completes on the main dispatch queue. If
    /// you give a *scheduler*, is completes on that scheduler.
    ///
    /// - parameter scheduler: The scheduler on which the completable completes.
    ///   Defaults to MainScheduler.instance.
    /// - parameter updates: A closure which writes in the database.
    public func write(
        observeOn scheduler: ImmediateSchedulerType = MainScheduler.instance,
        updates: @escaping (Database) throws -> Void)
        -> Completable
    {
        return flatMapWrite(
            observeOn: scheduler,
            updates: { db in
                do {
                    try updates(db)
                    return .empty()
                } catch {
                    return .error(error)
                }
        })
            .asCompletable()
    }
    
    /// Returns a Single that asynchronously writes into the database.
    ///
    ///     let newPlayerCount: Single<Int> = dbQueue.rx.writeAndReturn { db in
    ///         try Player(...).insert(db)
    ///         return try Player.fetchCount(db)
    ///     }
    ///
    /// By default, the single completes on the main dispatch queue. If
    /// you give a *scheduler*, is completes on that scheduler.
    ///
    /// - parameter scheduler: The scheduler on which the observable completes.
    ///   Defaults to MainScheduler.instance.
    /// - parameter updates: A closure which writes in the database.
    public func writeAndReturn<T>(
        observeOn scheduler: ImmediateSchedulerType = MainScheduler.instance,
        updates: @escaping (Database) throws -> T)
        -> Single<T>
    {
        return flatMapWrite(
            observeOn: scheduler,
            updates: { db in
                do {
                    return try .just(updates(db))
                } catch {
                    return .error(error)
                }
        })
            .asSingle()
    }
    
    /// Returns a Single that asynchronously writes into the database.
    ///
    ///     let newPlayerCount: Single<Int> = dbQueue.rx.write(
    ///         updates: { db in try Player(...).insert(db) },
    ///         thenRead: { db, _ in try Player.fetchCount(db) })
    ///
    /// By default, the single completes on the main dispatch queue. If
    /// you give a *scheduler*, is completes on that scheduler.
    ///
    /// - parameter scheduler: The scheduler on which the observable completes.
    ///   Defaults to MainScheduler.instance.
    /// - parameter updates: A closure which writes in the database.
    /// - parameter value: A closure which reads from the database.
    public func write<T, U>(
        observeOn scheduler: ImmediateSchedulerType = MainScheduler.instance,
        updates: @escaping (Database) throws -> T,
        thenRead value: @escaping (Database, T) throws -> U)
        -> Single<U>
    {
        return flatMapWrite(
            observeOn: scheduler,
            updates: { db in
                let updatesValue = try updates(db)
                return Observable.create { observer in
                    self.base.spawnConcurrentRead { db in<commit_msg>Fix failing test for asynchronous transactions that must be rollbacked in case of error<commit_after>import GRDB
import RxSwift

/// We want the `rx` joiner on DatabaseWriter.
/// Normally we'd use ReactiveCompatible. But ReactiveCompatible is unable to
/// define `rx` on existentials as well:
///
///     let writer: DatabaseWriter
///     writer.rx...
///
/// :nodoc:
extension DatabaseWriter {
    public var rx: Reactive<AnyDatabaseWriter> {
        return Reactive(AnyDatabaseWriter(self))
    }
}

extension Reactive where Base: DatabaseWriter {
    /// Returns an Observable that asynchronously writes into the database.
    ///
    /// For example:
    ///
    ///     // Observable<Int>
    ///     let newPlayerCount = dbQueue.rx.flatMapWrite { db in
    ///         try Player(...).insert(db)
    ///         let newPlayerCount = try Player.fetchAll(db)
    ///         return Observable.just(newPlayerCount)
    ///     }
    ///
    /// The database updates are executed inside a database transaction. If the
    /// transaction completes successfully, the observable returned from the
    /// closure is subscribed, from a database protected dispatch queue,
    /// immediately after the transaction has been committed.
    ///
    /// - parameter scheduler: The scheduler on which the observable completes.
    ///   Defaults to MainScheduler.instance.
    /// - parameter updates: A closure which writes in the database.
    func flatMapWrite<T>(
        observeOn scheduler: ImmediateSchedulerType = MainScheduler.instance,
        updates: @escaping (Database) throws -> Observable<T>)
        -> Observable<T>
    {
        let writer = base
        return Observable
            .create { observer in
                var disposable: Disposable?
                writer.asyncWriteWithoutTransaction { db in
                    var observable: Observable<T>?
                    do {
                        try db.inTransaction {
                            observable = try updates(db)
                            return .commit
                        }
                        // Subscribe after transaction
                        disposable = observable!.subscribe(observer)
                    } catch {
                        observer.onError(error)
                    }
                }
                return Disposables.create {
                    disposable?.dispose()
                }
            }
            .observeOn(scheduler)
    }
    
    /// Returns a Completable that asynchronously writes into the database.
    ///
    ///     let dbQueue = DatabaseQueue()
    ///     let completable: Completable = dbQueue.rx.write { db in
    ///         try Player(...).insert(db)
    ///     }
    ///
    /// By default, the completable completes on the main dispatch queue. If
    /// you give a *scheduler*, is completes on that scheduler.
    ///
    /// - parameter scheduler: The scheduler on which the completable completes.
    ///   Defaults to MainScheduler.instance.
    /// - parameter updates: A closure which writes in the database.
    public func write(
        observeOn scheduler: ImmediateSchedulerType = MainScheduler.instance,
        updates: @escaping (Database) throws -> Void)
        -> Completable
    {
        return flatMapWrite(
            observeOn: scheduler,
            updates: { db in
                try updates(db)
                return .empty()
        })
            .asCompletable()
    }
    
    /// Returns a Single that asynchronously writes into the database.
    ///
    ///     let newPlayerCount: Single<Int> = dbQueue.rx.writeAndReturn { db in
    ///         try Player(...).insert(db)
    ///         return try Player.fetchCount(db)
    ///     }
    ///
    /// By default, the single completes on the main dispatch queue. If
    /// you give a *scheduler*, is completes on that scheduler.
    ///
    /// - parameter scheduler: The scheduler on which the observable completes.
    ///   Defaults to MainScheduler.instance.
    /// - parameter updates: A closure which writes in the database.
    public func writeAndReturn<T>(
        observeOn scheduler: ImmediateSchedulerType = MainScheduler.instance,
        updates: @escaping (Database) throws -> T)
        -> Single<T>
    {
        return flatMapWrite(
            observeOn: scheduler,
            updates: { db in try .just(updates(db)) })
            .asSingle()
    }
    
    /// Returns a Single that asynchronously writes into the database.
    ///
    ///     let newPlayerCount: Single<Int> = dbQueue.rx.write(
    ///         updates: { db in try Player(...).insert(db) },
    ///         thenRead: { db, _ in try Player.fetchCount(db) })
    ///
    /// By default, the single completes on the main dispatch queue. If
    /// you give a *scheduler*, is completes on that scheduler.
    ///
    /// - parameter scheduler: The scheduler on which the observable completes.
    ///   Defaults to MainScheduler.instance.
    /// - parameter updates: A closure which writes in the database.
    /// - parameter value: A closure which reads from the database.
    public func write<T, U>(
        observeOn scheduler: ImmediateSchedulerType = MainScheduler.instance,
        updates: @escaping (Database) throws -> T,
        thenRead value: @escaping (Database, T) throws -> U)
        -> Single<U>
    {
        return flatMapWrite(
            observeOn: scheduler,
            updates: { db in
                let updatesValue = try updates(db)
                return Observable.create { observer in
                    self.base.spawnConcurrentRead { db in
{'fraction_non_alphanumeric': 0.09790699711007969, 'fraction_numerical': 0.0, 'mean_word_length': 1.655196465938154, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 45, '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': '4600727', 'n_tokens_mistral': 2898, 'n_tokens_neox': 2799, 'n_words': 931}