diff options
| -rw-r--r-- | plugins/mod_c2s.lua | 28 | ||||
| -rw-r--r-- | plugins/mod_s2s.lua | 27 |
2 files changed, 32 insertions, 23 deletions
diff --git a/plugins/mod_c2s.lua b/plugins/mod_c2s.lua index 09d4be08..e29ea6a0 100644 --- a/plugins/mod_c2s.lua +++ b/plugins/mod_c2s.lua @@ -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); diff --git a/plugins/mod_s2s.lua b/plugins/mod_s2s.lua index 638ace3d..8eb1565e 100644 --- a/plugins/mod_s2s.lua +++ b/plugins/mod_s2s.lua @@ -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); |