Merge pull request #27 from naxxster/fix-merge-subscription
Fix merge operator to produce a subscription.master
commit
c89f0de521
10
rx.lua
10
rx.lua
|
@ -1049,6 +1049,8 @@ function Observable:merge(...)
|
|||
table.insert(sources, 1, self)
|
||||
|
||||
return Observable.create(function(observer)
|
||||
local subscriptions = {}
|
||||
|
||||
local function onNext(...)
|
||||
return observer:onNext(...)
|
||||
end
|
||||
|
@ -1068,8 +1070,14 @@ function Observable:merge(...)
|
|||
end
|
||||
|
||||
for i = 1, #sources do
|
||||
sources[i]:subscribe(onNext, onError, onCompleted(i))
|
||||
subscriptions[i] = sources[i]:subscribe(onNext, onError, onCompleted(i))
|
||||
end
|
||||
|
||||
return Subscription.create(function ()
|
||||
for i = 1, #sources do
|
||||
if subscriptions[i] then subscriptions[i]:unsubscribe() end
|
||||
end
|
||||
end)
|
||||
end)
|
||||
end
|
||||
|
||||
|
|
|
@ -9,6 +9,8 @@ function Observable:merge(...)
|
|||
table.insert(sources, 1, self)
|
||||
|
||||
return Observable.create(function(observer)
|
||||
local subscriptions = {}
|
||||
|
||||
local function onNext(...)
|
||||
return observer:onNext(...)
|
||||
end
|
||||
|
@ -28,7 +30,13 @@ function Observable:merge(...)
|
|||
end
|
||||
|
||||
for i = 1, #sources do
|
||||
sources[i]:subscribe(onNext, onError, onCompleted(i))
|
||||
subscriptions[i] = sources[i]:subscribe(onNext, onError, onCompleted(i))
|
||||
end
|
||||
|
||||
return Subscription.create(function ()
|
||||
for i = 1, #sources do
|
||||
if subscriptions[i] then subscriptions[i]:unsubscribe() end
|
||||
end
|
||||
end)
|
||||
end)
|
||||
end
|
||||
|
|
|
@ -4,6 +4,22 @@ describe('merge', function()
|
|||
expect(observable).to.produce(1, 2, 3, 4, 5)
|
||||
end)
|
||||
|
||||
it('unsubscribes from all input observables', function()
|
||||
local observableA = Rx.Observable.create(function(observer)
|
||||
return
|
||||
end)
|
||||
|
||||
local unsubscribeB = spy()
|
||||
local subscriptionB = Rx.Subscription.create(unsubscribeB)
|
||||
local observableB = Rx.Observable.create(function(observer)
|
||||
return subscriptionB
|
||||
end)
|
||||
|
||||
local subscription = observableA:merge(observableB):subscribe()
|
||||
subscription:unsubscribe()
|
||||
expect(#unsubscribeB).to.equal(1)
|
||||
end)
|
||||
|
||||
it('produces values from all input observables, in order', function()
|
||||
local observableA = Rx.Subject.create()
|
||||
local observableB = Rx.Subject.create()
|
||||
|
|
Loading…
Reference in New Issue