Fix: 'close(force: true)' does not cause uncaught exception (#404)

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
 

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();
   }
 

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>()));
       });
     });
   });