如何使 RxJS Observable 在特定日期时间发出？

Question

我有一个 RxJS Observable 需要在特定时间重新计算，如 DateTime 对象数组所述（尽管出于此问题的目的，它们可能是 JavaScript Date 对象、纪元毫秒或其他任何东西代表一个特定的时刻）：

const changeTimes = [
    //            yyyy, mm, dd, hh, mm
    DateTime.utc( 2018, 10, 31, 21, 45 ),
    DateTime.utc( 2018, 10, 31, 21, 50 ),
    DateTime.utc( 2018, 10, 31, 22, 00 ),
    DateTime.utc( 2018, 10, 31, 23, 00 ),
    DateTime.utc( 2018, 10, 31, 23, 30 ),
];

我很难理解如何创建一个在这样一个数组中指定的时间发射的 Observable。

这是我在尝试回答我自己的问题时的想法：

• 我几乎肯定需要使用the delay operator，其中指定的延迟是“现在”和下一个未来日期时间之间的时间。
• 我需要确保“现在”在订阅时是最新的，而不是在创建 Observable 时——可能使用the defer operator——尽管我不想不必要地创建多个 Observable 实例，如果有的话多个订阅。
• 我不确定随着时间的推移如何迭代数组—the expand operator 可能是我需要的，但它调用了一些东西递归，我只是想迭代一个列表.
• The timer operator 似乎无关紧要，因为每个日期时间之间的持续时间不同。
• 我可以将每个日期时间映射到它自己的延迟 Observable 并通过merge 将它们全部返回，但是随着数组中日期时间数量的增加（可能有数百个），这变得非常低效，所以这绝对是最后的手段.

如何制作一个 RxJS Observable，它接受一个日期时间列表，然后在每个时间到达时发出，在最后一个完成？

Answer 1

我认为您在要点中总结的内容都是正确的。使用delay 似乎很明显，但它会使链条难以理解。

我想到的解决方案假设您在创建可观察链之前知道changeTimes 数组。例如，您可以创建自己的“可观察创建方法”，该方法将返回基于 setTimeout 发出的可观察对象（这只是“伪代码”，它不能正确计算日期）：

const schedule = (dates: Date[]): Observable<Date> => new Observable(observer => {
  // Sort the `dates` array from the earliest to the latest...

  let index = 0;
  let clearTimeout;

  const loop = () => {
    const now = new Date();
    const delay = dates[index] - now;

    clearTimeout = setTimeout(() => {
      observer.next(dates[index++]);

      if (index < dates.length) {
        loop();
      }
    }, delay);
  }

  loop();

  return () => clearTimeout(clearTimeout);
}); 

...

schedule(changeTimes)
  .subscribe(...)

您在merge 中提到的最后一个选项实际上并没有那么糟糕。我了解您担心它会创建大量订阅，但是如果您对 changeTimes 数组进行排序，然后使用 concat 而不是 merge 它将始终只保留一个活动订阅，即使您创建了 100 个可观察的。

Answer 2

这是一个工作示例：

import {Injectable} from '@angular/core';
import {Observable, Subject, timer} from 'rxjs';

@Injectable()
export class TimerService {

  futureDates: Date[] = [];
  futureDate: Date;
  notifier: Observable<string>;

  cycle = (observer) => {
    if (this.futureDates.length > 0) {
      this.futureDate = this.futureDates.shift();

      const msInFuture = this.futureDate.getTime() - Date.now();
      if (msInFuture < 0) {
        console.log(`date ${this.futureDate.toISOString()}
            expected to be in the future, but was ${msInFuture} msec in the past, so stopping`);

        observer.complete();
      } else {
        timer(msInFuture).subscribe(x => {
          observer.next(`triggered at ${new Date().toISOString()}`);
          this.cycle(observer);
        });
      }
    } else {
        observer. complete();
    }
  }

  getTimer(): Observable<string> {
    const now = new Date();
    const ms1 = now.getTime() + 10000;
    const ms2 = now.getTime() + 20000;
    this.futureDates.push(new Date(ms1));
    this.futureDates.push(new Date(ms2));

    this.notifier = new Observable(observer => {
      this.cycle(observer);
    });

    return this.notifier;
  }
}

在本例中，未来时间列表是在 getTimer() 方法中创建的，但您可以将日期数组传递给该方法。

关键是简单地存储日期，一次处理一个，在处理时，检查该日期距离未来多远，并为该毫秒数设置一个一次性 Rx 计时器。

Answer 3

给定一组DateTime 对象：

const changeTimes = [
    //            yyyy, mm, dd, hh, mm
    DateTime.utc( 2018, 10, 31, 21, 45 ),
    DateTime.utc( 2018, 10, 31, 21, 50 ),
    DateTime.utc( 2018, 10, 31, 22, 00 ),
    DateTime.utc( 2018, 10, 31, 23, 00 ),
    DateTime.utc( 2018, 10, 31, 23, 30 ),
];

或者更好的是，每次数组更改时都会发出一个数组的 Observable（这是我的场景中实际发生的情况，尽管我没有在问题中提及它，因为它不是严格相关的）：

const changeTimes$: Observable<DateTime[]> = /* ... */;

以下 Observable 将在订阅时立即发出下一个未来时间，在上一个未来时间过去时发出每个后续未来时间，然后以 null 完成：

const nextTime$ = changeTimes$.pipe(
    // sort DateTimes chronologically
    map(unsorted => [...unsorted].sort((x, y) => +x - +y),
    // remove duplicates
    map(duplicated => duplicated.filter((item, i) => !i || +item !== +duplicated[i - 1])),
    // convert each time to a delayed Observable
    map(times => [...times, null].map((time, i) => defer(() => of(time).pipe(
        // emit the first one immediately
        // emit all others at the previously emitted time
        delay(i === 0 ? 0 : +times[i - 1] - +DateTime.utc())
    )))),
    // combine into a single Observable
    switchMap(observables => concat(...observables)),
);

• 排序是必要的，因为每个内部 Observable（“等待 X 毫秒然后报告时间”）在前一个完成时订阅。
• 删除重复项并非绝对必要，但似乎可以满足要求。
• 使用defer 以便在订阅内部 Observable 时计算当前时间。
• concat 用于连续执行每个内部 Observable（感谢 martin），避免同时订阅列表中的每个时间的开销。

这种方式满足了我原来的需求：

我有一个需要在特定时间重新计算的 RxJS Observable，如 DateTime 对象数组所述...

如果我将它与需要使用the combineLatest operator 重新计算的数据结合起来，以在正确的时间触发重新计算：

const timeAwareData$ = combineLatest(timeUnawareData$, nextTime$).pipe(
    tap(() => console.log('either the data has changed or a time has been reached')),
    // ...
);

我不喜欢我的解决方案的地方

它为列表中的每个时间同时创建一个单独的内部 Observable。我觉得可以重构，使得每个内部 Observable 仅在前一个被销毁后创建。我们将不胜感激地收到任何改进提示。

Answer 4

您可能希望将您的解决方案建立在这些基础之上：

• Call a javascript function at a specific time of day
• Javascript: how to trigger at a specific time
• Running JavaScript at Specified Times: Timed JavaScript