MT#6407 support MUC on cluster

Change-Id: I6d8cc8976ca32bb2af379e2bce9aae38551c7928

plugins/mod_shard.lua

@@ -7,29 +7,76 @@
 --
 module:depends("sipwise_redis_sessions");
 local redis_sessions = module:shared("/*/sipwise_redis_sessions/redis_sessions");
+local redis_mucs;
 local jid_split = require "util.jid".split;
 local fire_event = prosody.events.fire_event;
 local st = require "util.stanza";
-
+local ut = require "util.table";
 local shard_name = module:get_option("shard_name", nil);
 if not shard_name then
     error("shard_name not configured", 0);
 end
 
-
 module:log("info", "%s added to shard %s", module.host, shard_name);
 
+local function build_query_result(host, rooms, stanza)
+    local xmlns = 'http://jabber.org/protocol/disco#items';
+    local s = stanza:query(xmlns);
+    for _,v in ipairs(rooms) do
+        s:tag("item", {jid=v.."@"..host}):up();
+    end
+    return stanza;
+end
+
+local function handle_room_event(event)
+    local to = event.stanza.attr.to;
+    local node, host, _ = jid_split(to);
+    local rhost;
+
+    if node then
+        module:log("debug", "looking up target room shard for %s", to);
+        rhost = redis_mucs.get_room_host(to);
+    else
+        local l = redis_mucs.get_rooms(host);
+        module:log("debug", "rooms: %s", ut.table.tostring(l));
+        local stanza = build_query_result(host, l, st.reply(event.stanza));
+        module:log("debug", "reply[%s]", tostring(stanza));
+        event.origin.send(stanza);
+        return true;
+    end
+
+    if not rhost then
+        module:log("debug", "room not found. Nothing to do");
+        return nil;
+    end
+
+    if rhost == shard_name then
+        module:log("debug", "room is hosted here. Nothing to do");
+        return nil
+    end
+
+    module:log("debug", "target shard for %s is %s", to, rhost);
+    fire_event("shard/send", { shard = rhost, stanza = event.stanza });
+    return true;
+end
 
 local function handle_event (event)
     local to = event.stanza.attr.to;
     local node, host, resource = jid_split(to);
     local stop_process_local;
 
-    if not node or not host then
+    if not host then
+        module:log("debug", "no host. Nothing to do here");
         return nil
     end
-    if host ~= module.host then
-        return nil
+
+    if ut.string.starts(host, 'conference.') then
+        if redis_mucs then
+            module:log("debug", "MUC %s detected", host);
+            return handle_room_event(event);
+        else
+            module:log("debug", "redis_mucs nill");
+        end
     end
 
     if resource and prosody.full_sessions[to] then
@@ -37,6 +84,11 @@ local function handle_event (event)
         return nil
     end
 
+    if not node then
+        module:log("debug", "no node. Nothing to do here");
+        return nil
+    end
+
     if prosody.bare_sessions[to] then
         module:log("debug", "%s has a bare session here."..
             " stanza will be processed here too", to);
@@ -62,6 +114,12 @@ local function handle_event (event)
     return stop_process_local;
 end
 
+local host = module:get_host();
+if module:get_host_type() == "component" then
+    module:depends("sipwise_redis_mucs");
+    redis_mucs = module:shared("/*/sipwise_redis_mucs/redis_mucs");
+    module:log("debug", "enable MUC for %s", host);
+end
 module:hook("iq/bare", handle_event, 1000);
 module:hook("iq/full", handle_event, 1000);
 module:hook("iq/host", handle_event, 1000);
@@ -71,3 +129,4 @@ module:hook("message/host", handle_event, 1000);
 module:hook("presence/bare", handle_event, 1000);
 module:hook("presence/full", handle_event, 1000);
 module:hook("presence/host", handle_event, 1000);
+module:log("debug", "hooked at %s", host);

plugins/mod_sipwise_redis_mucs.lua

