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.
87 lines
2.5 KiB
Lua
87 lines
2.5 KiB
Lua
local Observable = require "rx.observable"
|
|
local Observer = require "rx.observer"
|
|
local Subscription = require "rx.subscription"
|
|
local util = require "rx.util"
|
|
|
|
--- @class Subject
|
|
-- @description Subjects function both as an Observer and as an Observable. Subjects inherit all
|
|
-- Observable functions, including subscribe. Values can also be pushed to the Subject, which will
|
|
-- be broadcasted to any subscribed Observers.
|
|
local Subject = setmetatable({}, Observable)
|
|
Subject.__index = Subject
|
|
Subject.__tostring = util.constant("Subject")
|
|
|
|
--- Creates a new Subject.
|
|
-- @returns {Subject}
|
|
function Subject.create()
|
|
local self = {
|
|
observers = {},
|
|
stopped = false,
|
|
}
|
|
|
|
return setmetatable(self, Subject)
|
|
end
|
|
|
|
--- Creates a new Observer and attaches it to the Subject.
|
|
-- @arg {function|table} onNext|observer - A function called when the Subject produces a value or
|
|
-- an existing Observer to attach to the Subject.
|
|
-- @arg {function} onError - Called when the Subject terminates due to an error.
|
|
-- @arg {function} onCompleted - Called when the Subject completes normally.
|
|
function Subject:subscribe(onNext, onError, onCompleted)
|
|
local observer
|
|
|
|
if util.isa(onNext, Observer) then
|
|
observer = onNext
|
|
else
|
|
observer = Observer.create(onNext, onError, onCompleted)
|
|
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 Subject. They will be broadcasted to all Observers.
|
|
-- @arg {*...} values
|
|
function Subject:onNext(...)
|
|
if not self.stopped then
|
|
for i = #self.observers, 1, -1 do
|
|
self.observers[i]:onNext(...)
|
|
end
|
|
end
|
|
end
|
|
|
|
--- Signal to all Observers that an error has occurred.
|
|
-- @arg {string=} message - A string describing what went wrong.
|
|
function Subject:onError(message)
|
|
if not self.stopped then
|
|
for i = #self.observers, 1, -1 do
|
|
self.observers[i]:onError(message)
|
|
end
|
|
|
|
self.stopped = true
|
|
end
|
|
end
|
|
|
|
--- Signal to all Observers that the Subject will not produce any more values.
|
|
function Subject:onCompleted()
|
|
if not self.stopped then
|
|
for i = #self.observers, 1, -1 do
|
|
self.observers[i]:onCompleted()
|
|
end
|
|
|
|
self.stopped = true
|
|
end
|
|
end
|
|
|
|
Subject.__call = Subject.onNext
|
|
|
|
return Subject
|