Add isOpen and closed (#222)

Author: simolus3

lib/postgres.dart

@@ -87,6 +87,20 @@ class Sql {
 }
 
 abstract class Session {
+  /// Whether this connection is currently open.
+  ///
+  /// A [Connection] is open until it's closed (either by an explicit
+  /// [Connection.close] call or due to an unrecoverable error from the server).
+  /// Other sessions, such as transactions or connections borrowed from a pool,
+  /// may have a shorter lifetime.
+  ///
+  /// The [closed] future can be awaited to get notified when this session is
+  /// closing.
+  bool get isOpen;
+
+  /// A future that completes when [isOpen] turns false.
+  Future<void> get closed;
+
   /// Prepares a reusable statement from a [query].
   ///
   /// [query] must either be a [String] or a [Sql] object with types for

lib/src/pool/pool_impl.dart

@@ -34,6 +34,12 @@ class PoolImplementation<L> implements Pool<L> {
   PoolImplementation(this._selector, PoolSettings? settings)
       : _settings = ResolvedPoolSettings(settings);
 
+  @override
+  bool get isOpen => !_semaphore.isClosed;
+
+  @override
+  Future<void> get closed => _semaphore.done;
+
   @override
   Future<void> close() async {
     await _semaphore.close();
@@ -251,6 +257,12 @@ class _PoolConnection implements Connection {
     await _connection.close();
   }
 
+  @override
+  bool get isOpen => _connection.isOpen;
+
+  @override
+  Future<void> get closed => _connection.closed;
+
   @override
   Channels get channels {
     throw UnsupportedError(

lib/src/v3/connection.dart

@@ -38,12 +38,20 @@ abstract class _PgSessionBase implements Session {
   /// connection in the meantime.
   final _operationLock = pool.Pool(1);
 
-  bool _sessionClosed = false;
+  final Completer<void> _sessionClosedCompleter = Completer();
+
+  bool get _sessionClosed => _sessionClosedCompleter.isCompleted;
 
   PgConnectionImplementation get _connection;
   ResolvedSessionSettings get _settings;
   Encoding get encoding => _connection._settings.encoding;
 
+  void _closeSession() {
+    if (!_sessionClosed) {
+      _sessionClosedCompleter.complete();
+    }
+  }
+
   void _checkActive() {
     if (_sessionClosed) {
       throw PgException(
@@ -88,6 +96,12 @@ abstract class _PgSessionBase implements Session {
     });
   }
 
+  @override
+  bool get isOpen => !_sessionClosed && !_connection._isClosing;
+
+  @override
+  Future<void> get closed => _sessionClosedCompleter.future;
+
   @override
   Future<Result> execute(
     Object query, {
@@ -360,6 +374,9 @@ class PgConnectionImplementation extends _PgSessionBase implements Connection {
     }
   }
 
+  @override
+  Future<void> get closed => _channel.sink.done;
+
   @override
   Future<R> run<R>(
     Future<R> Function(Session session) fn, {
@@ -370,7 +387,7 @@ class PgConnectionImplementation extends _PgSessionBase implements Connection {
     // Unlike runTx, this doesn't need any locks. An active transaction changes
     // the state of the connection, this method does not. If methods requiring
     // locks are called by [fn], these methods will aquire locks as needed.
-    return Future<R>(() => fn(session));
+    return Future<R>(() => fn(session)).whenComplete(session._closeSession);
   }
 
   @override
@@ -866,7 +883,7 @@ class _TransactionSession extends _PgSessionBase {
       controller.stream.listen(items.add),
       true,
       cleanup: () {
-        _sessionClosed = true;
+        _closeSession();
         _connection._activeTransaction = null;
       },
     );

test/pool_test.dart

@@ -74,6 +74,22 @@ void main() {
       expect(await stmt.run([]), hasLength(3));
       await stmt.dispose();
     });
+
+    test('disables close()', () async {
+      late Connection leakedConnection;
+
+      await pool.withConnection((connection) async {
+        expect(connection.isOpen, isTrue);
+        await connection.close();
+        expect(connection.isOpen, isTrue);
+
+        leakedConnection = connection;
+      });
+
+      await pool.close();
+      expect(pool.isOpen, isFalse);
+      expect(leakedConnection.isOpen, isFalse);
+    });
   });
 
   withPostgresServer('limit pool connections', (server) {

test/transaction_test.dart

@@ -308,6 +308,23 @@ void main() {
       final total = await conn.execute('SELECT id FROM t');
       expect(total, []);
     });
+
+    test('isOpen and closed', () async {
+      late Session leakedTransaction;
+      var afterTransaction = false;
+
+      await conn.runTx(expectAsync1((session) async {
+        leakedTransaction = session;
+        expect(session.isOpen, isTrue);
+
+        // .closed should complete before the runTx future
+        expectLater(
+            session.closed.then((_) => afterTransaction), completion(isFalse));
+      }));
+      afterTransaction = true;
+
+      expect(leakedTransaction.isOpen, isFalse);
+    });
   });
 
   // A transaction can fail for three reasons: query error, exception in code, or a rollback.

test/v3_test.dart

@@ -224,9 +224,13 @@ void main() {
 
     test('run', () async {
       const returnValue = 'returned from run()';
+      late Session leakedSession;
 
       await expectLater(
         connection.run(expectAsync1((session) async {
+          expect(session.isOpen, isTrue);
+          leakedSession = session;
+
           await session
               .execute('CREATE TEMPORARY TABLE foo (id INTEGER PRIMARY KEY);');
           await session.execute('INSERT INTO foo VALUES (3);');
@@ -236,6 +240,8 @@ void main() {
         completion(returnValue),
       );
 
+      expect(leakedSession.isOpen, isFalse);
+
       final rows = await connection.execute('SELECT * FROM foo');
       expect(rows, [
         [3]
@@ -448,6 +454,21 @@ void main() {
       expect(incoming, contains(isA<DataRowMessage>()));
       expect(outgoing, contains(isA<QueryMessage>()));
     });
+
+    test('can close connection', () async {
+      expect(connection.isOpen, isTrue);
+      await connection.execute('SELECT 1');
+
+      var didFinishClose = false;
+      expect(
+          connection.closed.then((_) => didFinishClose), completion(isFalse));
+      await connection.close();
+      didFinishClose = true;
+
+      expect(connection.isOpen, isFalse);
+      expect(
+          () => connection.execute('SELECT 1'), throwsA(_isPostgresException));
+    });
   });
 
   withPostgresServer('can close connection after error conditions', (server) {