Mercurial > prosody-hg

comparison plugins/mod_s2s.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 63492b6e7fb6
children 99d2100d2918
comparison
equal deleted inserted replaced
13574:f29d15aef6f8 13575:750ff9f579e2
87 ); 87 );
88 88
89 local sessions = module:shared("sessions"); 89 local sessions = module:shared("sessions");
90 90
91 local runner_callbacks = {}; 91 local runner_callbacks = {};
92 local session_events = {};
92 93
93 local listener = {}; 94 local listener = {};
94 95
95 local log = module._log; 96 local log = module._log;
96 97
467 468
468 local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams"; 469 local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams";
469 470
470 function stream_callbacks.streamopened(session, attr) 471 function stream_callbacks.streamopened(session, attr)
471 -- run _streamopened in async context 472 -- run _streamopened in async context
472 session.thread:run({ stream = "opened", attr = attr }); 473 session.thread:run({ event = "streamopened", attr = attr });
473 end 474 end
474 475
475 function stream_callbacks._streamopened(session, attr) 476 function session_events.streamopened(session, event)
477 local attr = event.attr;
476 session.version = tonumber(attr.version) or 0; 478 session.version = tonumber(attr.version) or 0;
477 session.had_stream = true; -- Had a stream opened at least once 479 session.had_stream = true; -- Had a stream opened at least once
478 480
479 -- TODO: Rename session.secure to session.encrypted 481 -- TODO: Rename session.secure to session.encrypted
480 if session.secure == false then -- Set by mod_tls during STARTTLS handshake 482 if session.secure == false then -- Set by mod_tls during STARTTLS handshake
611 end 613 end
612 end 614 end
613 end 615 end
614 end 616 end
615 617
616 function stream_callbacks._streamclosed(session) 618 function session_events.streamclosed(session)
617 (session.log or log)("debug", "Received </stream:stream>"); 619 (session.log or log)("debug", "Received </stream:stream>");
618 session:close(false); 620 session:close(false);
619 end 621 end
620 622
623 function session_events.callback(session, event)
624 session.log("debug", "Running session callback %s", event.name);
625 event.callback(session, event);
626 end
627
621 function stream_callbacks.streamclosed(session, attr) 628 function stream_callbacks.streamclosed(session, attr)
622 -- run _streamclosed in async context 629 -- run _streamclosed in async context
623 session.thread:run({ stream = "closed", attr = attr }); 630 session.thread:run({ event = "streamclosed", attr = attr });
624 end 631 end
625 632
626 -- Some stream conditions indicate a problem on our end, e.g. that we sent 633 -- Some stream conditions indicate a problem on our end, e.g. that we sent
627 -- something invalid. Those should be investigated. Others are problems or 634 -- something invalid. Those should be investigated. Others are problems or
628 -- events in the remote host that don't affect us, or simply that the 635 -- events in the remote host that don't affect us, or simply that the
782 789
783 -- Session initialization logic shared by incoming and outgoing 790 -- Session initialization logic shared by incoming and outgoing
784 local function initialize_session(session) 791 local function initialize_session(session)
785 local stream = new_xmpp_stream(session, stream_callbacks, stanza_size_limit); 792 local stream = new_xmpp_stream(session, stream_callbacks, stanza_size_limit);
786 793
787 session.thread = runner(function (stanza) 794 session.thread = runner(function (item)
788 if st.is_stanza(stanza) then 795 if st.is_stanza(item) then
789 core_process_stanza(session, stanza); 796 core_process_stanza(session, item);
790 elseif stanza.stream == "opened" then 797 else
791 stream_callbacks._streamopened(session, stanza.attr); 798 session_events[item.event](session, item);
792 elseif stanza.stream == "closed" then
793 stream_callbacks._streamclosed(session, stanza.attr);
794 end 799 end
795 end, runner_callbacks, session); 800 end, runner_callbacks, session);
796 801
797 local log = session.log or log; 802 local log = session.log or log;
798 session.stream = stream; 803 session.stream = stream;