bjornbytes/RxLua
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #33 from naxxster/fix-merge-source-completed

Browse Source 
Fix merge operator to handle onCompleted of input observables not a d…
This commit is contained in:
Bjorn Swenson 2018-12-02 09:52:47 -08:00 committed by GitHub
commit 687c7e53b4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 20 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
rx.lua
View File

@ -1049,6 +1049,7 @@ function Observable:merge(...)
table.insert(sources, 1, self) table.insert(sources, 1, self)
return Observable.create(function(observer) return Observable.create(function(observer)
local completed = {}
local subscriptions = {} local subscriptions = {}
local function onNext(...) local function onNext(...)
@ -1061,9 +1062,9 @@ function Observable:merge(...)
local function onCompleted(i) local function onCompleted(i)
return function() return function()
sources[i] = nil table.insert(completed, i)
if not next(sources) then if #completed == #sources then
observer:onCompleted() observer:onCompleted()
end end
end end

5
src/operators/merge.lua
View File

@ -9,6 +9,7 @@ function Observable:merge(...)
table.insert(sources, 1, self) table.insert(sources, 1, self)
return Observable.create(function(observer) return Observable.create(function(observer)
local completed = {}
local subscriptions = {} local subscriptions = {}
local function onNext(...) local function onNext(...)
@ -21,9 +22,9 @@ function Observable:merge(...)
local function onCompleted(i) local function onCompleted(i)
return function() return function()
sources[i] = nil table.insert(completed, i)
if not next(sources) then if #completed == #sources then
observer:onCompleted() observer:onCompleted()
end end
end end

14
tests/merge.lua
View File

@ -20,6 +20,20 @@ describe('merge', function()
expect(#unsubscribeB).to.equal(1) expect(#unsubscribeB).to.equal(1)
end) end)
it('unsubscribes from all input observables included completed', function()
local observableA = Rx.Observable.empty()
local unsubscribeB = spy()
local subscriptionB = Rx.Subscription.create(unsubscribeB)
local observableB = Rx.Observable.create(function(observer)
return subscriptionB
end)
local subscription = observableA:merge(Rx.Observable.empty(), observableB):subscribe()
subscription:unsubscribe()
expect(#unsubscribeB).to.equal(1)
end)
it('produces values from all input observables, in order', function() it('produces values from all input observables, in order', function()
local observableA = Rx.Subject.create() local observableA = Rx.Subject.create()
local observableB = Rx.Subject.create() local observableB = Rx.Subject.create()