是什么触发了 combineLatest？

Question

我有一些观察结果。我需要知道是哪一个触发了订阅。

Observable.combineLatest(
      this.tournamentsService.getUpcoming(),
      this.favoriteService.getFavoriteTournaments(),
      this.teamsService.getTeamRanking(),
(tournament, favorite, team) => {
//what triggered combinelatest to run?
}).subscribe()

Answer 1

简短的回答是：你不知道。您可以实施一些解决方法，但这真的很丑陋，我建议您重新考虑用例为什么需要这个以及是否可以更改体系结构。还要记住，你的函数的第一次执行将在所有三个可观察量都发出至少 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();

如果你想根据触发的 observable 执行特定的操作，你应该利用每个 observable 并将它们用作单独的触发器，切换到组合触发器，它可能会稍微多一些代码，但它已经很多了更干净，你不会以丑陋的 if/else、switch/case-mess 和一些 hacky 解决方法结束 - plus 你甚至有机会使用 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();

Answer 2

您可以使用 scan 运算符将发出的值与任何先前发出的值进行比较，并且可以包含额外的数据来指示组合的可观察值的组件是否实际发生了变化。例如：

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}`);
  }
);

Answer 3

实现此目的的一种非常简洁且 "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()。

Answer 4

我在 Flutter 中使用 RxJava，我想使用运算符 combineLatest 组合 12 个不同的可观察对象。

我看到了一个函数原型，它带有一个可观察对象列表和一个实现。但我不确定该怎么做，我在实现调用方法时遇到了问题。请检查我的代码并执行必要的操作。

流> 获取城市=> _citiesController.stream;

流获取城市 => _cityController.stream;

流获取代理城市 => _agentcityController.stream;

流获取用户包 => _packagesController.stream;

流获取电子邮件 => _emailController.stream.transform(validateEmail);

流获取名字 => _firstNameController.stream.transform(validateFirstName);

流获取姓氏 => _lastNameController.stream.transform(validateLastName);

Stream get mobileNumber => _mobileNumberController.stream.transform(validateMobile);

Stream get dob => _dobController.stream.transform(validatedob);

流获取约会日期 => _appointmentdateController.stream.transform(validateappointmentDate);

串流获取密码 => _pincodeController.stream.transform(validatePincode);

流获取性别 => _genderController.stream;

流获取地址 => _addressController.stream.transform(validateAddress);

Stream get agentname => _agentnameController.stream.transform(validateAgentName);

Stream get validSubmission => 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。

Answer 5

解决这个问题的更好方法是使用可区分的联合类型。如果您的语言没有内置有区别的联合，您可以通过将构造函数设为私有并公开 n 可空 public 静态属性来创建一个，每个可观察对象一个。用 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 发出了一个值。

Answer 6

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':
            // ...
    }
})

是什么触发了 combineLatest？

what triggered combineLatest?

rxjs

observable

combinelatest