zip.js 7.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255
  1. import { ArrayObservable } from '../observable/ArrayObservable';
  2. import { isArray } from '../util/isArray';
  3. import { Subscriber } from '../Subscriber';
  4. import { OuterSubscriber } from '../OuterSubscriber';
  5. import { subscribeToResult } from '../util/subscribeToResult';
  6. import { iterator as Symbol_iterator } from '../symbol/iterator';
  7. /* tslint:enable:max-line-length */
  8. /**
  9. * @param observables
  10. * @return {Observable<R>}
  11. * @method zip
  12. * @owner Observable
  13. */
  14. export function zip(...observables) {
  15. return function zipOperatorFunction(source) {
  16. return source.lift.call(zipStatic(source, ...observables));
  17. };
  18. }
  19. /* tslint:enable:max-line-length */
  20. /**
  21. * Combines multiple Observables to create an Observable whose values are calculated from the values, in order, of each
  22. * of its input Observables.
  23. *
  24. * If the latest parameter is a function, this function is used to compute the created value from the input values.
  25. * Otherwise, an array of the input values is returned.
  26. *
  27. * @example <caption>Combine age and name from different sources</caption>
  28. *
  29. * let age$ = Observable.of<number>(27, 25, 29);
  30. * let name$ = Observable.of<string>('Foo', 'Bar', 'Beer');
  31. * let isDev$ = Observable.of<boolean>(true, true, false);
  32. *
  33. * Observable
  34. * .zip(age$,
  35. * name$,
  36. * isDev$,
  37. * (age: number, name: string, isDev: boolean) => ({ age, name, isDev }))
  38. * .subscribe(x => console.log(x));
  39. *
  40. * // outputs
  41. * // { age: 27, name: 'Foo', isDev: true }
  42. * // { age: 25, name: 'Bar', isDev: true }
  43. * // { age: 29, name: 'Beer', isDev: false }
  44. *
  45. * @param observables
  46. * @return {Observable<R>}
  47. * @static true
  48. * @name zip
  49. * @owner Observable
  50. */
  51. export function zipStatic(...observables) {
  52. const project = observables[observables.length - 1];
  53. if (typeof project === 'function') {
  54. observables.pop();
  55. }
  56. return new ArrayObservable(observables).lift(new ZipOperator(project));
  57. }
  58. export class ZipOperator {
  59. constructor(project) {
  60. this.project = project;
  61. }
  62. call(subscriber, source) {
  63. return source.subscribe(new ZipSubscriber(subscriber, this.project));
  64. }
  65. }
  66. /**
  67. * We need this JSDoc comment for affecting ESDoc.
  68. * @ignore
  69. * @extends {Ignored}
  70. */
  71. export class ZipSubscriber extends Subscriber {
  72. constructor(destination, project, values = Object.create(null)) {
  73. super(destination);
  74. this.iterators = [];
  75. this.active = 0;
  76. this.project = (typeof project === 'function') ? project : null;
  77. this.values = values;
  78. }
  79. _next(value) {
  80. const iterators = this.iterators;
  81. if (isArray(value)) {
  82. iterators.push(new StaticArrayIterator(value));
  83. }
  84. else if (typeof value[Symbol_iterator] === 'function') {
  85. iterators.push(new StaticIterator(value[Symbol_iterator]()));
  86. }
  87. else {
  88. iterators.push(new ZipBufferIterator(this.destination, this, value));
  89. }
  90. }
  91. _complete() {
  92. const iterators = this.iterators;
  93. const len = iterators.length;
  94. if (len === 0) {
  95. this.destination.complete();
  96. return;
  97. }
  98. this.active = len;
  99. for (let i = 0; i < len; i++) {
  100. let iterator = iterators[i];
  101. if (iterator.stillUnsubscribed) {
  102. this.add(iterator.subscribe(iterator, i));
  103. }
  104. else {
  105. this.active--; // not an observable
  106. }
  107. }
  108. }
  109. notifyInactive() {
  110. this.active--;
  111. if (this.active === 0) {
  112. this.destination.complete();
  113. }
  114. }
  115. checkIterators() {
  116. const iterators = this.iterators;
  117. const len = iterators.length;
  118. const destination = this.destination;
  119. // abort if not all of them have values
  120. for (let i = 0; i < len; i++) {
  121. let iterator = iterators[i];
  122. if (typeof iterator.hasValue === 'function' && !iterator.hasValue()) {
  123. return;
  124. }
  125. }
  126. let shouldComplete = false;
  127. const args = [];
  128. for (let i = 0; i < len; i++) {
  129. let iterator = iterators[i];
  130. let result = iterator.next();
  131. // check to see if it's completed now that you've gotten
  132. // the next value.
  133. if (iterator.hasCompleted()) {
  134. shouldComplete = true;
  135. }
  136. if (result.done) {
  137. destination.complete();
  138. return;
  139. }
  140. args.push(result.value);
  141. }
  142. if (this.project) {
  143. this._tryProject(args);
  144. }
  145. else {
  146. destination.next(args);
  147. }
  148. if (shouldComplete) {
  149. destination.complete();
  150. }
  151. }
  152. _tryProject(args) {
  153. let result;
  154. try {
  155. result = this.project.apply(this, args);
  156. }
  157. catch (err) {
  158. this.destination.error(err);
  159. return;
  160. }
  161. this.destination.next(result);
  162. }
  163. }
  164. class StaticIterator {
  165. constructor(iterator) {
  166. this.iterator = iterator;
  167. this.nextResult = iterator.next();
  168. }
  169. hasValue() {
  170. return true;
  171. }
  172. next() {
  173. const result = this.nextResult;
  174. this.nextResult = this.iterator.next();
  175. return result;
  176. }
  177. hasCompleted() {
  178. const nextResult = this.nextResult;
  179. return nextResult && nextResult.done;
  180. }
  181. }
  182. class StaticArrayIterator {
  183. constructor(array) {
  184. this.array = array;
  185. this.index = 0;
  186. this.length = 0;
  187. this.length = array.length;
  188. }
  189. [Symbol_iterator]() {
  190. return this;
  191. }
  192. next(value) {
  193. const i = this.index++;
  194. const array = this.array;
  195. return i < this.length ? { value: array[i], done: false } : { value: null, done: true };
  196. }
  197. hasValue() {
  198. return this.array.length > this.index;
  199. }
  200. hasCompleted() {
  201. return this.array.length === this.index;
  202. }
  203. }
  204. /**
  205. * We need this JSDoc comment for affecting ESDoc.
  206. * @ignore
  207. * @extends {Ignored}
  208. */
  209. class ZipBufferIterator extends OuterSubscriber {
  210. constructor(destination, parent, observable) {
  211. super(destination);
  212. this.parent = parent;
  213. this.observable = observable;
  214. this.stillUnsubscribed = true;
  215. this.buffer = [];
  216. this.isComplete = false;
  217. }
  218. [Symbol_iterator]() {
  219. return this;
  220. }
  221. // NOTE: there is actually a name collision here with Subscriber.next and Iterator.next
  222. // this is legit because `next()` will never be called by a subscription in this case.
  223. next() {
  224. const buffer = this.buffer;
  225. if (buffer.length === 0 && this.isComplete) {
  226. return { value: null, done: true };
  227. }
  228. else {
  229. return { value: buffer.shift(), done: false };
  230. }
  231. }
  232. hasValue() {
  233. return this.buffer.length > 0;
  234. }
  235. hasCompleted() {
  236. return this.buffer.length === 0 && this.isComplete;
  237. }
  238. notifyComplete() {
  239. if (this.buffer.length > 0) {
  240. this.isComplete = true;
  241. this.parent.notifyInactive();
  242. }
  243. else {
  244. this.destination.complete();
  245. }
  246. }
  247. notifyNext(outerValue, innerValue, outerIndex, innerIndex, innerSub) {
  248. this.buffer.push(innerValue);
  249. this.parent.checkIterators();
  250. }
  251. subscribe(value, index) {
  252. return subscribeToResult(this, this.observable, this, index);
  253. }
  254. }
  255. //# sourceMappingURL=zip.js.map