aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorMatthew Wild <mwild1@gmail.com>2012-07-22 17:08:09 +0100
committerMatthew Wild <mwild1@gmail.com>2012-07-22 17:08:09 +0100
commit8a965123f5c38071e14ee2f43fd203706ab56f95 (patch)
tree2593981ba2580f7c824501dc13223411d34fab49
parent92d6725339ce0233dffd2ecde606d684d05d4c93 (diff)
parent6aa575f09b858fef93101afbc7c776f138aca380 (diff)
downloadprosody-8a965123f5c38071e14ee2f43fd203706ab56f95.tar.gz
prosody-8a965123f5c38071e14ee2f43fd203706ab56f95.zip
Merge 0.9->trunk
-rw-r--r--core/sessionmanager.lua2
-rw-r--r--net/http.lua1
-rw-r--r--net/server_event.lua16
-rw-r--r--net/server_select.lua42
-rw-r--r--plugins/mod_c2s.lua27
-rw-r--r--plugins/mod_iq.lua5
-rw-r--r--plugins/mod_message.lua11
-rw-r--r--plugins/mod_presence.lua10
-rw-r--r--plugins/mod_s2s/mod_s2s.lua61
-rw-r--r--plugins/mod_s2s/s2sout.lib.lua120
10 files changed, 164 insertions, 131 deletions
diff --git a/core/sessionmanager.lua b/core/sessionmanager.lua
index 37c1626a..131c29f7 100644
--- a/core/sessionmanager.lua
+++ b/core/sessionmanager.lua
@@ -82,7 +82,7 @@ function retire_session(session)
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
diff --git a/net/http.lua b/net/http.lua
index f2061e00..02f3a265 100644
--- a/net/http.lua
+++ b/net/http.lua
@@ -203,7 +203,6 @@ function destroy_request(request)
if request.conn then
request.conn = nil;
request.handler:close()
- listener.ondisconnect(request.handler, "closed");
end
end
diff --git a/net/server_event.lua b/net/server_event.lua
index 03a7708c..3c4185af 100644
--- a/net/server_event.lua
+++ b/net/server_event.lua
@@ -249,7 +249,7 @@ do
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 @@ do
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
- self:_lock( true, true, true )
- self:_close()
- return true
- else -- wait for incomplete write request
+ 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
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
diff --git a/net/server_select.lua b/net/server_select.lua
index de637f70..c0f8742e 100644
--- a/net/server_select.lua
+++ b/net/server_select.lua
@@ -314,22 +314,28 @@ wrapconnection = function( server, listeners, socket, ip, serverport, clientport
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 @@ wrapconnection = function( server, listeners, socket, ip, serverport, clientport
local _handler = handler;
handler = nil
if disconnect then
- disconnect(_handler, "closed");
+ disconnect(_handler, false);
end
end
if server then
@@ -480,7 +486,7 @@ wrapconnection = function( server, listeners, socket, ip, serverport, clientport
_ = _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 @@ wrapconnection = function( server, listeners, socket, ip, serverport, clientport
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 @@ wrapconnection = function( server, listeners, socket, ip, serverport, clientport
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 @@ wrapconnection = function( server, listeners, socket, ip, serverport, clientport
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 @@ loop = function(once) -- this is the main loop of the program
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 @@ addtimer( function( )
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
diff --git a/plugins/mod_c2s.lua b/plugins/mod_c2s.lua
index 55c53e2d..75a6f689 100644
--- a/plugins/mod_c2s.lua
+++ b/plugins/mod_c2s.lua
@@ -24,6 +24,7 @@ local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams";
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 @@ local function session_close(session, reason)
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 @@ end
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
diff --git a/plugins/mod_iq.lua b/plugins/mod_iq.lua
index 6412ad11..8044a533 100644
--- a/plugins/mod_iq.lua
+++ b/plugins/mod_iq.lua
@@ -17,10 +17,7 @@ if module:get_host_type() == "local" then
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
diff --git a/plugins/mod_message.lua b/plugins/mod_message.lua
index ebff2fe7..0b0ad8e4 100644
--- a/plugins/mod_message.lua
+++ b/plugins/mod_message.lua
@@ -35,10 +35,13 @@ local function process_to_bare(bare, origin, stanza)
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
+ if sent then
+ return true;
end
- return true;
end
end
-- no resources are online
@@ -65,9 +68,7 @@ module:hook("message/full", function(data)
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);
diff --git a/plugins/mod_presence.lua b/plugins/mod_presence.lua
index 6d039d83..09a6f9f2 100644
--- a/plugins/mod_presence.lua
+++ b/plugins/mod_presence.lua
@@ -352,13 +352,15 @@ module:hook("resource-unbind", function(event)
-- 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);
diff --git a/plugins/mod_s2s/mod_s2s.lua b/plugins/mod_s2s/mod_s2s.lua
index f6c20606..f686fcfb 100644
--- a/plugins/mod_s2s/mod_s2s.lua
+++ b/plugins/mod_s2s/mod_s2s.lua
@@ -31,6 +31,7 @@ local cert_verify_identity = require "util.x509".verify_identity;
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 @@ function route_to_existing_session(event)
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 @@ function stream_callbacks.streamclosed(session)
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 @@ local function session_close(session, reason, remote_reason)
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 @@ local function initialize_session(session)
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 @@ end
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)
diff --git a/plugins/mod_s2s/s2sout.lib.lua b/plugins/mod_s2s/s2sout.lib.lua
index 6e1eb3d7..d2c6023e 100644
--- a/plugins/mod_s2s/s2sout.lib.lua
+++ b/plugins/mod_s2s/s2sout.lib.lua
@@ -173,85 +173,85 @@ function s2sout.try_connect(host_session, connect_host, connect_port, err)
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;
+ 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;
+ -- 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
+ -- end of CNAME resolving
- 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");
+ 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
- end
- 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};
+ 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
- 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
+ have_other_result = true;
end
- else
- have_other_result = true;
- end
- end, connect_host, "A", "IN");
+ end, connect_host, "A", "IN");
else
have_other_result = true;
end
if has_ipv6 then
- handle6 = adns.lookup(function (reply, err)
- handle6 = nil;
+ 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");
+ 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
- end
- 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};
+ 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
- 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
+ have_other_result = true;
end
- else
- have_other_result = true;
- end
- end, connect_host, "AAAA", "IN");
+ end, connect_host, "AAAA", "IN");
else
have_other_result = true;
end