6

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

例子：

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

numbers.listen((number) => print(number));

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

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

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

4

3 回答 3

11

与其像“如何将转换器动态插入流”那样思考，一种可能的方法是像“如何动态控制已经注入的转换器”那样思考。

以下是使用 a 的示例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
于 2014-09-18T02:08:40.663 回答
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 秒后。

于 2014-09-18T01:15:38.107 回答
0

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, []);
于 2019-11-27T14:08:06.027 回答