windowCount.js 1.4 KB

12345678910111213141516171819202122232425262728293031323334353637383940
  1. import { Subject } from '../Subject';
  2. import { operate } from '../util/lift';
  3. import { createOperatorSubscriber } from './OperatorSubscriber';
  4. export function windowCount(windowSize, startWindowEvery = 0) {
  5. const startEvery = startWindowEvery > 0 ? startWindowEvery : windowSize;
  6. return operate((source, subscriber) => {
  7. let windows = [new Subject()];
  8. let starts = [];
  9. let count = 0;
  10. subscriber.next(windows[0].asObservable());
  11. source.subscribe(createOperatorSubscriber(subscriber, (value) => {
  12. for (const window of windows) {
  13. window.next(value);
  14. }
  15. const c = count - windowSize + 1;
  16. if (c >= 0 && c % startEvery === 0) {
  17. windows.shift().complete();
  18. }
  19. if (++count % startEvery === 0) {
  20. const window = new Subject();
  21. windows.push(window);
  22. subscriber.next(window.asObservable());
  23. }
  24. }, () => {
  25. while (windows.length > 0) {
  26. windows.shift().complete();
  27. }
  28. subscriber.complete();
  29. }, (err) => {
  30. while (windows.length > 0) {
  31. windows.shift().error(err);
  32. }
  33. subscriber.error(err);
  34. }, () => {
  35. starts = null;
  36. windows = null;
  37. }));
  38. });
  39. }
  40. //# sourceMappingURL=windowCount.js.map