Subject.js 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134
  1. import { Observable } from './Observable';
  2. import { Subscription, EMPTY_SUBSCRIPTION } from './Subscription';
  3. import { ObjectUnsubscribedError } from './util/ObjectUnsubscribedError';
  4. import { arrRemove } from './util/arrRemove';
  5. import { errorContext } from './util/errorContext';
  6. export class Subject extends Observable {
  7. constructor() {
  8. super();
  9. this.closed = false;
  10. this.currentObservers = null;
  11. this.observers = [];
  12. this.isStopped = false;
  13. this.hasError = false;
  14. this.thrownError = null;
  15. }
  16. lift(operator) {
  17. const subject = new AnonymousSubject(this, this);
  18. subject.operator = operator;
  19. return subject;
  20. }
  21. _throwIfClosed() {
  22. if (this.closed) {
  23. throw new ObjectUnsubscribedError();
  24. }
  25. }
  26. next(value) {
  27. errorContext(() => {
  28. this._throwIfClosed();
  29. if (!this.isStopped) {
  30. if (!this.currentObservers) {
  31. this.currentObservers = Array.from(this.observers);
  32. }
  33. for (const observer of this.currentObservers) {
  34. observer.next(value);
  35. }
  36. }
  37. });
  38. }
  39. error(err) {
  40. errorContext(() => {
  41. this._throwIfClosed();
  42. if (!this.isStopped) {
  43. this.hasError = this.isStopped = true;
  44. this.thrownError = err;
  45. const { observers } = this;
  46. while (observers.length) {
  47. observers.shift().error(err);
  48. }
  49. }
  50. });
  51. }
  52. complete() {
  53. errorContext(() => {
  54. this._throwIfClosed();
  55. if (!this.isStopped) {
  56. this.isStopped = true;
  57. const { observers } = this;
  58. while (observers.length) {
  59. observers.shift().complete();
  60. }
  61. }
  62. });
  63. }
  64. unsubscribe() {
  65. this.isStopped = this.closed = true;
  66. this.observers = this.currentObservers = null;
  67. }
  68. get observed() {
  69. var _a;
  70. return ((_a = this.observers) === null || _a === void 0 ? void 0 : _a.length) > 0;
  71. }
  72. _trySubscribe(subscriber) {
  73. this._throwIfClosed();
  74. return super._trySubscribe(subscriber);
  75. }
  76. _subscribe(subscriber) {
  77. this._throwIfClosed();
  78. this._checkFinalizedStatuses(subscriber);
  79. return this._innerSubscribe(subscriber);
  80. }
  81. _innerSubscribe(subscriber) {
  82. const { hasError, isStopped, observers } = this;
  83. if (hasError || isStopped) {
  84. return EMPTY_SUBSCRIPTION;
  85. }
  86. this.currentObservers = null;
  87. observers.push(subscriber);
  88. return new Subscription(() => {
  89. this.currentObservers = null;
  90. arrRemove(observers, subscriber);
  91. });
  92. }
  93. _checkFinalizedStatuses(subscriber) {
  94. const { hasError, thrownError, isStopped } = this;
  95. if (hasError) {
  96. subscriber.error(thrownError);
  97. }
  98. else if (isStopped) {
  99. subscriber.complete();
  100. }
  101. }
  102. asObservable() {
  103. const observable = new Observable();
  104. observable.source = this;
  105. return observable;
  106. }
  107. }
  108. Subject.create = (destination, source) => {
  109. return new AnonymousSubject(destination, source);
  110. };
  111. export class AnonymousSubject extends Subject {
  112. constructor(destination, source) {
  113. super();
  114. this.destination = destination;
  115. this.source = source;
  116. }
  117. next(value) {
  118. var _a, _b;
  119. (_b = (_a = this.destination) === null || _a === void 0 ? void 0 : _a.next) === null || _b === void 0 ? void 0 : _b.call(_a, value);
  120. }
  121. error(err) {
  122. var _a, _b;
  123. (_b = (_a = this.destination) === null || _a === void 0 ? void 0 : _a.error) === null || _b === void 0 ? void 0 : _b.call(_a, err);
  124. }
  125. complete() {
  126. var _a, _b;
  127. (_b = (_a = this.destination) === null || _a === void 0 ? void 0 : _a.complete) === null || _b === void 0 ? void 0 : _b.call(_a);
  128. }
  129. _subscribe(subscriber) {
  130. var _a, _b;
  131. return (_b = (_a = this.source) === null || _a === void 0 ? void 0 : _a.subscribe(subscriber)) !== null && _b !== void 0 ? _b : EMPTY_SUBSCRIPTION;
  132. }
  133. }
  134. //# sourceMappingURL=Subject.js.map