mirrors/prosody
1
0
Fork 0
mirror of https://github.com/bjc/prosody.git synced 2025-04-03 21:27:38 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

mod_c2s, mod_s2s: Support for queuing callbacks to run in session thread

Browse source 
This allows certain session-specific code that needs to run in the async
context, but is itself triggered outside of that context (e.g. timers), to
be queued.

An example of this is the session destruction code of mod_smacks, when the
hibernation timeout is reached.
This commit is contained in:
Matthew Wild 2024-11-21 17:02:07 +00:00
parent a3b71e8fc9
commit 6bbfcd801a
2 changed files with 32 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

28
plugins/mod_c2s.lua
View file

@ -45,6 +45,7 @@ local hosts = prosody.hosts;
local stream_callbacks = { default_ns = "jabber:client" };
local listener = {};
local runner_callbacks = {};
local session_events = {};
local m_tls_params = module:metric(
"counter", "encrypted", "",
@ -76,11 +77,11 @@ local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'};
function stream_callbacks.streamopened(session, attr)
-- run _streamopened in async context
session.thread:run({ stream = "opened", attr = attr });
session.thread:run({ event = "streamopened", attr = attr });
end
function stream_callbacks._streamopened(session, attr)
local send = session.send;
function session_events.streamopened(session, event)
local send, attr = session.send, event.attr;
if not attr.to then
session:close{ condition = "improper-addressing",
text = "A 'to' attribute is required on stream headers" };
@ -162,14 +163,19 @@ end
function stream_callbacks.streamclosed(session, attr)
-- run _streamclosed in async context
session.thread:run({ stream = "closed", attr = attr });
session.thread:run({ event = "streamclosed", attr = attr });
end
function stream_callbacks._streamclosed(session)
function session_events.streamclosed(session)
session.log("debug", "Received </stream:stream>");
session:close(false);
end
function session_events.callback(session, event)
session.log("debug", "Running session callback %s", event.name);
event.callback(session, event);
end
function stream_callbacks.error(session, error, data)
if error == "no-stream" then
session.log("debug", "Invalid opening stream header (%s)", (data:gsub("^([^\1]+)\1", "{%1}")));
@ -350,13 +356,11 @@ function listener.onconnect(conn)
session.stream:reset();
end
session.thread = runner(function (stanza)
if st.is_stanza(stanza) then
core_process_stanza(session, stanza);
elseif stanza.stream == "opened" then
stream_callbacks._streamopened(session, stanza.attr);
elseif stanza.stream == "closed" then
stream_callbacks._streamclosed(session, stanza.attr);
session.thread = runner(function (item)
if st.is_stanza(item) then
core_process_stanza(session, item);
else
session_events[item.event](session, item);
end
end, runner_callbacks, session);

27
plugins/mod_s2s.lua
View file

@ -89,6 +89,7 @@ local m_tls_params = module:metric(
local sessions = module:shared("sessions");
local runner_callbacks = {};
local session_events = {};
local listener = {};
@ -469,10 +470,11 @@ local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams";
function stream_callbacks.streamopened(session, attr)
-- run _streamopened in async context
session.thread:run({ stream = "opened", attr = attr });
session.thread:run({ event = "streamopened", attr = attr });
end
function stream_callbacks._streamopened(session, attr)
function session_events.streamopened(session, event)
local attr = event.attr;
session.version = tonumber(attr.version) or 0;
session.had_stream = true; -- Had a stream opened at least once
@ -613,14 +615,19 @@ function stream_callbacks._streamopened(session, attr)
end
end
function stream_callbacks._streamclosed(session)
function session_events.streamclosed(session)
(session.log or log)("debug", "Received </stream:stream>");
session:close(false);
end
function session_events.callback(session, event)
session.log("debug", "Running session callback %s", event.name);
event.callback(session, event);
end
function stream_callbacks.streamclosed(session, attr)
-- run _streamclosed in async context
session.thread:run({ stream = "closed", attr = attr });
session.thread:run({ event = "streamclosed", attr = attr });
end
-- Some stream conditions indicate a problem on our end, e.g. that we sent
@ -784,13 +791,11 @@ end
local function initialize_session(session)
local stream = new_xmpp_stream(session, stream_callbacks, stanza_size_limit);
session.thread = runner(function (stanza)
if st.is_stanza(stanza) then
core_process_stanza(session, stanza);
elseif stanza.stream == "opened" then
stream_callbacks._streamopened(session, stanza.attr);
elseif stanza.stream == "closed" then
stream_callbacks._streamclosed(session, stanza.attr);
session.thread = runner(function (item)
if st.is_stanza(item) then
core_process_stanza(session, item);
else
session_events[item.event](session, item);
end
end, runner_callbacks, session);