Connection now exposes an AsyncStream for recieving server notices. Improvements to connection state tracking.

Author: willtemperley

Sources/SwiftPostgresClient/Connection+direct.swift

@@ -0,0 +1,43 @@
+//
+//  Connection+transaction.swift
+//  SwiftPostgresClient
+//
+//  Copyright 2025 Will Temperley and the SwiftPostgresClient contributors.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+/// Execution of queries without using extended query protocol.
+extension Connection {
+    
+    func execute(_ sql: String) async throws {
+        if state != .ready {
+            throw PostgresError.invalidState("Expected ready state, got \(state)")
+        }
+        let queryRequest = QueryRequest(query: sql)
+        try await sendRequest(queryRequest)
+        try await receiveResponse(type: CommandCompleteResponse.self)
+        try await receiveResponse(type: ReadyForQueryResponse.self)
+    }
+    
+    func query(_ sql: String) async throws -> ResultCursor {
+        if state != .ready {
+            throw PostgresError.invalidState("Expected ready state, got \(state)")
+        }
+        let queryRequest = QueryRequest(query: sql)
+        try await sendRequest(queryRequest)
+        try await receiveResponse(type: RowDescriptionResponse.self)
+        let response = await receiveResponse()
+        return ResultCursor(connection: self, portalName: "", extendedProtocol: false, initialResponse: response)
+    }
+}

Sources/SwiftPostgresClient/Connection+metadata.swift

@@ -0,0 +1,55 @@
+//
+//  Connection+metadata.swift
+//  SwiftPostgresClient
+//
+//  Copyright 2025 Will Temperley and the SwiftPostgresClient contributors.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+/// Methods to query connection status, using server state instead of attempting to track status client-side.
+extension Connection {
+    
+    func listPreparedStatements() async throws -> [String] {
+        
+        let text = "SELECT name FROM pg_prepared_statements"
+        
+        let cursor = try await query(text)
+        
+        var data: [String] = .init()
+        for try await row in cursor {
+            data.append(try row.columns[0].string())
+        }
+        try await receiveResponse(type: ReadyForQueryResponse.self)
+        return data
+    }
+    
+    func listOpenPortals() async throws -> [String] {
+        
+        let text = "SELECT name FROM pg_cursor"
+        
+        let cursor = try await query(text)
+        
+        var data: [String] = .init()
+        for try await row in cursor {
+            data.append(try row.columns[0].string())
+        }
+        try await receiveResponse(type: ReadyForQueryResponse.self)
+        return data
+    }
+
+    func statementClosed(_ name: String) async -> Bool {
+        !openStatements.contains(name)
+    }
+    
+}

Sources/SwiftPostgresClient/Connection+portal.swift

@@ -26,6 +26,11 @@ extension Connection {
         parameterValues: [PostgresValueConvertible?] = [],
         columnMetadata: Bool = false
     ) async throws -> Portal {
+        
+        if state == .closed {
+            throw PostgresError.connectionClosed
+        }
+        
         let portalName = UUID().uuidString
         let bindRequest = BindRequest(name: portalName, statement: statement, parameterValues: parameterValues)
         try await sendRequest(bindRequest)
@@ -39,27 +44,62 @@ extension Connection {
         if columnMetadata {
             metadata = try await retrieveColumnMetadata(portalName: portalName)
         }
-        portalStatus[portalName] = .open
 
         return Portal(name: portalName, metadata: metadata, statement: statement, connection: self)
     }
 
+    func query(
+        portalName: String,
+        statement: Statement,
+        metadata: [ColumnMetadata]?
+    ) async throws -> ResultCursor {
+        
+        state = .querySent
+        
+        let executeRequest = ExecuteRequest(portalName: portalName, statement: statement)
+        try await sendRequest(executeRequest)
+        
+        let flushRequest = FlushRequest()
+        try await sendRequest(flushRequest)
+        
+        // The first response of the query is evaluated eagerly to check its type.
+        let response = await receiveResponse()
+        
+        // A zero-result query immediately gives a CommandCompleteResponse
+        if response is CommandCompleteResponse {
+            try await cleanupPortal(name: portalName)
+        }
+
+        return ResultCursor(connection: self, portalName: portalName, metadata: metadata, initialResponse: response)
+    }
+    
+    func execute(
+        portalName: String,
+        statement: Statement
+    ) async throws -> CommandStatus {
+        
+        state = .querySent
+        
+        let executeRequest = ExecuteRequest(portalName: portalName, statement: statement)
+        try await sendRequest(executeRequest)
+        let flushRequest = FlushRequest()
+        try await sendRequest(flushRequest)
+
+        let response = try await receiveResponse(type: CommandCompleteResponse.self)
+        try await cleanupPortal(name: portalName)
+        return response.status
+    }
+    
     func cleanupPortal(name: String) async throws {
         // Close the portal
         try await sendRequest(ClosePortalRequest(name: name))
         try await sendRequest(FlushRequest())
         try await receiveResponse(type: CloseCompleteResponse.self)
 
-        let status = portalStatus.removeValue(forKey: name)
-        if status == nil {
-            throw PostgresError.protocolError("Missing portal status for \(name)")
-        }
-        
         // Finalize the transaction (unless already in BEGIN/COMMIT)
         try await sendRequest(SyncRequest())
 
         // transaction status always set on ReadyForQueryResponse
         try await receiveResponse(type: ReadyForQueryResponse.self)
     }
-
 }

