close method

  1. @override
Future<void> close()
override

Closes every SSE session, then the HTTP server, and never throws.

Sessions are closed from a snapshot: closing a response fires its done callback, which removes the session from the live map. Each waits for its pending write first — closing a response while a flush is in flight throws — bounded by _closeFlushLimit. A session that still fails to close is logged and the rest are closed regardless. The HTTP server is closed in finally, so the port is released whatever a session does. onClose completes when all of it is done.

Implementation

@override
Future<void> close() async {
  if (_closing) return _closeCompleter.future;
  _closing = true;
  final sessions = Map<String, HttpResponse>.of(_sessionClients);
  final pendingWrites = Map<String, Future<void>>.of(_sessionFlushChain);
  _sessionClients.clear();
  _sessionCompression.clear();
  try {
    await Future.wait(sessions.entries.map((entry) async {
      try {
        await (pendingWrites[entry.key] ?? Future<void>.value())
            .timeout(_closeFlushLimit, onTimeout: () {});
        await entry.value.close();
      } catch (e) {
        _logger.debug('[SSE] closing session ${entry.key} failed: $e');
      }
    }));
  } finally {
    _sessionFlushChain.clear();
    try {
      await _server?.close(force: true);
    } catch (e) {
      _logger.error('[SSE] closing the HTTP server failed: $e');
    }
    _server = null;
    if (!_messageController.isClosed) await _messageController.close();
    if (!_closeCompleter.isCompleted) _closeCompleter.complete();
  }
}