operators/count.js

  1. import { Observable } from '../Observable';
  2. import { filter } from './filter';
  4. /**
  5.  * Counts the number of values coming into the stream on complete
  6.  * 
  7.  * @memberof operators
  8.  *
  9.  * @param {Observable} source$
  10.  * @param {Function} [countCallback] a function to use to filter out values that you do not wish to count
  11.  * @returns {Observable}
  12.  */
  13. export const count = function (source$, countCallback = (value) => true) {
  14.   return new Observable(function ({next, error, complete }) {
  15.     let count = 0;
  16.     const subscription = filter(source$, countCallback).subscribe({
  17.       next () {
  18.         count++;
  19.       },
  20.       error,
  21.       complete () {
  22.         next(count);
  23.         complete();
  24.       }
  25.     });
  26.     
  27.     return () => subscription.unsubscribe();
  28.   });
  29. };
  31. Observable.count = count;
  32. Observable.prototype.count = function (countCallback) {
  33.   return count(this, countCallback);
  34. };