matty/peertube-plugin-livechat
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Chat Federation WIP:

Browse Source 
Trying to fix some non-working code. Still not working.
This commit is contained in:
John Livingston 2023-05-22 11:23:15 +02:00
parent ef3417dab0
commit cd3afe2b26
No known key found for this signature in database
GPG Key ID: B17B5640CE66CDBC
2 changed files with 110 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
prosody-modules/mod_s2s_peertubelivechat/mod_s2s_peertubelivechat.lua
View File

@ -140,7 +140,7 @@ local instance_url = assert(module:get_option_string("peertubelivechat_instance_
function discover_websocket_s2s(event) function discover_websocket_s2s(event)
local to_host = event.to_host; local to_host = event.to_host;
module:log("debug", "Trying to route to %s", to_host); module:log("debug", "Searching websocket s2s for remote host %s", to_host);
local f_s2s = io.open(path.join(server_infos_dir, to_host, 's2s'), "r"); local f_s2s = io.open(path.join(server_infos_dir, to_host, 's2s'), "r");
if f_s2s ~= nil then if f_s2s ~= nil then

168
prosody-modules/mod_websocket_s2s_peertubelivechat/mod_websocket_s2s_peertubelivechat.lua
View File

@ -47,6 +47,8 @@ local s2s_listener = portmanager.get_service("s2s").listener;
--- Session methods --- Session methods
local function session_open_stream(session, from, to) local function session_open_stream(session, from, to)
local log = session.log or log;
log("debug", "Opening stream from %s to %to.", from, to);
local attr = { local attr = {
xmlns = xmlns_framing, xmlns = xmlns_framing,
["xml:lang"] = "en", ["xml:lang"] = "en",
@ -203,12 +205,16 @@ local function validate_frame(frame, max_length)
end end
local function wrap_websocket(session, conn) local function wrap_websocket(session, conn)
local log = session.log or log;
local function websocket_close(code, message) local function websocket_close(code, message)
log("debug", "Websocket close, writing a build_close frame");
conn:write(build_close(code, message)); conn:write(build_close(code, message));
conn:close(); conn:close();
end end
local function websocket_handle_error(session, code, message) local function websocket_handle_error(session, code, message)
log("debug", "handling an error");
if code == 1009 then -- stanza size limit exceeded if code == 1009 then -- stanza size limit exceeded
-- we close the session, rather than the connection, -- we close the session, rather than the connection,
-- otherwise a resuming client will simply resend the -- otherwise a resuming client will simply resend the
@ -220,7 +226,7 @@ local function wrap_websocket(session, conn)
end end
local function handle_frame(frame) local function handle_frame(frame)
module:log("debug", "Websocket received frame: opcode=%0x, %i bytes", frame.opcode, #frame.data); log("debug", "Websocket received frame: opcode=%0x, %i bytes", frame.opcode, #frame.data);
-- Check frame makes sense -- Check frame makes sense
local frame_ok, err_status, err_text = validate_frame(frame, stanza_size_limit); local frame_ok, err_status, err_text = validate_frame(frame, stanza_size_limit);
@ -246,8 +252,10 @@ local function wrap_websocket(session, conn)
local frameBuffer = dbuffer.new(frame_buffer_limit, frame_fragment_limit); local frameBuffer = dbuffer.new(frame_buffer_limit, frame_fragment_limit);
add_filter(session, "bytes/in", function(data) add_filter(session, "bytes/in", function(data)
log("debug", "Calling the bytes/in filter");
if not frameBuffer:write(data) then if not frameBuffer:write(data) then
session.log("warn", "websocket frame buffer full - terminating session"); log("warn", "websocket frame buffer full - terminating session");
session:close({ condition = "resource-constraint", text = "frame buffer exceeded" }); session:close({ condition = "resource-constraint", text = "frame buffer exceeded" });
return; return;
end end
@ -279,6 +287,8 @@ local function wrap_websocket(session, conn)
end); end);
add_filter(session, "stanzas/out", function(stanza) add_filter(session, "stanzas/out", function(stanza)
log("debug", "Calling the stanzas/out filter");
stanza = st.clone(stanza); stanza = st.clone(stanza);
local attr = stanza.attr; local attr = stanza.attr;
attr.xmlns = attr.xmlns or xmlns_client; attr.xmlns = attr.xmlns or xmlns_client;
@ -289,6 +299,7 @@ local function wrap_websocket(session, conn)
end, -1000); end, -1000);
add_filter(session, "bytes/out", function(data) add_filter(session, "bytes/out", function(data)
log("debug", "Calling the bytes/out filter");
return build_frame({ FIN = true, opcode = 0x01, data = tostring(data)}); return build_frame({ FIN = true, opcode = 0x01, data = tostring(data)});
end); end);
end end
@ -303,7 +314,7 @@ function handle_request(event)
return module:fire_event("http-message", { return module:fire_event("http-message", {
response = event.response; response = event.response;
--- ---
title = "Prosody WebSocket endpoint"; title = "Prosody S2s WebSocket endpoint";
message = websocket_get_response_text; message = websocket_get_response_text;
warning = not (consider_websocket_secure or request.secure) and "This endpoint is not considered secure!" or nil; warning = not (consider_websocket_secure or request.secure) and "This endpoint is not considered secure!" or nil;
}) or websocket_get_response_body; }) or websocket_get_response_body;
@ -312,7 +323,7 @@ function handle_request(event)
local wants_xmpp = contains_token(request.headers.sec_websocket_protocol or "", "xmpp"); local wants_xmpp = contains_token(request.headers.sec_websocket_protocol or "", "xmpp");
if not wants_xmpp then if not wants_xmpp then
module:log("debug", "Client didn't want to talk XMPP, list of protocols was %s", request.headers.sec_websocket_protocol or "(empty)"); log("debug", "Client didn't want to talk XMPP, list of protocols was %s", request.headers.sec_websocket_protocol or "(empty)");
return 501; return 501;
end end
@ -341,7 +352,7 @@ function handle_request(event)
module:fire_event("websocket-session", { session = session, request = request }); module:fire_event("websocket-session", { session = session, request = request });
session.log("debug", "Sending WebSocket handshake"); log("debug", "Sending WebSocket handshake");
return ""; return "";
end end
@ -356,68 +367,91 @@ end
-- OUTGOING CONNECTIONS -- OUTGOING CONNECTIONS
local pending_ws_connection_methods = {}; -- local pending_ws_connection_methods = {};
local pending_ws_connection_mt = { -- local pending_ws_connection_mt = {
__name = "pending_ws_connection"; -- __name = "pending_ws_connection";
__index = pending_ws_connection_methods; -- __index = pending_ws_connection_methods;
__tostring = function (p) -- __tostring = function (p)
return "<pending websocket connection "..p.id.." to "..tostring(p.target_resolver.hostname)..">"; -- return "<pending websocket connection "..p.id.." to "..tostring(p.data.session.to_host)..">";
end; -- end;
}; -- };
function pending_ws_connection_methods:log(level, message, ...) -- function pending_ws_connection_methods:log(level, message, ...)
log(level, "[pending connection %s] "..message, self.id, ...); -- log(level, "[pending connection %s] "..message, self.id, ...);
end -- end
-- pending_ws_connections_map[ws_connection] = pending_connection -- function pending_ws_connection_methods:onopen()
local pending_ws_connections_map = {}; -- self:log("debug", "Pending WS Connection onopen event.");
local pending_ws_connection_listeners = {}; -- local conn = self.conn;
-- local session = self.data.session;
-- log("debug", "Successfully connected");
-- conn:setlistener(s2s_listener);
-- s2s_listener.onconnect(conn);
-- wrap_websocket(session, conn);
-- end
function pending_ws_connection_listeners.onopen(ws_connection) -- function pending_ws_connection_methods:onclose(code, message)
local p = pending_ws_connections_map[ws_connection]; -- self:log("debug", "Pending WS Connection onclose event.");
if not p then -- -- FIXME/TODO: what can i do??
if ws_connection.conn then -- -- if p.listeners.onfail then
module:log("warn", "Successful connection, but unexpected! Closing."); -- -- p.listeners.onfail(p.data, p.last_error or "unable to connect to websocket");
ws_connection.conn:close(); -- -- end
else -- end
module:log("error", "Successful connection, but unexpected, and no conn attribute!");
end
return;
end
pending_ws_connections_map[ws_connection] = nil;
local conn = ws_connection.conn;
p:log("debug", "Successfully connected");
conn:setlistener(p.listeners, p.data);
p.listeners.onconnect(conn);
wrap_websocket(session, conn);
end
function pending_ws_connection_listeners.onclose(ws_connection, reason) -- -- pending_ws_connections_map[ws_connection] = pending_connection
local p = pending_ws_connections_map[ws_connection]; -- local pending_ws_connections_map = {};
if not p then -- local pending_ws_connection_listeners = {};
module:log("warn", "Failed connection, but unexpected!");
return; -- function pending_ws_connection_listeners.onopen(ws_connection)
end -- log("debug", "Pending WS Connection onopen event.");
p.last_error = reason or "unknown reason";
p:log("debug", "Connection attempt failed: %s", p.last_error); -- local p = pending_ws_connections_map[ws_connection];
if p.listeners.onfail then -- if not p then
p.listeners.onfail(p.data, p.last_error or p.target_resolver.last_error or "unable to connect to websocket"); -- if ws_connection.conn then
end -- module:log("warn", "Successful connection, but unexpected! Closing.");
end -- ws_connection.conn:close();
-- else
-- module:log("error", "Successful connection, but unexpected, and no conn attribute!");
-- end
-- return;
-- end
-- pending_ws_connections_map[ws_connection] = nil;
-- local conn = ws_connection.conn;
-- p:log("debug", "Successfully connected");
-- conn:setlistener(p.listeners, p.data);
-- p.listeners.onconnect(conn);
-- wrap_websocket(session, conn);
-- end
-- function pending_ws_connection_listeners.onclose(ws_connection, reason)
-- log("debug", "Pending WS Connection onclose event.");
-- local p = pending_ws_connections_map[ws_connection];
-- if not p then
-- module:log("warn", "Failed connection, but unexpected!");
-- return;
-- end
-- p.last_error = reason or "unknown reason";
-- p:log("debug", "Connection attempt failed: %s", p.last_error);
-- if p.listeners.onfail then
-- p.listeners.onfail(p.data, p.last_error or "unable to connect to websocket");
-- end
-- end
function route_to_new_session(event) function route_to_new_session(event)
local from_host, to_host, stanza = event.from_host, event.to_host, event.stanza; local from_host, to_host, stanza = event.from_host, event.to_host, event.stanza;
module:log("debug", "Trying to route to %s", to_host); log("debug", "Trying to route to %s", to_host);
local ws_properties = module:fire_event("discover-websocket-s2s", { to_host = to_host }); local ws_properties = module:fire_event("discover-websocket-s2s", { to_host = to_host });
if not ws_properties then if not ws_properties then
module:log("debug", "No websocket s2s capabilities from remote host %s", to_host); log("debug", "No websocket s2s capabilities from remote host %s", to_host);
return; return;
end end
module:log("debug", "Found a Websocket endpoint for s2s communications to remote host %s", to_host); log("debug", "Found a Websocket endpoint for s2s communications to remote host %s", to_host);
local session = s2s_new_outgoing(from_host, to_host); local session = s2s_new_outgoing(from_host, to_host);
session.version = 1;
-- Store in buffer -- Store in buffer
session.bounce_sendq = bounce_sendq; session.bounce_sendq = bounce_sendq;
@ -434,13 +468,29 @@ function route_to_new_session(event)
ex["headers"] = ws_properties.extra_headers or {}; ex["headers"] = ws_properties.extra_headers or {};
ex["protocol"] = "xmpp"; ex["protocol"] = "xmpp";
module:log("debug", "Starting the websocket connection process"); session.log("debug", "Starting the s2s websocket connection process");
local p = setmetatable({ local log = session.log;
id = new_id();
listeners = portmanager.get_service("s2s").listener; local function onopen(s)
data = { session = session }; log("debug", "Outgoing Websocket S2S: Successfully connected");
}, pending_ws_connection_mt); local conn = s.conn;
local ws_connection = websocket.connect(ws_properties['url'], ex, p); session.conn = conn;
wrap_websocket(session, conn);
conn:setlistener(s2s_listener);
s2s_listener.register_outgoing(conn, session);
s2s_listener.onconnect(conn);
end
local function onclose(s, code, message)
log("debug", "Pending WS Connection onclose event.");
-- FIXME: is this ok?
s2s_destroy_session(session, message or code or "unable to connect to websocket");
end
local ws_connection = websocket.connect(ws_properties['url'], ex, {
onopen = onopen;
onclose = onclose;
});
return true; return true;
end end