1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
-- Copyright (C) 2016-2018 Kim Alvefur
--
-- This project is MIT/X11 licensed. Please see the
-- COPYING file in the source package for more information.
--
module:depends"csi"
local jid = require "util.jid";
local st = require "util.stanza";
local dt = require "util.datetime";
local filters = require "util.filters";
local queue_size = module:get_option_number("csi_queue_size", 256);
module:hook("csi-is-stanza-important", function (event)
local stanza = event.stanza;
if not st.is_stanza(stanza) then
return true;
end
local st_name = stanza.name;
if not st_name then return false; end
local st_type = stanza.attr.type;
if st_name == "presence" then
if st_type == nil or st_type == "unavailable" then
return false;
end
return true;
elseif st_name == "message" then
if st_type == "headline" then
return false;
end
if stanza:get_child("sent", "urn:xmpp:carbons:2") then
return true;
end
local forwarded = stanza:find("{urn:xmpp:carbons:2}received/{urn:xmpp:forward:0}/{jabber:client}message");
if forwarded then
stanza = forwarded;
end
if stanza:get_child("body") then
return true;
end
if stanza:get_child("subject") then
return true;
end
if stanza:get_child("encryption", "urn:xmpp:eme:0") then
return true;
end
return false;
end
return true;
end, -1);
local function with_timestamp(stanza, from)
if st.is_stanza(stanza) and stanza.attr.xmlns == nil and stanza.name ~= "iq" then
stanza = st.clone(stanza);
stanza:add_direct_child(st.stanza("delay", {xmlns = "urn:xmpp:delay", from = from, stamp = dt.datetime()}));
end
return stanza;
end
local function manage_buffer(stanza, session)
local ctr = session.csi_counter or 0;
if ctr >= queue_size or module:fire_event("csi-is-stanza-important", { stanza = stanza, session = session }) then
session.conn:resume_writes();
else
stanza = with_timestamp(stanza, jid.join(session.username, session.host))
end
session.csi_counter = ctr + 1;
return stanza;
end
local function flush_buffer(data, session)
session.log("debug", "Client sent something, flushing buffer once");
session.conn:resume_writes();
return data;
end
function enable_optimizations(session)
if session.conn and session.conn and session.conn.pause_writes then
session.conn:pause_writes();
filters.add_filter(session, "stanzas/out", manage_buffer);
filters.add_filter(session, "bytes/in", flush_buffer);
else
session.log("warn", "Session connection does not support write pausing");
end
end
function disble_optimizations(session)
if session.conn and session.conn and session.conn.resume_writes then
filters.remove_filter(session, "stanzas/out", manage_buffer);
filters.remove_filter(session, "bytes/in", flush_buffer);
session.conn:resume_writes();
end
end
module:hook("csi-client-inactive", function (event)
local session = event.origin;
enable_optimizations(session);
end);
module:hook("csi-client-active", function (event)
local session = event.origin;
disble_optimizations(session);
end);
module:hook("c2s-ondrain", function (event)
local session = event.session;
if session.state == "inactive" and session.conn and session.conn and session.conn.pause_writes then
session.csi_counter = 0;
session.conn:pause_writes();
session.log("debug", "Buffer flushed, resuming inactive mode");
end
end);
function module.load()
for _, user_session in pairs(prosody.hosts[module.host].sessions) do
for _, session in pairs(user_session.sessions) do
if session.state == "inactive" then
enable_optimizations(session);
end
end
end
end
function module.unload()
for _, user_session in pairs(prosody.hosts[module.host].sessions) do
for _, session in pairs(user_session.sessions) do
if session.state == "inactive" then
disble_optimizations(session);
end
end
end
end
|
