RxJSでユニットテスト

gistfile1.coffee

#! /usr/bin/env coffee

assert = require('assert')
Rx = require('rx')

collectionAssert = (expected, actual) ->
  if expected.length != actual.length
    assert(false, 'Not eual length.')

  isOk = true
  for i in [0 .. expected.length]
    isOk = Rx.internals.isEqual(expected[i], actual[i])
    break unless isOk

  assert(isOk)

nClick = (stream, scheduler) ->
  stream
    .buffer stream.throttle(250, scheduler)
    .map (x) -> x.length
    .filter (n) -> n >= 2

onNext = Rx.ReactiveTest.onNext
onCompleted = Rx.ReactiveTest.onCompleted

scheduler = new Rx.TestScheduler()
xs = scheduler.createHotObservable(
  onNext(250, true)
  onNext(300, true)
  onNext(1000, true)
)

res = scheduler.startWithTiming(() ->
  nClick(xs, scheduler)
, 50, 150, 1600)

collectionAssert(res.messages, [
  onNext(550, 2)
])