Compare commits

...
Sign in to create a new pull request.

2 commits

Author SHA1 Message Date
Андрей Бабушкин
5072513047 removed AGENTS_CONNECTED event 2025-04-03 18:09:12 +02:00
Андрей Бабушкин
5da51662af add agent info to assist and tracker 2025-04-03 18:09:03 +02:00
5 changed files with 35 additions and 60 deletions

View file

@ -19,15 +19,16 @@ const EVENTS_DEFINITION = {
}
};
EVENTS_DEFINITION.emit = {
NEW_AGENT: "NEW_AGENT",
NO_AGENTS: "NO_AGENT",
AGENT_DISCONNECT: "AGENT_DISCONNECTED",
AGENTS_CONNECTED: "AGENTS_CONNECTED",
NO_SESSIONS: "SESSION_DISCONNECTED",
SESSION_ALREADY_CONNECTED: "SESSION_ALREADY_CONNECTED",
SESSION_RECONNECTED: "SESSION_RECONNECTED",
UPDATE_EVENT: EVENTS_DEFINITION.listen.UPDATE_EVENT,
WEBRTC_CONFIG: "WEBRTC_CONFIG",
NEW_AGENT: "NEW_AGENT",
NO_AGENTS: "NO_AGENT",
AGENT_DISCONNECT: "AGENT_DISCONNECTED",
AGENTS_CONNECTED: "AGENTS_CONNECTED",
AGENTS_INFO_CONNECTED: "AGENTS_INFO_CONNECTED",
NO_SESSIONS: "SESSION_DISCONNECTED",
SESSION_ALREADY_CONNECTED: "SESSION_ALREADY_CONNECTED",
SESSION_RECONNECTED: "SESSION_RECONNECTED",
UPDATE_EVENT: EVENTS_DEFINITION.listen.UPDATE_EVENT,
WEBRTC_CONFIG: "WEBRTC_CONFIG",
};
const BASE_sessionInfo = {

View file

@ -42,7 +42,7 @@ const findSessionSocketId = async (io, roomId, tabId) => {
};
async function getRoomData(io, roomID) {
let tabsCount = 0, agentsCount = 0, tabIDs = [], agentIDs = [], config = null;
let tabsCount = 0, agentsCount = 0, tabIDs = [], agentIDs = [], config = null, agentInfos = [];
const connected_sockets = await io.in(roomID).fetchSockets();
if (connected_sockets.length > 0) {
for (let socket of connected_sockets) {
@ -52,6 +52,7 @@ async function getRoomData(io, roomID) {
} else {
agentsCount++;
agentIDs.push(socket.id);
agentInfos.push({ ...socket.handshake.query.agentInfo, socketId: socket.id });
if (socket.handshake.query.config !== undefined) {
config = socket.handshake.query.config;
}
@ -60,8 +61,10 @@ async function getRoomData(io, roomID) {
} else {
tabsCount = -1;
agentsCount = -1;
agentInfos = [];
agentIDs = [];
}
return {tabsCount, agentsCount, tabIDs, agentIDs, config};
return {tabsCount, agentsCount, tabIDs, agentIDs, config, agentInfos};
}
function processNewSocket(socket) {
@ -81,7 +84,7 @@ async function onConnect(socket) {
IncreaseOnlineConnections(socket.handshake.query.identity);
const io = getServer();
const {tabsCount, agentsCount, tabIDs, agentIDs, config} = await getRoomData(io, socket.handshake.query.roomId);
const {tabsCount, agentsCount, tabIDs, agentInfos, agentIDs, config} = await getRoomData(io, socket.handshake.query.roomId);
if (socket.handshake.query.identity === IDENTITIES.session) {
// Check if session with the same tabID already connected, if so, refuse new connexion
@ -105,6 +108,7 @@ async function onConnect(socket) {
logger.debug(`notifying new session about agent-existence`);
io.to(socket.id).emit(EVENTS_DEFINITION.emit.WEBRTC_CONFIG, config);
io.to(socket.id).emit(EVENTS_DEFINITION.emit.AGENTS_CONNECTED, agentIDs);
io.to(socket.id).emit(EVENTS_DEFINITION.emit.AGENTS_INFO_CONNECTED, agentInfos);
socket.to(socket.handshake.query.roomId).emit(EVENTS_DEFINITION.emit.SESSION_RECONNECTED, socket.id);
}
} else if (tabsCount <= 0) {

View file

@ -348,7 +348,7 @@ export default class AssistManager {
);
this.canvasReceiver = new CanvasReceiver(
this.peerID,
this.config,
this.getIceServers(),
this.getNode,
{
...this.session.agentInfo,

View file

@ -28,7 +28,7 @@ export default class CanvasReceiver {
// sendSignal for sending signals (offer/answer/ICE)
constructor(
private readonly peerIdPrefix: string,
private readonly config: RTCIceServer[] | null,
private readonly config: RTCIceServer[],
private readonly getNode: MessageManager['getNode'],
private readonly agentInfo: Record<string, any>,
private readonly socket: Socket,
@ -66,9 +66,7 @@ export default class CanvasReceiver {
id: string,
): Promise<void> {
const pc = new RTCPeerConnection({
iceServers: this.config
? this.config
: [{ urls: 'stun:stun.l.google.com:19302' }],
iceServers: this.config,
});
// Save the connection
@ -177,44 +175,6 @@ function spawnVideo(stream: MediaStream, node: VElement) {
return videoEl;
}
function spawnDebugVideo(stream: MediaStream, node: VElement) {
const video = document.createElement('video');
video.id = 'canvas-or-testing';
video.style.border = '1px solid red';
video.setAttribute('autoplay', 'true');
video.setAttribute('muted', 'true');
video.setAttribute('playsinline', 'true');
video.setAttribute('crossorigin', 'anonymous');
const coords = node.node.getBoundingClientRect();
Object.assign(video.style, {
position: 'absolute',
left: `${coords.left}px`,
top: `${coords.top}px`,
width: `${coords.width}px`,
height: `${coords.height}px`,
});
video.width = coords.width;
video.height = coords.height;
video.srcObject = stream;
document.body.appendChild(video);
video
.play()
.then(() => {
console.debug('started streaming canvas');
})
.catch((e) => {
console.error(e);
const waiter = () => {
void video.play();
document.removeEventListener('click', waiter);
};
document.addEventListener('click', waiter);
});
}
function checkId(id: string, cId: string): boolean {
return id.includes(cId);
}

View file

@ -26,6 +26,7 @@ interface AgentInfo {
name: string;
peerId: string;
query: string;
socketId?: string;
}
export interface Options {
@ -436,11 +437,11 @@ export default class Assist {
}
});
socket.on("AGENTS_CONNECTED", (ids: string[]) => {
socket.on("AGENTS_INFO_CONNECTED", (agentsInfo: AgentInfo[]) => {
this.cleanCanvasConnections();
ids.forEach((id) => {
const agentInfo = this.agents[id]?.agentInfo;
this.agents[id] = {
agentsInfo.forEach((agentInfo) => {
if (!agentInfo.socketId) return;
this.agents[agentInfo.socketId] = {
agentInfo,
onDisconnect: this.options.onAgentConnect?.(agentInfo),
};
@ -448,6 +449,7 @@ export default class Assist {
if (this.app.active()) {
this.assistDemandedRestart = true;
this.app.stop();
this.app.clearBuffers();
this.app.waitStatus(0).then(() => {
this.app.allowAppStart();
setTimeout(() => {
@ -457,7 +459,7 @@ export default class Assist {
this.assistDemandedRestart = false;
})
.then(() => {
this.remoteControl?.reconnect(ids);
this.remoteControl?.reconnect(Object.keys(this.agents));
})
.catch((e) => app.debug.error(e));
}, 100);
@ -522,6 +524,13 @@ export default class Assist {
} catch (e) {
app.debug.error("Error adding ICE candidate", e);
}
} else {
this.iceCandidatesBuffer.set(
data.id,
this.iceCandidatesBuffer
.get(data.id)
?.concat([data.candidate]) || [data.candidate]
);
}
}
);
@ -889,6 +898,7 @@ export default class Assist {
iceServers: this.config,
});
this.setupPeerListeners(uniqueId);
this.applyBufferedIceCandidates(uniqueId);
stream.getTracks().forEach((track) => {
this.canvasPeers[uniqueId]?.addTrack(track, stream);