Mercurial > prosody-hg

comparison plugins/mod_s2s/mod_s2s.lua @ 4970:eeff01224865

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Merge 0.9->trunk
author Matthew Wild <mwild1@gmail.com>
date Sun, 22 Jul 2012 17:08:09 +0100
parents 15183193c6a6
children 29bdf68ad142
comparison
equal deleted inserted replaced
4958:1f1c74ac9666 4970:eeff01224865
29 local cert_verify_identity = require "util.x509".verify_identity; 29 local cert_verify_identity = require "util.x509".verify_identity;
30 30
31 local s2sout = module:require("s2sout"); 31 local s2sout = module:require("s2sout");
32 32
33 local connect_timeout = module:get_option_number("s2s_timeout", 60); 33 local connect_timeout = module:get_option_number("s2s_timeout", 60);
34 local stream_close_timeout = module:get_option_number("s2s_close_timeout", 5);
34 35
35 local sessions = module:shared("sessions"); 36 local sessions = module:shared("sessions");
36 37
37 local log = module._log; 38 local log = module._log;
38 39
95 -- FIXME 96 -- FIXME
96 if host.from_host ~= from_host then 97 if host.from_host ~= from_host then
97 log("error", "WARNING! This might, possibly, be a bug, but it might not..."); 98 log("error", "WARNING! This might, possibly, be a bug, but it might not...");
98 log("error", "We are going to send from %s instead of %s", tostring(host.from_host), tostring(from_host)); 99 log("error", "We are going to send from %s instead of %s", tostring(host.from_host), tostring(from_host));
99 end 100 end
100 host.sends2s(stanza); 101 if host.sends2s(stanza) then
101 host.log("debug", "stanza sent over "..host.type); 102 host.log("debug", "stanza sent over "..host.type);
102 return true; 103 return true;
104 end
103 end 105 end
104 end 106 end
105 end 107 end
106 108
107 -- Create a new outgoing session for a stanza 109 -- Create a new outgoing session for a stanza
287 end 289 end
288 290
289 function stream_callbacks.streamclosed(session) 291 function stream_callbacks.streamclosed(session)
290 (session.log or log)("debug", "Received </stream:stream>"); 292 (session.log or log)("debug", "Received </stream:stream>");
291 session:close(); 293 session:close();
292 end
293
294 function stream_callbacks.streamdisconnected(session, err)
295 if err and err ~= "closed" and session.direction == "outgoing" and session.notopen then
296 (session.log or log)("debug", "s2s connection attempt failed: %s", err);
297 if s2sout.attempt_connection(session, err) then
298 (session.log or log)("debug", "...so we're going to try another target");
299 return true; -- Session lives for now
300 end
301 end
302 (session.log or log)("info", "s2s disconnected: %s->%s (%s)", tostring(session.from_host), tostring(session.to_host), tostring(err or "closed"));
303 s2s_destroy_session(session, err);
304 end 294 end
305 295
306 function stream_callbacks.error(session, error, data) 296 function stream_callbacks.error(session, error, data)
307 if error == "no-stream" then 297 if error == "no-stream" then
308 session:close("invalid-namespace"); 298 session:close("invalid-namespace");
372 session.sends2s(reason); 362 session.sends2s(reason);
373 end 363 end
374 end 364 end
375 end 365 end
376 session.sends2s("</stream:stream>"); 366 session.sends2s("</stream:stream>");
377 if session.notopen or not session.conn:close() then 367
378 session.conn:close(true); -- Force FIXME: timer? 368 function session.sends2s() return false; end
379 end 369
380 session.conn:close(); 370 local reason = remote_reason or (reason and (reason.text or reason.condition)) or reason or "stream closed";
381 listener.ondisconnect(session.conn, remote_reason or (reason and (reason.text or reason.condition)) or reason or "stream closed"); 371 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);
372
373 -- Authenticated incoming stream may still be sending us stanzas, so wait for </stream:stream> from remote
374 local conn = session.conn;
375 if not session.notopen and session.type == "s2sin" then
376 add_task(stream_close_timeout, function ()
377 if not session.destroyed then
378 session.log("warn", "Failed to receive a stream close response, closing connection anyway...");
379 s2s_destroy_session(session, reason);
380 conn:close();
381 end
382 end);
383 else
384 s2s_destroy_session(session, reason);
385 conn:close(); -- Close immediately, as this is an outgoing connection or is not authed
386 end
382 end 387 end
383 end 388 end
384 389
385 -- Session initialization logic shared by incoming and outgoing 390 -- Session initialization logic shared by incoming and outgoing
386 local function initialize_session(session) 391 local function initialize_session(session)
411 local handlestanza = stream_callbacks.handlestanza; 416 local handlestanza = stream_callbacks.handlestanza;
412 function session.dispatch_stanza(session, stanza) 417 function session.dispatch_stanza(session, stanza)
413 return handlestanza(session, stanza); 418 return handlestanza(session, stanza);
414 end 419 end
415 420
416 local conn = session.conn;
417 add_task(connect_timeout, function () 421 add_task(connect_timeout, function ()
418 if session.conn ~= conn or session.connecting 422 if session.type == "s2sin" or session.type == "s2sout" then
419 or session.type == "s2sin" or session.type == "s2sout" then 423 return; -- Ok, we're connected
420 return; -- Ok, we're connect[ed|ing]
421 end 424 end
422 -- Not connected, need to close session and clean up 425 -- Not connected, need to close session and clean up
423 (session.log or log)("debug", "Destroying incomplete session %s->%s due to inactivity", 426 (session.log or log)("debug", "Destroying incomplete session %s->%s due to inactivity",
424 session.from_host or "(unknown)", session.to_host or "(unknown)"); 427 session.from_host or "(unknown)", session.to_host or "(unknown)");
425 session:close("connection-timeout"); 428 session:close("connection-timeout");
472 end 475 end
473 476
474 function listener.ondisconnect(conn, err) 477 function listener.ondisconnect(conn, err)
475 local session = sessions[conn]; 478 local session = sessions[conn];
476 if session then 479 if session then
477 if stream_callbacks.streamdisconnected(session, err) then 480 if err and session.direction == "outgoing" and session.notopen then
478 return; -- Connection lives, for now 481 (session.log or log)("debug", "s2s connection attempt failed: %s", err);
479 end 482 if s2sout.attempt_connection(session, err) then
480 end 483 (session.log or log)("debug", "...so we're going to try another target");
481 sessions[conn] = nil; 484 return; -- Session lives for now
485 end
486 end
487 (session.log or log)("debug", "s2s disconnected: %s->%s (%s)", tostring(session.from_host), tostring(session.to_host), tostring(err or "connection closed"));
488 s2s_destroy_session(session, err);
489 sessions[conn] = nil;
490 end
482 end 491 end
483 492
484 function listener.register_outgoing(conn, session) 493 function listener.register_outgoing(conn, session)
485 session.direction = "outgoing"; 494 session.direction = "outgoing";
486 sessions[conn] = session; 495 sessions[conn] = session;