Software / code / prosody
Changeset
4970:eeff01224865
Merge 0.9->trunk
| author | Matthew Wild <mwild1@gmail.com> |
|---|---|
| date | Sun, 22 Jul 2012 17:08:09 +0100 |
| parents | 4958:1f1c74ac9666 (current diff) 4969:15183193c6a6 (diff) |
| children | 4971:bfc52b9137c8 |
| files | |
| diffstat | 10 files changed, 170 insertions(+), 137 deletions(-) [+] |
line wrap: on
line diff
--- a/core/sessionmanager.lua Sun Jul 22 16:17:33 2012 +0100 +++ b/core/sessionmanager.lua Sun Jul 22 17:08:09 2012 +0100 @@ -82,7 +82,7 @@ end end - function session.send(data) log("debug", "Discarding data sent to resting session: %s", tostring(data)); end + function session.send(data) log("debug", "Discarding data sent to resting session: %s", tostring(data)); return false; end function session.data(data) log("debug", "Discarding data received from resting session: %s", tostring(data)); end return setmetatable(session, resting_session); end
--- a/net/http.lua Sun Jul 22 16:17:33 2012 +0100 +++ b/net/http.lua Sun Jul 22 17:08:09 2012 +0100 @@ -203,7 +203,6 @@ if request.conn then request.conn = nil; request.handler:close() - listener.ondisconnect(request.handler, "closed"); end end
--- a/net/server_event.lua Sun Jul 22 16:17:33 2012 +0100 +++ b/net/server_event.lua Sun Jul 22 17:08:09 2012 +0100 @@ -249,7 +249,7 @@ return true end function interface_mt:_destroy() -- close this interface + events and call last listener - debug( "closing client with id:", self.id ) + debug( "closing client with id:", self.id, self.fatalerror ) self:_lock( true, true, true ) -- first of all, lock the interface to avoid further actions local _ _ = self.eventread and self.eventread:close( ) -- close events; this must be called outside of the event callbacks! @@ -328,22 +328,22 @@ end return true end - function interface_mt:close(now) + function interface_mt:close() if self.nointerface then return nil, "locked"; end debug( "try to close client connection with id:", self.id ) if self.type == "client" then self.fatalerror = "client to close" - if ( not self.eventwrite ) or now then -- try to close immediately + if self.eventwrite then -- wait for incomplete write request + self:_lock( true, true, false ) + debug "closing delayed until writebuffer is empty" + return nil, "writebuffer not empty, waiting" + else -- close now self:_lock( true, true, true ) self:_close() return true - else -- wait for incomplete write request - self:_lock( true, true, false ) - debug "closing delayed until writebuffer is empty" - return nil, "writebuffer not empty, waiting" end else - debug( "try to close server with id:", tostring(self.id), "args:", tostring(now) ) + debug( "try to close server with id:", tostring(self.id)) self.fatalerror = "server to close" self:_lock( true ) self:_close( 0 ) -- add new event to remove the server interface
--- a/net/server_select.lua Sun Jul 22 16:17:33 2012 +0100 +++ b/net/server_select.lua Sun Jul 22 17:08:09 2012 +0100 @@ -314,22 +314,28 @@ end return false, "setoption not implemented"; end - handler.close = function( self, forced ) + handler.force_close = function ( self ) + if bufferqueuelen ~= 0 then + out_put("discarding unwritten data for ", tostring(ip), ":", tostring(clientport)) + for i = bufferqueuelen, 1, -1 do + bufferqueue[i] = nil; + end + bufferqueuelen = 0; + end + return self:close(); + end + handler.close = function( self ) if not handler then return true; end _readlistlen = removesocket( _readlist, socket, _readlistlen ) _readtimes[ handler ] = nil if bufferqueuelen ~= 0 then - if not ( forced or fatalerror ) then - handler.sendbuffer( ) - if bufferqueuelen ~= 0 then -- try again... - if handler then - handler.write = nil -- ... but no further writing allowed - end - toclose = true - return false + handler.sendbuffer() -- Try now to send any outstanding data + if bufferqueuelen ~= 0 then -- Still not empty, so we'll try again later + if handler then + handler.write = nil -- ... but no further writing allowed end - else - send( socket, table_concat( bufferqueue, "", 1, bufferqueuelen ), 1, bufferlen ) -- forced send + toclose = true + return false end end if socket then @@ -347,7 +353,7 @@ local _handler = handler; handler = nil if disconnect then - disconnect(_handler, "closed"); + disconnect(_handler, false); end end if server then @@ -480,7 +486,7 @@ _ = _cleanqueue and clean( bufferqueue ) --out_put( "server.lua: sended '", buffer, "', bytes: ", tostring(succ), ", error: ", tostring(err), ", part: ", tostring(byte), ", to: ", tostring(ip), ":", tostring(clientport) ) else - succ, err, count = false, "closed", 0; + succ, err, count = false, "unexpected close", 0; end if succ then -- sending succesful bufferqueuelen = 0 @@ -491,7 +497,7 @@ drain(handler) end _ = needtls and handler:starttls(nil) - _ = toclose and handler:close( ) + _ = toclose and handler:force_close( ) return true elseif byte and ( err == "timeout" or err == "wantwrite" ) then -- want write buffer = string_sub( buffer, byte + 1, bufferlen ) -- new buffer @@ -504,7 +510,7 @@ out_put( "server.lua: client ", tostring(ip), ":", tostring(clientport), " write error: ", tostring(err) ) fatalerror = true disconnect( handler, err ) - _ = handler and handler:close( ) + _ = handler and handler:force_close( ) return false end end @@ -547,7 +553,7 @@ end out_put( "server.lua: ssl handshake error: ", tostring(err or "handshake too long") ) disconnect( handler, "ssl handshake failed" ) - _ = handler and handler:close( true ) -- forced disconnect + _ = handler and handler:force_close() return false, err -- handshake failed end ) @@ -810,7 +816,7 @@ end for handler, err in pairs( _closelist ) do handler.disconnect( )( handler, err ) - handler:close( true ) -- forced disconnect + handler:force_close() -- forced disconnect end clean( _closelist ) _currenttime = luasocket_gettime( ) @@ -896,7 +902,7 @@ if os_difftime( _currenttime - timestamp ) > _sendtimeout then --_writetimes[ handler ] = nil handler.disconnect( )( handler, "send timeout" ) - handler:close( true ) -- forced disconnect + handler:force_close() -- forced disconnect end end for handler, timestamp in pairs( _readtimes ) do
--- a/plugins/mod_c2s.lua Sun Jul 22 16:17:33 2012 +0100 +++ b/plugins/mod_c2s.lua Sun Jul 22 17:08:09 2012 +0100 @@ -24,6 +24,7 @@ local log = module._log; local c2s_timeout = module:get_option_number("c2s_timeout"); +local stream_close_timeout = module:get_option_number("c2s_close_timeout", 5); local opt_keepalives = module:get_option_boolean("tcp_keepalives", false); local sessions = module:shared("sessions"); @@ -143,8 +144,27 @@ end end session.send("</stream:stream>"); - session.conn:close(); - listener.ondisconnect(session.conn, (reason and (reason.text or reason.condition)) or reason or "session closed"); + + function session.send() return false; end + + local reason = (reason and (reason.text or reason.condition)) or reason or "session closed"; + session.log("info", "c2s stream for %s closed: %s", session.full_jid or ("<"..session.ip..">"), reason); + + -- Authenticated incoming stream may still be sending us stanzas, so wait for </stream:stream> from remote + local conn = session.conn; + if reason == "session closed" and not session.notopen and session.type == "c2s" then + -- Grace time to process data from authenticated cleanly-closed stream + add_task(stream_close_timeout, function () + if not session.destroyed then + session.log("warn", "Failed to receive a stream close response, closing connection anyway..."); + sm_destroy_session(session, reason); + conn:close(); + end + end); + else + sm_destroy_session(session, reason); + conn:close(); + end end end @@ -208,10 +228,9 @@ function listener.ondisconnect(conn, err) local session = sessions[conn]; if session then - (session.log or log)("info", "Client disconnected: %s", err); + (session.log or log)("info", "Client disconnected: %s", err or "connection closed"); sm_destroy_session(session, err); sessions[conn] = nil; - session = nil; end end
--- a/plugins/mod_iq.lua Sun Jul 22 16:17:33 2012 +0100 +++ b/plugins/mod_iq.lua Sun Jul 22 17:08:09 2012 +0100 @@ -17,10 +17,7 @@ local origin, stanza = data.origin, data.stanza; local session = full_sessions[stanza.attr.to]; - if session then - -- TODO fire post processing event - session.send(stanza); - else -- resource not online + if not (session and session.send(stanza)) then if stanza.attr.type == "get" or stanza.attr.type == "set" then origin.send(st.error_reply(stanza, "cancel", "service-unavailable")); end
--- a/plugins/mod_message.lua Sun Jul 22 16:17:33 2012 +0100 +++ b/plugins/mod_message.lua Sun Jul 22 17:08:09 2012 +0100 @@ -35,10 +35,13 @@ if user then -- some resources are connected local recipients = user.top_resources; if recipients then + local sent; for i=1,#recipients do - recipients[i].send(stanza); + sent = recipients[i].send(stanza) or sent; end - return true; + if sent then + return true; + end end end -- no resources are online @@ -65,9 +68,7 @@ local origin, stanza = data.origin, data.stanza; local session = full_sessions[stanza.attr.to]; - if session then - -- TODO fire post processing event - session.send(stanza); + if session and session.send(stanza) then return true; else -- resource not online return process_to_bare(jid_bare(stanza.attr.to), origin, stanza);
--- a/plugins/mod_presence.lua Sun Jul 22 16:17:33 2012 +0100 +++ b/plugins/mod_presence.lua Sun Jul 22 17:08:09 2012 +0100 @@ -352,13 +352,15 @@ -- Send unavailable presence if session.presence then local pres = st.presence{ type = "unavailable" }; - if not(err) or err == "closed" then err = "connection closed"; end - pres:tag("status"):text("Disconnected: "..err):up(); + if err then + pres:tag("status"):text("Disconnected: "..err):up(); + end session:dispatch_stanza(pres); elseif session.directed then local pres = st.presence{ type = "unavailable", from = session.full_jid }; - if not(err) or err == "closed" then err = "connection closed"; end - pres:tag("status"):text("Disconnected: "..err):up(); + if err then + pres:tag("status"):text("Disconnected: "..err):up(); + end for jid in pairs(session.directed) do pres.attr.to = jid; core_post_stanza(session, pres, true);
--- a/plugins/mod_s2s/mod_s2s.lua Sun Jul 22 16:17:33 2012 +0100 +++ b/plugins/mod_s2s/mod_s2s.lua Sun Jul 22 17:08:09 2012 +0100 @@ -31,6 +31,7 @@ local s2sout = module:require("s2sout"); local connect_timeout = module:get_option_number("s2s_timeout", 60); +local stream_close_timeout = module:get_option_number("s2s_close_timeout", 5); local sessions = module:shared("sessions"); @@ -97,9 +98,10 @@ log("error", "WARNING! This might, possibly, be a bug, but it might not..."); log("error", "We are going to send from %s instead of %s", tostring(host.from_host), tostring(from_host)); end - host.sends2s(stanza); - host.log("debug", "stanza sent over "..host.type); - return true; + if host.sends2s(stanza) then + host.log("debug", "stanza sent over "..host.type); + return true; + end end end end @@ -291,18 +293,6 @@ session:close(); end -function stream_callbacks.streamdisconnected(session, err) - if err and err ~= "closed" and session.direction == "outgoing" and session.notopen then - (session.log or log)("debug", "s2s connection attempt failed: %s", err); - if s2sout.attempt_connection(session, err) then - (session.log or log)("debug", "...so we're going to try another target"); - return true; -- Session lives for now - end - end - (session.log or log)("info", "s2s disconnected: %s->%s (%s)", tostring(session.from_host), tostring(session.to_host), tostring(err or "closed")); - s2s_destroy_session(session, err); -end - function stream_callbacks.error(session, error, data) if error == "no-stream" then session:close("invalid-namespace"); @@ -374,11 +364,26 @@ end end session.sends2s("</stream:stream>"); - if session.notopen or not session.conn:close() then - session.conn:close(true); -- Force FIXME: timer? + + function session.sends2s() return false; end + + local reason = remote_reason or (reason and (reason.text or reason.condition)) or reason or "stream closed"; + session.log("info", "%s s2s stream %s->%s closed: %s", session.direction, session.from_host or "(unknown host)", session.to_host or "(unknown host)", reason); + + -- Authenticated incoming stream may still be sending us stanzas, so wait for </stream:stream> from remote + local conn = session.conn; + if not session.notopen and session.type == "s2sin" then + add_task(stream_close_timeout, function () + if not session.destroyed then + session.log("warn", "Failed to receive a stream close response, closing connection anyway..."); + s2s_destroy_session(session, reason); + conn:close(); + end + end); + else + s2s_destroy_session(session, reason); + conn:close(); -- Close immediately, as this is an outgoing connection or is not authed end - session.conn:close(); - listener.ondisconnect(session.conn, remote_reason or (reason and (reason.text or reason.condition)) or reason or "stream closed"); end end @@ -413,11 +418,9 @@ return handlestanza(session, stanza); end - local conn = session.conn; add_task(connect_timeout, function () - if session.conn ~= conn or session.connecting - or session.type == "s2sin" or session.type == "s2sout" then - return; -- Ok, we're connect[ed|ing] + if session.type == "s2sin" or session.type == "s2sout" then + return; -- Ok, we're connected end -- Not connected, need to close session and clean up (session.log or log)("debug", "Destroying incomplete session %s->%s due to inactivity", @@ -474,11 +477,17 @@ function listener.ondisconnect(conn, err) local session = sessions[conn]; if session then - if stream_callbacks.streamdisconnected(session, err) then - return; -- Connection lives, for now + if err and session.direction == "outgoing" and session.notopen then + (session.log or log)("debug", "s2s connection attempt failed: %s", err); + if s2sout.attempt_connection(session, err) then + (session.log or log)("debug", "...so we're going to try another target"); + return; -- Session lives for now + end end + (session.log or log)("debug", "s2s disconnected: %s->%s (%s)", tostring(session.from_host), tostring(session.to_host), tostring(err or "connection closed")); + s2s_destroy_session(session, err); + sessions[conn] = nil; end - sessions[conn] = nil; end function listener.register_outgoing(conn, session)
--- a/plugins/mod_s2s/s2sout.lib.lua Sun Jul 22 16:17:33 2012 +0100 +++ b/plugins/mod_s2s/s2sout.lib.lua Sun Jul 22 17:08:09 2012 +0100 @@ -173,85 +173,85 @@ local have_other_result = not(has_ipv4) or not(has_ipv6) or false; if has_ipv4 then - handle4 = adns.lookup(function (reply, err) - handle4 = nil; - - -- COMPAT: This is a compromise for all you CNAME-(ab)users :) - if not (reply and reply[#reply] and reply[#reply].a) then - local count = max_dns_depth; - reply = dns.peek(connect_host, "CNAME", "IN"); - while count > 0 and reply and reply[#reply] and not reply[#reply].a and reply[#reply].cname do - log("debug", "Looking up %s (DNS depth is %d)", tostring(reply[#reply].cname), count); - reply = dns.peek(reply[#reply].cname, "A", "IN") or dns.peek(reply[#reply].cname, "CNAME", "IN"); - count = count - 1; - end - end - -- end of CNAME resolving + handle4 = adns.lookup(function (reply, err) + handle4 = nil; - if reply and reply[#reply] and reply[#reply].a then - for _, ip in ipairs(reply) do - log("debug", "DNS reply for %s gives us %s", connect_host, ip.a); - IPs[#IPs+1] = new_ip(ip.a, "IPv4"); + -- COMPAT: This is a compromise for all you CNAME-(ab)users :) + if not (reply and reply[#reply] and reply[#reply].a) then + local count = max_dns_depth; + reply = dns.peek(connect_host, "CNAME", "IN"); + while count > 0 and reply and reply[#reply] and not reply[#reply].a and reply[#reply].cname do + log("debug", "Looking up %s (DNS depth is %d)", tostring(reply[#reply].cname), count); + reply = dns.peek(reply[#reply].cname, "A", "IN") or dns.peek(reply[#reply].cname, "CNAME", "IN"); + count = count - 1; + end end - end + -- end of CNAME resolving - if have_other_result then - if #IPs > 0 then - rfc3484_dest(host_session.ip_hosts, sources); - for i = 1, #IPs do - IPs[i] = {ip = IPs[i], port = connect_port}; - end - host_session.ip_choice = 0; - s2sout.try_next_ip(host_session); - else - log("debug", "DNS lookup failed to get a response for %s", connect_host); - host_session.ip_hosts = nil; - if not s2sout.attempt_connection(host_session, "name resolution failed") then -- Retry if we can - log("debug", "No other records to try for %s - destroying", host_session.to_host); - err = err and (": "..err) or ""; - s2s_destroy_session(host_session, "DNS resolution failed"..err); -- End of the line, we can't + if reply and reply[#reply] and reply[#reply].a then + for _, ip in ipairs(reply) do + log("debug", "DNS reply for %s gives us %s", connect_host, ip.a); + IPs[#IPs+1] = new_ip(ip.a, "IPv4"); end end - else - have_other_result = true; - end - end, connect_host, "A", "IN"); + + if have_other_result then + if #IPs > 0 then + rfc3484_dest(host_session.ip_hosts, sources); + for i = 1, #IPs do + IPs[i] = {ip = IPs[i], port = connect_port}; + end + host_session.ip_choice = 0; + s2sout.try_next_ip(host_session); + else + log("debug", "DNS lookup failed to get a response for %s", connect_host); + host_session.ip_hosts = nil; + if not s2sout.attempt_connection(host_session, "name resolution failed") then -- Retry if we can + log("debug", "No other records to try for %s - destroying", host_session.to_host); + err = err and (": "..err) or ""; + s2s_destroy_session(host_session, "DNS resolution failed"..err); -- End of the line, we can't + end + end + else + have_other_result = true; + end + end, connect_host, "A", "IN"); else have_other_result = true; end if has_ipv6 then - handle6 = adns.lookup(function (reply, err) - handle6 = nil; - - if reply and reply[#reply] and reply[#reply].aaaa then - for _, ip in ipairs(reply) do - log("debug", "DNS reply for %s gives us %s", connect_host, ip.aaaa); - IPs[#IPs+1] = new_ip(ip.aaaa, "IPv6"); - end - end + handle6 = adns.lookup(function (reply, err) + handle6 = nil; - if have_other_result then - if #IPs > 0 then - rfc3484_dest(host_session.ip_hosts, sources); - for i = 1, #IPs do - IPs[i] = {ip = IPs[i], port = connect_port}; - end - host_session.ip_choice = 0; - s2sout.try_next_ip(host_session); - else - log("debug", "DNS lookup failed to get a response for %s", connect_host); - host_session.ip_hosts = nil; - if not s2sout.attempt_connection(host_session, "name resolution failed") then -- Retry if we can - log("debug", "No other records to try for %s - destroying", host_session.to_host); - err = err and (": "..err) or ""; - s2s_destroy_session(host_session, "DNS resolution failed"..err); -- End of the line, we can't + if reply and reply[#reply] and reply[#reply].aaaa then + for _, ip in ipairs(reply) do + log("debug", "DNS reply for %s gives us %s", connect_host, ip.aaaa); + IPs[#IPs+1] = new_ip(ip.aaaa, "IPv6"); end end - else - have_other_result = true; - end - end, connect_host, "AAAA", "IN"); + + if have_other_result then + if #IPs > 0 then + rfc3484_dest(host_session.ip_hosts, sources); + for i = 1, #IPs do + IPs[i] = {ip = IPs[i], port = connect_port}; + end + host_session.ip_choice = 0; + s2sout.try_next_ip(host_session); + else + log("debug", "DNS lookup failed to get a response for %s", connect_host); + host_session.ip_hosts = nil; + if not s2sout.attempt_connection(host_session, "name resolution failed") then -- Retry if we can + log("debug", "No other records to try for %s - destroying", host_session.to_host); + err = err and (": "..err) or ""; + s2s_destroy_session(host_session, "DNS resolution failed"..err); -- End of the line, we can't + end + end + else + have_other_result = true; + end + end, connect_host, "AAAA", "IN"); else have_other_result = true; end
