Repositorio del curso CCOM4030 el semestre B91 del proyecto Artesanías con el Instituto de Cultura

bufferTime.js 5.3KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141
  1. import { async } from '../scheduler/async';
  2. import { Subscriber } from '../Subscriber';
  3. import { isScheduler } from '../util/isScheduler';
  4. export function bufferTime(bufferTimeSpan) {
  5. let length = arguments.length;
  6. let scheduler = async;
  7. if (isScheduler(arguments[arguments.length - 1])) {
  8. scheduler = arguments[arguments.length - 1];
  9. length--;
  10. }
  11. let bufferCreationInterval = null;
  12. if (length >= 2) {
  13. bufferCreationInterval = arguments[1];
  14. }
  15. let maxBufferSize = Number.POSITIVE_INFINITY;
  16. if (length >= 3) {
  17. maxBufferSize = arguments[2];
  18. }
  19. return function bufferTimeOperatorFunction(source) {
  20. return source.lift(new BufferTimeOperator(bufferTimeSpan, bufferCreationInterval, maxBufferSize, scheduler));
  21. };
  22. }
  23. class BufferTimeOperator {
  24. constructor(bufferTimeSpan, bufferCreationInterval, maxBufferSize, scheduler) {
  25. this.bufferTimeSpan = bufferTimeSpan;
  26. this.bufferCreationInterval = bufferCreationInterval;
  27. this.maxBufferSize = maxBufferSize;
  28. this.scheduler = scheduler;
  29. }
  30. call(subscriber, source) {
  31. return source.subscribe(new BufferTimeSubscriber(subscriber, this.bufferTimeSpan, this.bufferCreationInterval, this.maxBufferSize, this.scheduler));
  32. }
  33. }
  34. class Context {
  35. constructor() {
  36. this.buffer = [];
  37. }
  38. }
  39. class BufferTimeSubscriber extends Subscriber {
  40. constructor(destination, bufferTimeSpan, bufferCreationInterval, maxBufferSize, scheduler) {
  41. super(destination);
  42. this.bufferTimeSpan = bufferTimeSpan;
  43. this.bufferCreationInterval = bufferCreationInterval;
  44. this.maxBufferSize = maxBufferSize;
  45. this.scheduler = scheduler;
  46. this.contexts = [];
  47. const context = this.openContext();
  48. this.timespanOnly = bufferCreationInterval == null || bufferCreationInterval < 0;
  49. if (this.timespanOnly) {
  50. const timeSpanOnlyState = { subscriber: this, context, bufferTimeSpan };
  51. this.add(context.closeAction = scheduler.schedule(dispatchBufferTimeSpanOnly, bufferTimeSpan, timeSpanOnlyState));
  52. }
  53. else {
  54. const closeState = { subscriber: this, context };
  55. const creationState = { bufferTimeSpan, bufferCreationInterval, subscriber: this, scheduler };
  56. this.add(context.closeAction = scheduler.schedule(dispatchBufferClose, bufferTimeSpan, closeState));
  57. this.add(scheduler.schedule(dispatchBufferCreation, bufferCreationInterval, creationState));
  58. }
  59. }
  60. _next(value) {
  61. const contexts = this.contexts;
  62. const len = contexts.length;
  63. let filledBufferContext;
  64. for (let i = 0; i < len; i++) {
  65. const context = contexts[i];
  66. const buffer = context.buffer;
  67. buffer.push(value);
  68. if (buffer.length == this.maxBufferSize) {
  69. filledBufferContext = context;
  70. }
  71. }
  72. if (filledBufferContext) {
  73. this.onBufferFull(filledBufferContext);
  74. }
  75. }
  76. _error(err) {
  77. this.contexts.length = 0;
  78. super._error(err);
  79. }
  80. _complete() {
  81. const { contexts, destination } = this;
  82. while (contexts.length > 0) {
  83. const context = contexts.shift();
  84. destination.next(context.buffer);
  85. }
  86. super._complete();
  87. }
  88. _unsubscribe() {
  89. this.contexts = null;
  90. }
  91. onBufferFull(context) {
  92. this.closeContext(context);
  93. const closeAction = context.closeAction;
  94. closeAction.unsubscribe();
  95. this.remove(closeAction);
  96. if (!this.closed && this.timespanOnly) {
  97. context = this.openContext();
  98. const bufferTimeSpan = this.bufferTimeSpan;
  99. const timeSpanOnlyState = { subscriber: this, context, bufferTimeSpan };
  100. this.add(context.closeAction = this.scheduler.schedule(dispatchBufferTimeSpanOnly, bufferTimeSpan, timeSpanOnlyState));
  101. }
  102. }
  103. openContext() {
  104. const context = new Context();
  105. this.contexts.push(context);
  106. return context;
  107. }
  108. closeContext(context) {
  109. this.destination.next(context.buffer);
  110. const contexts = this.contexts;
  111. const spliceIndex = contexts ? contexts.indexOf(context) : -1;
  112. if (spliceIndex >= 0) {
  113. contexts.splice(contexts.indexOf(context), 1);
  114. }
  115. }
  116. }
  117. function dispatchBufferTimeSpanOnly(state) {
  118. const subscriber = state.subscriber;
  119. const prevContext = state.context;
  120. if (prevContext) {
  121. subscriber.closeContext(prevContext);
  122. }
  123. if (!subscriber.closed) {
  124. state.context = subscriber.openContext();
  125. state.context.closeAction = this.schedule(state, state.bufferTimeSpan);
  126. }
  127. }
  128. function dispatchBufferCreation(state) {
  129. const { bufferCreationInterval, bufferTimeSpan, subscriber, scheduler } = state;
  130. const context = subscriber.openContext();
  131. const action = this;
  132. if (!subscriber.closed) {
  133. subscriber.add(context.closeAction = scheduler.schedule(dispatchBufferClose, bufferTimeSpan, { subscriber, context }));
  134. action.schedule(state, bufferCreationInterval);
  135. }
  136. }
  137. function dispatchBufferClose(arg) {
  138. const { subscriber, context } = arg;
  139. subscriber.closeContext(context);
  140. }
  141. //# sourceMappingURL=bufferTime.js.map