Mercurial > prosody-hg

comparison plugins/mod_c2s.lua @ 13575:750ff9f579e2

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
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 2159a206684e
children 1b2de94f0580
comparison
equal deleted inserted replaced
13574:f29d15aef6f8 13575:750ff9f579e2
43 local hosts = prosody.hosts; 43 local hosts = prosody.hosts;
44 44
45 local stream_callbacks = { default_ns = "jabber:client" }; 45 local stream_callbacks = { default_ns = "jabber:client" };
46 local listener = {}; 46 local listener = {};
47 local runner_callbacks = {}; 47 local runner_callbacks = {};
48 local session_events = {};
48 49
49 local m_tls_params = module:metric( 50 local m_tls_params = module:metric(
50 "counter", "encrypted", "", 51 "counter", "encrypted", "",
51 "Encrypted connections", 52 "Encrypted connections",
52 {"protocol"; "cipher"} 53 {"protocol"; "cipher"}
74 --- Stream events handlers 75 --- Stream events handlers
75 local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'}; 76 local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'};
76 77
77 function stream_callbacks.streamopened(session, attr) 78 function stream_callbacks.streamopened(session, attr)
78 -- run _streamopened in async context 79 -- run _streamopened in async context
79 session.thread:run({ stream = "opened", attr = attr }); 80 session.thread:run({ event = "streamopened", attr = attr });
80 end 81 end
81 82
82 function stream_callbacks._streamopened(session, attr) 83 function session_events.streamopened(session, event)
83 local send = session.send; 84 local send, attr = session.send, event.attr;
84 if not attr.to then 85 if not attr.to then
85 session:close{ condition = "improper-addressing", 86 session:close{ condition = "improper-addressing",
86 text = "A 'to' attribute is required on stream headers" }; 87 text = "A 'to' attribute is required on stream headers" };
87 return; 88 return;
88 end 89 end
160 end 161 end
161 end 162 end
162 163
163 function stream_callbacks.streamclosed(session, attr) 164 function stream_callbacks.streamclosed(session, attr)
164 -- run _streamclosed in async context 165 -- run _streamclosed in async context
165 session.thread:run({ stream = "closed", attr = attr }); 166 session.thread:run({ event = "streamclosed", attr = attr });
166 end 167 end
167 168
168 function stream_callbacks._streamclosed(session) 169 function session_events.streamclosed(session)
169 session.log("debug", "Received </stream:stream>"); 170 session.log("debug", "Received </stream:stream>");
170 session:close(false); 171 session:close(false);
172 end
173
174 function session_events.callback(session, event)
175 session.log("debug", "Running session callback %s", event.name);
176 event.callback(session, event);
171 end 177 end
172 178
173 function stream_callbacks.error(session, error, data) 179 function stream_callbacks.error(session, error, data)
174 if error == "no-stream" then 180 if error == "no-stream" then
175 session.log("debug", "Invalid opening stream header (%s)", (data:gsub("^([^\1]+)\1", "{%1}"))); 181 session.log("debug", "Invalid opening stream header (%s)", (data:gsub("^([^\1]+)\1", "{%1}")));
348 function session.reset_stream() 354 function session.reset_stream()
349 session.notopen = true; 355 session.notopen = true;
350 session.stream:reset(); 356 session.stream:reset();
351 end 357 end
352 358
353 session.thread = runner(function (stanza) 359 session.thread = runner(function (item)
354 if st.is_stanza(stanza) then 360 if st.is_stanza(item) then
355 core_process_stanza(session, stanza); 361 core_process_stanza(session, item);
356 elseif stanza.stream == "opened" then 362 else
357 stream_callbacks._streamopened(session, stanza.attr); 363 session_events[item.event](session, item);
358 elseif stanza.stream == "closed" then
359 stream_callbacks._streamclosed(session, stanza.attr);
360 end 364 end
361 end, runner_callbacks, session); 365 end, runner_callbacks, session);
362 366
363 local filter = session.filter; 367 local filter = session.filter;
364 function session.data(data) 368 function session.data(data)