windowTime.js 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147
  1. import { Subject } from '../Subject';
  2. import { async } from '../scheduler/async';
  3. import { Subscriber } from '../Subscriber';
  4. import { isNumeric } from '../util/isNumeric';
  5. import { isScheduler } from '../util/isScheduler';
  6. export function windowTime(windowTimeSpan) {
  7. let scheduler = async;
  8. let windowCreationInterval = null;
  9. let maxWindowSize = Number.POSITIVE_INFINITY;
  10. if (isScheduler(arguments[3])) {
  11. scheduler = arguments[3];
  12. }
  13. if (isScheduler(arguments[2])) {
  14. scheduler = arguments[2];
  15. }
  16. else if (isNumeric(arguments[2])) {
  17. maxWindowSize = arguments[2];
  18. }
  19. if (isScheduler(arguments[1])) {
  20. scheduler = arguments[1];
  21. }
  22. else if (isNumeric(arguments[1])) {
  23. windowCreationInterval = arguments[1];
  24. }
  25. return function windowTimeOperatorFunction(source) {
  26. return source.lift(new WindowTimeOperator(windowTimeSpan, windowCreationInterval, maxWindowSize, scheduler));
  27. };
  28. }
  29. class WindowTimeOperator {
  30. constructor(windowTimeSpan, windowCreationInterval, maxWindowSize, scheduler) {
  31. this.windowTimeSpan = windowTimeSpan;
  32. this.windowCreationInterval = windowCreationInterval;
  33. this.maxWindowSize = maxWindowSize;
  34. this.scheduler = scheduler;
  35. }
  36. call(subscriber, source) {
  37. return source.subscribe(new WindowTimeSubscriber(subscriber, this.windowTimeSpan, this.windowCreationInterval, this.maxWindowSize, this.scheduler));
  38. }
  39. }
  40. class CountedSubject extends Subject {
  41. constructor() {
  42. super(...arguments);
  43. this._numberOfNextedValues = 0;
  44. }
  45. next(value) {
  46. this._numberOfNextedValues++;
  47. super.next(value);
  48. }
  49. get numberOfNextedValues() {
  50. return this._numberOfNextedValues;
  51. }
  52. }
  53. /**
  54. * We need this JSDoc comment for affecting ESDoc.
  55. * @ignore
  56. * @extends {Ignored}
  57. */
  58. class WindowTimeSubscriber extends Subscriber {
  59. constructor(destination, windowTimeSpan, windowCreationInterval, maxWindowSize, scheduler) {
  60. super(destination);
  61. this.destination = destination;
  62. this.windowTimeSpan = windowTimeSpan;
  63. this.windowCreationInterval = windowCreationInterval;
  64. this.maxWindowSize = maxWindowSize;
  65. this.scheduler = scheduler;
  66. this.windows = [];
  67. const window = this.openWindow();
  68. if (windowCreationInterval !== null && windowCreationInterval >= 0) {
  69. const closeState = { subscriber: this, window, context: null };
  70. const creationState = { windowTimeSpan, windowCreationInterval, subscriber: this, scheduler };
  71. this.add(scheduler.schedule(dispatchWindowClose, windowTimeSpan, closeState));
  72. this.add(scheduler.schedule(dispatchWindowCreation, windowCreationInterval, creationState));
  73. }
  74. else {
  75. const timeSpanOnlyState = { subscriber: this, window, windowTimeSpan };
  76. this.add(scheduler.schedule(dispatchWindowTimeSpanOnly, windowTimeSpan, timeSpanOnlyState));
  77. }
  78. }
  79. _next(value) {
  80. const windows = this.windows;
  81. const len = windows.length;
  82. for (let i = 0; i < len; i++) {
  83. const window = windows[i];
  84. if (!window.closed) {
  85. window.next(value);
  86. if (window.numberOfNextedValues >= this.maxWindowSize) {
  87. this.closeWindow(window);
  88. }
  89. }
  90. }
  91. }
  92. _error(err) {
  93. const windows = this.windows;
  94. while (windows.length > 0) {
  95. windows.shift().error(err);
  96. }
  97. this.destination.error(err);
  98. }
  99. _complete() {
  100. const windows = this.windows;
  101. while (windows.length > 0) {
  102. const window = windows.shift();
  103. if (!window.closed) {
  104. window.complete();
  105. }
  106. }
  107. this.destination.complete();
  108. }
  109. openWindow() {
  110. const window = new CountedSubject();
  111. this.windows.push(window);
  112. const destination = this.destination;
  113. destination.next(window);
  114. return window;
  115. }
  116. closeWindow(window) {
  117. window.complete();
  118. const windows = this.windows;
  119. windows.splice(windows.indexOf(window), 1);
  120. }
  121. }
  122. function dispatchWindowTimeSpanOnly(state) {
  123. const { subscriber, windowTimeSpan, window } = state;
  124. if (window) {
  125. subscriber.closeWindow(window);
  126. }
  127. state.window = subscriber.openWindow();
  128. this.schedule(state, windowTimeSpan);
  129. }
  130. function dispatchWindowCreation(state) {
  131. const { windowTimeSpan, subscriber, scheduler, windowCreationInterval } = state;
  132. const window = subscriber.openWindow();
  133. const action = this;
  134. let context = { action, subscription: null };
  135. const timeSpanState = { subscriber, window, context };
  136. context.subscription = scheduler.schedule(dispatchWindowClose, windowTimeSpan, timeSpanState);
  137. action.add(context.subscription);
  138. action.schedule(state, windowCreationInterval);
  139. }
  140. function dispatchWindowClose(state) {
  141. const { subscriber, window, context } = state;
  142. if (context && context.action && context.subscription) {
  143. context.action.remove(context.subscription);
  144. }
  145. subscriber.closeWindow(window);
  146. }
  147. //# sourceMappingURL=windowTime.js.map