File

mod_compression_unsafe/mod_compression_unsafe.lua @ 5571:ca3c2d11823c

mod_pubsub_feeds: Track latest timestamp seen in feeds instead of last poll This should ensure that an entry that has a publish timestmap after the previously oldest post, but before the time of the last poll check, is published to the node. Previously if an entry would be skipped if it was published at 13:00 with a timestamp of 12:30, where the last poll was at 12:45. For feeds that lack a timestamp, it now looks for the first post that is not published, assuming that the feed is in reverse chronological order, then iterates back up from there.
author Kim Alvefur <zash@zash.se>
date Sun, 25 Jun 2023 16:27:55 +0200
parent 2795:226693a22fc9
line wrap: on
line source

-- Prosody IM
-- Copyright (C) 2009-2012 Tobias Markmann
--
-- This project is MIT/X11 licensed. Please see the
-- COPYING file in the source package for more information.
--

local st = require "util.stanza";
local zlib = require "zlib";
local pcall = pcall;
local tostring = tostring;

local xmlns_compression_feature = "http://jabber.org/features/compress"
local xmlns_compression_protocol = "http://jabber.org/protocol/compress"
local xmlns_stream = "http://etherx.jabber.org/streams";
local compression_stream_feature = st.stanza("compression", {xmlns=xmlns_compression_feature}):tag("method"):text("zlib"):up();
local add_filter = require "util.filters".add_filter;

local compression_level = module:get_option_number("compression_level", 7);

if not compression_level or compression_level < 1 or compression_level > 9 then
	module:log("warn", "Invalid compression level in config: %s", tostring(compression_level));
	module:log("warn", "Module loading aborted. Compression won't be available.");
	return;
end

module:hook("stream-features", function(event)
	local origin, features = event.origin, event.features;
	if not origin.compressed and (origin.type == "c2s" or origin.type == "c2s_unbound") then
		features:add_child(compression_stream_feature);
	end
end);

module:hook("s2s-stream-features", function(event)
	local origin, features = event.origin, event.features;
	if not origin.compressed and origin.type == "s2sin" then
		features:add_child(compression_stream_feature);
	end
end);

-- Hook to activate compression if remote server supports it.
module:hook_stanza(xmlns_stream, "features",
		function (session, stanza)
			if not session.compressed and session.type == "s2sout" then
				-- does remote server support compression?
				local comp_st = stanza:get_child("compression", xmlns_compression_feature);
				if comp_st then
					-- do we support the mechanism
					for a in comp_st:childtags("method") do
						local algorithm = a:get_text();
						if algorithm == "zlib" then
							session.sends2s(st.stanza("compress", {xmlns=xmlns_compression_protocol}):tag("method"):text("zlib"))
							session.log("debug", "Enabled compression using zlib.")
							return true;
						end
					end
					session.log("debug", "Remote server supports no compression algorithm we support.")
				end
			end
		end
, 250);


-- returns either nil or a fully functional ready to use inflate stream
local function get_deflate_stream(session)
	local status, deflate_stream = pcall(zlib.deflate, compression_level);
	if status == false then
		local error_st = st.stanza("failure", {xmlns=xmlns_compression_protocol}):tag("setup-failed");
		(session.sends2s or session.send)(error_st);
		session.log("error", "Failed to create zlib.deflate filter.");
		module:log("error", "%s", tostring(deflate_stream));
		return
	end
	return deflate_stream
end

-- returns either nil or a fully functional ready to use inflate stream
local function get_inflate_stream(session)
	local status, inflate_stream = pcall(zlib.inflate);
	if status == false then
		local error_st = st.stanza("failure", {xmlns=xmlns_compression_protocol}):tag("setup-failed");
		(session.sends2s or session.send)(error_st);
		session.log("error", "Failed to create zlib.inflate filter.");
		module:log("error", "%s", tostring(inflate_stream));
		return
	end
	return inflate_stream
end

-- setup compression for a stream
local function setup_compression(session, deflate_stream)
	add_filter(session, "bytes/out", function(t)
		local status, compressed, eof = pcall(deflate_stream, tostring(t), 'sync');
		if status == false then
			module:log("warn", "%s", tostring(compressed));
			session:close({
				condition = "undefined-condition";
				text = compressed;
				extra = st.stanza("failure", {xmlns="http://jabber.org/protocol/compress"}):tag("processing-failed");
			});
			return;
		end
		return compressed;
	end);
end

-- setup decompression for a stream
local function setup_decompression(session, inflate_stream)
	add_filter(session, "bytes/in", function(data)
		local status, decompressed, eof = pcall(inflate_stream, data);
		if status == false then
			module:log("warn", "%s", tostring(decompressed));
			session:close({
				condition = "undefined-condition";
				text = decompressed;
				extra = st.stanza("failure", {xmlns="http://jabber.org/protocol/compress"}):tag("processing-failed");
			});
			return;
		end
		return decompressed;
	end);
end

module:hook("stanza/http://jabber.org/protocol/compress:compressed", function(event)
	local session = event.origin;
	
	if session.type == "s2sout" then
		session.log("debug", "Activating compression...")
		-- create deflate and inflate streams
		local deflate_stream = get_deflate_stream(session);
		if not deflate_stream then return true; end
		
		local inflate_stream = get_inflate_stream(session);
		if not inflate_stream then return true; end
		
		-- setup compression for session.w
		setup_compression(session, deflate_stream);
			
		-- setup decompression for session.data
		setup_decompression(session, inflate_stream);
		session:reset_stream();
		session:open_stream(session.from_host, session.to_host);
		session.compressed = true;
		return true;
	end
end);

module:hook("stanza/http://jabber.org/protocol/compress:failure", function(event)
	local err = event.stanza:get_child();
	(event.origin.log or module._log)("warn", "Compression setup failed (%s)", err and err.name or "unknown reason");
	return true;
end);

module:hook("stanza/http://jabber.org/protocol/compress:compress", function(event)
	local session, stanza = event.origin, event.stanza;

	if session.type == "c2s" or session.type == "s2sin" then
		-- fail if we are already compressed
		if session.compressed then
			local error_st = st.stanza("failure", {xmlns=xmlns_compression_protocol}):tag("setup-failed");
			(session.sends2s or session.send)(error_st);
			session.log("debug", "Client tried to establish another compression layer.");
			return true;
		end
		
		-- checking if the compression method is supported
		local method = stanza:get_child_text("method");
		if method == "zlib" then
			session.log("debug", "zlib compression enabled.");
			
			-- create deflate and inflate streams
			local deflate_stream = get_deflate_stream(session);
			if not deflate_stream then return true; end
			
			local inflate_stream = get_inflate_stream(session);
			if not inflate_stream then return true; end
			
			(session.sends2s or session.send)(st.stanza("compressed", {xmlns=xmlns_compression_protocol}));
			session:reset_stream();
			
			-- setup compression for session.w
			setup_compression(session, deflate_stream);
				
			-- setup decompression for session.data
			setup_decompression(session, inflate_stream);
			
			session.compressed = true;
		elseif method then
			session.log("debug", "%s compression selected, but we don't support it.", tostring(method));
			local error_st = st.stanza("failure", {xmlns=xmlns_compression_protocol}):tag("unsupported-method");
			(session.sends2s or session.send)(error_st);
		else
			(session.sends2s or session.send)(st.stanza("failure", {xmlns=xmlns_compression_protocol}):tag("setup-failed"));
		end
		return true;
	end
end);