dart - 取消流onData

Question

我有一个事件总线，可以处理我应用程序中的所有中央事件。我有一个特殊情况，即我有一系列异步 Action 只想执行一次(发生特殊事件时)，所以我必须启动一个异步函数，使其他函数失控，这些事件将触发一个事件。我的第二个 Action ，依此类推。

所以我需要启动 Action 一，然后听事件总线等待 Action 一触发(非直接)触发将启动 Action 二的事件，依此类推...

自然地，一旦序列中的每个元素都执行了，我想停止监听触发它的事件。

为此，我想像了一个ConsumerOnce(event，action)函数，该函数将订阅总线，等待预期的事件，在接收到事件时执行该 Action ，并在 Action 启动后立即取消订阅(异步)

  final StreamController<Map<PlaceParam, dynamic>> _controller =
  new StreamController<Map<PlaceParam, dynamic>>.broadcast();

  void consumeOnce(PlaceParam param, Function executeOnce) {
    StreamSubscription subscription = _controller.stream.listen((Map<PlaceParam, dynamic> params) {
      if(params.containsKey(param)) {
        executeOnce();
        subscription.cancel(); //can't access, too early: not created yet
      }
    });
  }

问题是我无法在回调主体中访问变量 subscription ，因为当时尚未创建变量

由于监听器不会按照订阅顺序执行任何担保，因此我无法注册另一个将删除我的订阅的订阅者(即使执行顺序得到保证，无论如何，我都会找到无法删除的订阅来找到自己:负责删除我的原始订阅)。

有什么想法吗？

这种模式可以解决我的问题，但是我觉得它并不优雅:

@Injectable()
class EventBus<K, V> {
  final StreamController<Map<PlaceParam, dynamic>> _controller =
  new StreamController<Map<PlaceParam, dynamic>>.broadcast();

  Future<Null> fire(Map<PlaceParam, dynamic> params) async {
    await _controller.add(params);
  }

  Stream<Map<PlaceParam, dynamic>> getBus() {
    return _controller.stream;
  }


  void consumeOnce(PlaceParam param, Function executeOnce) {
    SubscriptionRemover remover = new SubscriptionRemover(param, executeOnce);
    StreamSubscription subscription = _controller.stream.listen(remover.executeOnce);
    remover.subscription = subscription;
  }
}

class SubscriptionRemover {
  PlaceParam param;
  Function executeOnce;
  StreamSubscription subscription;

  SubscriptionRemover(this.param, this.executeOnce);

  void execute(Map<PlaceParam, dynamic> params) {
    if (params.containsKey(param)) {
      executeOnce();
      subscription.cancel();
    }
  }
}

但我不太喜欢，因为从理论上讲，该事件可能在两个调用之间发生:

    StreamSubscription subscription = _controller.stream.listen(remover.executeOnce); //event may occur now!!!
    remover.subscription = subscription;

我认为存在一种方法:_controller.stream.remove(Function fn)会更加直接和清晰。

我对吗？还是有我没想到的方法？

Answer 1

The issue is that I can't access the variable subscription in the body of my callback, since it is still not created at the time

没错-即使知道订阅本身将存在，您也无法访问subscription变量。 Dart不允许变量声明引用自己。当变量仅在尚未执行的闭包中引用时，这有时会很烦人。

解决方案是在进行侦听之后预先声明变量或更新onData -listener:

// Pre-declare variable (can't be final, though).
StreamSubscription<Map<PlaceParam, dynamic>> subscription;
subscription = stream.listen((event) {
  .... subscription.cancel();
});

要么

final subscription = stream.listen(null);
subscription.onData((event) {  // Update onData after listening.
  .... subscription.cancel(); ....
});

在某些情况下，您还不能访问订阅对象，但这仅在流中断Stream契约并在收听时立即开始发送事件的情况下才可能。流不能这样做，它们必须等到以后的微任务才传递第一个事件，这样称为listen的代码才有时间接收订阅并将其分配给变量。使用同步流 Controller 可能违反契约(Contract)(这是为什么应谨慎使用同步流 Controller 的原因之一)。