Address PR feedbacks

Author: mariosangiorgio

Sources/PostgresNIO/Connection/PostgresConnection.swift

@@ -462,15 +462,15 @@ extension PostgresConnection {
     }
 
     /// Execute a prepared statement, taking care of the preparation when necessary
-    public func execute<P: PreparedStatement>(
-        _ preparedStatement: P,
+    public func execute<Statement: PostgresPreparedStatement>(
+        _ preparedStatement: Statement,
         logger: Logger
-    ) async throws -> AsyncThrowingMapSequence<PostgresRowSequence, P.Row>
+    ) async throws -> AsyncThrowingMapSequence<PostgresRowSequence, Statement.Row>
     {
         let promise = self.channel.eventLoop.makePromise(of: PSQLRowStream.self)
         let task = HandlerTask.executePreparedStatement(.init(
-            name: String(reflecting: P.self),
-            sql: P.sql,
+            name: String(reflecting: Statement.self),
+            sql: Statement.sql,
             bindings: preparedStatement.makeBindings(),
             logger: logger,
             promise: promise
@@ -481,6 +481,27 @@ extension PostgresConnection {
             .get()
             .map { try preparedStatement.decodeRow($0) }
     }
+
+    /// Execute a prepared statement, taking care of the preparation when necessary
+    public func execute<Statement: PostgresPreparedStatement>(
+        _ preparedStatement: Statement,
+        logger: Logger
+    ) async throws -> String
+    where Statement.Row == ()
+    {
+        let promise = self.channel.eventLoop.makePromise(of: PSQLRowStream.self)
+        let task = HandlerTask.executePreparedStatement(.init(
+            name: String(reflecting: Statement.self),
+            sql: Statement.sql,
+            bindings: preparedStatement.makeBindings(),
+            logger: logger,
+            promise: promise
+        ))
+        self.channel.write(task, promise: nil)
+        return try await promise.futureResult
+            .map { $0.commandTag }
+            .get()
+    }
 }
 
 // MARK: EventLoopFuture interface

Sources/PostgresNIO/New/Connection State Machine/PreparedStatementStateMachine.swift

@@ -6,29 +6,30 @@ struct PreparedStatementStateMachine {
         case prepared(RowDescription?)
         case error(PSQLError)
     }
+
+    var preparedStatements: [String: State]
 
-    enum Action {
+    init() {
+        self.preparedStatements = [:]
+    }
+
+    enum LookupAction {
         case prepareStatement
         case waitForAlreadyInFlightPreparation
+        case executeStatement(RowDescription?)
         case executePendingStatements([PreparedStatementContext], RowDescription?)
         case returnError([PreparedStatementContext], PSQLError)
     }
-    
-    var preparedStatements: [String: State]
-    
-    init() {
-        self.preparedStatements = [:]
-    }
-    
-    mutating func lookup(name: String, context: PreparedStatementContext) -> Action {
+
+    mutating func lookup(name: String, context: PreparedStatementContext) -> LookupAction {
         if let state = self.preparedStatements[name] {
             switch state {
             case .preparing(var statements):
                 statements.append(context)
                 self.preparedStatements[name] = .preparing(statements)
                 return .waitForAlreadyInFlightPreparation
             case .prepared(let rowDescription):
-                return .executePendingStatements([context], rowDescription)
+                return .executeStatement(rowDescription)
             case .error(let error):
                 return .returnError([context], error)
             }
@@ -37,11 +38,15 @@ struct PreparedStatementStateMachine {
             return .prepareStatement
         }
     }
-    
+
+    enum PreparationCompleteAction {
+        case executePendingStatements([PreparedStatementContext], RowDescription?)
+    }
+
     mutating func preparationComplete(
         name: String,
         rowDescription: RowDescription?
-    ) -> Action {
+    ) -> PreparationCompleteAction {
         guard case .preparing(let statements) = self.preparedStatements[name] else {
             preconditionFailure("Preparation completed for an unexpected statement")
         }
@@ -58,7 +63,11 @@ struct PreparedStatementStateMachine {
         }
     }
 
-    mutating func errorHappened(name: String, error: PSQLError) -> Action {
+    enum ErrorHappenedAction {
+        case returnError([PreparedStatementContext], PSQLError)
+    }
+    
+    mutating func errorHappened(name: String, error: PSQLError) -> ErrorHappenedAction {
         guard case .preparing(let statements) = self.preparedStatements[name] else {
             preconditionFailure("Preparation completed for an unexpected statement")
         }

Sources/PostgresNIO/New/PostgresChannelHandler.swift

@@ -237,10 +237,11 @@ final class PostgresChannelHandler: ChannelDuplexHandler {
                 return
             }
         case .executePreparedStatement(let preparedStatement):
-            switch self.preparedStatementState.lookup(
+            let action = self.preparedStatementState.lookup(
                 name: preparedStatement.name,
                 context: preparedStatement
-            ) {
+            )
+            switch action {
             case .prepareStatement:
                 let promise = self.eventLoop.makePromise(of: RowDescription?.self)
                 promise.futureResult.whenSuccess { rowDescription in
@@ -267,6 +268,15 @@ final class PostgresChannelHandler: ChannelDuplexHandler {
                 // The state machine already keeps track of this
                 // and will execute the statement as soon as it's prepared
                 return
+            case .executeStatement(let rowDescription):
+                psqlTask = .extendedQuery(.init(
+                    executeStatement: .init(
+                        name: preparedStatement.name,
+                        binds: preparedStatement.bindings,
+                        rowDescription: rowDescription),
+                    logger: preparedStatement.logger,
+                    promise: preparedStatement.promise
+                ))
             case .executePendingStatements(let pendingStatements, let rowDescription):
                 for statement in pendingStatements {
                     let action = self.state.enqueue(task: .extendedQuery(.init(

Sources/PostgresNIO/New/PreparedStatement.swift

@@ -5,7 +5,7 @@
 ///
 /// As an example, consider this struct:
 /// ```swift
-/// struct Example: PreparedStatement {
+/// struct Example: PostgresPreparedStatement {
 ///     static var sql = "SELECT pid, datname FROM pg_stat_activity WHERE state = $1"
 ///     typealias Row = (Int, String)
 ///
@@ -25,7 +25,7 @@
 ///
 /// Structs conforming to this protocol can then be used with `PostgresConnection.execute(_ preparedStatement:, logger:)`,
 /// which will take care of preparing the statement on the server side and executing it.
-public protocol PreparedStatement {
+public protocol PostgresPreparedStatement: Sendable {
     /// The type rows returned by the statement will be decoded into
     associatedtype Row
 

Tests/IntegrationTests/AsyncTests.swift

@@ -321,7 +321,7 @@ final class AsyncPostgresConnectionTests: XCTestCase {
         defer { XCTAssertNoThrow(try eventLoopGroup.syncShutdownGracefully()) }
         let eventLoop = eventLoopGroup.next()
 
-        struct TestPreparedStatement: PreparedStatement {
+        struct TestPreparedStatement: PostgresPreparedStatement {
             static var sql = "SELECT pid, datname FROM pg_stat_activity WHERE state = $1"
             typealias Row = (Int, String)