create-hot-cold-observable.js

/*eslint-env jasmine*/

import Rx from 'rxjs';

fdescribe('TestScheduler', () => {
  
  describe('createHotObservable()', () => {
    it('should create a hot observable emmiting the values given as marble strings', () => {
      //given
      const scheduler = new Rx.TestScheduler(null);
      const source = scheduler.createHotObservable('--a---b-cd-|');
      const expected = ['ab', 'cd'];
  
      //when
      source
        .bufferCount(2)
        .map((array) => array.join(''))
        .subscribe((val) => {
          expect(val).toEqual(expected.shift());
        });
        
      scheduler.flush();
      expect(expected.length).toBe(0);
    });
  });
  
  describe('createColdObservable()', () => {
    it('should create a cold observable emmiting the values given as marble strings', () => {
      //given
      const scheduler = new Rx.TestScheduler(null);
      const source = scheduler.createColdObservable('--a---b|');
      const expected = ['a', 'b'];
  
      //when
      source
        .subscribe((val) => {
          expect(val).toEqual(expected.shift());
        });
        
      scheduler.flush();
      expect(expected.length).toBe(0);
    });
  });
});

expectObservable.js

/*eslint-env jasmine*/

import Rx from 'rxjs';

function assertEquals (actual, expected) {
    //we will use jasmine's api for the assertion:
    expect(actual).toEqual(expected);
}

describe('TestScheduler', () => {
  describe('expectObservable()', () => {
    it('should simplify the result check of given observables', () => {
      //given
      const scheduler = new Rx.TestScheduler(assertEquals);
      const source = scheduler.createHotObservable('--a-^--b-c');

      //then
      scheduler.expectObservable(source).toBe('---bc');
      
      scheduler.flush();
    });
  });
});

schedule.js

/*eslint-env jasmine*/

import Rx from 'rxjs';

describe('TestScheduler', () => {
  describe('schedule()', () => {
    it('should schedule the subscription on the right time', () => {
      //given
      const scheduler = new Rx.TestScheduler(null);
      const source = scheduler.createHotObservable('--a-^--b-c|');
      const results = [];

      //when
      scheduler.schedule(() => {
        source.subscribe((val) => {
          results.push(val);
        }, null, () => {
          results.push('done');  
        });
      });
              
      scheduler.flush();
      expect(results).toEqual(['b', 'c', 'done']);
    });
    
    it('should allow scheduling of non ending streams', () => {
      //given
      const scheduler = new Rx.TestScheduler(null);
      const source = scheduler.createHotObservable('--a-^--b-c');
      const results = [];

      //when
      scheduler.schedule(() => {
        source.subscribe((val) => {
          results.push(val);
        }, null, () => {
          results.push('done');  
        });
      });
              
      scheduler.flush();
      expect(results).toEqual(['b', 'c']);
    });
  });
});