Flutter 中的异步事件处理之_SyncStreamController、_AsyncStreamController源码解析(二)

2023-09-19 20:56:49

了解 [SyncStreamController] 和 [AsyncStreamController] 的工作原理，可以帮助我们更好地理解 Flutter 中异步事件处理的机制，并为我们构建更加健壮的应用程序打下坚实的基础。

Flutter 中的异步事件处理

在 Flutter 中，异步事件处理是一个非常重要的概念。它允许我们在应用程序中处理那些需要等待一段时间才能完成的任务，比如网络请求、文件操作等。

Flutter 中异步事件处理主要有两种方式：

使用 Future 对象 ：Future 对象表示一个异步操作的结果。它可以被其他代码等待，并在操作完成后得到结果。
使用 Stream 对象 ：Stream 对象表示一个异步事件流。它可以被其他代码订阅，并在事件发生时收到通知。

[SyncStreamController] 和 [AsyncStreamController]

[SyncStreamController] 和 [AsyncStreamController] 是 Flutter 中用于创建 [Stream] 对象的两个类。它们都实现了 [StreamController] 接口，并提供了不同的功能。

[SyncStreamController] ：创建一个同步 [Stream] 对象。这意味着 [Stream] 中的事件将立即传递给订阅者。
[AsyncStreamController] ：创建一个异步 [Stream] 对象。这意味着 [Stream] 中的事件将在稍后传递给订阅者。

[SyncStreamController] 源码解析

class _SyncStreamController<T> extends StreamControllerBase<T> {
  _SyncStreamController({
    void onListen(),
    void onPause(),
    void onResume(),
    void onCancel(),
  }) : super(
          onListen: onListen,
          onPause: onPause,
          onResume: onResume,
          onCancel: onCancel,
        );

  @override
  void add(T event) {
    if (_state & _STREAM_WAS_CLOSED != 0) {
      throw StateError('Cannot add to a closed stream');
    }
    if (_state & _STATE_PAUSED != 0) {
      _addPendingEvent(event);
      return;
    }
    _publishEvent(event);
  }

  @override
  void addError(Object error, [StackTrace? stackTrace]) {
    if (_state & _STREAM_WAS_CLOSED != 0) {
      throw StateError('Cannot add to a closed stream');
    }
    if (_state & _STATE_PAUSED != 0) {
      _addPendingEvent(error);
      return;
    }
    _addError(error, stackTrace);
  }
}

[SyncStreamController] 的源码非常简单，它只实现了 [StreamControllerBase] 接口中定义的基本方法。

[add] ：将一个事件添加到 [Stream] 中。
[addError] ：将一个错误添加到 [Stream] 中。

[AsyncStreamController] 源码解析

class _AsyncStreamController<T> extends StreamControllerBase<T> {
  _AsyncStreamController({
    void onListen(),
    void onPause(),
    void onResume(),
    void onCancel(),
  }) : super(
          onListen: onListen,
          onPause: onPause,
          onResume: onResume,
          onCancel: onCancel,
        );

  @override
  void add(T event) {
    _runGuarded(() {
      if (_state & _STREAM_WAS_CLOSED != 0) {
        throw StateError('Cannot add to a closed stream');
      }
      if (_state & _STATE_PAUSED != 0) {
        _addPendingEvent(event);
        return;
      }
      _asyncAdd(event);
    });
  }

  @override
  void addError(Object error, [StackTrace? stackTrace]) {
    _runGuarded(() {
      if (_state & _STREAM_WAS_CLOSED != 0) {
        throw StateError('Cannot add to a closed stream');
      }
      if (_state & _STATE_PAUSED != 0) {
        _addPendingEvent(error);
        return;
      }
      _asyncAddError(error, stackTrace);
    });
  }
}