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