File

net/connect.lua @ 12960:31b22cc221b5

mod_pubsub, mod_pep: Support per-node configurable inclusion of publisher This matches ejabberd's behaviour, using the 'pubsub#itemreply' config option. Although the current definition of this option in the specification is not as clear as it could be, I think matching what existing deployments do is the best option to resolve the ambiguity and reduce fragmentation. We should update the spec to be clearer about how to use and interpret this option. The 'expose_publisher' option for mod_pubsub is now an override (always expose or never expose). If unset, it will use the per-node config (which defaults to not exposing). Thanks to Link Mauve, edhelas and goffi for sparking this feature.
author Matthew Wild <mwild1@gmail.com>
date Wed, 22 Mar 2023 11:39:19 +0000
parent 12639:6d9ee0a3eb4b
child 12974:ba409c67353b
line wrap: on
line source

local server = require "net.server";
local log = require "util.logger".init("net.connect");
local new_id = require "util.id".short;
local timer = require "util.timer";

-- FIXME RFC 6724
-- FIXME Error propagation from resolvers doesn't work
-- FIXME #1428 Reuse DNS resolver object between service and basic resolver
-- FIXME #1429 Close DNS resolver object when done

local pending_connection_methods = {};
local pending_connection_mt = {
	__name = "pending_connection";
	__index = pending_connection_methods;
	__tostring = function (p)
		return "<pending connection "..p.id.." to "..tostring(p.target_resolver.hostname)..">";
	end;
};

function pending_connection_methods:log(level, message, ...)
	log(level, "[pending connection %s] "..message, self.id, ...);
end

-- pending_connections_map[conn] = pending_connection
local pending_connections_map = {};

local pending_connection_listeners = {};

local function attempt_connection(p)
	p:log("debug", "Checking for targets...");
	p.target_resolver:next(function (conn_type, ip, port, extra, more_targets_available)
		if not conn_type then
			-- No more targets to try
			p:log("debug", "No more connection targets to try", p.target_resolver.last_error);
			if next(p.conns) == nil then
				p:log("debug", "No more targets, no pending connections. Connection failed.");
				if p.listeners.onfail then
					p.listeners.onfail(p.data, p.last_error or p.target_resolver.last_error or "unable to resolve service");
				end
			else
				p:log("debug", "One or more connection attempts are still pending. Waiting for now.");
			end
			return;
		end
		p:log("debug", "Next target to try is %s:%d", ip, port);
		local conn, err = server.addclient(ip, port, pending_connection_listeners, p.options.pattern or "*a",
			extra and extra.sslctx or p.options.sslctx, conn_type, extra);
		if not conn then
			log("debug", "Connection attempt failed immediately: %s", err);
			p.last_error = err or "unknown reason";
			return attempt_connection(p);
		end
		p.conns[conn] = true;
		pending_connections_map[conn] = p;
		if more_targets_available then
			timer.add_task(0.250, function ()
				if not p.connected then
					p:log("debug", "Still not connected, making parallel connection attempt...");
					attempt_connection(p);
				end
			end);
		end
	end);
end

function pending_connection_listeners.onconnect(conn)
	local p = pending_connections_map[conn];
	if not p then
		log("warn", "Successful connection, but unexpected! Closing.");
		conn:close();
		return;
	end
	pending_connections_map[conn] = nil;
	if p.connected then
		-- We already succeeded in connecting
		p.conns[conn] = nil;
		conn:close();
		return;
	end
	p.connected = true;
	p:log("debug", "Successfully connected");
	conn:setlistener(p.listeners, p.data);
	return p.listeners.onconnect(conn);
end

function pending_connection_listeners.ondisconnect(conn, reason)
	local p = pending_connections_map[conn];
	if not p then
		log("warn", "Failed connection, but unexpected!");
		return;
	end
	p.conns[conn] = nil;
	pending_connections_map[conn] = nil;
	p.last_error = reason or "unknown reason";
	p:log("debug", "Connection attempt failed: %s", p.last_error);
	if p.connected then
		p:log("debug", "Connection already established, ignoring failure");
	elseif next(p.conns) == nil then
		p:log("debug", "No pending connection attempts, and not yet connected");
		attempt_connection(p);
	else
		p:log("debug", "Other attempts are still pending, ignoring failure");
	end
end

local function connect(target_resolver, listeners, options, data)
	local p = setmetatable({
		id = new_id();
		target_resolver = target_resolver;
		listeners = assert(listeners);
		options = options or {};
		data = data;
		conns = {};
	}, pending_connection_mt);

	p:log("debug", "Starting connection process");
	attempt_connection(p);
end

return {
	connect = connect;
};