Mercurial > prosody-modules

comparison mod_smacks/mod_smacks.lua @ 603:efc9d88b70ab

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
merged.
author Marco Cirillo <maranda@lightwitch.org>
date Thu, 09 Feb 2012 00:54:39 +0000
parents 7693724881b3
children ce39df945de1
comparison
equal deleted inserted replaced
601:00590d492a5b 603:efc9d88b70ab
5 local math_min = math.min; 5 local math_min = math.min;
6 local os_time = os.time; 6 local os_time = os.time;
7 local tonumber, tostring = tonumber, tostring; 7 local tonumber, tostring = tonumber, tostring;
8 local add_filter = require "util.filters".add_filter; 8 local add_filter = require "util.filters".add_filter;
9 local timer = require "util.timer"; 9 local timer = require "util.timer";
10 local datetime = require "util.datetime";
10 11
11 local xmlns_sm = "urn:xmpp:sm:2"; 12 local xmlns_sm = "urn:xmpp:sm:2";
12 local xmlns_errors = "urn:ietf:params:xml:ns:xmpp-stanzas"; 13 local xmlns_errors = "urn:ietf:params:xml:ns:xmpp-stanzas";
14 local xmlns_delay = "urn:xmpp:delay";
13 15
14 local sm_attr = { xmlns = xmlns_sm }; 16 local sm_attr = { xmlns = xmlns_sm };
15 17
16 local resume_timeout = module:get_option("smacks_hibernation_time", 300); 18 local resume_timeout = module:get_option("smacks_hibernation_time", 300);
17 local s2s_smacks = module:get_option_boolean("smacks_enabled_s2s", false); 19 local s2s_smacks = module:get_option_boolean("smacks_enabled_s2s", false);
18 local max_unacked_stanzas = 0; 20 local max_unacked_stanzas = 0;
19 21
20 local session_registry = {}; 22 local session_registry = {};
21 23
24 local function can_do_smacks(session, advertise_only)
25 if session.smacks then return false, "unexpected-request", "Stream management is already enabled"; end
26
27 local session_type = session.type;
28 if session_type == "c2s" then
29 if not(advertise_only) and not(session.resource) then -- Fail unless we're only advertising sm
30 return false, "unexpected-request", "Client must bind a resource before enabling stream management";
31 end
32 return true;
33 elseif s2s_smacks and (session_type == "s2sin" or session_type == "s2sout") then
34 return true;
35 end
36 return false, "service-unavailable", "Stream management is not available for this stream";
37 end
38
22 module:hook("stream-features", 39 module:hook("stream-features",
23 function (event) 40 function (event)
24 event.features:tag("sm", sm_attr):tag("optional"):up():up(); 41 if can_do_smacks(event.origin, true) then
42 event.features:tag("sm", sm_attr):tag("optional"):up():up();
43 end
25 end); 44 end);
26 45
27 module:hook("s2s-stream-features", 46 module:hook("s2s-stream-features",
28 function (event) 47 function (event)
29 local origin = event.origin; 48 if can_do_smacks(event.origin, true) then
30 if s2s_smacks and (origin.type == "s2sin" or origin.type == "s2sout") then
31 event.features:tag("sm", sm_attr):tag("optional"):up():up(); 49 event.features:tag("sm", sm_attr):tag("optional"):up():up();
32 end 50 end
33 end); 51 end);
34 52
35 module:hook_stanza("http://etherx.jabber.org/streams", "features", 53 module:hook_stanza("http://etherx.jabber.org/streams", "features",
36 function (session, stanza) 54 function (session, stanza)
37 if s2s_smacks and (session.type == "s2sin" or session.type == "s2sout") 55 if can_do_smacks(session) and stanza:get_child("sm", xmlns_sm) then
38 and not session.smacks and stanza:get_child("sm", xmlns_sm) then
39 session.sends2s(st.stanza("enable", sm_attr)); 56 session.sends2s(st.stanza("enable", sm_attr));
40 end 57 end
41 end); 58 end);
42 59
43 local function wrap_session(session, resume) 60 local function wrap_session(session, resume)
53 70
54 local _send = session.sends2s or session.send; 71 local _send = session.sends2s or session.send;
55 local function new_send(stanza) 72 local function new_send(stanza)
56 local attr = stanza.attr; 73 local attr = stanza.attr;
57 if attr and not attr.xmlns then -- Stanza in default stream namespace 74 if attr and not attr.xmlns then -- Stanza in default stream namespace
58 queue[#queue+1] = st.clone(stanza); 75 local cached_stanza = st.clone(stanza);
76
77 if cached_stanza and cached_stanza:get_child("delay", xmlns_delay) == nil then
78 cached_stanza = cached_stanza:tag("delay", { xmlns = xmlns_delay, from = session.host, stamp = datetime.datetime()});
79 end
80
81 queue[#queue+1] = cached_stanza;
59 end 82 end
60 local ok, err = _send(stanza); 83 local ok, err = _send(stanza);
61 if ok and #queue > max_unacked_stanzas and not session.awaiting_ack then 84 if ok and #queue > max_unacked_stanzas and not session.awaiting_ack then
62 session.awaiting_ack = true; 85 session.awaiting_ack = true;
63 return _send(st.stanza("r", sm_attr)); 86 return _send(st.stanza("r", sm_attr));
84 107
85 return session; 108 return session;
86 end 109 end
87 110
88 module:hook_stanza(xmlns_sm, "enable", function (session, stanza) 111 module:hook_stanza(xmlns_sm, "enable", function (session, stanza)
112 local ok, err, err_text = can_do_smacks(session);
113 if not ok then
114 session.log("warn", "Failed to enable smacks: %s", err_text); -- TODO: XEP doesn't say we can send error text, should it?
115 session.send(st.stanza("failed", { xmlns = xmlns_sm }):tag(err, { xmlns = xmlns_errors}));
116 return true;
117 end
118
89 module:log("debug", "Enabling stream management"); 119 module:log("debug", "Enabling stream management");
90 session.smacks = true; 120 session.smacks = true;
91 121
92 wrap_session(session); 122 wrap_session(session);
93 123
179 module:log("warn", "::%s", tostring(queue[i])); 209 module:log("warn", "::%s", tostring(queue[i]));
180 end 210 end
181 handle_unacked_stanzas(session); 211 handle_unacked_stanzas(session);
182 end 212 end
183 else 213 else
214 session.log("debug", "mod_smacks hibernating session for up to %d seconds", resume_timeout);
184 local hibernate_time = os_time(); -- Track the time we went into hibernation 215 local hibernate_time = os_time(); -- Track the time we went into hibernation
185 session.hibernating = hibernate_time; 216 session.hibernating = hibernate_time;
186 local resumption_token = session.resumption_token; 217 local resumption_token = session.resumption_token;
187 timer.add_task(resume_timeout, function () 218 timer.add_task(resume_timeout, function ()
219 session.log("debug", "mod_smacks hibernation timeout reached...");
188 -- We need to check the current resumption token for this resource 220 -- We need to check the current resumption token for this resource
189 -- matches the smacks session this timer is for in case it changed 221 -- matches the smacks session this timer is for in case it changed
190 -- (for example, the client may have bound a new resource and 222 -- (for example, the client may have bound a new resource and
191 -- started a new smacks session, or not be using smacks) 223 -- started a new smacks session, or not be using smacks)
192 local curr_session = hosts[session.host].sessions[session.username].sessions[session.resource]; 224 local curr_session = hosts[session.host].sessions[session.username].sessions[session.resource];
193 if curr_session.resumption_token == resumption_token 225 if curr_session.resumption_token == resumption_token
194 -- Check the hibernate time still matches what we think it is, 226 -- Check the hibernate time still matches what we think it is,
195 -- otherwise the session resumed and re-hibernated. 227 -- otherwise the session resumed and re-hibernated.
196 and session.hibernating == hibernate_time then 228 and session.hibernating == hibernate_time then
229 session.log("debug", "Destroying session for hibernating too long");
197 session_registry[session.resumption_token] = nil; 230 session_registry[session.resumption_token] = nil;
198 session.resumption_token = nil; 231 session.resumption_token = nil;
199 -- This recursion back into our destroy handler is to 232 -- This recursion back into our destroy handler is to
200 -- make sure we still handle any queued stanzas 233 -- make sure we still handle any queued stanzas
201 sessionmanager.destroy_session(session); 234 sessionmanager.destroy_session(session);
235 else
236 session.log("debug", "Session resumed before hibernation timeout, all is well")
202 end 237 end
203 end); 238 end);
204 return; -- Postpone destruction for now 239 return; -- Postpone destruction for now
205 end 240 end
206 241
210 245
211 module:hook_stanza(xmlns_sm, "resume", function (session, stanza) 246 module:hook_stanza(xmlns_sm, "resume", function (session, stanza)
212 local id = stanza.attr.previd; 247 local id = stanza.attr.previd;
213 local original_session = session_registry[id]; 248 local original_session = session_registry[id];
214 if not original_session then 249 if not original_session then
250 session.log("debug", "Tried to resume non-existent session with id %s", id);
215 session.send(st.stanza("failed", sm_attr) 251 session.send(st.stanza("failed", sm_attr)
216 :tag("item-not-found", { xmlns = xmlns_errors }) 252 :tag("item-not-found", { xmlns = xmlns_errors })
217 ); 253 );
218 elseif session.username == original_session.username 254 elseif session.username == original_session.username
219 and session.host == original_session.host then 255 and session.host == original_session.host then
256 session.log("debug", "mod_smacks resuming existing session...");
220 -- TODO: All this should move to sessionmanager (e.g. session:replace(new_session)) 257 -- TODO: All this should move to sessionmanager (e.g. session:replace(new_session))
221 original_session.ip = session.ip; 258 original_session.ip = session.ip;
222 original_session.conn = session.conn; 259 original_session.conn = session.conn;
223 original_session.send = session.send; 260 original_session.send = session.send;
224 original_session.stream = session.stream; 261 original_session.stream = session.stream;