@@ -0,0 +1,98 @@
+-- Prosody IM
+-- Copyright (C) 2014-2015 Sipwise GmbH <development@sipwise.com>
+--
+-- This project is MIT/X11 licensed. Please see the
+-- COPYING file in the source package for more information.
+--
+module:set_global();
+local ut = require "util.table";
+local jid = require "util.jid";
+local redis = require 'redis';
+local redis_config = {
+	port = 6739, host = "127.0.0.1",
+	server_id = "0", redis_db = "2"
+};
+local redis_client;
+
+local redis_mucs = module:shared("redis_mucs");
+
+local function test_connection()
+	if not redis_client then return nil end;
+	local ok, _ = pcall(redis_client.ping, redis_client);
+	if not ok then
+		redis_client = nil;
+	end
+end
+
+local function client_connect()
+	redis_client = redis.connect(redis_config.host, redis_config.port);
+	if redis_config.redis_db then
+		redis_client:select(redis_config.redis_db);
+	end
+end
+
+local function muc_created(event)
+	local room = event.room;
+	local node, host, _ = jid.split(room.jid)
+
+	module:log("debug", "muc-room-created %s", room.jid);
+	module:log("debug", "save [%s]=%s", room.jid, redis_config.server_id);
+	if not test_connection() then client_connect() end
+	redis_client:set(room.jid, redis_config.server_id);
+	module:log("debug", "append [%s]=>%s:%s", host,
+		redis_config.server_id, node);
+    redis_client:sadd(host, redis_config.server_id..":"..node);
+end
+
+local function muc_destroyed(event)
+	local room = event.room;
+	local node, host, _ = jid.split(room.jid)
+
+	module:log("debug", "muc-room-destroyed %s", room.jid);
+	module:log("debug", "remove [%s]=%s", room.jid, redis_config.server_id);
+	if not test_connection() then client_connect() end
+	redis_client:del(room.jid);
+	module:log("debug", "remove [%s]=>%s:%s", host,
+		redis_config.server_id, node);
+    redis_client:srem(host, redis_config.server_id..":"..node);
+end
+
+local function split_key(key)
+	local t = ut.string.explode(':', key);
+	return t[1], t[2];
+end
+
+function redis_mucs.get_rooms(host)
+	local res = {};
+	local l, r;
+
+	module:log("debug", "search rooms at %s host", host);
+	if not test_connection() then client_connect() end
+	l = redis_client:smembers(host);
+
+	for _,v in pairs(l) do
+		_, r = split_key(v);
+		ut.table.add(res, r);
+	end
+	return res;
+end
+
+function redis_mucs.get_room_host(room_jid)
+	local node, domain = jid.split(room_jid);
+	local bare_jid = node.."@"..domain;
+
+	module:log("debug", "search room:%s host", bare_jid);
+	if not test_connection() then client_connect() end
+	return redis_client:get(bare_jid);
+end
+
+
+function module.load()
+	redis_config = module:get_option("redis_sessions_auth", redis_config);
+end
+
+function module.add_host(module)
+	module:hook("muc-room-created", muc_created, 200);
+	module:hook("muc-room-destroyed", muc_destroyed, 200);
+	module:log("debug", "hooked at %s", module:get_host());
+end

plugins/mod_sipwise_redis_sessions.lua

@@ -1,5 +1,5 @@
 -- Prosody IM
--- Copyright (C) 2014 Sipwise GmbH <development@sipwise.com>
+-- Copyright (C) 2014-2015 Sipwise GmbH <development@sipwise.com>
 --
 -- This project is MIT/X11 licensed. Please see the
 -- COPYING file in the source package for more information.
@@ -20,7 +20,7 @@ local redis_sessions = module:shared("redis_sessions");
 
 local function test_connection()
 	if not redis_client then return nil end;
-	local ok, err = pcall(redis_client.ping, redis_client);
+	local ok, _ = pcall(redis_client.ping, redis_client);
 	if not ok then
 		redis_client = nil;
 	end
@@ -49,7 +49,7 @@ local function resource_bind(event)
 end
 
 local function resource_unbind(event)
-	local session, err = event.session, event.error;
+	local session, _ = event.session, event.error;
 	local node, domain, resource = jid.split(session.full_jid);
 	local full_jid, bare_jid = session.full_jid, node.."@"..domain;
 
@@ -67,7 +67,7 @@ local function split_key(key)
 end
 
 function redis_sessions.get_hosts(j)
-	local node, domain, resource = jid.split(j);
+	local node, domain = jid.split(j);
 	local bare_jid = node.."@"..domain;
 	local res = {};
 	local l, h, r;
