         self.writer = try writer(directoryURL.appendingPathComponent("db.sqlite").path)
     }
     
+    @discardableResult
+    func test(with testClosure: (Writer) throws -> ()) rethrows -> Self {
+        try testClosure(writer)
+        return self
+    }
+    
     deinit {
+        self.writer = nil
         try! FileManager.default.removeItem(at: directoryURL)
     }
 }

            }
        }
        
        var changed1 = false
        var changed2 = false
        var changed3 = false
        
        let request1 = SQLRequest("SELECT * FROM table1")
        let request2 = SQLRequest("SELECT id, a FROM table1")
        let request3 = SQLRequest("SELECT table1.id, table1.a, table2.a FROM table1 JOIN table2 ON table1.id = table2.id")
        
        request1.rx.selection(in: writer).subscribe(onNext: { _ in changed1 = true }).addDisposableTo(disposeBag)
        request2.rx.selection(in: writer).subscribe(onNext: { _ in changed2 = true }).addDisposableTo(disposeBag)
        request3.rx.selection(in: writer).subscribe(onNext: { _ in changed3 = true }).addDisposableTo(disposeBag)
        
        // Subscription immediately triggers an event
        XCTAssertTrue(changed1)
        XCTAssertTrue(changed2)
        XCTAssertTrue(changed3)
        
        try writer.write { db in
            func reset() {
                changed1 = false
                changed2 = false
                changed3 = false
            }
            
            
            // Transaction triggers an event for concerned requests
            reset()
            try db.inTransaction {
                try db.execute("INSERT INTO table1 (id, a, b) VALUES (NULL, 0, 0)")
                XCTAssertFalse(changed1)
                XCTAssertFalse(changed2)
                XCTAssertFalse(changed3)
                return .commit
            }
            XCTAssertTrue(changed1)
            XCTAssertTrue(changed2)
            XCTAssertTrue(changed3)
            
            // Transaction triggers an event for concerned requests
            reset()
            try db.execute("INSERT INTO table2 (id, a, b) VALUES (NULL, 0, 0)")
            XCTAssertFalse(changed1)
            XCTAssertFalse(changed2)
            XCTAssertTrue(changed3)
            
            // Transaction triggers an event for concerned requests
            reset()
            try db.execute("UPDATE table1 SET a = 1")
            XCTAssertTrue(changed1)
            XCTAssertTrue(changed2)
            XCTAssertTrue(changed3)
            
            // Transaction triggers an event for concerned requests
            reset()
            try db.execute("UPDATE table1 SET b = 1")
            XCTAssertTrue(changed1)
            XCTAssertFalse(changed2)
            XCTAssertFalse(changed3)
            
            // Transaction triggers an event for concerned requests
            reset()
            try db.execute("UPDATE table2 SET a = 1")
            XCTAssertFalse(changed1)
            XCTAssertFalse(changed2)
            XCTAssertTrue(changed3)
            
            // Transaction triggers an event for concerned requests
            reset()
            try db.execute("UPDATE table2 SET b = 1")
            XCTAssertFalse(changed1)
            XCTAssertFalse(changed2)
            XCTAssertFalse(changed3)
        }
    }
    
    func testRequestRxSelection() throws {
        let testQueue = try TestDatabase { try DatabaseQueue(path: $0) }
        try testRequestRxSelection(writer: testQueue.writer)
        
        let testPool = try TestDatabase { try DatabasePool(path: $0) }
        try testRequestRxSelection(writer: testPool.writer)
    }
    
    func testRequestRxFetchAll(writer: DatabaseWriter) throws {
        class Person: Record {
            var id: Int64?
            var name: String
            
            init(id: Int64?, name: String) {
                self.id = id
                self.name = name
                super.init()
            }
            
            required init(row: Row) {
                id = row.value(named: "id")
                name = row.value(named: "name")
                super.init(row: row)
            }
            
            override class var databaseTableName: String { return "persons" }
            
            override var persistentDictionary: [String : DatabaseValueConvertible?] {
                return ["id": id, "name": name]
            }
            
            override func didInsert(with rowID: Int64, for column: String?) {
                id = rowID
            }
        }
        
        try writer.write { db in
            try db.create(table: "persons") { t in
                t.column("id", .integer).primaryKey()

                t.column("b", .integer)
            }
            
            try selectionInfos.append(db.makeSelectStatement("SELECT * FROM table1").selectionInfo)
            try selectionInfos.append(db.makeSelectStatement("SELECT id, a FROM table1").selectionInfo)
            try selectionInfos.append(db.makeSelectStatement("SELECT table1.id, table1.a, table2.a FROM table1 JOIN table2 ON table1.id = table2.id").selectionInfo)
        }
        
        var changes = selectionInfos.map { _ in false }
        for (index, selectionInfo) in selectionInfos.enumerated() {
            selectionInfo.rx
                .selection(in: writer)
                .subscribe(onNext: { _ in changes[index] = true })
                .addDisposableTo(disposeBag)
        }
        
        // Subscription immediately triggers an event
        XCTAssertEqual(changes, [true, true, true])
        
        try writer.write { db in
            func reset() { changes = changes.map { _ in false } }
            
            // Transaction triggers an event for concerned selection
            reset()
            try db.inTransaction {
                try db.execute("INSERT INTO table1 (id, a, b) VALUES (NULL, 0, 0)")
                XCTAssertEqual(changes, [false, false, false])
                return .commit
            }
            XCTAssertEqual(changes, [true, true, true])
            
            // Transaction triggers an event for concerned selection
            reset()
            try db.execute("INSERT INTO table2 (id, a, b) VALUES (NULL, 0, 0)")
            XCTAssertEqual(changes, [false, false, true])
            
            // Transaction triggers an event for concerned selection
            reset()
            try db.execute("UPDATE table1 SET a = 1")
            XCTAssertEqual(changes, [true, true, true])
            
            // Transaction triggers an event for concerned selection
            reset()
            try db.execute("UPDATE table1 SET b = 1")
            XCTAssertEqual(changes, [true, false, false])
            
            // Transaction triggers an event for concerned selection
            reset()
            try db.execute("UPDATE table2 SET a = 1")
            XCTAssertEqual(changes, [false, false, true])
            
            // Transaction triggers an event for concerned selection
            reset()
            try db.execute("UPDATE table2 SET b = 1")
            XCTAssertEqual(changes, [false, false, false])
        }
    }
}

extension ReactiveRequestTests {
    func testRequestRxSelection() throws {
        try TestDatabase({ try DatabaseQueue(path: $0) }).test(with: testRequestRxSelection)
        try TestDatabase({ try DatabasePool(path: $0) }).test(with: testRequestRxSelection)
    }
    
    func testRequestRxSelection(writer: DatabaseWriter) throws {
