Dart Streams：无需重复即可获得下一个值

Question

我正在想办法从 Dart 中的 Streams 中获取后续值而不会重复：

  // should resolve to first value received in stream
  Future f1 = mygetNextValueFn(stream); 
  // should resolve to second
  Future f2 = mygetNextValueFn(stream); 

这里的想法是让f1 和f2 解析为流中的后续值，而不是解析为相同的值。如果我从stream 中提取一个值作为未来的f，我想确保解析f 的流值对于f 是唯一的。

我可以使用包装类来做到这一点：

import "dart:async";
import "dart:collection";

class StreamWrapper<T> {
  Queue<Completer<T>> _completers;
  Queue<T> _values;

  _addValue(T value) {
    _values.add(value);
    _update();
  }

  _update() {
    while (_completers.isNotEmpty && _values.isNotEmpty) {
      Completer<T> completer = _completers.removeFirst();
      T value = _values.removeFirst();
      completer.complete(value);
    }
  }

  Future<T> next() {
    var complete = new Completer<T>();
    _completers.add(complete);
    _update();
    return complete.future;
  }

  StreamWrapper(Stream<T> stream) {
    this._completers = new Queue<Completer<T>>();
    this._values = new Queue<T>();

    stream.listen(_addValue);
  }
}

然后可以确保来自streamWrapperInstance.next() 的 Futures 将始终解析为唯一的流输出值。这有点骇人听闻，有没有更标准的方法可以为单个订阅者流做到这一点？

Answer 1

编辑

根据您的新解释，这应该更适合您的问题：

import 'dart:async';

int _i = 0;

Future myGetNextValueFn(Stream s) {
  Completer c = new Completer();
  s.elementAt(_i++).then(c.complete);
  return c.future;
}

main() async {
  StreamController controller = new StreamController.broadcast();
  Stream stream = controller.stream;

  Future value1 = myGetNextValueFn(stream);
  Future value2 = myGetNextValueFn(stream);

  value1.then((val) => print('value1: $val'));
  value2.then((val) => print('value2: $val (1)'));
  value2.then((val) => print('value2: $val (2)'));

  controller.add('a');
  controller.add('b');
}

打印：

值1：a
值2：b (1)
值2：b（2）

旧

这两个想法怎么样：

import 'dart:async';

main() {
  StreamController controller = new StreamController.broadcast();
  Stream stream = controller.stream;

  // Idea 1
  stream.first.then((_) => print('1: $_'));
  stream.skip(1).first.then((_) => print('2: $_'));

  // Idea 2
  stream.elementAt(0).then((_) => print('1: $_'));
  stream.elementAt(1).then((_) => print('2: $_'));

  controller.add('a');
  controller.add('b');
}

打印：

1：一个
1：一个
2：b
2：b

或者，如果您想要使用 await/async 的解决方案：

import 'dart:async';

handler(Stream s) async {
  var x = await s.first;
  print('done for $x');
}

main() async {
  StreamController controller = new StreamController.broadcast();
  Stream stream = controller.stream;

  new Timer(new Duration(seconds: 2), () {
    controller.add('a');
    new Timer(new Duration(seconds: 2), () {
      controller.add('b');
    });
  });

  await handler(stream);
  await handler(stream);
}

打印：

为a完成
为 b 完成
完成

如果您只有一个订阅流，请使用stream.asBroadcastStream() 获取广播流！