@@ -86,12 +86,11 @@ function redis_sessions.get_hosts(j)
 end
 
 function module.load()
-	module:log("debug", "load");
 	redis_config = module:get_option("redis_sessions_auth", redis_config);
-	redis_enable = client_connect();
 end
 
 function module.add_host(module)
 	module:hook("resource-bind", resource_bind, 200);
 	module:hook("resource-unbind", resource_unbind, 200);
+	module:log("debug", "hooked at %s", module:get_host());
 end

plugins/mod_sipwise_vhosts_sql.lua

@@ -1,5 +1,5 @@
 -- Load vhosts from DB on startup of Prosody XMPP server.
--- Copyright (C) 2013 Sipwise GmbH <development@sipwise.com>
+-- Copyright (C) 2013-2015 Sipwise GmbH <development@sipwise.com>
 --
 -- This project is MIT/X11 licensed. Please see the
 -- COPYING file in the source package for more information.
@@ -7,7 +7,6 @@
 
 module:set_global();
 
-local log = require "util.logger".init("sipwise_vhosts_sql");
 local DBI = require "DBI"
 local hostmanager = require "core.hostmanager";
 local configmanager = require "core.configmanager";
@@ -38,7 +37,8 @@ local function connect()
 		);
 		prosody.lock_globals();
 		if not dbh then
-			module:log("debug", "Database connection failed: %s", tostring(err));
+			module:log("debug",
+				"Database connection failed: %s", tostring(err));
 			return nil, err;
 		end
 		module:log("debug", "Successfully connected to database");
@@ -50,13 +50,15 @@ end
 
 do -- process options to get a db connection
 	params = params or { driver = "SQLite3" };
-	
+
 	if params.driver == "SQLite3" then
-		params.database = resolve_relative_path(prosody.paths.data or ".", params.database or "prosody.sqlite");
+		params.database = configmanager.resolve_relative_path(
+			prosody.paths.data or ".",
+			params.database or "prosody.sqlite");
 	end
-	
-	assert(params.driver and params.database, "Both the SQL driver and the database need to be specified");
-	
+
+	assert(params.driver and params.database,
+		"Both the SQL driver and the database need to be specified");
 	assert(connect());
 end
 
@@ -68,25 +70,39 @@ local function getsql(sql, ...)
 	-- do prepared statement stuff
 	local stmt, err = connection:prepare(sql);
 	if not stmt and not test_connection() then error("connection failed"); end
-	if not stmt then module:log("error", "QUERY FAILED: %s %s", err, debug.traceback()); return nil, err; end
+	if not stmt then
+		module:log("error", "QUERY FAILED: %s %s", err, debug.traceback());
+		return nil, err;
+	end
 	-- run query
-	local ok, err = stmt:execute(...);
+	local ok
+	ok, err = stmt:execute(...);
 	if not ok and not test_connection() then error("connection failed"); end
 	if not ok then return nil, err; end
-	
+
 	return stmt;
 end
 
 local function load_vhosts_from_db()
-	local stmt, err = getsql("SELECT `domain` FROM `domain`");
+	local stmt, _ = getsql("SELECT `domain` FROM `domain`");
 	local host_config = { enable = true };
 	if stmt then
 		for row in stmt:rows(true) do
-			module:log("debug", "load_vhosts_from_db: activate host %s", row.domain);
+			module:log("debug", "load_vhosts_from_db: activate host %s",
+				row.domain);
 			hostmanager.activate(row.domain, host_config);
-			module:log("debug", "load_vhosts_from_db: activate implicit search.%s", row.domain);
-			hostmanager.activate("search."..row.domain, { component_module = "sipwise_vjud" });
-			configmanager.set("conference."..row.domain, "component_module", "muc");
+
+			module:log("debug",
+				"load_vhosts_from_db: activate implicit search.%s",
+				row.domain);
+			configmanager.set("conference."..row.domain, "component_module",
+				"sipwise_vjud");
+			hostmanager.activate("search."..row.domain);
+
+			configmanager.set("conference."..row.domain, "component_module",
+				"muc");
+			configmanager.set("conference."..row.domain, "modules_enabled",
+				{ "sipwise_redis_mucs", "shard" });
 			local conference_config = configmanager.getconfig()["conference."..row.domain];
 			conference_config['restrict_room_creation'] = 'local';
 			hostmanager.activate("conference."..row.domain, conference_config);

