RxLua/tests/switch.lua

describe('switch', function()
  it('errors when the source errors', function()
    expect(Rx.Observable.throw():switch()).to.produce.error()
  end)

  it('errors when an Observable produced by the source errors', function()
    local observable = Rx.Observable.create(function(observer)
      observer:onNext(Rx.Observable.throw())
      observer:onCompleted()
    end)

    expect(observable:switch()).to.produce.error()
  end)

  it('produces the values produced by the latest Observable produced by the source', function()
    local a = Rx.Subject.create()
    local b = Rx.Subject.create()
    local c = Rx.Subject.create()

    local onNext, onError, onCompleted = observableSpy(a:switch())

    b:onNext(1)
    a:onNext(b)
    b:onNext(2)
    b:onNext(3)
    c:onNext(7)
    a:onNext(c)
    b:onNext(4)
    c:onNext(8)
    b:onCompleted()
    c:onNext(9)
    c:onCompleted()
    a:onCompleted()

    expect(onNext).to.equal({{2}, {3}, {8}, {9}})
    expect(#onError).to.equal(0)
    expect(#onCompleted).to.equal(1)
  end)

  it('should unsubscribe from inner subscription too', function()
    local unsubscribeA = spy()
    local observableA = Rx.Observable.create(function(observer)
      return Rx.Subscription.create(unsubscribeA)
    end)

    local subject = Rx.Subject.create()
    local subscription = subject:switch():subscribe()
    subject:onNext(observableA)
    subscription:unsubscribe()
    expect(#unsubscribeA).to.equal(1)
  end)
end)