From 527f53c17db2bf2942bb67862a66bdce9db6983b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Istv=C3=A1n=20So=C3=B3s?= <isoos@users.noreply.github.com>
Date: Sun, 22 Dec 2024 19:13:43 +0100
Subject: [PATCH] Fix: 'close(force: true)' does not cause uncaught exception
 (#404)

---
 CHANGELOG.md               |  1 +
 lib/src/v3/connection.dart |  9 +++------
 test/v3_close_test.dart    | 14 ++++----------
 3 files changed, 8 insertions(+), 16 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 780985c..6386403 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,7 @@
 ## 3.4.6
 
 - Fix: do not close connection after an empty statement.
+- Fix: `close(force: true)` does not cause uncaught exception.
 
 ## 3.4.5
 
diff --git a/lib/src/v3/connection.dart b/lib/src/v3/connection.dart
index e3106ef..09479b3 100644
--- a/lib/src/v3/connection.dart
+++ b/lib/src/v3/connection.dart
@@ -583,7 +583,8 @@ class PgConnectionImplementation extends _PgSessionBase implements Connection {
 
   @override
   Future<void> close({bool force = false}) async {
-    await _close(force, null);
+    final ex = force ? PgException('Connection closed.') : null;
+    await _close(force, ex);
   }
 
   Future<void> _close(bool interruptRunning, PgException? cause,
@@ -594,9 +595,6 @@ class PgConnectionImplementation extends _PgSessionBase implements Connection {
       try {
         if (interruptRunning) {
           _pending?.handleConnectionClosed(cause);
-          if (!_socketIsBroken) {
-            _channel.sink.add(const TerminateMessage());
-          }
         } else {
           // Wait for the previous operation to complete by using the lock
           await _operationLock.withResource(() {
@@ -823,8 +821,6 @@ class _PgResultStreamSubscription
   Future<ResultSchema> get schema => _schema.future;
 
   Future<void> _completeQuery() async {
-    _done.complete();
-
     // Make sure the affectedRows and schema futures complete with something
     // after the query is done, even if we didn't get a row description
     // message.
@@ -834,6 +830,7 @@ class _PgResultStreamSubscription
     if (!_schema.isCompleted) {
       _schema.complete(ResultSchema(const []));
     }
+    _done.complete();
     await _controller.close();
   }
 
diff --git a/test/v3_close_test.dart b/test/v3_close_test.dart
index c882f9d..ad3d577 100644
--- a/test/v3_close_test.dart
+++ b/test/v3_close_test.dart
@@ -108,8 +108,7 @@ void main() {
         await Future.delayed(const Duration(milliseconds: 100));
         await expectConn1ClosesForcefully(conn);
 
-        // TODO: this should be throwing PgException
-        await expectLater(() => rs, isNotNull);
+        await expectLater(() => rs, throwsA(isA<PgException>()));
       });
     });
 
@@ -120,17 +119,13 @@ void main() {
         final rs = conn.runTx((tx) async {
           started.complete();
           await runLongQuery(tx);
-        })
-            // Ignore async error, it will fail when the connection is closed and it tries to do COMMIT
-            // TODO: remove this ignore
-            .ignore();
+        });
         // let it start
         await started.future;
         await Future.delayed(const Duration(milliseconds: 100));
         await expectConn1ClosesForcefully(conn);
 
-        // TODO: this should be throwing PgException
-        await expectLater(() => rs, isNotNull);
+        await expectLater(() => rs, throwsA(isA<PgException>()));
       });
     });
 
@@ -147,8 +142,7 @@ void main() {
         await Future.delayed(const Duration(milliseconds: 100));
         await expectConn1ClosesForcefully(conn);
 
-        // TODO: this should be throwing PgException
-        await expectLater(() => rs, isNotNull);
+        await expectLater(() => rs, throwsA(isA<PgException>()));
       });
     });
   });