hunk.code_window:     }
       return indexOf * TestScheduler.frameTimeFactor;
     }
     createColdObservable<T>(marbles: string, values?: any, error?: any): ColdObservable<T> {
       if (marbles.indexOf('^') !== -1) {
         throw new Error('cold observable cannot have subscription offset "^"');
       }
       if (marbles.indexOf('!') !== -1) {
  
hunk.labels: keep keep keep keep replace keep keep keep keep
hunk.after_edit:   /**
      * @param marbles A diagram in the marble DSL. Letters map to keys in `values` if provided.
      * @param values Values to use for the letters in `marbles`. If ommitted, the letters themselves are used.
      * @param error The error to use for the `#` marble (if present).
      */
     createColdObservable<T = string>(marbles: string, values?: { [marble: string]: T }, error?: any): ColdObservable<T> {
  
hunk.file_path: src/internal/testing/TestScheduler.ts
hunk.type: replace
file: import { Observable } from '../Observable';
  import { Notification } from '../Notification';
  import { ColdObservable } from './ColdObservable';
  import { HotObservable } from './HotObservable';
  import { TestMessage } from './TestMessage';
  import { SubscriptionLog } from './SubscriptionLog';
  import { Subscription } from '../Subscription';
  import { VirtualTimeScheduler, VirtualAction } from '../scheduler/VirtualTimeScheduler';
  import { AsyncScheduler } from '../scheduler/AsyncScheduler';
  
  const defaultMaxFrame: number = 750;
  
  export interface RunHelpers {
    cold: typeof TestScheduler.prototype.createColdObservable;
    hot: typeof TestScheduler.prototype.createHotObservable;
    flush: typeof TestScheduler.prototype.flush;
    expectObservable: typeof TestScheduler.prototype.expectObservable;
    expectSubscriptions: typeof TestScheduler.prototype.expectSubscriptions;
  }
  
  interface FlushableTest {
    ready: boolean;
    actual?: any[];
    expected?: any[];
  }
  
  export type observableToBeFn = (marbles: string, values?: any, errorValue?: any) => void;
  export type subscriptionLogsToBeFn = (marbles: string | string[]) => void;
  
  export class TestScheduler extends VirtualTimeScheduler {
    public readonly hotObservables: HotObservable<any>[] = [];
    public readonly coldObservables: ColdObservable<any>[] = [];
    private flushTests: FlushableTest[] = [];
    private runMode = false;
  
    constructor(public assertDeepEqual: (actual: any, expected: any) => boolean | void) {
      super(VirtualAction, defaultMaxFrame);
    }
  
    createTime(marbles: string): number {
      const indexOf: number = marbles.indexOf('|');
      if (indexOf === -1) {
        throw new Error('marble diagram for time should have a completion marker "|"');
      }
      return indexOf * TestScheduler.frameTimeFactor;
    }
  
    createColdObservable<T>(marbles: string, values?: any, error?: any): ColdObservable<T> {
      if (marbles.indexOf('^') !== -1) {
        throw new Error('cold observable cannot have subscription offset "^"');
      }
      if (marbles.indexOf('!') !== -1) {
        throw new Error('cold observable cannot have unsubscription marker "!"');
      }
      const messages = TestScheduler.parseMarbles(marbles, values, error, undefined, this.runMode);
      const cold = new ColdObservable<T>(messages, this);
      this.coldObservables.push(cold);
      return cold;
    }
  
    createHotObservable<T>(marbles: string, values?: any, error?: any): HotObservable<T> {
      if (marbles.indexOf('!') !== -1) {
        throw new Error('hot observable cannot have unsubscription marker "!"');
      }
      const messages = TestScheduler.parseMarbles(marbles, values, error, undefined, this.runMode);
      const subject = new HotObservable<T>(messages, this);
      this.hotObservables.push(subject);
      return subject;
    }
  
    private materializeInnerObservable(observable: Observable<any>,
                                       outerFrame: number): TestMessage[] {
      const messages: TestMessage[] = [];
      observable.subscribe((value) => {
        messages.push({ frame: this.frame - outerFrame, notification: Notification.createNext(value) });
      }, (err) => {
        messages.push({ frame: this.frame - outerFrame, notification: Notification.createError(err) });
      }, () => {
        messages.push({ frame: this.frame - outerFrame, notification: Notification.createComplete() });
      });
      return messages;
    }
  
    expectObservable(observable: Observable<any>,
                     unsubscriptionMarbles: string = null): ({ toBe: observableToBeFn }) {
      const actual: TestMessage[] = [];
      const flushTest: FlushableTest = { actual, ready: false };
      const unsubscriptionFrame = TestScheduler
        .parseMarblesAsSubscriptions(unsubscriptionMarbles, this.runMode).unsubscribedFrame;
      let subscription: Subscription;
  
      this.schedule(() => {
        subscription = observable.subscribe(x => {
          let value = x;
          // Support Observable-of-Observables
          if (x instanceof Observable) {
            value = this.materializeInnerObservable(value, this.frame);
          }
          actual.push({ frame: this.frame, notification: Notification.createNext(value) });
        }, (err) => {
          actual.push({ frame: this.frame, notification: Notification.createError(err) });
        }, () => {
          actual.push({ frame: this.frame, notification: Notification.createComplete() });
        });
      }, 0);
  
      if (unsubscriptionFrame !== Number.POSITIVE_INFINITY) {
        this.schedule(() => subscription.unsubscribe(), unsubscriptionFrame);
      }
  
      this.flushTests.push(flushTest);
      const { runMode } = this;
  
      return {
        toBe(marbles: string, values?: any, errorValue?: any) {
          flushTest.ready = true;
          flushTest.expected = TestScheduler.parseMarbles(marbles, values, errorValue, true, runMode);
        }
      };
    }
  
    expectSubscriptions(actualSubscriptionLogs: SubscriptionLog[]): ({ toBe: subscriptionLogsToBeFn }) {
      const flushTest: FlushableTest = { actual: actualSubscriptionLogs, ready: false };
      this.flushTests.push(flushTest);
      const { runMode } = this;
      return {
        toBe(marbles: string | string[]) {
          const marblesArray: string[] = (typeof marbles === 'string') ? [marbles] : marbles;
          flushTest.ready = true;
          flushTest.expected = marblesArray.map(marbles =>
            TestScheduler.parseMarblesAsSubscriptions(marbles, runMode)
          );
        }
      };
    }
  
    flush() {
      const hotObservables = this.hotObservables;
      while (hotObservables.length > 0) {
        hotObservables.shift().setup();
      }
  
      super.flush();
  
      this.flushTests = this.flushTests.filter(test => {
        if (test.ready) {
          this.assertDeepEqual(test.actual, test.expected);
          return false;
        }
        return true;
      });
    }
  
    /** @nocollapse */
    static parseMarblesAsSubscriptions(marbles: string, runMode = false): SubscriptionLog {
      if (typeof marbles !== 'string') {
        return new SubscriptionLog(Number.POSITIVE_INFINITY);
      }
      const len = marbles.length;
      let groupStart = -1;
      let subscriptionFrame = Number.POSITIVE_INFINITY;
      let unsubscriptionFrame = Number.POSITIVE_INFINITY;
      let frame = 0;
  
      for (let i = 0; i < len; i++) {
        let nextFrame = frame;
        const advanceFrameBy = (count: number) => {
          nextFrame += count * this.frameTimeFactor;
        };
        const c = marbles[i];
        switch (c) {
          case ' ':
            // Whitespace no longer advances time
            if (!runMode) {
              advanceFrameBy(1);
            }
            break;
          case '-':
            advanceFrameBy(1);
            break;
          case '(':
            groupStart = frame;
            advanceFrameBy(1);
            break;
          case ')':
            groupStart = -1;
            advanceFrameBy(1);
            break;
          case '^':
            if (subscriptionFrame !== Number.POSITIVE_INFINITY) {
              throw new Error('found a second subscription point \'^\' in a ' +
                'subscription marble diagram. There can only be one.');
            }
            subscriptionFrame = groupStart > -1 ? groupStart : frame;
            advanceFrameBy(1);
            break;
          case '!':
            if (unsubscriptionFrame !== Number.POSITIVE_INFINITY) {
              throw new Error('found a second subscription point \'^\' in a ' +
                'subscription marble diagram. There can only be one.');
            }
            unsubscriptionFrame = groupStart > -1 ? groupStart : frame;
            break;
          default:
            // time progression syntax
            if (runMode && c.match(/^[0-9]$/)) {
              // Time progression must be preceeded by at least one space
              // if it's not at the beginning of the diagram
              if (i === 0 || marbles[i - 1] === ' ') {
                const buffer = marbles.slice(i);
                const match = buffer.match(/^([0-9]+(?:\.[0-9]+)?)(ms|s|m) /);
                if (match) {
                  i += match[0].length - 1;
                  const duration = parseFloat(match[1]);
                  const unit = match[2];
                  let durationInMs: number;
  
                  switch (unit) {
                    case 'ms':
                      durationInMs = duration;
                      break;
                    case 's':
                      durationInMs = duration * 1000;
                      break;
                    case 'm':
                      durationInMs = duration * 1000 * 60;
                      break;
                    default:
                      break;
                  }
  
                  advanceFrameBy(durationInMs / this.frameTimeFactor);
                  break;
                }
              }
            }
  
            throw new Error('there can only be \'^\' and \'!\' markers in a ' +
              'subscription marble diagram. Found instead \'' + c + '\'.');
        }
  
        frame = nextFrame;
      }
  
      if (unsubscriptionFrame < 0) {
        return new SubscriptionLog(subscriptionFrame);
      } else {
        return new SubscriptionLog(subscriptionFrame, unsubscriptionFrame);
      }
    }
  
    /** @nocollapse */
    static parseMarbles(marbles: string,
                        values?: any,
                        errorValue?: any,
                        materializeInnerObservables: boolean = false,
                        runMode = false): TestMessage[] {
      if (marbles.indexOf('!') !== -1) {
        throw new Error('conventional marble diagrams cannot have the ' +
          'unsubscription marker "!"');
      }
      const len = marbles.length;
      const testMessages: TestMessage[] = [];
      const subIndex = runMode ? marbles.replace(/^[ ]+/, '').indexOf('^') : marbles.indexOf('^');
      let frame = subIndex === -1 ? 0 : (subIndex * -this.frameTimeFactor);
      const getValue = typeof values !== 'object' ?
        (x: any) => x :
        (x: any) => {
          // Support Observable-of-Observables
          if (materializeInnerObservables && values[x] instanceof ColdObservable) {
            return values[x].messages;
          }
          return values[x];
        };
      let groupStart = -1;
  
      for (let i = 0; i < len; i++) {
        let nextFrame = frame;
        const advanceFrameBy = (count: number) => {
          nextFrame += count * this.frameTimeFactor;
        };
  
        let notification: Notification<any>;
        const c = marbles[i];
        switch (c) {
          case ' ':
            // Whitespace no longer advances time
            if (!runMode) {
              advanceFrameBy(1);
            }
            break;
          case '-':
            advanceFrameBy(1);
            break;
          case '(':
            groupStart = frame;
            advanceFrameBy(1);
            break;
          case ')':
            groupStart = -1;
            advanceFrameBy(1);
            break;
          case '|':
            notification = Notification.createComplete();
            advanceFrameBy(1);
            break;
          case '^':
            advanceFrameBy(1);
            break;
          case '#':
            notification = Notification.createError(errorValue || 'error');
            advanceFrameBy(1);
            break;
          default:
            // Might be time progression syntax, or a value literal
            if (runMode && c.match(/^[0-9]$/)) {
              // Time progression must be preceeded by at least one space
              // if it's not at the beginning of the diagram
              if (i === 0 || marbles[i - 1] === ' ') {
                const buffer = marbles.slice(i);
                const match = buffer.match(/^([0-9]+(?:\.[0-9]+)?)(ms|s|m) /);
                if (match) {
                  i += match[0].length - 1;
                  const duration = parseFloat(match[1]);
                  const unit = match[2];
                  let durationInMs: number;
  
                  switch (unit) {
                    case 'ms':
                      durationInMs = duration;
                      break;
                    case 's':
                      durationInMs = duration * 1000;
                      break;
                    case 'm':
                      durationInMs = duration * 1000 * 60;
                      break;
                    default:
                      break;
                  }
  
                  advanceFrameBy(durationInMs / this.frameTimeFactor);
                  break;
                }
              }
            }
  
            notification = Notification.createNext(getValue(c));
            advanceFrameBy(1);
            break;
        }
  
        if (notification) {
          testMessages.push({ frame: groupStart > -1 ? groupStart : frame, notification });
        }
  
        frame = nextFrame;
      }
      return testMessages;
    }
  
    run<T>(callback: (helpers: RunHelpers) => T): T {
      const prevFrameTimeFactor = TestScheduler.frameTimeFactor;
      const prevMaxFrames = this.maxFrames;
  
      TestScheduler.frameTimeFactor = 1;
      this.maxFrames = Number.POSITIVE_INFINITY;
      this.runMode = true;
      AsyncScheduler.delegate = this;
  
      const helpers = {
        cold: this.createColdObservable.bind(this),
        hot: this.createHotObservable.bind(this),
        flush: this.flush.bind(this),
        expectObservable: this.expectObservable.bind(this),
        expectSubscriptions: this.expectSubscriptions.bind(this),
      };
      try {
        const ret = callback(helpers);
        this.flush();
        return ret;
      } finally {
        TestScheduler.frameTimeFactor = prevFrameTimeFactor;
        this.maxFrames = prevMaxFrames;
        this.runMode = false;
        AsyncScheduler.delegate = undefined;
      }
    }
  }
  