…iftPostgresClient/Connection+query.swift …ostgresClient/Connection+statement.swiftSources/SwiftPostgresClient/Connection+query.swift renamed to Sources/SwiftPostgresClient/Connection+statement.swift Sources/SwiftPostgresClient/Connection+query.swift renamed to Sources/SwiftPostgresClient/Connection+statement.swift

@@ -30,6 +30,10 @@ extension Connection {
     /// - Returns: a `PreparedStatement` ready for parameter binding.
     public func prepareStatement(text: String) async throws -> PreparedStatement {
 
+        if self.state != .ready {
+            throw PostgresError.invalidState("Expected connection to be in state .ready, but was \(self.state)")
+        }
+        
         let statement = Statement(text: text)
         let parseRequest = ParseRequest(statement: statement)
         try await sendRequest(parseRequest)
@@ -39,34 +43,12 @@ extension Connection {
 
         try await receiveResponse(type: ParseCompleteResponse.self)
 
+        self.openStatements.insert(statement.name)
         return PreparedStatement(name: statement.name, statement: statement, connection: self)
     }
 
-    func query(
-        portalName: String,
-        statement: Statement,
-        metadata: [ColumnMetadata]?
-    ) async throws -> ResultCursor {
-        
-        state = .querySent
-        
-        let executeRequest = ExecuteRequest(portalName: portalName, statement: statement)
-        try await sendRequest(executeRequest)
-        
-        let flushRequest = FlushRequest()
-        try await sendRequest(flushRequest)
-        
-        // The first response of the query is evaluated eagerly to check its
-        // type.
-        // An zero-result query could immediately give a CommandCompleteResponse
-        // This also prevents some issues with undrained message queues.
-        // TODO: check error response
-        let response = await receiveResponse()
-
-        return ResultCursor(connection: self, portalName: portalName, metadata: metadata, initialResponse: response)
-    }
-    
-    public func closeStatement(name: String) async throws {
+    /// Close a prepared statement
+    func closeStatement(name: String) async throws {
 
         let closeStatementRequest = CloseStatementRequest(name: name)
         try await sendRequest(closeStatementRequest)
@@ -75,9 +57,10 @@ extension Connection {
         try await sendRequest(flushRequest)
 
         try await receiveResponse(type: CloseCompleteResponse.self)
+        self.openStatements.remove(name)
     }
 
-    // used for decoding structs using column names
+    /// Returns metadata used for decoding structs via column names
     func retrieveColumnMetadata(portalName: String) async throws -> [ColumnMetadata]? {
 
         var columns: [ColumnMetadata]? = nil

Sources/SwiftPostgresClient/Connection+transaction.swift

@@ -17,8 +17,21 @@
 //  limitations under the License.
 //
 
+/// Transaction convenience methods
 extension Connection {
 
+    public func beginTransaction() async throws {
+        try await execute("BEGIN;")
+    }
+    
+    public func commitTransaction() async throws {
+        try await execute("COMMIT;")
+    }
+    
+    func rollback() async throws {
+        try await execute("ROLLBACK;")
+    }
+
     /// Runs a block within a transaction. Commits on success, rolls back on error.
     func withTransaction<T>(_ operation: @Sendable () async throws -> T) async throws -> T where T: Sendable {
         try await beginTransaction()
@@ -35,26 +48,4 @@ extension Connection {
             throw error
         }
     }
-    
-    public func beginTransaction() async throws {
-        try await executeSimpleQuery("BEGIN;")
-    }
-    
-    public func commitTransaction() async throws {
-        try await executeSimpleQuery("COMMIT;")
-    }
-    
-    func rollback() async throws {
-        try await executeSimpleQuery("ROLLBACK;")
-    }
-    
-    func executeSimpleQuery(_ sql: String) async throws {
-        let queryRequest = QueryRequest(query: sql)
-        try await sendRequest(queryRequest)
-        try await receiveResponse(type: CommandCompleteResponse.self)
-        try await receiveResponse(type: ReadyForQueryResponse.self)
-#if DEBUG
-        print("Transaction status: \(transactionStatus)")
-#endif
-    }
 }