plugins/mod_sipwise_vjud.lua

@@ -1,16 +1,17 @@
 --
--- Copyright (C) 2013 Sipwise GmbH <development@sipwise.com>
+-- Copyright (C) 2013-2015 Sipwise GmbH <development@sipwise.com>
 --
 -- This project is MIT/X11 licensed. Please see the
 -- COPYING file in the source package for more information.
 --
-local nodeprep = require "util.encodings".stringprep.nodeprep;
-local jid_split = require "util.jid".split;
-local sql = module:require("sql");
+module:set_global();
+
+local ut_jid = require "util.jid";
+local mod_sql = module:require("sql");
 local st = require "util.stanza";
 local template = require "util.template";
-local array = require "util.array";
 local rex = require "rex_pcre";
+local prosodyctl = require "util.prosodyctl"
 
 local get_reply = template[[
   <query xmlns='jabber:iq:search'>
@@ -55,18 +56,16 @@ SELECT username,domain FROM kamailio.dbaliases
 WHERE alias_username=?;
 ]];
 
-local mod_sql = module:require("sql");
 local params = module:get_option("auth_sql", {
-	driver = "MySQL", 
-	database = "provisioning", 
-	username = "prosody", 
-	password = "PW_PROSODY", 
-	host = "localhost" 
+	driver = "MySQL",
+	database = "provisioning",
+	username = "prosody",
+	password = "PW_PROSODY",
+	host = "localhost"
 });
-engine = mod_sql:create_engine(params);
-engine:execute("SET NAMES 'utf8' COLLATE 'utf8_bin';");
+local engine;
 
-function normalize_number(user, host, number)
+local function normalize_number(user, host, number)
 	local locale_info = {};
 	for row in engine:select(locale_query, user, host) do
 		locale_info["caller_"..row[1]] = row[2];
@@ -89,7 +88,7 @@ function normalize_number(user, host, number)
 			table.insert(replacement_regexes, { patt, repl });
 		end
 	end
-	
+
 	for _, rule in ipairs(replacement_regexes) do
 		local new_number, n_matches = rex.gsub(number, rule[1], rule[2]);
 		if n_matches > 0 then
@@ -100,7 +99,7 @@ function normalize_number(user, host, number)
 	return number;
 end
 
-function search_by_number(number)
+local function search_by_number(number)
 	local results = {};
 	for result in engine:select(lookup_query, number) do
 		table.insert(results, result[1].."@"..result[2]);
@@ -117,7 +116,7 @@ module:hook("iq/host/jabber:iq:search:query", function(event)
 	if stanza.attr.type == "get" then
 		return origin.send(st.reply(stanza):add_child(get_reply));
 	else
-		local user, host = jid_split(stanza.attr.from);
+		local user, host = ut_jid.split(stanza.attr.from);
 		local number = stanza.tags[1]:get_child_text("nick");
 
 		-- Reconnect to DB if necessary
@@ -139,7 +138,6 @@ module:hook("iq/host/jabber:iq:search:query", function(event)
 end);
 
 function module.command(arg)
-	local jid = require "util.jid";
 	local warn = prosodyctl.show_warning;
 	local command = arg[1];
 	if not command then
@@ -153,7 +151,7 @@ function module.command(arg)
 			return 1;
 		end
 		local user_jid, number = arg[1], arg[2];
-		local user, host = jid.prepped_split(user_jid);
+		local user, host = ut_jid.prepped_split(user_jid);
 		if not (user and host) then
 			warn("Invalid JID: "..user_jid);
 			return 1;
@@ -172,3 +170,15 @@ function module.command(arg)
 	end
 	return 0;
 end
+
+function module.load()
+	engine = mod_sql:create_engine(params);
+	engine:execute("SET NAMES 'utf8' COLLATE 'utf8_bin';");
+end
+
+function module.add_host(module)
+	if module:get_host_type() ~= "component" then
+		error("Don't load mod_sipwise_vjud manually,"..
+			" it should be for a component", 0);
+	end
+end