util/pubsub.lua

   1 local events = require "util.events";
   2 local t_remove = table.remove;
   3
   4 module("pubsub", package.seeall);
   5
   6 local service = {};
   7 local service_mt = { __index = service };
   8
   9 local default_config = { __index = {
  10         broadcaster = function () end;
  11         get_affiliation = function () end;
  12         capabilities = {};
  13 } };
  14 local default_node_config = { __index = {
  15         ["pubsub#max_items"] = "20";
  16 } };
  17
  18 function new(config)
  19         config = config or {};
  20         return setmetatable({
  21                 config = setmetatable(config, default_config);
  22                 node_defaults = setmetatable(config.node_defaults or {}, default_node_config);
  23                 affiliations = {};
  24                 subscriptions = {};
  25                 nodes = {};
  26                 data = {};
  27                 events = events.new();
  28         }, service_mt);
  29 end
  30
  31 function service:jids_equal(jid1, jid2)
  32         local normalize = self.config.normalize_jid;
  33         return normalize(jid1) == normalize(jid2);
  34 end
  35
  36 function service:may(node, actor, action)
  37         if actor == true then return true; end
  38
  39         local node_obj = self.nodes[node];
  40         local node_aff = node_obj and node_obj.affiliations[actor];
  41         local service_aff = self.affiliations[actor]
  42                          or self.config.get_affiliation(actor, node, action)
  43                          or "none";
  44
  45         -- Check if node allows/forbids it
  46         local node_capabilities = node_obj and node_obj.capabilities;
  47         if node_capabilities then
  48                 local caps = node_capabilities[node_aff or service_aff];
  49                 if caps then
  50                         local can = caps[action];
  51                         if can ~= nil then
  52                                 return can;
  53                         end
  54                 end
  55         end
  56
  57         -- Check service-wide capabilities instead
  58         local service_capabilities = self.config.capabilities;
  59         local caps = service_capabilities[node_aff or service_aff];
  60         if caps then
  61                 local can = caps[action];
  62                 if can ~= nil then
  63                         return can;
  64                 end
  65         end
  66
  67         return false;
  68 end
  69
  70 function service:set_affiliation(node, actor, jid, affiliation)
  71         -- Access checking
  72         if not self:may(node, actor, "set_affiliation") then
  73                 return false, "forbidden";
  74         end
  75         --
  76         local node_obj = self.nodes[node];
  77         if not node_obj then
  78                 return false, "item-not-found";
  79         end
  80         node_obj.affiliations[jid] = affiliation;
  81         local _, jid_sub = self:get_subscription(node, true, jid);
  82         if not jid_sub and not self:may(node, jid, "be_unsubscribed") then
  83                 local ok, err = self:add_subscription(node, true, jid);
  84                 if not ok then
  85                         return ok, err;
  86                 end
  87         elseif jid_sub and not self:may(node, jid, "be_subscribed") then
  88                 local ok, err = self:add_subscription(node, true, jid);
  89                 if not ok then
  90                         return ok, err;
  91                 end
  92         end
  93         return true;
  94 end
  95
  96 function service:add_subscription(node, actor, jid, options)
  97         -- Access checking
  98         local cap;
  99         if actor == true or jid == actor or self:jids_equal(actor, jid) then
 100                 cap = "subscribe";
 101         else
 102                 cap = "subscribe_other";
 103         end
 104         if not self:may(node, actor, cap) then
 105                 return false, "forbidden";
 106         end
 107         if not self:may(node, jid, "be_subscribed") then
 108                 return false, "forbidden";
 109         end
 110         --
 111         local node_obj = self.nodes[node];
 112         if not node_obj then
 113                 if not self.config.autocreate_on_subscribe then
 114                         return false, "item-not-found";
 115                 else
 116                         local ok, err = self:create(node, true);
 117                         if not ok then
 118                                 return ok, err;
 119                         end
 120                         node_obj = self.nodes[node];
 121                 end
 122         end
 123         node_obj.subscribers[jid] = options or true;
 124         local normal_jid = self.config.normalize_jid(jid);
 125         local subs = self.subscriptions[normal_jid];
 126         if subs then
 127                 if not subs[jid] then
 128                         subs[jid] = { [node] = true };
 129                 else
 130                         subs[jid][node] = true;
 131                 end
 132         else
 133                 self.subscriptions[normal_jid] = { [jid] = { [node] = true } };
 134         end
 135         self.events.fire_event("subscription-added", { node = node, jid = jid, normalized_jid = normal_jid, options = options });
 136         return true;
 137 end
 138
 139 function service:remove_subscription(node, actor, jid)
 140         -- Access checking
 141         local cap;
 142         if actor == true or jid == actor or self:jids_equal(actor, jid) then
 143                 cap = "unsubscribe";
 144         else
 145                 cap = "unsubscribe_other";
 146         end
 147         if not self:may(node, actor, cap) then
 148                 return false, "forbidden";
 149         end
 150         if not self:may(node, jid, "be_unsubscribed") then
 151                 return false, "forbidden";
 152         end
 153         --
 154         local node_obj = self.nodes[node];
 155         if not node_obj then
 156                 return false, "item-not-found";
 157         end
 158         if not node_obj.subscribers[jid] then
 159                 return false, "not-subscribed";
 160         end
 161         node_obj.subscribers[jid] = nil;
 162         local normal_jid = self.config.normalize_jid(jid);
 163         local subs = self.subscriptions[normal_jid];
 164         if subs then
 165                 local jid_subs = subs[jid];
 166                 if jid_subs then
 167                         jid_subs[node] = nil;
 168                         if next(jid_subs) == nil then
 169                                 subs[jid] = nil;
 170                         end
 171                 end
 172                 if next(subs) == nil then
 173                         self.subscriptions[normal_jid] = nil;
 174                 end
 175         end
 176         self.events.fire_event("subscription-removed", { node = node, jid = jid, normalized_jid = normal_jid });
 177         return true;
 178 end
 179
 180 function service:remove_all_subscriptions(actor, jid)
 181         local normal_jid = self.config.normalize_jid(jid);
 182         local subs = self.subscriptions[normal_jid]
 183         subs = subs and subs[jid];
 184         if subs then
 185                 for node in pairs(subs) do
 186                         self:remove_subscription(node, true, jid);
 187                 end
 188         end
 189         return true;
 190 end
 191
 192 function service:get_subscription(node, actor, jid)
 193         -- Access checking
 194         local cap;
 195         if actor == true or jid == actor or self:jids_equal(actor, jid) then
 196                 cap = "get_subscription";
 197         else
 198                 cap = "get_subscription_other";
 199         end
 200         if not self:may(node, actor, cap) then
 201                 return false, "forbidden";
 202         end
 203         --
 204         local node_obj = self.nodes[node];
 205         if not node_obj then
 206                 return false, "item-not-found";
 207         end
 208         return true, node_obj.subscribers[jid];
 209 end
 210
 211 function service:create(node, actor, options)
 212         -- Access checking
 213         if not self:may(node, actor, "create") then
 214                 return false, "forbidden";
 215         end
 216         --
 217         if self.nodes[node] then
 218                 return false, "conflict";
 219         end
 220
 221         self.data[node] = {};
 222         self.nodes[node] = {
 223                 name = node;
 224                 subscribers = {};
 225                 config = setmetatable(options or {}, {__index=self.node_defaults});
 226                 affiliations = {};
 227         };
 228         setmetatable(self.nodes[node], { __index = { data = self.data[node] } }); -- COMPAT
 229         self.events.fire_event("node-created", { node = node, actor = actor });
 230         local ok, err = self:set_affiliation(node, true, actor, "owner");
 231         if not ok then
 232                 self.nodes[node] = nil;
 233                 self.data[node] = nil;
 234         end
 235         return ok, err;
 236 end
 237
 238 function service:delete(node, actor)
 239         -- Access checking
 240         if not self:may(node, actor, "delete") then
 241                 return false, "forbidden";
 242         end
 243         --
 244         local node_obj = self.nodes[node];
 245         if not node_obj then
 246                 return false, "item-not-found";
 247         end
 248         self.nodes[node] = nil;
 249         self.data[node] = nil;
 250         self.events.fire_event("node-deleted", { node = node, actor = actor });
 251         self.config.broadcaster("delete", node, node_obj.subscribers);
 252         return true;
 253 end
 254
 255 local function remove_item_by_id(data, id)
 256         if not data[id] then return end
 257         data[id] = nil;
 258         for i, _id in ipairs(data) do
 259                 if id == _id then
 260                         t_remove(data, i);
 261                         return i;
 262                 end
 263         end
 264 end
 265
 266 local function trim_items(data, max)
 267         max = tonumber(max);
 268         if not max or #data <= max then return end
 269         repeat
 270                 data[t_remove(data, 1)] = nil;
 271         until #data <= max
 272 end
 273
 274 function service:publish(node, actor, id, item)
 275         -- Access checking
 276         if not self:may(node, actor, "publish") then
 277                 return false, "forbidden";
 278         end
 279         --
 280         local node_obj = self.nodes[node];
 281         if not node_obj then
 282                 if not self.config.autocreate_on_publish then
 283                         return false, "item-not-found";
 284                 end
 285                 local ok, err = self:create(node, true);
 286                 if not ok then
 287                         return ok, err;
 288                 end
 289                 node_obj = self.nodes[node];
 290         end
 291         local node_data = self.data[node];
 292         remove_item_by_id(node_data, id);
 293         node_data[#node_data + 1] = id;
 294         node_data[id] = item;
 295         trim_items(node_data, node_obj.config["pubsub#max_items"]);
 296         self.events.fire_event("item-published", { node = node, actor = actor, id = id, item = item });
 297         self.config.broadcaster("items", node, node_obj.subscribers, item, actor);
 298         return true;
 299 end
 300
 301 function service:retract(node, actor, id, retract)
 302         -- Access checking
 303         if not self:may(node, actor, "retract") then
 304                 return false, "forbidden";
 305         end
 306         --
 307         local node_obj = self.nodes[node];
 308         if (not node_obj) or (not self.data[node][id]) then
 309                 return false, "item-not-found";
 310         end
 311         self.events.fire_event("item-retracted", { node = node, actor = actor, id = id });
 312         remove_item_by_id(self.data[node], id);
 313         if retract then
 314                 self.config.broadcaster("items", node, node_obj.subscribers, retract);
 315         end
 316         return true
 317 end
 318
 319 function service:purge(node, actor, notify)
 320         -- Access checking
 321         if not self:may(node, actor, "retract") then
 322                 return false, "forbidden";
 323         end
 324         --
 325         local node_obj = self.nodes[node];
 326         if not node_obj then
 327                 return false, "item-not-found";
 328         end
 329         self.data[node] = {}; -- Purge
 330         self.events.fire_event("node-purged", { node = node, actor = actor });
 331         if notify then
 332                 self.config.broadcaster("purge", node, node_obj.subscribers);
 333         end
 334         return true
 335 end
 336
 337 function service:get_items(node, actor, id)
 338         -- Access checking
 339         if not self:may(node, actor, "get_items") then
 340                 return false, "forbidden";
 341         end
 342         --
 343         local node_obj = self.nodes[node];
 344         if not node_obj then
 345                 return false, "item-not-found";
 346         end
 347         if id then -- Restrict results to a single specific item
 348                 return true, { id, [id] = self.data[node][id] };
 349         else
 350                 return true, self.data[node];
 351         end
 352 end
 353
 354 function service:get_nodes(actor)
 355         -- Access checking
 356         if not self:may(nil, actor, "get_nodes") then
 357                 return false, "forbidden";
 358         end
 359         --
 360         return true, self.nodes;
 361 end
 362
 363 function service:get_subscriptions(node, actor, jid)
 364         -- Access checking
 365         local cap;
 366         if actor == true or jid == actor or self:jids_equal(actor, jid) then
 367                 cap = "get_subscriptions";
 368         else
 369                 cap = "get_subscriptions_other";
 370         end
 371         if not self:may(node, actor, cap) then
 372                 return false, "forbidden";
 373         end
 374         --
 375         local node_obj;
 376         if node then
 377                 node_obj = self.nodes[node];
 378                 if not node_obj then
 379                         return false, "item-not-found";
 380                 end
 381         end
 382         local normal_jid = self.config.normalize_jid(jid);
 383         local subs = self.subscriptions[normal_jid];
 384         -- We return the subscription object from the node to save
 385         -- a get_subscription() call for each node.
 386         local ret = {};
 387         if subs then
 388                 for jid, subscribed_nodes in pairs(subs) do
 389                         if node then -- Return only subscriptions to this node
 390                                 if subscribed_nodes[node] then
 391                                         ret[#ret+1] = {
 392                                                 node = node;
 393                                                 jid = jid;
 394                                                 subscription = node_obj.subscribers[jid];
 395                                         };
 396                                 end
 397                         else -- Return subscriptions to all nodes
 398                                 local nodes = self.nodes;
 399                                 for subscribed_node in pairs(subscribed_nodes) do
 400                                         ret[#ret+1] = {
 401                                                 node = subscribed_node;
 402                                                 jid = jid;
 403                                                 subscription = nodes[subscribed_node].subscribers[jid];
 404                                         };
 405                                 end
 406                         end
 407                 end
 408         end
 409         return true, ret;
 410 end
 411
 412 -- Access models only affect 'none' affiliation caps, service/default access level...
 413 function service:set_node_capabilities(node, actor, capabilities)
 414         -- Access checking
 415         if not self:may(node, actor, "configure") then
 416                 return false, "forbidden";
 417         end
 418         --
 419         local node_obj = self.nodes[node];
 420         if not node_obj then
 421                 return false, "item-not-found";
 422         end
 423         node_obj.capabilities = capabilities;
 424         return true;
 425 end
 426
 427 function service:set_node_config(node, actor, new_config)
 428         if not self:may(node, actor, "configure") then
 429                 return false, "forbidden";
 430         end
 431
 432         local node_obj = self.nodes[node];
 433         if not node_obj then
 434                 return false, "item-not-found";
 435         end
 436
 437         for k,v in pairs(new_config) do
 438                 node_obj.config[k] = v;
 439         end
 440         trim_items(self.data[node], node_obj.config["pubsub#max_items"]);
 441
 442         return true;
 443 end
 444
 445 return _M;