Dart:如何将转换器附加到现有流中

Question

我正在寻找一种方法，以编程方式将转换器添加到已经在侦听的现有流中。

示例：

Stream numbers = new Stream.fromIterable([0,1,2,3]);
numbers.listen((number) => print(number));

现在，为了响应一些UI事件，我想通过添加映射转换器来修改这个流，就像我最初写的那样：

numbers.where((number) => number % 2 == 0);

从现在起，所有现有的侦听器都应该只接收偶数，而不会中断。如何做到这一点？

Answer 1

与其像"我如何将转换器动态插入流中"这样思考，不如像"我怎样动态控制已经注入的转换器"这样思考。

以下是使用StreamTransformer:的示例

var onlySendEvenNumbers = false; // controlled by some UI event handler
var originalStream = makeStreamOfStuff();
originalStream = originalStream.transform(new StreamTransformer.fromHandlers(
  handleData: (int value, EventSink<int> sink) {
    if (onlySendEvenNumber) {
      if (value.isEven) {
        sink.add(value);
      }
    } else {
      sink.add(value);
    }
}));
originalStream.listen(print);  // listen on events like normal

Answer 2

我可以想到的一种方法是用一个调用另一个函数的函数过滤Stream：

var filter = (n) => true;
Stream numbers = new String.fromIterable([0, 1, 2, 3]).where((n) => filter(n));

然后，当你想改变过滤：

filter = (n) => n % 2 == 0;

一个具体的例子：

import 'dart:async';
main() {
  var filter = (n) => true;
  Stream numbers = new Stream.periodic(new Duration(seconds: 1), (n) => n)
      .where((n) => filter(n));
  numbers.listen((n) => print(n));
  new Future.delayed(new Duration(seconds: 4)).then((_) {
    filter = (n) => n % 2 == 0;
  });
}

这将打印：

0
1
2
3
4
6
8
10
12

依此类推，仅适用于偶数，4秒后。

Answer 3

rxdart的combineLatest2怎么样？

它将两个流合并，每次更改两个流时都会发出。

您可以使用Switch类在有条件的情况下打开/关闭。

class XsBloc {
  Api _api = Api();
  BehaviorSubject<List<X>> _xs = BehaviorSubject();
  BehaviorSubject<Switcher> _switcher =
      BehaviorSubject<Switcher>.seeded(Switcher(false, []));
  XsBloc() {
    Observable.combineLatest2<List<X>, Switcher, List<X>>(
        _api.xs(), _switcher, (xs, s) {
      if (s.isOn == true) {
        return xs.where((x) => s.conditions.contains(x.id)).toList();
      } else {
        return xs;
      }
    }).listen((x) => _xs.add(x));
  }
  Stream<List<X>> get xs => _xs;
  ValueObservable<Switcher> get switcher =>
      _switcher.stream;
  Function(Switcher) get setSwitcher => _switcher.sink.add;
}
class Switcher {
  final bool isOn;
  final List<String> conditions;
  Switcher(this.isOn, this.conditions);
}
var bloc = XsBloc();
bloc.setSwitcher(true, ['A', 'B']);
bloc.setSwitcher(false, []);
bloc.setSwitcher(true, []);