[prosody-dev] [PATCH v2] mod_muc_eventsource: history push support

[senya]

Here is another version of the patch which supports reconnection of
EventSource clients.

-- 
You received this message because you are subscribed to the Google Groups 
"prosody-dev" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to prosody-dev+unsubscr...@googlegroups.com.
To post to this group, send email to prosody-dev@googlegroups.com.
Visit this group at https://groups.google.com/group/prosody-dev.
For more options, visit https://groups.google.com/d/optout.

# HG changeset patch
# User Senya <se...@kinetiksoft.com>
# Date 1522707948 -10800
#      Tue Apr 03 01:25:48 2018 +0300
# Node ID f4951ff530b127464245de71fb980d2101ab0ba1
# Parent  ba6466fa68233ea7739bfe70202c8d7522323413
mod_muc_eventsource: history push support

This patch adds optional feature to send history upon a new connection
to the EventSource node of a client.

Also instead of sending JSON object as it was in the original version,
patch changes the module to send actual XMPP stanzas via EventSource
because they contain more information (like timestamps) and this way it
is up to the client software to parse and use any part of the stanza.

The history push supports EventSource reconnection.

diff -r ba6466fa6823 -r f4951ff530b1 mod_muc_eventsource/mod_muc_eventsource.lua
--- a/mod_muc_eventsource/mod_muc_eventsource.lua	Mon Apr 02 16:15:24 2018 +0200
+++ b/mod_muc_eventsource/mod_muc_eventsource.lua	Tue Apr 03 01:25:48 2018 +0300
@@ -4,8 +4,14 @@
 local jid_split = require "util.jid".split;
 local json = require "util.json";
 
+local options = module:get_option("muc_eventsource", {});
+local do_push_history = options.push_history;
 local streams = {};
 
+local st = require "util.stanza";
+local archive_store = "muc_log";
+local archive = module:open_store(archive_store, "archive");
+
 function client_closed(response)
 	local node = response._eventsource_node;
 	module:log("debug", "Destroying client for %q", node);
@@ -15,6 +21,46 @@
 	end
 end
 
+local function get_room_from_jid(node, host)
+  local jid = node .. "@" .. host;
+	local component = hosts[host];
+	if component then
+		local muc = component.modules.muc
+		if muc and rawget(muc,"get_room_from_jid") then
+			-- We're running >0.10 (new MUC API)
+			return muc.get_room_from_jid(jid);
+		else
+			return
+		end
+	end
+end
+
+function render_unique_id(stanza)
+	local stanza_id = stanza:get_child("stanza-id", "urn:xmpp:sid:0");
+	if stanza_id and stanza_id.attr.id then
+		return "id: "..stanza_id.attr.id.."\n";
+	end
+	return "";
+end
+
+function stanza_to_es_message(stanza)
+	return "data: "..tostring(stanza).."\n"..render_unique_id(stanza).."\n";
+end
+
+function push_history(room_name, conn, since)
+	local event = {
+		room = get_room_from_jid(room_name, module:get_host());
+		next_stanza = function() end;
+		since = since;
+	};
+
+	module:fire_event("muc-get-history", event);
+
+	for msg in event.next_stanza, event do
+		conn:write(stanza_to_es_message(msg));
+	end
+end
+
 function serve_stream(event, node)
 	local response = event.response;
 
@@ -45,6 +91,26 @@
 	end
 	clientlist[response] = response.conn;
 
+	if do_push_history then
+		local since = nil;
+		local last_id = event.request.headers.last_event_id;
+		if last_id then
+			local query = st.stanza("query", {id = last_id; xmlns = 'urn:xmpp:mam:2'});
+			local data = archive:find(node, {});
+
+			if data then
+				for id, item, when in data do
+					if id == last_id then
+						since = when + 1;
+						break;
+					end
+				end
+			end
+		end
+
+		push_history(node, response.conn, since);
+	end
+
 	return true;
 end
 
@@ -59,13 +125,7 @@
 	if not body then
 		return;
 	end
-	local nick = select(3, jid_split(stanza.attr.from));
-	-- Encode body and broadcast to eventsource subscribers
-	local json_data = json.encode({
-		nick = nick;
-		body = body;
-	});
-	local data = "data: "..json_data:gsub("\n", "\ndata: \n").."\n\n";
+	local data = stanza_to_es_message(stanza);
 	for response, conn in pairs(clientlist) do
 		conn:write(data);
 	end
@@ -78,5 +138,4 @@
 	};
 });
 
-
-module:hook("muc-broadcast-message", handle_message);
+module:hook("muc-broadcast-message", handle_message, -10);

signature.asc
Description: OpenPGP digital signature