diff options
| -rw-r--r-- | plugins/mod_c2s.lua | 28 | ||||
| -rw-r--r-- | plugins/mod_s2s.lua | 27 | ||||
| -rw-r--r-- | plugins/mod_smacks.lua | 18 | ||||
| -rw-r--r-- | util-src/pposix.c | 32 |
4 files changed, 75 insertions, 30 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); diff --git a/plugins/mod_smacks.lua b/plugins/mod_smacks.lua index 2b131031..65d410e0 100644 --- a/plugins/mod_smacks.lua +++ b/plugins/mod_smacks.lua @@ -541,13 +541,17 @@ module:hook("pre-resource-unbind", function (event) return end - prosody.main_thread:run(function () - session.log("debug", "Destroying session for hibernating too long"); - save_old_session(session); - session.resumption_token = nil; - sessionmanager.destroy_session(session, "Hibernating too long"); - sessions_expired(1); - end); + session.thread:run({ + event = "callback"; + name = "mod_smacks/destroy_hibernating"; + callback = function () + session.log("debug", "Destroying session for hibernating too long"); + save_old_session(session); + session.resumption_token = nil; + sessionmanager.destroy_session(session, "Hibernating too long"); + sessions_expired(1); + end; + }); end); if session.conn then local conn = session.conn; diff --git a/util-src/pposix.c b/util-src/pposix.c index 23afd7b1..502e5326 100644 --- a/util-src/pposix.c +++ b/util-src/pposix.c @@ -694,6 +694,37 @@ static int lc_pipe(lua_State *L) { return 2; } +/* This helper function is adapted from Lua 5.3's liolib.c */ +static int stdio_fclose (lua_State *L) { + int res = -1; + luaL_Stream *p = ((luaL_Stream *)luaL_checkudata(L, 1, LUA_FILEHANDLE)); + if (p->f == NULL) { + return 0; + } + res = fclose(p->f); + p->f = NULL; + return luaL_fileresult(L, (res == 0), NULL); +} + +static int lc_fdopen(lua_State *L) { + int fd = luaL_checkinteger(L, 1); + const char *mode = luaL_checkstring(L, 2); + + luaL_Stream *file = (luaL_Stream *)lua_newuserdata(L, sizeof(luaL_Stream)); + file->closef = stdio_fclose; + file->f = fdopen(fd, mode); + + if (!file->f) { + luaL_pushfail(L); + lua_pushstring(L, strerror(errno)); + return 2; + } + + luaL_getmetatable(L, LUA_FILEHANDLE); + lua_setmetatable(L, -2); + return 1; +} + static int lc_uname(lua_State *L) { struct utsname uname_info; @@ -911,6 +942,7 @@ int luaopen_prosody_util_pposix(lua_State *L) { { "mkdir", lc_mkdir }, { "pipe", lc_pipe }, + { "fdopen", lc_fdopen }, { "setrlimit", lc_setrlimit }, { "getrlimit", lc_getrlimit }, |