6352ca6efffa3ab5ca421ced334f7e76476130ba4c1f01685710a98706b8ca33575160047c67280012db5001807c3779759a1333763327bdf671b38326cf84 1.9 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152
  1. import { Observable } from '../Observable';
  2. import { Subscription } from '../Subscription';
  3. import { Scheduler } from '../Scheduler';
  4. import { TestMessage } from './TestMessage';
  5. import { SubscriptionLog } from './SubscriptionLog';
  6. import { SubscriptionLoggable } from './SubscriptionLoggable';
  7. import { applyMixins } from '../util/applyMixins';
  8. import { Subscriber } from '../Subscriber';
  9. import { observeNotification } from '../Notification';
  10. export class ColdObservable<T> extends Observable<T> implements SubscriptionLoggable {
  11. public subscriptions: SubscriptionLog[] = [];
  12. scheduler: Scheduler;
  13. // @ts-ignore: Property has no initializer and is not definitely assigned
  14. logSubscribedFrame: () => number;
  15. // @ts-ignore: Property has no initializer and is not definitely assigned
  16. logUnsubscribedFrame: (index: number) => void;
  17. constructor(public messages: TestMessage[], scheduler: Scheduler) {
  18. super(function (this: Observable<T>, subscriber: Subscriber<any>) {
  19. const observable: ColdObservable<T> = this as any;
  20. const index = observable.logSubscribedFrame();
  21. const subscription = new Subscription();
  22. subscription.add(
  23. new Subscription(() => {
  24. observable.logUnsubscribedFrame(index);
  25. })
  26. );
  27. observable.scheduleMessages(subscriber);
  28. return subscription;
  29. });
  30. this.scheduler = scheduler;
  31. }
  32. scheduleMessages(subscriber: Subscriber<any>) {
  33. const messagesLength = this.messages.length;
  34. for (let i = 0; i < messagesLength; i++) {
  35. const message = this.messages[i];
  36. subscriber.add(
  37. this.scheduler.schedule(
  38. (state) => {
  39. const { message: { notification }, subscriber: destination } = state!;
  40. observeNotification(notification, destination);
  41. },
  42. message.frame,
  43. { message, subscriber }
  44. )
  45. );
  46. }
  47. }
  48. }
  49. applyMixins(ColdObservable, [SubscriptionLoggable]);