Software / code / prosody
Changeset
13575:750ff9f579e2
mod_c2s, mod_s2s: Support for queuing callbacks to run in session thread
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.
| author | Matthew Wild <mwild1@gmail.com> |
|---|---|
| date | Thu, 21 Nov 2024 17:02:07 +0000 |
| parents | 13574:f29d15aef6f8 |
| children | 13576:5d8fc1c4eb7c |
| files | plugins/mod_c2s.lua plugins/mod_s2s.lua |
| diffstat | 2 files changed, 32 insertions(+), 23 deletions(-) [+] |
line wrap: on
line diff
--- a/plugins/mod_c2s.lua Wed Nov 20 12:08:59 2024 +0000 +++ b/plugins/mod_c2s.lua Thu Nov 21 17:02:07 2024 +0000 @@ -45,6 +45,7 @@ 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 @@ 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 @@ 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 @@ 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);
--- a/plugins/mod_s2s.lua Wed Nov 20 12:08:59 2024 +0000 +++ b/plugins/mod_s2s.lua Thu Nov 21 17:02:07 2024 +0000 @@ -89,6 +89,7 @@ local sessions = module:shared("sessions"); local runner_callbacks = {}; +local session_events = {}; local listener = {}; @@ -469,10 +470,11 @@ 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 @@ 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 @@ 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);
