forkJoin.js 2.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465
  1. import { Observable } from '../Observable';
  2. import { isArray } from '../util/isArray';
  3. import { EMPTY } from './empty';
  4. import { subscribeToResult } from '../util/subscribeToResult';
  5. import { OuterSubscriber } from '../OuterSubscriber';
  6. import { map } from '../operators/map';
  7. export function forkJoin(...sources) {
  8. let resultSelector;
  9. if (typeof sources[sources.length - 1] === 'function') {
  10. resultSelector = sources.pop();
  11. }
  12. if (sources.length === 1 && isArray(sources[0])) {
  13. sources = sources[0];
  14. }
  15. if (sources.length === 0) {
  16. return EMPTY;
  17. }
  18. if (resultSelector) {
  19. return forkJoin(sources).pipe(map(args => resultSelector(...args)));
  20. }
  21. return new Observable(subscriber => {
  22. return new ForkJoinSubscriber(subscriber, sources);
  23. });
  24. }
  25. class ForkJoinSubscriber extends OuterSubscriber {
  26. constructor(destination, sources) {
  27. super(destination);
  28. this.sources = sources;
  29. this.completed = 0;
  30. this.haveValues = 0;
  31. const len = sources.length;
  32. this.values = new Array(len);
  33. for (let i = 0; i < len; i++) {
  34. const source = sources[i];
  35. const innerSubscription = subscribeToResult(this, source, null, i);
  36. if (innerSubscription) {
  37. this.add(innerSubscription);
  38. }
  39. }
  40. }
  41. notifyNext(outerValue, innerValue, outerIndex, innerIndex, innerSub) {
  42. this.values[outerIndex] = innerValue;
  43. if (!innerSub._hasValue) {
  44. innerSub._hasValue = true;
  45. this.haveValues++;
  46. }
  47. }
  48. notifyComplete(innerSub) {
  49. const { destination, haveValues, values } = this;
  50. const len = values.length;
  51. if (!innerSub._hasValue) {
  52. destination.complete();
  53. return;
  54. }
  55. this.completed++;
  56. if (this.completed !== len) {
  57. return;
  58. }
  59. if (haveValues === len) {
  60. destination.next(values);
  61. }
  62. destination.complete();
  63. }
  64. }
  65. //# sourceMappingURL=forkJoin.js.map