diff options
Diffstat (limited to 'util/pubsub.lua')
| -rw-r--r-- | util/pubsub.lua | 200 |
1 files changed, 160 insertions, 40 deletions
diff --git a/util/pubsub.lua b/util/pubsub.lua index 1db917d8..74fa2be4 100644 --- a/util/pubsub.lua +++ b/util/pubsub.lua @@ -1,32 +1,75 @@ local events = require "util.events"; local cache = require "util.cache"; -local service = {}; -local service_mt = { __index = service }; +local service_mt = {}; -local default_config = { __index = { - itemstore = function (config) return cache.new(tonumber(config["pubsub#max_items"])) end; +local default_config = { + itemstore = function (config, _) return cache.new(config["max_items"]) end; broadcaster = function () end; + itemcheck = function () return true; end; get_affiliation = function () end; + normalize_jid = function (jid) return jid; end; capabilities = {}; -} }; -local default_node_config = { __index = { - ["pubsub#max_items"] = "20"; -} }; +}; +local default_config_mt = { __index = default_config }; + +local default_node_config = { + ["persist_items"] = false; + ["max_items"] = 20; +}; +local default_node_config_mt = { __index = default_node_config }; + +-- Storage helper functions + +local function load_node_from_store(service, node_name) + local node = service.config.nodestore:get(node_name); + node.config = setmetatable(node.config or {}, {__index=service.node_defaults}); + return node; +end + +local function save_node_to_store(service, node) + return service.config.nodestore:set(node.name, { + name = node.name; + config = node.config; + subscribers = node.subscribers; + affiliations = node.affiliations; + }); +end +local function delete_node_in_store(service, node_name) + return service.config.nodestore:set(node_name, nil); +end + +-- Create and return a new service object local function new(config) config = config or {}; - return setmetatable({ - config = setmetatable(config, default_config); - node_defaults = setmetatable(config.node_defaults or {}, default_node_config); + + local service = setmetatable({ + config = setmetatable(config, default_config_mt); + node_defaults = setmetatable(config.node_defaults or {}, default_node_config_mt); affiliations = {}; subscriptions = {}; nodes = {}; data = {}; events = events.new(); }, service_mt); + + -- Load nodes from storage, if we have a store and it supports iterating over stored items + if config.nodestore and config.nodestore.users then + for node_name in config.nodestore:users() do + service.nodes[node_name] = load_node_from_store(service, node_name); + service.data[node_name] = config.itemstore(service.nodes[node_name].config, node_name); + end + end + + return service; end +--- Service methods + +local service = {}; +service_mt.__index = service; + function service:jids_equal(jid1, jid2) local normalize = self.config.normalize_jid; return normalize(jid1) == normalize(jid2); @@ -36,7 +79,8 @@ function service:may(node, actor, action) if actor == true then return true; end local node_obj = self.nodes[node]; - local node_aff = node_obj and node_obj.affiliations[actor]; + local node_aff = node_obj and (node_obj.affiliations[actor] + or node_obj.affiliations[self.config.normalize_jid(actor)]); local service_aff = self.affiliations[actor] or self.config.get_affiliation(actor, node, action) or "none"; @@ -76,7 +120,18 @@ function service:set_affiliation(node, actor, jid, affiliation) if not node_obj then return false, "item-not-found"; end + jid = self.config.normalize_jid(jid); + local old_affiliation = node_obj.affiliations[jid]; node_obj.affiliations[jid] = affiliation; + + if self.config.nodestore then + local ok, err = save_node_to_store(self, node_obj); + if not ok then + node_obj.affiliations[jid] = old_affiliation; + return ok, "internal-server-error"; + end + end + local _, jid_sub = self:get_subscription(node, true, jid); if not jid_sub and not self:may(node, jid, "be_unsubscribed") then local ok, err = self:add_subscription(node, true, jid); @@ -119,6 +174,7 @@ function service:add_subscription(node, actor, jid, options) node_obj = self.nodes[node]; end end + local old_subscription = node_obj.subscribers[jid]; node_obj.subscribers[jid] = options or true; local normal_jid = self.config.normalize_jid(jid); local subs = self.subscriptions[normal_jid]; @@ -131,6 +187,16 @@ function service:add_subscription(node, actor, jid, options) else self.subscriptions[normal_jid] = { [jid] = { [node] = true } }; end + + if self.config.nodestore then + local ok, err = save_node_to_store(self, node_obj); + if not ok then + node_obj.subscribers[jid] = old_subscription; + self.subscriptions[normal_jid][jid][node] = old_subscription and true or nil; + return ok, "internal-server-error"; + end + end + self.events.fire_event("subscription-added", { node = node, jid = jid, normalized_jid = normal_jid, options = options }); return true; end @@ -157,6 +223,7 @@ function service:remove_subscription(node, actor, jid) if not node_obj.subscribers[jid] then return false, "not-subscribed"; end + local old_subscription = node_obj.subscribers[jid]; node_obj.subscribers[jid] = nil; local normal_jid = self.config.normalize_jid(jid); local subs = self.subscriptions[normal_jid]; @@ -172,19 +239,17 @@ function service:remove_subscription(node, actor, jid) self.subscriptions[normal_jid] = nil; end end - self.events.fire_event("subscription-removed", { node = node, jid = jid, normalized_jid = normal_jid }); - return true; -end -function service:remove_all_subscriptions(actor, jid) - local normal_jid = self.config.normalize_jid(jid); - local subs = self.subscriptions[normal_jid] - subs = subs and subs[jid]; - if subs then - for node in pairs(subs) do - self:remove_subscription(node, true, jid); + if self.config.nodestore then + local ok, err = save_node_to_store(self, node_obj); + if not ok then + node_obj.subscribers[jid] = old_subscription; + self.subscriptions[normal_jid][jid][node] = old_subscription and true or nil; + return ok, "internal-server-error"; end end + + self.events.fire_event("subscription-removed", { node = node, jid = jid, normalized_jid = normal_jid }); return true; end @@ -223,14 +288,27 @@ function service:create(node, actor, options) config = setmetatable(options or {}, {__index=self.node_defaults}); affiliations = {}; }; - self.data[node] = self.config.itemstore(self.nodes[node].config); + + if self.config.nodestore then + local ok, err = save_node_to_store(self, self.nodes[node]); + if not ok then + self.nodes[node] = nil; + return ok, "internal-server-error"; + end + end + + self.data[node] = self.config.itemstore(self.nodes[node].config, node); self.events.fire_event("node-created", { node = node, actor = actor }); - local ok, err = self:set_affiliation(node, true, actor, "owner"); - if not ok then - self.nodes[node] = nil; - self.data[node] = nil; + if actor ~= true then + local ok, err = self:set_affiliation(node, true, actor, "owner"); + if not ok then + self.nodes[node] = nil; + self.data[node] = nil; + return ok, err; + end end - return ok, err; + + return true; end function service:delete(node, actor) @@ -244,9 +322,21 @@ function service:delete(node, actor) return false, "item-not-found"; end self.nodes[node] = nil; + if self.data[node] and self.data[node].clear then + self.data[node]:clear(); + end self.data[node] = nil; + + if self.config.nodestore then + local ok, err = delete_node_in_store(self, node); + if not ok then + self.nodes[node] = nil; + return ok, err; + end + end + self.events.fire_event("node-deleted", { node = node, actor = actor }); - self.config.broadcaster("delete", node, node_obj.subscribers); + self.config.broadcaster("delete", node, node_obj.subscribers, nil, actor, node_obj, self); return true; end @@ -267,13 +357,17 @@ function service:publish(node, actor, id, item) end node_obj = self.nodes[node]; end + if not self.config.itemcheck(item) then + return nil, "internal-server-error"; + end local node_data = self.data[node]; local ok = node_data:set(id, item); if not ok then return nil, "internal-server-error"; end + if type(ok) == "string" then id = ok; end self.events.fire_event("item-published", { node = node, actor = actor, id = id, item = item }); - self.config.broadcaster("items", node, node_obj.subscribers, item, actor); + self.config.broadcaster("items", node, node_obj.subscribers, item, actor, node_obj, self); return true; end @@ -293,7 +387,7 @@ function service:retract(node, actor, id, retract) end self.events.fire_event("item-retracted", { node = node, actor = actor, id = id }); if retract then - self.config.broadcaster("items", node, node_obj.subscribers, retract); + self.config.broadcaster("items", node, node_obj.subscribers, retract, actor, node_obj, self); end return true end @@ -308,10 +402,14 @@ function service:purge(node, actor, notify) if not node_obj then return false, "item-not-found"; end - self.data[node] = self.config.itemstore(self.nodes[node].config); + if self.data[node] and self.data[node].clear then + self.data[node]:clear() + else + self.data[node] = self.config.itemstore(self.nodes[node].config, node); + end self.events.fire_event("node-purged", { node = node, actor = actor }); if notify then - self.config.broadcaster("purge", node, node_obj.subscribers); + self.config.broadcaster("purge", node, node_obj.subscribers, nil, actor, node_obj, self); end return true end @@ -327,7 +425,11 @@ function service:get_items(node, actor, id) return false, "item-not-found"; end if id then -- Restrict results to a single specific item - return true, { id, [id] = self.data[node]:get(id) }; + local with_id = self.data[node]:get(id); + if not with_id then + return true, { }; + end + return true, { id, [id] = with_id }; else local data = {} for key, value in self.data[node]:items() do @@ -338,6 +440,15 @@ function service:get_items(node, actor, id) end end +function service:get_last_item(node, actor) + -- Access checking + if not self:may(node, actor, "get_items") then + return false, "forbidden"; + end + -- + return true, self.data[node]:tail(); +end + function service:get_nodes(actor) -- Access checking if not self:may(nil, actor, "get_nodes") then @@ -421,14 +532,23 @@ function service:set_node_config(node, actor, new_config) return false, "item-not-found"; end - for k,v in pairs(new_config) do - node_obj.config[k] = v; + local old_config = node_obj.config; + node_obj.config = setmetatable(new_config, {__index=self.node_defaults}); + + if self.config.nodestore then + local ok, err = save_node_to_store(self, node_obj); + if not ok then + node_obj.config = old_config; + return ok, "internal-server-error"; + end end - local new_data = self.config.itemstore(self.nodes[node].config); - for key, value in self.data[node]:items() do - new_data:set(key, value); + + if old_config["persist_items"] ~= node_obj.config["persist_items"] then + self.data[node] = self.config.itemstore(self.nodes[node].config, node); + elseif old_config["max_items"] ~= node_obj.config["max_items"] then + self.data[node]:resize(self.nodes[node].config["max_items"]); end - self.data[node] = new_data; + return true; end |