feat(advanceto): implement advanceTo interface

- closes #36

Author: kwonoj

src/index.ts

@@ -27,6 +27,10 @@ export type RxSandboxInstance = {
    * Flush out currently scheduled observables, fill values returned by `getMarbles`.
    */
   flush: typeof TestScheduler.prototype.flush;
+  /**
+   * Flush out currently scheduled observables, only until reaches frame specfied.
+   */
+  advanceTo: typeof TestScheduler.prototype.advanceTo;
   /**
    * Get array of observable value's metadata TestMessage<T> from observable
    * created via `hot` or `cold`. Returned array will be filled once scheduler flushes
@@ -53,10 +57,12 @@ export interface RxSandbox {
    *
    * @param {boolean} [autoFlush] Flush scheduler automatically when `getMarbles` is being called. False by default.
    * @param {number} [frameTimeFactor] Custom frametime factor for virtual time frame. 1 by default.
-   *
+   * @param {number} [maxFrameValue] Maximum frame value of marble diagram can be read.
+   * If marble has value over max frame, it'll be ignored when scheduler flushes out.
+   * 1000 * frameTimeFactory by default.
    * @return {RxSandboxInstance} instance of test scheduler interfaces.
    */
-  create(autoFlush?: boolean, frameTimeFactor?: number): RxSandboxInstance;
+  create(autoFlush?: boolean, frameTimeFactor?: number, maxFrameValue?: number): RxSandboxInstance;
   /**
    * Utility assertion method to assert marble based observable test messages.
    * By default return values of sandbox functions are plain object works with
@@ -79,17 +85,18 @@ export interface RxSandbox {
 }
 
 const rxSandbox: RxSandbox = {
-  create: (autoFlush: boolean = false, frameTimeFactor: number = 1) => {
-    const scheduler = new TestScheduler(autoFlush, frameTimeFactor);
+  create: (autoFlush: boolean = false, frameTimeFactor: number = 1, maxFrameValue = 1000) => {
+    const scheduler = new TestScheduler(autoFlush, frameTimeFactor, maxFrameValue);
 
     return {
       hot: scheduler.createHotObservable.bind(scheduler) as typeof scheduler.createHotObservable,
       cold: scheduler.createColdObservable.bind(scheduler) as typeof scheduler.createColdObservable,
       flush: scheduler.flush.bind(scheduler) as typeof scheduler.flush,
+      advanceTo: scheduler.advanceTo.bind(scheduler) as typeof scheduler.advanceTo,
       getMessages: scheduler.getMessages.bind(scheduler) as typeof scheduler.getMessages,
       e: <T = string>(marble: string, value?: { [key: string]: T } | null, error?: any) =>
-        parseObservableMarble(marble, value, error, true, frameTimeFactor),
-      s: (marble: string) => parseSubscriptionMarble(marble, frameTimeFactor)
+        parseObservableMarble(marble, value, error, true, frameTimeFactor, frameTimeFactor * maxFrameValue),
+      s: (marble: string) => parseSubscriptionMarble(marble, frameTimeFactor, frameTimeFactor * maxFrameValue)
     };
   },
   marbleAssert: marbleAssert

src/marbles/parseObservableMarble.ts

@@ -17,7 +17,8 @@ const parseObservableMarble = <T = string>(
   value?: { [key: string]: T } | null,
   error?: any,
   materializeInnerObservables: boolean = false,
-  frameTimeFactor = 1
+  frameTimeFactor = 1,
+  maxFrame = 1000
 ): Readonly<Array<TestMessage<T | Array<TestMessage<T>>>>> => {
   if (marble.indexOf(SubscriptionMarbleToken.UNSUBSCRIBE) !== -1) {
     throw new Error(`Observable marble cannot have unsubscription marker ${SubscriptionMarbleToken.UNSUBSCRIBE}`);
@@ -29,7 +30,7 @@ const parseObservableMarble = <T = string>(
   const frameOffset = subscriptionIndex < 0 ? 0 : -subscriptionIndex;
 
   const values = marbleTokenArray.reduce(
-    observableTokenParseReducer(value || null, error, materializeInnerObservables, frameTimeFactor),
+    observableTokenParseReducer(value || null, error, materializeInnerObservables, frameTimeFactor, maxFrame),
     {
       currentTimeFrame: frameOffset,
       messages: [],

src/marbles/parseSubscriptionMarble.ts

@@ -2,13 +2,13 @@ import { SubscriptionLog } from 'rxjs/testing/SubscriptionLog';
 import { ObservableMarbleToken } from './ObservableMarbleToken';
 import { subscriptionTokenParseReducer } from './tokenParseReducer';
 
-const parseSubscriptionMarble = (marble: string | null, frameTimeFactor: number = 1) => {
+const parseSubscriptionMarble = (marble: string | null, frameTimeFactor: number = 1, maxFrame = 1000) => {
   if (!marble) {
     return new SubscriptionLog(Number.POSITIVE_INFINITY);
   }
 
   const marbleTokenArray = Array.from(marble).filter(token => token !== ObservableMarbleToken.NOOP);
-  const value = marbleTokenArray.reduce(subscriptionTokenParseReducer(frameTimeFactor), {
+  const value = marbleTokenArray.reduce(subscriptionTokenParseReducer(frameTimeFactor, maxFrame), {
     currentTimeFrame: 0,
     subscriptionFrame: Number.POSITIVE_INFINITY,
     unsubscriptionFrame: Number.POSITIVE_INFINITY,

src/marbles/tokenParseReducer.ts

@@ -115,8 +115,13 @@ const observableTokenParseReducer = <T>(
   value: { [key: string]: T } | null,
   error: any,
   materializeInnerObservables: boolean,
-  frameTimeFactor: number
+  frameTimeFactor: number,
+  maxFrame: number
 ) => (acc: ObservableTokenParseAccumulator<T>, token: any) => {
+  if (acc.currentTimeFrame >= maxFrame) {
+    return acc;
+  }
+
   let message: TestMessage<T | Array<TestMessage<T>>> | null = null;
 
   switch (token) {
@@ -168,10 +173,14 @@ const observableTokenParseReducer = <T>(
  * Reducer to traverse subscription marble diagram to generate SubscriptionLog metadata.
  * @param frameTimeFactor Custom timeframe factor
  */
-const subscriptionTokenParseReducer = (frameTimeFactor: number) => (
+const subscriptionTokenParseReducer = (frameTimeFactor: number, maxFrame: number) => (
   acc: SubscriptionTokenParseAccumulator,
   token: string
 ) => {
+  if (acc.currentTimeFrame >= maxFrame) {
+    return acc;
+  }
+
   switch (token) {
     case SubscriptionMarbleToken.SUBSCRIBE:
       acc.subscriptionFrame = acc.currentTimeFrame;

src/scheduler/TestScheduler.ts

@@ -1,7 +1,8 @@
 import { Notification } from 'rxjs/Notification';
 import { Observable } from 'rxjs/Observable';
-import { VirtualTimeScheduler } from 'rxjs/scheduler/VirtualTimeScheduler';
+import { AsyncAction } from 'rxjs/scheduler/AsyncAction';
 import { VirtualAction } from 'rxjs/scheduler/VirtualTimeScheduler';
+import { VirtualTimeScheduler } from 'rxjs/scheduler/VirtualTimeScheduler';
 import { Subscription } from 'rxjs/Subscription';
 import { ColdObservable } from 'rxjs/testing/ColdObservable';
 import { HotObservable } from 'rxjs/testing/HotObservable';
@@ -18,19 +19,20 @@ class TestScheduler extends VirtualTimeScheduler {
   private readonly coldObservables: Array<ColdObservable<any>> = [];
   private readonly hotObservables: Array<HotObservable<any>> = [];
   private flushed: boolean = false;
+  private flushing: boolean = false;
+
+  private readonly _maxFrame: number;
+  public get maxFrame(): number {
+    return this._maxFrame;
+  }
 
-  constructor(private readonly autoFlush = false, private readonly frameTimeFactor = 1) {
+  constructor(private readonly autoFlush: boolean, private readonly frameTimeFactor: number, maxFrameValue: number) {
     super(VirtualAction, Number.POSITIVE_INFINITY);
+    this._maxFrame = maxFrameValue * frameTimeFactor;
   }
 
   public flush(): void {
-    const hotObservables = this.hotObservables;
-    while (hotObservables.length > 0) {
-      hotObservables.shift()!.setup();
-    }
-
-    super.flush();
-    this.flushed = true;
+    this.flushUntil();
   }
 
   public getMessages<T = string>(observable: Observable<T>, unsubscriptionMarbles: string | null = null) {
@@ -87,7 +89,7 @@ class TestScheduler extends VirtualTimeScheduler {
 
     const messages = Array.isArray(marbleValue)
       ? marbleValue
-      : parseObservableMarble(marbleValue, value, error, false, this.frameTimeFactor) as any;
+      : parseObservableMarble(marbleValue, value, error, false, this.frameTimeFactor, this._maxFrame) as any;
     const observable = new ColdObservable<T>(messages as Array<TestMessage<T | Array<TestMessage<T>>>>, this);
     this.coldObservables.push(observable);
     return observable;
@@ -104,12 +106,26 @@ class TestScheduler extends VirtualTimeScheduler {
 
     const messages = Array.isArray(marbleValue)
       ? marbleValue
-      : parseObservableMarble(marbleValue, value, error, false, this.frameTimeFactor) as any;
+      : parseObservableMarble(marbleValue, value, error, false, this.frameTimeFactor, this._maxFrame) as any;
     const subject = new HotObservable<T>(messages as Array<TestMessage<T | Array<TestMessage<T>>>>, this);
     this.hotObservables.push(subject);
+    subject.setup();
     return subject;
   }
 
+  public advanceTo(toFrame: number): void {
+    if (this.autoFlush) {
+      throw new Error('Cannot advance frame manually with autoflushing scheduler');
+    }
+
+    if (toFrame < 0 || toFrame < this.frame) {
+      throw new Error(`Cannot advance frame, given frame is either negative or smaller than current frame`);
+    }
+
+    this.flushUntil(toFrame);
+    this.frame = toFrame;
+  }
+
   private materializeInnerObservable<T>(observable: Observable<any>, outerFrame: number): Array<TestMessage<T>> {
     const innerObservableMetadata: Array<TestMessage<T>> = [];
     const pushMetaData = (notification: Notification<T>) =>
@@ -123,6 +139,45 @@ class TestScheduler extends VirtualTimeScheduler {
 
     return innerObservableMetadata;
   }
+
+  private peek(): AsyncAction<any> | null {
+    const { actions } = this;
+    return actions && actions.length > 0 ? actions[0] : null;
+  }
+
+  private flushUntil(toFrame: number = this.maxFrame): void {
+    if (this.flushing) {
+      return;
+    }
+
+    this.flushing = true;
+
+    const { actions } = this;
+    let error: any;
+    let action: AsyncAction<any> | null | undefined = null;
+
+    while (this.flushing && (action = this.peek()) && action.delay <= toFrame) {
+      const action: AsyncAction<any> = actions.shift()!;
+      this.frame = action.delay;
+
+      if ((error = action.execute(action.state, action.delay))) {
+        break;
+      }
+    }
+
+    this.flushing = false;
+
+    if (toFrame >= this.maxFrame) {
+      this.flushed = true;
+    }
+
+    if (error) {
+      while ((action = actions.shift())) {
+        action.unsubscribe();
+      }
+      throw error;
+    }
+  }
 }
 
 export { TestScheduler };