Fix Sendable warnings (#536)

Author: fabianfett

Package.swift

@@ -20,7 +20,7 @@ let package = Package(
     dependencies: [
         .package(url: "https://github.com/apple/swift-atomics.git", from: "1.2.0"),
         .package(url: "https://github.com/apple/swift-collections.git", from: "1.0.4"),
-        .package(url: "https://github.com/apple/swift-nio.git", from: "2.59.0"),
+        .package(url: "https://github.com/apple/swift-nio.git", from: "2.81.0"),
         .package(url: "https://github.com/apple/swift-nio-transport-services.git", from: "1.19.0"),
         .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.25.0"),
         .package(url: "https://github.com/apple/swift-crypto.git", "2.0.0" ..< "4.0.0"),

Tests/IntegrationTests/PostgresNIOTests.swift

@@ -1032,28 +1032,6 @@ final class PostgresNIOTests: XCTestCase {
         }
     }
 
-    func testRemoteTLSServer() {
-        // postgres://uymgphwj:[email protected]:5432/uymgphwj
-        var conn: PostgresConnection?
-        let logger = Logger(label: "test")
-        let sslContext = try! NIOSSLContext(configuration: .makeClientConfiguration())
-        let config = PostgresConnection.Configuration(
-            host: "elmer.db.elephantsql.com",
-            port: 5432,
-            username: "uymgphwj",
-            password: "7_tHbREdRwkqAdu4KoIS7hQnNxr8J1LA",
-            database: "uymgphwj",
-            tls: .require(sslContext)
-        )
-        XCTAssertNoThrow(conn = try PostgresConnection.connect(on: eventLoop, configuration: config, id: 0, logger: logger).wait())
-        defer { XCTAssertNoThrow( try conn?.close().wait() ) }
-        var rows: [PostgresRow]?
-        XCTAssertNoThrow(rows = try conn?.simpleQuery("SELECT version()").wait())
-        XCTAssertEqual(rows?.count, 1)
-        let row = rows?.first?.makeRandomAccess()
-        XCTAssertEqual(row?[data: "version"].string?.contains("PostgreSQL"), true)
-    }
-
     @available(*, deprecated, message: "Test deprecated functionality")
     func testFailingTLSConnectionClosesConnection() {
         // There was a bug (https://github.com/vapor/postgres-nio/issues/133) where we would hit

Tests/PostgresNIOTests/New/PostgresChannelHandlerTests.swift

@@ -124,7 +124,7 @@ class PostgresChannelHandlerTests: XCTestCase {
             handler
         ], loop: self.eventLoop)
         let eventHandler = TestEventHandler()
-        try embedded.pipeline.addHandler(eventHandler, position: .last).wait()
+        try embedded.pipeline.syncOperations.addHandler(eventHandler, position: .last)
 
         embedded.connect(to: try .init(ipAddress: "0.0.0.0", port: 5432), promise: nil)
         XCTAssertTrue(embedded.isActive)

Tests/PostgresNIOTests/New/PostgresConnectionTests.swift

@@ -40,10 +40,10 @@ class PostgresConnectionTests: XCTestCase {
 
     func testOptionsAreSentOnTheWire() async throws {
         let eventLoop = NIOAsyncTestingEventLoop()
-        let channel = await NIOAsyncTestingChannel(handlers: [
-            ReverseByteToMessageHandler(PSQLFrontendMessageDecoder()),
-            ReverseMessageToByteHandler(PSQLBackendMessageEncoder()),
-        ], loop: eventLoop)
+        let channel = try await NIOAsyncTestingChannel(loop: eventLoop) { channel in
+            try channel.pipeline.syncOperations.addHandlers(ReverseByteToMessageHandler(PSQLFrontendMessageDecoder()))
+            try channel.pipeline.syncOperations.addHandlers(ReverseMessageToByteHandler(PSQLBackendMessageEncoder()))
+        }
         try await channel.connect(to: .makeAddressResolvingHost("localhost", port: 5432))
 
         let configuration = {
@@ -640,10 +640,10 @@ class PostgresConnectionTests: XCTestCase {
 
     func makeTestConnectionWithAsyncTestingChannel() async throws -> (PostgresConnection, NIOAsyncTestingChannel) {
         let eventLoop = NIOAsyncTestingEventLoop()
-        let channel = await NIOAsyncTestingChannel(handlers: [
-            ReverseByteToMessageHandler(PSQLFrontendMessageDecoder()),
-            ReverseMessageToByteHandler(PSQLBackendMessageEncoder()),
-        ], loop: eventLoop)
+        let channel = try await NIOAsyncTestingChannel(loop: eventLoop) { channel in
+            try channel.pipeline.syncOperations.addHandlers(ReverseByteToMessageHandler(PSQLFrontendMessageDecoder()))
+            try channel.pipeline.syncOperations.addHandlers(ReverseMessageToByteHandler(PSQLBackendMessageEncoder()))
+        }
         try await channel.connect(to: .makeAddressResolvingHost("localhost", port: 5432))
 
         let configuration = PostgresConnection.Configuration(

Tests/PostgresNIOTests/New/PostgresRowSequenceTests.swift

@@ -1,25 +1,25 @@
 import Atomics
 import NIOEmbedded
-import Dispatch
+import NIOPosix
 import XCTest
 @testable import PostgresNIO
 import NIOCore
 import Logging
 
 final class PostgresRowSequenceTests: XCTestCase {
     let logger = Logger(label: "PSQLRowStreamTests")
-    let eventLoop = EmbeddedEventLoop()
 
     func testBackpressureWorks() async throws {
         let dataSource = MockRowDataSource()
+        let embeddedEventLoop = EmbeddedEventLoop()
         let stream = PSQLRowStream(
             source: .stream(
                 [
                     .init(name: "test", tableOID: 0, columnAttributeNumber: 0, dataType: .int8, dataTypeSize: 8, dataTypeModifier: 0, format: .binary)
                 ], 
                 dataSource
             ),
-            eventLoop: self.eventLoop,
+            eventLoop: embeddedEventLoop,
             logger: self.logger
         )
 
@@ -41,14 +41,15 @@ final class PostgresRowSequenceTests: XCTestCase {
 
     func testCancellationWorksWhileIterating() async throws {
         let dataSource = MockRowDataSource()
+        let embeddedEventLoop = EmbeddedEventLoop()
         let stream = PSQLRowStream(
             source: .stream(
                 [
                     .init(name: "test", tableOID: 0, columnAttributeNumber: 0, dataType: .int8, dataTypeSize: 8, dataTypeModifier: 0, format: .binary)
                 ],
                 dataSource
             ),
-            eventLoop: self.eventLoop,
+            eventLoop: embeddedEventLoop,
             logger: self.logger
         )
 
@@ -72,14 +73,15 @@ final class PostgresRowSequenceTests: XCTestCase {
 
     func testCancellationWorksBeforeIterating() async throws {
         let dataSource = MockRowDataSource()
+        let embeddedEventLoop = EmbeddedEventLoop()
         let stream = PSQLRowStream(
             source: .stream(
                 [
                     .init(name: "test", tableOID: 0, columnAttributeNumber: 0, dataType: .int8, dataTypeSize: 8, dataTypeModifier: 0, format: .binary)
                 ],
                 dataSource
             ),
-            eventLoop: self.eventLoop,
+            eventLoop: embeddedEventLoop,
             logger: self.logger
         )
 
@@ -97,14 +99,15 @@ final class PostgresRowSequenceTests: XCTestCase {
 
     func testDroppingTheSequenceCancelsTheSource() async throws {
         let dataSource = MockRowDataSource()
+        let embeddedEventLoop = EmbeddedEventLoop()
         let stream = PSQLRowStream(
             source: .stream(
                 [
                     .init(name: "test", tableOID: 0, columnAttributeNumber: 0, dataType: .int8, dataTypeSize: 8, dataTypeModifier: 0, format: .binary)
                 ],
                 dataSource
             ),
-            eventLoop: self.eventLoop,
+            eventLoop: embeddedEventLoop,
             logger: self.logger
         )
 
@@ -117,14 +120,15 @@ final class PostgresRowSequenceTests: XCTestCase {
 
     func testStreamBasedOnCompletedQuery() async throws {
         let dataSource = MockRowDataSource()
+        let embeddedEventLoop = EmbeddedEventLoop()
         let stream = PSQLRowStream(
             source: .stream(
                 [
                     .init(name: "test", tableOID: 0, columnAttributeNumber: 0, dataType: .int8, dataTypeSize: 8, dataTypeModifier: 0, format: .binary)
                 ],
                 dataSource
             ),
-            eventLoop: self.eventLoop,
+            eventLoop: embeddedEventLoop,
             logger: self.logger
         )
 
@@ -144,14 +148,15 @@ final class PostgresRowSequenceTests: XCTestCase {
 
     func testStreamIfInitializedWithAllData() async throws {
         let dataSource = MockRowDataSource()
+        let embeddedEventLoop = EmbeddedEventLoop()
         let stream = PSQLRowStream(
             source: .stream(
                 [
                     .init(name: "test", tableOID: 0, columnAttributeNumber: 0, dataType: .int8, dataTypeSize: 8, dataTypeModifier: 0, format: .binary)
                 ],
                 dataSource
             ),
-            eventLoop: self.eventLoop,
+            eventLoop: embeddedEventLoop,
             logger: self.logger
         )
 
@@ -172,14 +177,15 @@ final class PostgresRowSequenceTests: XCTestCase {
 
     func testStreamIfInitializedWithError() async throws {
         let dataSource = MockRowDataSource()
+        let embeddedEventLoop = EmbeddedEventLoop()
         let stream = PSQLRowStream(
             source: .stream(
                 [
                     .init(name: "test", tableOID: 0, columnAttributeNumber: 0, dataType: .int8, dataTypeSize: 8, dataTypeModifier: 0, format: .binary)
                 ],
                 dataSource
             ),
-            eventLoop: self.eventLoop,
+            eventLoop: embeddedEventLoop,
             logger: self.logger
         )
 
@@ -200,29 +206,30 @@ final class PostgresRowSequenceTests: XCTestCase {
 
     func testSucceedingRowContinuationsWorks() async throws {
         let dataSource = MockRowDataSource()
+        let eventLoop = NIOSingletons.posixEventLoopGroup.next()
         let stream = PSQLRowStream(
             source: .stream(
                 [
                     .init(name: "test", tableOID: 0, columnAttributeNumber: 0, dataType: .int8, dataTypeSize: 8, dataTypeModifier: 0, format: .binary)
                 ],
                 dataSource
             ),
-            eventLoop: self.eventLoop,
+            eventLoop: eventLoop,
             logger: self.logger
         )
 
-        let rowSequence = stream.asyncSequence()
+        let rowSequence = try await eventLoop.submit { stream.asyncSequence() }.get()
         var rowIterator = rowSequence.makeAsyncIterator()
 
-        DispatchQueue.main.asyncAfter(deadline: .now() + .seconds(1)) {
+        eventLoop.scheduleTask(in: .seconds(1)) {
             let dataRows: [DataRow] = (0..<1).map { [ByteBuffer(integer: Int64($0))] }
             stream.receive(dataRows)
         }
 
         let row1 = try await rowIterator.next()
         XCTAssertEqual(try row1?.decode(Int.self), 0)
 
-        DispatchQueue.main.asyncAfter(deadline: .now() + .seconds(1)) {
+        eventLoop.scheduleTask(in: .seconds(1)) {
             stream.receive(completion: .success("SELECT 1"))
         }
 
@@ -232,29 +239,30 @@ final class PostgresRowSequenceTests: XCTestCase {
 
     func testFailingRowContinuationsWorks() async throws {
         let dataSource = MockRowDataSource()
+        let eventLoop = NIOSingletons.posixEventLoopGroup.next()
         let stream = PSQLRowStream(
             source: .stream(
                 [
                     .init(name: "test", tableOID: 0, columnAttributeNumber: 0, dataType: .int8, dataTypeSize: 8, dataTypeModifier: 0, format: .binary)
                 ],
                 dataSource
             ),
-            eventLoop: self.eventLoop,
+            eventLoop: eventLoop,
             logger: self.logger
         )
 
-        let rowSequence = stream.asyncSequence()
+        let rowSequence = try await eventLoop.submit { stream.asyncSequence() }.get()
         var rowIterator = rowSequence.makeAsyncIterator()
 
-        DispatchQueue.main.asyncAfter(deadline: .now() + .seconds(1)) {
+        eventLoop.scheduleTask(in: .seconds(1)) {
             let dataRows: [DataRow] = (0..<1).map { [ByteBuffer(integer: Int64($0))] }
             stream.receive(dataRows)
         }
 
         let row1 = try await rowIterator.next()
         XCTAssertEqual(try row1?.decode(Int.self), 0)
 
-        DispatchQueue.main.asyncAfter(deadline: .now() + .seconds(1)) {
+        eventLoop.scheduleTask(in: .seconds(1)) {
             stream.receive(completion: .failure(PSQLError.serverClosedConnection(underlying: nil)))
         }
 
@@ -268,14 +276,15 @@ final class PostgresRowSequenceTests: XCTestCase {
 
     func testAdaptiveRowBufferShrinksAndGrows() async throws {
         let dataSource = MockRowDataSource()
+        let embeddedEventLoop = EmbeddedEventLoop()
         let stream = PSQLRowStream(
             source: .stream(
                 [
                     .init(name: "test", tableOID: 0, columnAttributeNumber: 0, dataType: .int8, dataTypeSize: 8, dataTypeModifier: 0, format: .binary)
                 ],
                 dataSource
             ),
-            eventLoop: self.eventLoop,
+            eventLoop: embeddedEventLoop,
             logger: self.logger
         )
 
@@ -332,14 +341,15 @@ final class PostgresRowSequenceTests: XCTestCase {
 
     func testAdaptiveRowShrinksToMin() async throws {
         let dataSource = MockRowDataSource()
+        let embeddedEventLoop = EmbeddedEventLoop()
         let stream = PSQLRowStream(
             source: .stream(
                 [
                     .init(name: "test", tableOID: 0, columnAttributeNumber: 0, dataType: .int8, dataTypeSize: 8, dataTypeModifier: 0, format: .binary)
                 ],
                 dataSource
             ),
-            eventLoop: self.eventLoop,
+            eventLoop: embeddedEventLoop,
             logger: self.logger
         )
 
@@ -386,14 +396,15 @@ final class PostgresRowSequenceTests: XCTestCase {
 
     func testStreamBufferAcceptsNewRowsEventhoughItDidntAskForIt() async throws {
         let dataSource = MockRowDataSource()
+        let embeddedEventLoop = EmbeddedEventLoop()
         let stream = PSQLRowStream(
             source: .stream(
                 [
                     .init(name: "test", tableOID: 0, columnAttributeNumber: 0, dataType: .int8, dataTypeSize: 8, dataTypeModifier: 0, format: .binary)
                 ],
                 dataSource
             ),
-            eventLoop: self.eventLoop,
+            eventLoop: embeddedEventLoop,
             logger: self.logger
         )