Merge 0.9->trunk

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