You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
104 lines
3.1 KiB
Lua
104 lines
3.1 KiB
Lua
local Observable = require "rx.observable"
|
|
local Observer = require "rx.observer"
|
|
local Subscription = require "rx.subscription"
|
|
local util = require "rx.util"
|
|
|
|
--- @class AsyncSubject
|
|
-- @description AsyncSubjects are subjects that produce either no values or a single value. If
|
|
-- multiple values are produced via onNext, only the last one is used. If onError is called, then
|
|
-- no value is produced and onError is called on any subscribed Observers. If an Observer
|
|
-- subscribes and the AsyncSubject has already terminated, the Observer will immediately receive the
|
|
-- value or the error.
|
|
local AsyncSubject = setmetatable({}, Observable)
|
|
AsyncSubject.__index = AsyncSubject
|
|
AsyncSubject.__tostring = util.constant("AsyncSubject")
|
|
|
|
--- Creates a new AsyncSubject.
|
|
-- @returns {AsyncSubject}
|
|
function AsyncSubject.create()
|
|
local self = {
|
|
observers = {},
|
|
stopped = false,
|
|
value = nil,
|
|
errorMessage = nil,
|
|
}
|
|
|
|
return setmetatable(self, AsyncSubject)
|
|
end
|
|
|
|
--- Creates a new Observer and attaches it to the AsyncSubject.
|
|
-- @arg {function|table} onNext|observer - A function called when the AsyncSubject produces a value
|
|
-- or an existing Observer to attach to the AsyncSubject.
|
|
-- @arg {function} onError - Called when the AsyncSubject terminates due to an error.
|
|
-- @arg {function} onCompleted - Called when the AsyncSubject completes normally.
|
|
function AsyncSubject:subscribe(onNext, onError, onCompleted)
|
|
local observer
|
|
|
|
if util.isa(onNext, Observer) then
|
|
observer = onNext
|
|
else
|
|
observer = Observer.create(onNext, onError, onCompleted)
|
|
end
|
|
|
|
if self.value then
|
|
observer:onNext(util.unpack(self.value))
|
|
observer:onCompleted()
|
|
return
|
|
elseif self.errorMessage then
|
|
observer:onError(self.errorMessage)
|
|
return
|
|
end
|
|
|
|
table.insert(self.observers, observer)
|
|
|
|
return Subscription.create(function()
|
|
for i = 1, #self.observers do
|
|
if self.observers[i] == observer then
|
|
table.remove(self.observers, i)
|
|
return
|
|
end
|
|
end
|
|
end)
|
|
end
|
|
|
|
--- Pushes zero or more values to the AsyncSubject.
|
|
-- @arg {*...} values
|
|
function AsyncSubject:onNext(...)
|
|
if not self.stopped then
|
|
self.value = util.pack(...)
|
|
end
|
|
end
|
|
|
|
--- Signal to all Observers that an error has occurred.
|
|
-- @arg {string=} message - A string describing what went wrong.
|
|
function AsyncSubject:onError(message)
|
|
if not self.stopped then
|
|
self.errorMessage = message
|
|
|
|
for i = 1, #self.observers do
|
|
self.observers[i]:onError(self.errorMessage)
|
|
end
|
|
|
|
self.stopped = true
|
|
end
|
|
end
|
|
|
|
--- Signal to all Observers that the AsyncSubject will not produce any more values.
|
|
function AsyncSubject:onCompleted()
|
|
if not self.stopped then
|
|
for i = 1, #self.observers do
|
|
if self.value then
|
|
self.observers[i]:onNext(util.unpack(self.value))
|
|
end
|
|
|
|
self.observers[i]:onCompleted()
|
|
end
|
|
|
|
self.stopped = true
|
|
end
|
|
end
|
|
|
|
AsyncSubject.__call = AsyncSubject.onNext
|
|
|
|
return AsyncSubject
|