Software /
code /
prosody
File
plugins/mod_csi_simple.lua @ 10682:62ef68f95b6f 0.11
mod_mam,mod_muc_mam: Allow other work to be performed during archive cleanup (fixes #1504)
This lets Prosody handle socket related work between each step in the
cleanup in order to prevent the server from being completely blocked
during this.
An async storage backend would not need this but those are currently
rare.
author | Kim Alvefur <zash@zash.se> |
---|---|
date | Wed, 11 Mar 2020 21:15:01 +0100 |
parent | 9771:bf92f37de137 |
child | 9909:3229be01a08a |
child | 10816:b4c2c7fae7c0 |
line wrap: on
line source
-- 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 new_queue = require "util.queue".new; local function new_pump(output, ...) -- luacheck: ignore 212/self local q = new_queue(...); local flush = true; function q:pause() flush = false; end function q:resume() flush = true; return q:flush(); end local push = q.push; function q:push(item) local ok = push(self, item); if not ok then q:flush(); output(item, self); elseif flush then return q:flush(); end return true; end function q:flush() local item = self:pop(); while item do output(item, self); item = self:pop(); end return true; end return q; end 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); module:hook("csi-client-inactive", function (event) local session = event.origin; if session.pump then session.pump:pause(); else local bare_jid = jid.join(session.username, session.host); local send = session.send; session._orig_send = send; local pump = new_pump(session.send, queue_size); pump:pause(); session.pump = pump; function session.send(stanza) if session.state == "active" or module:fire_event("csi-is-stanza-important", { stanza = stanza, session = session }) then pump:flush(); send(stanza); else 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 = bare_jid, stamp = dt.datetime()})); end pump:push(stanza); end return true; end end end); module:hook("csi-client-active", function (event) local session = event.origin; if session.pump then session.pump:resume(); end end);