我有一些可观察的。我需要知道是哪一个触发了订阅。
Observable.combineLatest(
this.tournamentsService.getUpcoming(),
this.favoriteService.getFavoriteTournaments(),
this.teamsService.getTeamRanking(),
(tournament, favorite, team) => {
//what triggered combinelatest to run?
}).subscribe()
实现此目的的一种非常干净且“rx”的方式是使用时间戳运算符http://reactivex.io/documentation/operators/timestamp.html
示例代码
sourceObservable
.pipe(
timestamp(), // wraps the source items in object with timestamp of emit
combineLatest( otherObservable.pipe( timestamp() ), function( source, other ) {
if( source.timestamp > other.timestamp ) {
// source emitted and triggered combineLatest
return source.value;
}
else {
// other emitted and triggered combineLatest
return other.value;
}
} ),
)
如果按时间戳对它们进行排序涉及两个以上的可观察对象，则combineLatest()可以检测是哪一个触发了combineLatest().
简短的回答是：你不知道。您可以实施一些解决方法，但这确实很难看，我建议重新考虑用例为什么需要它，也许您可以更改架构。还要记住，你的函数的第一次执行将在所有三个可观察对象都发出至少 1 个值之后。
无论如何 - 一个可能的解决方法可能是：
let trigger = "";
Observable.combineLatest(
this.tournamentsService.getUpcoming().do(() => trigger = "tournament"),
this.favoriteService.getFavoriteTournaments().do(() => trigger = "favTournament"),
this.teamsService.getTeamRanking().do(() => trigger = "teamRanking"),
(tournament, favorite, team) => {
console.log(`triggered by ${trigger}`);
}).subscribe();
如果您想根据触发的可观察对象执行特定操作，您应该利用每个可观察对象并将它们用作单独的触发器，切换到组合触发器，它可能会稍微多一些代码，但它更干净，你不会最终陷入丑陋的 if/else、switch/case-message 以及一些 hacky 变通办法 -另外，您甚至有机会使用async-pipe 而不是手动订阅所有内容并更新局部变量（无论如何这是一个不好的做法） ：
下面是一些示例代码，它看起来像这样：
let upcoming$ = this.tournamentsService.getUpcoming();
let favorite$ = this.favoriteService.getFavoriteTournaments();
let rankings$ = this.teamsService.getTeamRanking();
let allData$ = Observable.combineLatest(
upcoming$, favorite$, rankings$,
(tournament, favorite, team) => {
return {tournament, favorite, team};
}
);
// initial call -> this SHOULD be redundant,
// but since I don't know your code in detail
// i've put this in - if you can remove it or not
// depends on the order your data coming in
allData$
.take(1)
.do(({tournament, favorite, team}) => {
this.displayMatches(...);
this.sortByFavorites(...);
this.fillWithRanking(...);
})
.subscribe();
// individual update triggers
upcoming$
.switchMapTo(allData$.take(1))
.do(({tournament, favorite, team}) => this.displayMatches(...))
.subscribe();
favorite$
.switchMapTo(allData$.take(1))
.do(({tournament, favorite, team}) => this.sortByFavorites(...))
.subscribe();
rankings$
.switchMapTo(allData$.take(1))
.do(({tournament, favorite, team}) => this.fillWithRanking(...))
.subscribe();
您可以使用scan操作符将发出的值与任何先前发出的值进行比较，并且可以包含额外的数据，指示组合的 observable 的组件是否实际发生了变化。例如：
let combined = Observable
.combineLatest(
this.tournamentsService.getUpcoming(),
this.favoriteService.getFavoriteTournaments(),
this.teamsService.getTeamRanking()
)
.scan((acc, values) => [
...values,
acc[0] !== values[0],
acc[1] !== values[1],
acc[2] !== values[2]
], []);
combined.subscribe(
([tournament, favorite, team, tournamentChanged, favoriteChanged, teamChanged]) => {
console.log(`tournament = ${tournament}; changed = ${tournamentChanged}`);
console.log(`favorite = ${favorite}; changed = ${favoriteChanged}`);
console.log(`team = ${team}; changed = ${teamChanged}`);
}
);
combineLatest([
this.obs1.pipe(tap(() => (this.trigger = 'obs1'))),
this.obs2.pipe(tap(() => (this.trigger = 'obs2'))),
])
.subscribe(([obs1, obs2]) => {
switch (this.trigger) {
case 'obs1':
// ...
case 'obs2':
// ...
}
})
解决此问题的更好方法是使用可区分的联合类型。如果您的语言没有内置的可区分联合，您可以通过将构造函数设为私有并公开可以n为空的公共静态属性来创建一个，每个可观察到一个。用 C# 编写这个，因为我更喜欢这种语言，但它应该很容易翻译。注意可以为空的字符串。如果您的语言不支持可空值，请使用其他机制来指示是否设置了值。
private class DiscriminatedUnion
{
private DiscriminatedUnion(string? property1, string? property2)
{
Property1 = property1;
Property2 = property2;
}
public string? Property1 { get; }
public string? Property2 { get; }
public static DiscrimintatedUnion FromObservable1(string property1)
{
return new DiscriminatedUnion(property1, null);
}
public static DiscrimintatedUnion FromObservable2(string property2)
{
return new DiscriminatedUnion(null, property2);
}
}
private IObservable<DiscriminatedUnion> CreateCombination()
{
var firstObservable = tournamentsService
.getUpcoming()
.Select(x => DiscriminatedUnion.FromObservable1(x));
var secondObservable = favoriteService
.getFavoriteTournaments()
.Select(x => DiscriminatedUnion.FromObservable2(x));
return Observable
CombineLatest(firstObservable, secondObservable);
}
因此，现在您可以质疑从CreateCombination()哪个 observable 发出值返回的可区分联合。
我在 Flutter 中使用 RxJava，我想使用操作符 combineLatest 组合 12 个不同的 observables。
我看到了一个函数原型，它带有一个可观察对象列表和一个实现。但我不知道该怎么做，我在实现 call 方法时遇到了麻烦。请检查我的代码并做必要的事情。
流> 获取城市 => _citiesController.stream;
流获取城市 => _cityController.stream;
流获取 agentcity => _agentcityController.stream;
流获取用户包 => _packagesController.stream;
流获取电子邮件 => _emailController.stream.transform(validateEmail);
流获取名字 => _firstNameController.stream.transform(validateFirstName);
流获取姓氏 => _lastNameController.stream.transform(validateLastName);
流获取 mobileNumber => _mobileNumberController.stream.transform(validateMobile);
流获取 dob => _dobController.stream.transform(validatedob);
流获取约会日期 => _appointmentdateController.stream.transform(validateappointmentDate);
流获取密码 => _pincodeController.stream.transform(validatePincode);
流获取性别 => _genderController.stream;
流获取地址 => _addressController.stream.transform(validateAddress);
流获取代理名称 => _agentnameController.stream.transform(validateAgentName);
流获取有效提交 => Observable.combineLatest9(
email,
firstName,
mobileNumber,
pincode,
dob,
address,
agentname,
_genderController.stream,
_cityController.stream,
_agentcityController.stream,
_packagesController.stream,
_appointmentdateController.stream,
(e, f, m, p, d, a, an, g, c, ac, pc, ad) => true,
);
请让我知道如何在我的 Flutter 代码中使用 combineLatest。