RxDart，将列表的每个项目映射到另一个来自永无止境的流的对象

Question

我一直在努力寻找一个很好的方法来做这件事，但我没有运气。

这里是问题的简化版本：

import 'package:rxdart/rxdart.dart';


/// Input a list of integers [0,1,2,3,4]
/// map each of those integers to the corresponding index in the map
/// if the map updates, the output should update too.
///
/// The output should be a list of Strings:
/// ["Hi from 1", "Hi from 2"; "Hi from 3", "Hi from 4", "Hi from 5"]
BehaviorSubject<Map<int, String>> subject = BehaviorSubject(
    seedValue: {
      1: "Hi from 1",
      2: "Hi from 2",
      3: "Hi from 3",
      4: "Hi from 4",
      5: "Hi from 5",
    }
);

void main() {
  Observable.fromIterable([1, 2, 3, 4, 5])
      .flatMap((index) => subject.stream.map((map) => map[index]))
      .toList().asObservable()
      .listen((data) {
    print("List of data incoming $data");
  });
}

当运行这样时，什么都不打印。这是因为主题永远不会完成，因此 toList() 永远不会完成构建列表。

用 Observable.just(index + 2) 替换主题确实有效，因为 Observable 完成并且 toList() 能够收集它们。

但预期的行为是每次更改主题时示例都应发出新的字符串列表。

如有任何帮助，我们将不胜感激，

谢谢！

Answer 1

您可能想改用 combineLatest

BehaviorSubject<Map<int, String>> subject = BehaviorSubject(seedValue: {
  1: "Hi from 1",
  2: "Hi from 2",
  3: "Hi from 3",
  4: "Hi from 4",
  5: "Hi from 5",
});

void main() {
  Observable.combineLatest2(
      Observable.just([1, 2, 3, 4, 5]), subject.stream, combiner)
    ..listen(
      (data) {
        print("List of data incoming $data");
      },
    );
}

Iterable<String> combiner(List<int> indexes, Map<int, String> map) {
  return indexes.map((index) => map[index]);
}

Answer 2

编辑：发布后不久，我意识到这完全取决于数据的公开方式：

如果您有一个方法，其签名类似于 Observable<String> getStringForInt(int number)，并且您必须为列表中的每个项目调用它，那么我的解决方案适合您。注意如果上述方法总是订阅同一个流，更改该流将导致多次发射（因为 combineLatest 更新每个项目）。

但是，如果您有权访问整个数据容器（如地图

原答案

好的，事实证明，通过 rxDart 的实现不太可能实现这一目标。

combineLatest 构造函数最多只支持 9 个流。

但从那以后，结合最新的 n-streams 已经实现，这个问题的解决方案看起来像这样：

Map<int, BehaviorSubject<String>> subject2 = {
  1: BehaviorSubject<String>(seedValue: "Subject 1"),
  2: BehaviorSubject<String>(seedValue: "Subject 2"),
  3: BehaviorSubject<String>(seedValue: "Subject 3"),
  4: BehaviorSubject<String>(seedValue: "Subject 4"),
  5: BehaviorSubject<String>(seedValue: "Subject 5"),
};

void main() async {

  Observable.fromIterable([1, 2, 3, 4, 5])
      .toList().asObservable()
      .flatMap((numbers) =>
      Observable.combineLatest<String, List<String>>(numbers.map((index) => subject2[index]), (strings) => strings))
      .listen((data) {
    print("List of data incoming $data");
  });

  await Future.delayed(Duration(seconds: 2));
  subject2[1].add("I'm 42 now");
}

这将打印：

List of data incoming [Subject 1, Subject 2, Subject 3, Subject 4, Subject 5]
List of data incoming [I'm 42 now, Subject 2, Subject 3, Subject 4, Subject 5]

RxDart，将列表的每个项目映射到另一个来自永无止境的流的对象

RxDart, mapping each item of a list to another object coming from a never ending stream

stream

dart

rxdart