mod_pubsub: It's aliiiive!

2 files changed, 129 insertions, 0 deletions

diff --git a/plugins/mod_pubsub.lua b/plugins/mod_pubsub.lua
new file mode 100644
index 00000000..94e18f03
--- /dev/null
+++ b/plugins/mod_pubsub.lua
@@ -0,0 +1,94 @@
+local pubsub = require "util.pubsub";
+local st = require "util.stanza";
+local jid_bare = require "util.jid".bare;
+local uuid_generate = require "util.uuid".generate;
+
+require "core.modulemanager".load(module.host, "iq");
+
+local xmlns_pubsub = "http://jabber.org/protocol/pubsub";
+local xmlns_pubsub_errors = "http://jabber.org/protocol/pubsub#errors";
+local xmlns_pubsub_event = "http://jabber.org/protocol/pubsub#event";
+
+local service;
+
+local handlers = {};
+
+function handle_pubsub_iq(event)
+	local origin, stanza = event.origin, event.stanza;
+	local pubsub = stanza.tags[1];
+	local action = pubsub.tags[1];
+	local handler = handlers[stanza.attr.type.."_"..action.name];
+	if handler then
+		handler(origin, stanza, action);
+		return true;
+	end
+end
+
+local pubsub_errors = {
+	["invalid-jid"] = { "modify", "bad-request", nil, "invalid-jid" };
+	["item-not-found"] = { "cancel", "item-not-found" };
+};
+function pubsub_error_reply(stanza, error)
+	local e = pubsub_errors[error];
+	local reply = st.error_reply(stanza, unpack(e, 1, 3));
+	if e[4] then
+		reply:tag(e[4], { xmlns = xmlns_pubsub_errors }):up();
+	end
+	return reply;
+end
+
+function handlers.set_subscribe(origin, stanza, subscribe)
+	local node, jid = subscribe.attr.node, subscribe.attr.jid;
+	if jid_bare(jid) ~= jid_bare(stanza.attr.from) then
+		return origin.send(pubsub_error_reply(stanza, "invalid-jid"));
+	end
+	local ok, ret = service:add_subscription(node, stanza.attr.from, jid);
+	local reply;
+	if ok then
+		reply = st.reply(stanza)
+			:tag("pubsub", { xmlns = xmlns_pubsub })
+				:tag("subscription", {
+					node = node,
+					jid = jid,
+					subscription = "subscribed"
+				});
+	else
+		reply = pubsub_error_reply(stanza, ret);
+	end
+	return origin.send(reply);
+end
+
+function handlers.set_publish(origin, stanza, publish)
+	local node = publish.attr.node;
+	local item = publish:get_child("item");
+	local id = (item and item.attr.id) or uuid_generate();
+	local ok, ret = service:publish(node, stanza.attr.from, id, item);
+	local reply;
+	if ok then
+		reply = st.reply(stanza)
+			:tag("pubsub", { xmlns = xmlns_pubsub })
+				:tag("publish", { node = node })
+					:tag("item", { id = id });
+	else
+		reply = pubsub_error_reply(stanza, ret);
+	end
+	return origin.send(reply);
+end
+
+function simple_broadcast(node, jids, item)
+	local message = st.message({ from = module.host, type = "headline" })
+		:tag("event", { xmlns = xmlns_pubsub_event })
+			:tag("items", { node = node })
+				:add_child(item);
+	for jid in pairs(jids) do
+		module:log("debug", "Sending notification to %s", jid);
+		message.attr.to = jid;
+		core_post_stanza(hosts[module.host], message);
+	end
+end
+
+service = pubsub.new({
+	broadcaster = simple_broadcast
+});
+
+module:hook("iq/host/http://jabber.org/protocol/pubsub:pubsub", handle_pubsub_iq);
diff --git a/util/pubsub.lua b/util/pubsub.lua
new file mode 100644
index 00000000..34859fdb
--- /dev/null
+++ b/util/pubsub.lua
@@ -0,0 +1,35 @@
+module("pubsub", package.seeall);
+
+local service = {};
+local service_mt = { __index = service };
+
+function new(cb)
+	return setmetatable({ cb = cb or {}, nodes = {} }, service_mt);
+end
+
+function service:add_subscription(node, actor, jid)
+	local node_obj = self.nodes[node];
+	if not node_obj then
+		return false, "item-not-found";
+	end
+	node_obj.subscribers[jid] = true;
+	return true;
+end
+
+function service:remove_subscription(node, actor, jid)
+	self.nodes[node].subscribers[jid] = nil;
+	return true;
+end
+
+function service:publish(node, actor, id, item)
+	local node_obj = self.nodes[node];
+	if not node_obj then
+		node_obj = { name = node, subscribers = {}, config = {} };
+		self.nodes[node] = node_obj;
+	end
+	node_obj.data = item;
+	self.cb.broadcaster(node, node_obj.subscribers, item);
+	return true;
+end
+
+return _M;