From dffba219d6a3bd371a600913d94ef22b828fc3f5 Mon Sep 17 00:00:00 2001 From: hippoz Date: Fri, 21 May 2021 01:33:47 +0300 Subject: [PATCH] add broken webrtc media server and experiment system --- brainlet/api/v2/gateway/index.js | 89 +++++++++++++++++++++++- brainlet/api/v2/gateway/messageparser.js | 7 +- brainlet/app/gatewaytest/app.js | 6 +- brainlet/experiments.js | 6 ++ brainlet/secret.js | 2 +- resources/Docs/DOCS.md | 4 +- 6 files changed, 108 insertions(+), 6 deletions(-) create mode 100644 brainlet/experiments.js diff --git a/brainlet/api/v2/gateway/index.js b/brainlet/api/v2/gateway/index.js index c54b305..9005751 100644 --- a/brainlet/api/v2/gateway/index.js +++ b/brainlet/api/v2/gateway/index.js @@ -1,7 +1,9 @@ const websockets = require("ws"); const EventEmitter = require("events"); const uuid = require("uuid"); +const werift = require("werift"); +const { experiments } = require("../../../experiments"); const User = require("../../../models/User"); const Channel = require("../../../models/Channel"); const { parseMessage, opcodeSeparator, getOpcodeByName } = require("./messageparser"); @@ -9,6 +11,31 @@ const { checkToken } = require("../../../common/auth/authfunctions"); const pingCheckDelay = 10000; +class GatewayRTCConnection { + constructor(ws, server) { + this.ws = ws; + this.server = server; + + this.connection = new werift.RTCPeerConnection(); + this.connection.onnegotiationneeded.subscribe(() => this.renegotiate); + } + + async answer(offer) { + await this.connection.setRemoteDescription(offer); + await this.connection.setLocalDescription(await this.connection.createAnswer()); + return this.connection.localDescription; + } + + async addTrack(track) { + if (!track) return; + this.connection.addTrack(track); + } + + async renegotiate() { + this.ws.send(this.server.packet("EVENT_RENEGOTIATE_REQUIRED", {})); + } +} + class GatewayServer extends EventEmitter { constructor({ server }) { super(); @@ -63,7 +90,7 @@ class GatewayServer extends EventEmitter { ws.channels = channels.map(x => x._id); - ws.send(this.packet("YOO_ACK", { session_id: ws.session.sessionId, channels, user: { username: user.username, _id: user._id } })); + ws.send(this.packet("YOO_ACK", { session_id: ws.session.sessionId, channels, user: { username: user.username, _id: user._id }, __global_experiments: experiments })); console.log(`gateway: user ${user.username}: handshake complete`); } catch (e) { console.log("gateway:", e); @@ -97,6 +124,56 @@ class GatewayServer extends EventEmitter { break; } + case "ACTION_VOICE_REQUEST_SESSION": { + if (!experiments.voiceSFUTesting) return; + // just send ourselves as the voice server lol + ws.send(this.packet("EVENT_VOICE_ASSIGN_SERVER", { + reportTo: "/gateway", + channel: message.data.channel + })); + break; + } + case "ACTION_VOICE_CONNECTION_REQUEST": { + if (!this.authMessage(ws)) return; + if (!experiments.voiceSFUTesting) return; + + const offer = message.data.offer; + const channel = message.data.channel; + if (!channel || !channel._id) throw new Error("Client did not send a valid channel in ACTION_VOICE_CONNECTION_REQUEST"); + if (!offer) throw new Error("Client did not send any offer in ACTION_VOICE_CONNECTION_REQUEST"); + + let isNewConnection = true; + if (ws.rtc) { + isNewConnection = false; + } else { + ws.rtc = new GatewayRTCConnection(ws, this); + } + + ws.send(this.packet("EVENT_VOICE_CONNECTION_ANSWER", { + answer: await ws.rtc.answer(offer) + })); + + if (isNewConnection) { + this.inChannel(channel._id, (otherWs) => { + //if (!this.clientReady(otherWs)) return; + if (!this.clientRTCReady(otherWs)) return; + if (otherWs.session.user._id === ws.session.user._id) return; // Don't perform the actions below on ourselves + + const otherReceivers = otherWs.rtc.connection.getReceivers(); + for (let i = 0; i < otherReceivers.length; i++) { + const otherRecevier = otherReceivers[i]; + ws.rtc.addTrack(otherRecevier.tracks[0]); + } + const myReceviers = ws.rtc.connection.getReceivers(); + for (let i = 0; i < myReceviers.length; i++) { + const myReceiver = myReceviers[i]; + otherWs.rtc.addTrack(myReceiver.tracks[0]); + } + otherWs.rtc.renegotiate(); + }); + ws.rtc.renegotiate(); + } + } } } catch(e) { console.error("gateway:", e); @@ -113,10 +190,20 @@ GatewayServer.prototype.broadcast = function(channelId, data) { }); }; +GatewayServer.prototype.inChannel = function(channelId, func) { + this.wss.clients.forEach((client) => { + if (this.clientReady(client) && client.channels.includes(channelId)) func(client); + }); +}; + GatewayServer.prototype.clientReady = function(ws) { return ws.readyState === websockets.OPEN && ws.session && ws.session.authenticated; }; +GatewayServer.prototype.clientRTCReady = function(ws) { + return ws.rtc; // TODO: add more checks +}; + GatewayServer.prototype.authMessage = function(ws) { if (!this.clientReady(ws)) { ws.close(4007, "Not authenticated."); diff --git a/brainlet/api/v2/gateway/messageparser.js b/brainlet/api/v2/gateway/messageparser.js index a5a0b52..6b54e90 100644 --- a/brainlet/api/v2/gateway/messageparser.js +++ b/brainlet/api/v2/gateway/messageparser.js @@ -3,7 +3,12 @@ const opcodes = { 1: { name: "YOO", data: "JSON" }, 2: { name: "YOO_ACK", data: "JSON" }, 3: { name: "ACTION_CREATE_MESSAGE", data: "JSON" }, - 4: { name: "EVENT_CREATE_MESSAGE", data: "JSON" } + 4: { name: "EVENT_CREATE_MESSAGE", data: "JSON" }, + 21: { name: "ACTION_VOICE_REQUEST_SESSION", data: "JSON" }, + 22: { name: "EVENT_VOICE_ASSIGN_SERVER", data: "JSON" }, + 23: { name: "ACTION_VOICE_CONNECTION_REQUEST", data: "JSON" }, + 24: { name: "EVENT_VOICE_CONNECTION_ANSWER", data: "JSON" }, + 25: { name: "EVENT_RENEGOTIATE_REQUIRED", data: "JSON" } }; const opcodeSeparator = "@"; diff --git a/brainlet/app/gatewaytest/app.js b/brainlet/app/gatewaytest/app.js index 446c77b..90fa8f0 100644 --- a/brainlet/app/gatewaytest/app.js +++ b/brainlet/app/gatewaytest/app.js @@ -3,7 +3,11 @@ const opcodes = { 1: { name: "YOO", data: "JSON" }, 2: { name: "YOO_ACK", data: "JSON" }, 3: { name: "ACTION_CREATE_MESSAGE", data: "JSON" }, - 4: { name: "EVENT_CREATE_MESSAGE", data: "JSON" } + 4: { name: "EVENT_CREATE_MESSAGE", data: "JSON" }, + 21: { name: "ACTION_VOICE_REQUEST_SESSION", data: "JSON" }, + 22: { name: "EVENT_VOICE_ASSIGN_SERVER", data: "JSON" }, + 23: { name: "ACTION_VOICE_CONNECTION_REQUEST", data: "JSON" }, + 24: { name: "EVENT_VOICE_CONNECTION_ANSWER", data: "JSON" } }; const opcodeSeparator = "@"; diff --git a/brainlet/experiments.js b/brainlet/experiments.js new file mode 100644 index 0000000..2971298 --- /dev/null +++ b/brainlet/experiments.js @@ -0,0 +1,6 @@ +module.exports = { + experiments: { + voiceSFUTesting: false, + userListTest: false + } +}; diff --git a/brainlet/secret.js b/brainlet/secret.js index 8cc0847..89ea062 100755 --- a/brainlet/secret.js +++ b/brainlet/secret.js @@ -1,5 +1,5 @@ module.exports = { - jwtPrivateKey: "KEY" + jwtPrivateKey: "KjEY", }; // Set default values diff --git a/resources/Docs/DOCS.md b/resources/Docs/DOCS.md index be9cc13..afd81c7 100644 --- a/resources/Docs/DOCS.md +++ b/resources/Docs/DOCS.md @@ -168,7 +168,7 @@ JSON data format: Voice server signaling is done through a websocket gateway. This gateway is specified by the `reportTo` property in EVENT_VOICE_ASSIGN_SERVER. -## 50:ACTION_VOICE_CONNECTION_REQUEST +## 23:ACTION_VOICE_CONNECTION_REQUEST *Client to server* @@ -178,7 +178,7 @@ Voice server signaling is done through a websocket gateway. This gateway is spec | channel | An object that contains "_id", the id of the channel to connect to | | offer | An SDP payload; the webrtc offer from the client | -## 51:EVENT_VOICE_CONNECTION_ANSWER +## 24:EVENT_VOICE_CONNECTION_ANSWER *Auth required, Server to client*