我有两个流,Stream<A>
和Stream<B>
。我有一个类型为C
的构造函数,它接受一个A
和一个B
。如何将这两个Stream
合并成一个Stream<C>
?
我有两个流,Stream<A>
和Stream<B>
。我有一个类型为C
的构造函数,它接受一个A
和一个B
。如何将这两个Stream
合并成一个Stream<C>
?
import 'dart:async' show Stream;
import 'package:async/async.dart' show StreamGroup;
main() async {
var s1 = stream(10);
var s2 = stream(20);
var s3 = StreamGroup.merge([s1, s2]);
await for(int val in s3) {
print(val);
}
}
Stream<int> stream(int min) async* {
int i = min;
while(i < min + 10) {
yield i++;
}
}
参见 http://news.dartlang.org/2016/03/unboxing-packages-async-part-2.html
打印
10
20
11
21
12
22
13
23
14
24
15
25
16
26
17
27
18
28
19
29
你可以使用 package:async
中的StreamZip
将两个流组合成一对流,然后从中创建 C
对象。
import "package:async" show StreamZip;
...
Stream<C> createCs(Stream<A> as, Stream<B> bs) =>
new StreamZip([as, bs]).map((ab) => new C(ab[0], ab[1]));
StreamZip
只会发出成对的数据。如果输入流 A 和 B 的数据发射速率不同,C 会等待直到两个流都发射了一条数据后再将它们合并成一对数据进行发射。这可能符合人们的预期,也可能不符合。 - WSBTC
对象。除非每次收到 A
或 B
时都创建一个新的 C
,并使用另一个值的旧值。如果在第一个 B
之前收到两个 A
,仍然会出现问题。 - lrncombineLatest
通过等待每个流至少发出一个值来解决此问题:https://pub.dev/documentation/stream_transform/latest/stream_transform/CombineLatest/combineLatest.html - masterwokStream<A>
或Stream<B>
发出事件时做出反应,并使用两个流的最新值,则使用combineLatest
。Stream<C> merge(Stream<A> streamA, Stream<B> streamB) {
return streamA
.combineLatest(streamB, (a, b) => new C(a, b));
}
对于需要合并两个以上不同类型流并在任何一个流更新时获取所有最新值的人。
import 'package:stream_transform/stream_transform.dart';
Stream<List> combineLatest(Iterable<Stream> streams) {
final Stream<Object> first = streams.first.cast<Object>();
final List<Stream<Object>> others = [...streams.skip(1)];
return first.combineLatestAll(others);
}
streamA: a----b------------------c--------d---|
streamB: --1---------2-----------------|
streamC: -------&----------%---|
combined: -------b1&--b2&---b2%---c2%------d2%-|
为什么不使用StreamZip
?因为StreamZip
会产生以下问题:
streamA: a----b------------------c--------d---|
streamB: --1---------2-----------------|
streamC: -------&----------%---|
combined: -------a1&-------b2%--|
使用方法:
Stream<T> sA;
Stream<K> sB;
Stream<Y> sC;
combineLatest([sA, sB, sC]).map((data) {
T resA = data[0];
K resB = data[1];
Y resC = data[2];
return D(resA, resB, resC);
});
使用rxdart
,您可以使用CombineLatestStream
来实现您想要的功能。请注意,新流在所有流至少发出一个事件之前不会返回任何值:
CombineLatestStream.list()
创建组合流:import 'package:rxdart/rxdart.dart';
Stream<A> s1 = Stream.fromIterable([A()]);
Stream<B> s2 = Stream.fromIterable([B()]);
Stream<dynamic> s3 = CombineLatestStream.list<dynamic>([s1, s2])
..listen(
(value) {
A a = value[0];
B b = value[1];
},
);
由于Dart不支持联合类型,CombineLatestStream.list()
的一个缺点是来自不同类型流的事件之后应该进行转换(由于List<dynamic>
)。另一种方法是使用CombineLatestStream.combine2()
(例如使用创建Tuple2
的combiner
)来保持类型。
asyncExpand
。 Stream<UserModel?> getCurrentUserModelStream() {
return FirebaseAuth.instance.authStateChanges().asyncExpand<UserModel?>(
(currentUser) {
if (currentUser == null) {
return Stream.value(null);
}
return FirebaseFirestore.instance
.collection('users')
.doc(currentUser.uid)
.snapshots()
.map((doc) {
final userData = doc.data();
if (userData == null) {
return null;
}
return UserModel.fromJson(userData);
});
},
);
}
StreamController
设置正确,它在暂停时不应该产生事件(根据用例可能不可能)。 - Günter ZöchbauerStream<A>
和Stream<B>
上的更改数量完全相等并且在时间上匹配。这些流的数据源是Firebase数据库中的不同点,因此更新是在随机的时间点发生的,而且绝对不会以相同的数量出现。所以,实际上并不符合要求。无论如何,我还是接受它,因为它回答了提出的问题。 - Brett