proto-socket/dart/lib/src/base_client.dart

import 'dart:async';
import 'package:meta/meta.dart';

import 'communicator.dart';
import 'heartbeat_mixin.dart';

abstract class BaseClient<Self extends BaseClient<Self>> extends Communicator
    with HeartbeatMixin {
  late final Self _self;
  final List<void Function(Self)> _disconnectListeners = [];
  bool _closeCalled = false;

  @protected
  void initSelf(Self self) {
    _self = self;
  }

  void addDisconnectListener(void Function(Self) fn) {
    if (!_disconnectListeners.contains(fn)) {
      _disconnectListeners.add(fn);
    }
  }

  void removeDisconnectListener(void Function(Self) fn) {
    _disconnectListeners.remove(fn);
  }

  void onDisconnected(dynamic _) {
    unawaited(_closeWithGrace(false));
  }

  @override
  Future<void> close() async {
    await _closeWithGrace(true);
  }

  Future<void> _closeWithGrace(bool graceful) async {
    if (_closeCalled) return;
    _closeCalled = true;

    if (graceful) {
      await super.close();
    } else {
      super.shutdown();
    }

    stopHeartbeat();
    await closeTransport();

    final listeners = List<void Function(Self)>.from(_disconnectListeners);
    _disconnectListeners.clear();
    for (final fn in listeners) {
      fn(_self);
    }
  }

  @protected
  Future<void> closeTransport();

  bool get isSourcePaused;
}