2b3f3e85fec5a1590529cd9af7418fc0a3f7e9a5f6ba0b7ba28cb19e16d7eedc41dee8a7fb507f3cedefdb0c5172b96ca489e949eb8de88905f1f2036f9adb 2.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667
  1. import { MonoTypeOperatorFunction, SchedulerLike } from '../types';
  2. import { operate } from '../util/lift';
  3. /**
  4. * Asynchronously subscribes Observers to this Observable on the specified {@link SchedulerLike}.
  5. *
  6. * With `subscribeOn` you can decide what type of scheduler a specific Observable will be using when it is subscribed to.
  7. *
  8. * Schedulers control the speed and order of emissions to observers from an Observable stream.
  9. *
  10. * ![](subscribeOn.png)
  11. *
  12. * ## Example
  13. *
  14. * Given the following code:
  15. *
  16. * ```ts
  17. * import { of, merge } from 'rxjs';
  18. *
  19. * const a = of(1, 2, 3);
  20. * const b = of(4, 5, 6);
  21. *
  22. * merge(a, b).subscribe(console.log);
  23. *
  24. * // Outputs
  25. * // 1
  26. * // 2
  27. * // 3
  28. * // 4
  29. * // 5
  30. * // 6
  31. * ```
  32. *
  33. * Both Observable `a` and `b` will emit their values directly and synchronously once they are subscribed to.
  34. *
  35. * If we instead use the `subscribeOn` operator declaring that we want to use the {@link asyncScheduler} for values emitted by Observable `a`:
  36. *
  37. * ```ts
  38. * import { of, subscribeOn, asyncScheduler, merge } from 'rxjs';
  39. *
  40. * const a = of(1, 2, 3).pipe(subscribeOn(asyncScheduler));
  41. * const b = of(4, 5, 6);
  42. *
  43. * merge(a, b).subscribe(console.log);
  44. *
  45. * // Outputs
  46. * // 4
  47. * // 5
  48. * // 6
  49. * // 1
  50. * // 2
  51. * // 3
  52. * ```
  53. *
  54. * The reason for this is that Observable `b` emits its values directly and synchronously like before
  55. * but the emissions from `a` are scheduled on the event loop because we are now using the {@link asyncScheduler} for that specific Observable.
  56. *
  57. * @param scheduler The {@link SchedulerLike} to perform subscription actions on.
  58. * @param delay A delay to pass to the scheduler to delay subscriptions
  59. * @return A function that returns an Observable modified so that its
  60. * subscriptions happen on the specified {@link SchedulerLike}.
  61. */
  62. export function subscribeOn<T>(scheduler: SchedulerLike, delay: number = 0): MonoTypeOperatorFunction<T> {
  63. return operate((source, subscriber) => {
  64. subscriber.add(scheduler.schedule(() => source.subscribe(subscriber), delay));
  65. });
  66. }