root/webRtc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【m】分窗渲染测试没问题

Browse Source
This commit is contained in:
stary
2026-04-24 23:03:41 +08:00
parent b66b639df0
commit eba0d07c83
4 changed files with 403 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
WebApp/client/public/onebyone/store.js
View File

@@ -326,15 +326,15 @@ class CallStateManager {
{ googTypingNoiseDetection: true } // 启用打字噪声检测
]
};
this.renderstreaming = new RenderStreaming(signaling, config); // 创建WebRTC连接管理实例
this.renderstreaming.onNewPeer = (connectionId) => {
console.log(`New peer created for ${connectionId}, adding local tracks`);
this.renderstreaming = new RenderStreaming(signaling, config);
this.renderstreaming.onNewPeer = (participantId) => {
console.log(`New peer created for ${participantId}, adding local tracks`);
if (this.state.localStream) {
const tracks = this.state.localStream.getTracks();
for (const track of tracks) {
this.renderstreaming.addTransceiver(track, { direction: 'sendonly' });
this.renderstreaming.addTransceiver(track, { direction: 'sendonly' }, participantId);
}
this.setCodecPreferences();
this.setCodecPreferences(participantId);
}
};
// 连接建立回调
@@ -375,44 +375,39 @@ this.renderstreaming.onNewPeer = (connectionId) => {
};
// participant离开回调（host收到，房间仍然存在）
this.renderstreaming.onParticipantLeft = (connectionId) => {
console.log(`Participant left: ${connectionId}, room still active`);
// 更新远程用户状态，但不关闭房间
this.renderstreaming.onParticipantLeft = (participantId) => {
console.log(`Participant left: ${participantId}, room still active`);
this.updateRemoteUserStatus('offline');
this.updateRemoteUserNetworkQuality('no_signal');
showNotification('对方已离开通话', 'warning');
// 清理该 participant 的远端流
if (this.state.remoteStreams[connectionId]) {
this.state.remoteStreams[connectionId].getTracks().forEach(track => track.stop());
delete this.state.remoteStreams[connectionId];
if (this.state.remoteStreams[participantId]) {
this.state.remoteStreams[participantId].getTracks().forEach(track => track.stop());
delete this.state.remoteStreams[participantId];
}
// 同时清理单路远端流（兼容）
if (this.state.remoteStream) {
this.state.remoteStream.getTracks().forEach(track => track.stop());
this.state.remoteStream = null;
}
// 通知UI更新
this.notify({ type: 'PARTICIPANT_LEFT', connectionId: connectionId });
// 通知UI更新，用participantId作为connectionId传给renderer
this.notify({ type: 'PARTICIPANT_LEFT', connectionId: participantId });
};
// 轨道事件回调
this.renderstreaming.onTrackEvent = (data) => {
const direction = data.transceiver.direction;
if (direction == "sendrecv" || direction == "recvonly") {
// 获取当前连接的远端流
const trackConnectionId = this.connectionId;
// Host端: 每个participant有独立的远端流
// Participant端: 只有一个host的远端流
// 使用participantId区分不同participant的流
const trackParticipantId = data.participantId || this.connectionId;
const isHost = this.role === 'host';
// 获取或创建对应的远端流
let targetStream = null;
if (isHost) {
// Host端: 按 connectionId 管理多路远端流
if (!this.state.remoteStreams[trackConnectionId]) {
this.state.remoteStreams[trackConnectionId] = new MediaStream();
// Host端: 按 participantId 管理多路远端流
if (!this.state.remoteStreams[trackParticipantId]) {
this.state.remoteStreams[trackParticipantId] = new MediaStream();
}
targetStream = this.state.remoteStreams[trackConnectionId];
targetStream = this.state.remoteStreams[trackParticipantId];
} else {
// Participant端: 使用单一远端流
if (this.state.remoteStream == null) {
@@ -423,22 +418,19 @@ this.renderstreaming.onNewPeer = (connectionId) => {
// 检查是否已经有相同类型的轨道
const existingTracks = targetStream.getTracks().filter(track => track.kind === data.track.kind);
// 移除旧的轨道
existingTracks.forEach(track => {
targetStream.removeTrack(track);
console.log('Removed old track:', track.kind);
});
// 添加新的轨道
targetStream.addTrack(data.track);
console.log('Added new track:', data.track.kind, 'to stream:', trackConnectionId);
console.log('Added new track:', data.track.kind, 'for participant:', trackParticipantId);
// 通知UI远程流已更新
this.notify({
type: 'REMOTE_STREAM_OBTAINED',
stream: targetStream,
connectionId: trackConnectionId,
connectionId: trackParticipantId,
isHost: isHost
});
console.log('Notified UI about remote stream update');
@@ -585,29 +577,24 @@ this.renderstreaming.onNewPeer = (connectionId) => {
/**
* 设置编解码器偏好
*/
setCodecPreferences() {
/** @type {RTCRtpCodecCapability[] | null} */
setCodecPreferences(participantId) {
let selectedCodecs = null;
// 获取视频编解码器能力
const { codecs } = RTCRtpSender.getCapabilities('video');
if (codecs && codecs.length > 0) {
// 优先选择H.264编解码器
const h264Codec = codecs.find(c => c.mimeType === 'video/H264');
if (h264Codec) {
selectedCodecs = [h264Codec];
}
}
if (selectedCodecs == null) return;
if (selectedCodecs == null) {
return;
}
// 获取视频收发器并设置编解码器偏好
if (this.renderstreaming) {
const transceivers = this.renderstreaming.getTransceivers().filter(t => t.receiver.track.kind == "video");
const transceivers = this.renderstreaming.getTransceivers(participantId);
if (transceivers && transceivers.length > 0) {
transceivers.forEach(t => t.setCodecPreferences(selectedCodecs));
const videoTransceivers = transceivers.filter(t => t.receiver.track.kind == "video");
if (videoTransceivers && videoTransceivers.length > 0) {
videoTransceivers.forEach(t => t.setCodecPreferences(selectedCodecs));
}
}
}
}

270
WebApp/client/src/renderstreaming.js
View File

@@ -5,26 +5,26 @@ function uuid4() {
var temp_url = URL.createObjectURL(new Blob());
var uuid = temp_url.toString();
URL.revokeObjectURL(temp_url);
return uuid.split(/[:/]/g).pop().toLowerCase(); // remove prefixes
return uuid.split(/[:/]/g).pop().toLowerCase();
}
export class RenderStreaming {
/**
* @param signaling signaling class
* @param {RTCConfiguration} config
*/
constructor(signaling, config) {
this._peer = null;
this._peer = null; // participant端：单一peer
this._peers = new Map(); // host端：多peer Map (participantId → Peer)
this._connectionId = null;
this.onConnect = function (connectionId) { Logger.log(`Connect peer on ${connectionId}.`); };
this._participantId = null; // 自己的participantId
this._isHost = false;
this.onConnect = function (connectionId, data) { Logger.log(`Connect peer on ${connectionId}.`); };
this.onDisconnect = function (connectionId) { Logger.log(`Disconnect peer on ${connectionId}.`); };
this.onGotOffer = function (connectionId) { Logger.log(`On got Offer on ${connectionId}.`); };
this.onGotAnswer = function (connectionId) { Logger.log(`On got Answer on ${connectionId}.`); };
this.onTrackEvent = function (data) { Logger.log(`OnTrack event peer with data:${data}`); };
this.onAddChannel = function (data) { Logger.log(`onAddChannel event peer with data:${data}`); };
this.onMessage = function (data) { Logger.log(`On message: ${data}`); };
this.onParticipantLeft = function (connectionId) { Logger.log(`Participant left on ${connectionId}.`); };
this.onNewPeer = function (connectionId) { Logger.log(`New peer created for ${connectionId}.`); };
this.onParticipantLeft = function (participantId) { Logger.log(`Participant left: ${participantId}.`); };
this.onParticipantJoined = function (participantId) { Logger.log(`Participant joined: ${participantId}.`); };
this.onNewPeer = function (participantId) { Logger.log(`New peer created for ${participantId}.`); };
this._config = config;
this._signaling = signaling;
this._signaling.addEventListener('connect', this._onConnect.bind(this));
@@ -34,12 +34,21 @@ export class RenderStreaming {
this._signaling.addEventListener('candidate', this._onIceCandidate.bind(this));
this._signaling.addEventListener('on-message', this._onMessage.bind(this));
this._signaling.addEventListener('participant-left', this._onParticipantLeft.bind(this));
this._signaling.addEventListener('participant-joined', this._onParticipantJoined.bind(this));
}
async _onConnect(e) {
const data = e.detail;
if (this._connectionId == data.connectionId) {
this._preparePeerConnection(this._connectionId, data.polite);
this._participantId = data.participantId;
this._isHost = data.role === 'host';
if (!this._isHost) {
// participant端：立即创建单一peer并开始协商
this._preparePeerConnection(this._connectionId, data.polite, null);
}
// host端：不在connect时创建peer，等participant加入后再创建
this.onConnect(data.connectionId, data);
}
}
@@ -52,50 +61,91 @@ export class RenderStreaming {
this._peer.close();
this._peer = null;
}
// 关闭所有host端peers
this._peers.forEach((peer, participantId) => {
peer.close();
});
this._peers.clear();
}
}
async _onOffer(e) {
const offer = e.detail;
// 如果已有Peer但ICE连接已断开，需要重建Peer
if (this._peer && this._peer.pc && this._peer.pc.iceConnectionState === 'disconnected') {
Logger.log('ICE disconnected, resetting PeerConnection for new offer');
this._peer.close();
this._peer = null;
}
if (!this._peer) {
this._preparePeerConnection(offer.connectionId, offer.polite);
}
const desc = new RTCSessionDescription({ sdp: offer.sdp, type: "offer" });
try {
await this._peer.onGotDescription(offer.connectionId, desc);
} catch (error) {
Logger.warn(`Error happen on GotDescription that description.\n Message: ${error}\n RTCSdpType:${desc.type}\n sdp:${desc.sdp}`);
return;
const participantId = offer.participantId;
if (this._isHost) {
// host端：为该participant创建或复用peer
let peer = this._peers.get(participantId);
if (!peer || (peer.pc && peer.pc.iceConnectionState === 'disconnected')) {
if (peer) peer.close();
peer = this._preparePeerConnection(this._connectionId, offer.polite, participantId);
}
const desc = new RTCSessionDescription({ sdp: offer.sdp, type: "offer" });
try {
await peer.onGotDescription(this._connectionId, desc);
} catch (error) {
Logger.warn(`Error on GotDescription for participant ${participantId}: ${error}`);
}
} else {
// participant端：使用单一peer
if (this._peer && this._peer.pc && this._peer.pc.iceConnectionState === 'disconnected') {
this._peer.close();
this._peer = null;
}
if (!this._peer) {
this._preparePeerConnection(offer.connectionId, offer.polite, null);
}
const desc = new RTCSessionDescription({ sdp: offer.sdp, type: "offer" });
try {
await this._peer.onGotDescription(offer.connectionId, desc);
} catch (error) {
Logger.warn(`Error on GotDescription: ${error}`);
}
}
}
async _onAnswer(e) {
const answer = e.detail;
const participantId = answer.participantId;
const desc = new RTCSessionDescription({ sdp: answer.sdp, type: "answer" });
if (this._peer) {
if (this._isHost && participantId) {
// host端：路由到对应participant的peer
const peer = this._peers.get(participantId);
if (peer) {
try {
await peer.onGotDescription(this._connectionId, desc);
} catch (error) {
Logger.warn(`Error on GotDescription answer for ${participantId}: ${error}`);
}
}
} else if (this._peer) {
// participant端
try {
await this._peer.onGotDescription(answer.connectionId, desc);
} catch (error) {
Logger.warn(`Error happen on GotDescription that description.\n Message: ${error}\n RTCSdpType:${desc.type}\n sdp:${desc.sdp}`);
return;
Logger.warn(`Error on GotDescription answer: ${error}`);
}
}
}
async _onIceCandidate(e) {
const candidate = e.detail;
const participantId = candidate.participantId;
const iceCandidate = new RTCIceCandidate({ candidate: candidate.candidate, sdpMid: candidate.sdpMid, sdpMLineIndex: candidate.sdpMLineIndex });
if (this._peer) {
if (this._isHost && participantId) {
// host端：路由到对应participant的peer
const peer = this._peers.get(participantId);
if (peer) {
await peer.onGotCandidate(this._connectionId, iceCandidate);
}
} else if (this._peer) {
// participant端
await this._peer.onGotCandidate(candidate.connectionId, iceCandidate);
}
}
// 在 RenderStreaming 类中添加
async _onMessage(e) {
const data = e.detail;
this.onMessage(data);
@@ -103,13 +153,32 @@ export class RenderStreaming {
async _onParticipantLeft(e) {
const data = e.detail;
Logger.log(`Participant left: ${data.connectionId}`);
this.onParticipantLeft(data.connectionId);
const participantId = data.participantId;
Logger.log(`Participant left: ${participantId}`);
// 关闭该participant的peer
if (this._peers.has(participantId)) {
const peer = this._peers.get(participantId);
peer.close();
this._peers.delete(participantId);
}
this.onParticipantLeft(participantId);
}
/**
* if not set argument, a generated uuid is used.
* @param {string | null} connectionId
*/
async _onParticipantJoined(e) {
const data = e.detail;
const participantId = data.participantId;
Logger.log(`Participant joined: ${participantId}`);
// host端：为新participant创建peer
if (this._isHost && !this._peers.has(participantId)) {
this._preparePeerConnection(this._connectionId, false, participantId);
}
this.onParticipantJoined(participantId);
}
async createConnection(connectionId) {
this._connectionId = connectionId ? connectionId : uuid4();
await this._signaling.createConnection(this._connectionId);
@@ -118,90 +187,106 @@ export class RenderStreaming {
async deleteConnection() {
await this._signaling.deleteConnection(this._connectionId);
}
// 在 RenderStreaming 类中添加
_preparePeerConnection(connectionId, polite) {
if (this._peer) {
Logger.log('Close current PeerConnection');
this._peer.close();
this._peer = null;
_preparePeerConnection(connectionId, polite, participantId) {
// host端多peer模式：participantId标识目标participant
// participant端单peer模式：participantId为null
const peer = new Peer(connectionId, polite, this._config);
// 保存peer
if (participantId) {
if (this._peers.has(participantId)) {
const oldPeer = this._peers.get(participantId);
oldPeer.close();
}
this._peers.set(participantId, peer);
} else {
if (this._peer) {
this._peer.close();
}
this._peer = peer;
}
// Create peerConnection with proxy server and set up handlers
this._peer = new Peer(connectionId, polite, this._config);
// this._peer.addEventListener('disconnect', () => {
// this.onDisconnect(`Receive disconnect message from peer. connectionId:${connectionId}`);
// });
this._peer.addEventListener('trackevent', (e) => {
// 事件处理：附加participantId用于路由
peer.addEventListener('trackevent', (e) => {
const data = e.detail;
data.participantId = participantId;
this.onTrackEvent(data);
});
this._peer.addEventListener('adddatachannel', (e) => {
peer.addEventListener('adddatachannel', (e) => {
const data = e.detail;
this.onAddChannel(data);
});
this._peer.addEventListener('ongotoffer', (e) => {
peer.addEventListener('ongotoffer', (e) => {
const id = e.detail.connectionId;
this.onGotOffer(id);
});
this._peer.addEventListener('ongotanswer', (e) => {
peer.addEventListener('ongotanswer', (e) => {
const id = e.detail.connectionId;
this.onGotAnswer(id);
});
this._peer.addEventListener('sendoffer', (e) => {
peer.addEventListener('sendoffer', (e) => {
const offer = e.detail;
this._signaling.sendOffer(offer.connectionId, offer.sdp);
this._signaling.sendOffer(offer.connectionId, offer.sdp, participantId);
});
this._peer.addEventListener('sendanswer', (e) => {
peer.addEventListener('sendanswer', (e) => {
const answer = e.detail;
this._signaling.sendAnswer(answer.connectionId, answer.sdp);
this._signaling.sendAnswer(answer.connectionId, answer.sdp, participantId);
});
this._peer.addEventListener('sendcandidate', (e) => {
peer.addEventListener('sendcandidate', (e) => {
const candidate = e.detail;
this._signaling.sendCandidate(candidate.connectionId, candidate.candidate, candidate.sdpMid, candidate.sdpMLineIndex);
this._signaling.sendCandidate(candidate.connectionId, candidate.candidate, candidate.sdpMid, candidate.sdpMLineIndex, participantId);
});
this.onNewPeer(connectionId);
return this._peer;
this.onNewPeer(participantId || connectionId);
return peer;
}
/**
* @returns {Promise<RTCStatsReport> | null}
*/
async getStats() {
return await this._peer.getStats(this._connectionId);
async getStats(participantId) {
if (this._isHost && participantId) {
const peer = this._peers.get(participantId);
return peer ? await peer.getStats(this._connectionId) : null;
}
return this._peer ? await this._peer.getStats(this._connectionId) : null;
}
/**
* @param {string} label
* @returns {RTCDataChannel | null}
*/
createDataChannel(label) {
return this._peer.createDataChannel(this._connectionId, label);
createDataChannel(label, participantId) {
if (this._isHost && participantId) {
const peer = this._peers.get(participantId);
return peer ? peer.createDataChannel(this._connectionId, label) : null;
}
return this._peer ? this._peer.createDataChannel(this._connectionId, label) : null;
}
/**
* @param {MediaStreamTrack} track
* @returns {RTCRtpSender | null}
*/
addTrack(track) {
return this._peer.addTrack(this._connectionId, track);
addTrack(track, participantId) {
if (this._isHost && participantId) {
const peer = this._peers.get(participantId);
return peer ? peer.addTrack(this._connectionId, track) : null;
}
return this._peer ? this._peer.addTrack(this._connectionId, track) : null;
}
/**
* @param {MediaStreamTrack | string} trackOrKind
* @param {RTCRtpTransceiverInit | null} init
* @returns {RTCRtpTransceiver | null}
*/
addTransceiver(trackOrKind, init) {
return this._peer.addTransceiver(this._connectionId, trackOrKind, init);
addTransceiver(trackOrKind, init, participantId) {
if (this._isHost && participantId) {
const peer = this._peers.get(participantId);
return peer ? peer.addTransceiver(this._connectionId, trackOrKind, init) : null;
}
return this._peer ? this._peer.addTransceiver(this._connectionId, trackOrKind, init) : null;
}
/**
* @returns {RTCRtpTransceiver[] | null}
*/
getTransceivers() {
return this._peer.getTransceivers(this._connectionId);
getTransceivers(participantId) {
if (this._isHost && participantId) {
const peer = this._peers.get(participantId);
return peer ? peer.getTransceivers(this._connectionId) : null;
}
return this._peer ? this._peer.getTransceivers(this._connectionId) : null;
}
sendMessage(message) {
@@ -209,6 +294,7 @@ export class RenderStreaming {
this._signaling.sendMessage(this._connectionId, message);
}
}
async start() {
await this._signaling.start();
}
@@ -218,10 +304,14 @@ export class RenderStreaming {
this._peer.close();
this._peer = null;
}
this._peers.forEach((peer) => {
peer.close();
});
this._peers.clear();
if (this._signaling) {
await this._signaling.stop();
this._signaling = null;
}
}
}
}

22
WebApp/client/src/signaling.js
View File

@@ -182,13 +182,13 @@ export class WebSocketSignaling extends EventTarget {
this.dispatchEvent(new CustomEvent('disconnect', { detail: msg }));
break;
case "offer":
this.dispatchEvent(new CustomEvent('offer', { detail: { connectionId: msg.from, sdp: msg.data.sdp, polite: msg.data.polite } }));
this.dispatchEvent(new CustomEvent('offer', { detail: { connectionId: msg.from, sdp: msg.data.sdp, polite: msg.data.polite, participantId: msg.participantId } }));
break;
case "answer":
this.dispatchEvent(new CustomEvent('answer', { detail: { connectionId: msg.from, sdp: msg.data.sdp } }));
this.dispatchEvent(new CustomEvent('answer', { detail: { connectionId: msg.from, sdp: msg.data.sdp, participantId: msg.participantId } }));
break;
case "candidate":
this.dispatchEvent(new CustomEvent('candidate', { detail: { connectionId: msg.from, candidate: msg.data.candidate, sdpMLineIndex: msg.data.sdpMLineIndex, sdpMid: msg.data.sdpMid } }));
this.dispatchEvent(new CustomEvent('candidate', { detail: { connectionId: msg.from, candidate: msg.data.candidate, sdpMLineIndex: msg.data.sdpMLineIndex, sdpMid: msg.data.sdpMid, participantId: msg.participantId } }));
break;
case "on-message":
this.dispatchEvent(new CustomEvent('on-message', { detail: msg.data }));
@@ -196,8 +196,10 @@ export class WebSocketSignaling extends EventTarget {
case "participant-left":
this.dispatchEvent(new CustomEvent('participant-left', { detail: msg }));
break;
case "participant-joined":
this.dispatchEvent(new CustomEvent('participant-joined', { detail: msg }));
break;
case "broadcast":
// 处理服务器广播的消息
this.dispatchEvent(new CustomEvent('on-message', { detail: msg.message }));
break;
default:
@@ -231,28 +233,28 @@ export class WebSocketSignaling extends EventTarget {
this.websocket.send(sendJson);
}
sendOffer(connectionId, sdp) {
sendOffer(connectionId, sdp, participantId) {
const data = { 'sdp': sdp, 'connectionId': connectionId };
const sendJson = JSON.stringify({ type: "offer", from: connectionId, data: data });
const sendJson = JSON.stringify({ type: "offer", from: connectionId, data: data, participantId: participantId || '' });
Logger.log(sendJson);
this.websocket.send(sendJson);
}
sendAnswer(connectionId, sdp) {
sendAnswer(connectionId, sdp, participantId) {
const data = { 'sdp': sdp, 'connectionId': connectionId };
const sendJson = JSON.stringify({ type: "answer", from: connectionId, data: data });
const sendJson = JSON.stringify({ type: "answer", from: connectionId, data: data, participantId: participantId || '' });
Logger.log(sendJson);
this.websocket.send(sendJson);
}
sendCandidate(connectionId, candidate, sdpMLineIndex, sdpMid) {
sendCandidate(connectionId, candidate, sdpMLineIndex, sdpMid, participantId) {
const data = {
'candidate': candidate,
'sdpMLineIndex': sdpMLineIndex,
'sdpMid': sdpMid,
'connectionId': connectionId
};
const sendJson = JSON.stringify({ type: "candidate", from: connectionId, data: data });
const sendJson = JSON.stringify({ type: "candidate", from: connectionId, data: data, participantId: participantId || '' });
Logger.log(sendJson);
this.websocket.send(sendJson);
}

347
WebApp/src/class/websockethandler.ts
View File

@@ -112,32 +112,26 @@ function broadcastToGroup(connectionId: string, senderWs: WebSocket, message: an
* @param ws WebSocket连接实例
*/
function remove(ws: WebSocket): void {
// 获取连接的所有连接ID
const connectionIds = clients.get(ws);
if (!connectionIds) return;
// 遍历所有连接ID
connectionIds.forEach(connectionId => {
const group = connectionGroup.get(connectionId);
if (group) {
if (group.host === ws) {
// host断开连接，通知所有participants房间已关闭
group.participants.forEach(participantWs => {
participantWs.send(JSON.stringify({ type: "disconnect", connectionId: connectionId, reason: "host-left" }));
});
// 删除整个连接组
connectionGroup.delete(connectionId);
} else {
// participant断开连接，从participants中移除并通知host
group.participants.delete(ws);
group.host.send(JSON.stringify({ type: "participant-left", connectionId: connectionId }));
// 包含participantId，让host能识别是哪个participant离开
group.host.send(JSON.stringify({ type: "participant-left", connectionId: connectionId, participantId: (ws as any).participantId }));
}
}
// 记录删除连接ID的日志
console.log(`Remove connectionId: ${connectionId}`);
});
// 从客户端映射中删除
clients.delete(ws);
}
@@ -149,30 +143,27 @@ function remove(ws: WebSocket): void {
*/
function onConnect(ws: WebSocket, connectionId: string): void {
let polite = true;
// 处理私有模式
// 为每个WebSocket生成唯一的participantId
const participantId = (ws as any).participantId = (ws as any).participantId || `p_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;
if (isPrivate) {
if (connectionGroup.has(connectionId)) {
const group = connectionGroup.get(connectionId);
// 已有host，新连接作为participant加入
group.participants.add(ws);
console.log(`Participant joined connectionId: ${connectionId}, total participants: ${group.participants.size}`);
console.log(`Participant ${participantId} joined connectionId: ${connectionId}, total participants: ${group.participants.size}`);
// 通知host有新participant加入
group.host.send(JSON.stringify({ type: "participant-joined", connectionId: connectionId, participantId: participantId }));
} else {
// 第一个连接成为host
connectionGroup.set(connectionId, { host: ws, participants: new Set<WebSocket>() });
polite = false;
console.log(`Host created connectionId: ${connectionId}`);
}
}
// 获取或创建连接ID集合
const connectionIds = getOrCreateConnectionIds(ws);
// 添加连接ID
connectionIds.add(connectionId);
// 发送连接成功消息（包含角色信息）
const role = polite ? 'participant' : 'host';
ws.send(JSON.stringify({ type: "connect", connectionId: connectionId, polite: polite, role: role }));
//启用心跳包
//AddHeartbeat(ws, connectionId);
ws.send(JSON.stringify({ type: "connect", connectionId: connectionId, polite: polite, role: role, participantId: participantId }));
}
/**
@@ -220,25 +211,35 @@ function onDisconnect(ws: WebSocket, connectionId: string): void {
* @param message 消息数据
*/
function onOffer(ws: WebSocket, message: any): void {
// 获取连接ID
const connectionId = message.connectionId as string;
// 创建新的offer
const newOffer = new Offer(message.sdp, Date.now(), false);
// 处理私有模式
if (isPrivate) {
if (connectionGroup.has(connectionId)) {
const group = connectionGroup.get(connectionId);
const senderParticipantId = (ws as any).participantId;
const targetParticipantId = message.participantId;
if (group.host === ws) {
// host发送offer，转发给所有participants
// host发送offer给特定participant（多peer模式下按participantId路由）
newOffer.polite = true;
group.participants.forEach(participantWs => {
participantWs.send(JSON.stringify({ from: connectionId, to: "", type: "offer", data: newOffer }));
});
if (targetParticipantId) {
// 路由到指定participant
group.participants.forEach(participantWs => {
if ((participantWs as any).participantId === targetParticipantId) {
participantWs.send(JSON.stringify({ from: connectionId, to: "", type: "offer", data: newOffer, participantId: targetParticipantId }));
}
});
} else {
// 兼容：无目标时广播给所有participants
group.participants.forEach(participantWs => {
const pid = (participantWs as any).participantId;
participantWs.send(JSON.stringify({ from: connectionId, to: "", type: "offer", data: newOffer, participantId: pid }));
});
}
} else {
// participant发送offer，转发给host
// participant发送offer给host，携带该participant的participantId
newOffer.polite = true;
group.host.send(JSON.stringify({ from: connectionId, to: "", type: "offer", data: newOffer }));
group.host.send(JSON.stringify({ from: connectionId, to: "", type: "offer", data: newOffer, participantId: senderParticipantId }));
}
}
return;
@@ -264,30 +265,37 @@ function onOffer(ws: WebSocket, message: any): void {
* @param message 消息数据
*/
function onAnswer(ws: WebSocket, message: any): void {
// 获取连接ID
const connectionId = message.connectionId as string;
// 获取或创建连接ID集合
const connectionIds = getOrCreateConnectionIds(ws);
// 添加连接ID
connectionIds.add(connectionId);
// 创建新的answer
const newAnswer = new Answer(message.sdp, Date.now());
// 检查连接组是否存在
if (!connectionGroup.has(connectionId)) {
return;
}
const group = connectionGroup.get(connectionId);
const senderParticipantId = (ws as any).participantId;
// 从answer消息中获取目标participantId（host回复时指定）
const targetParticipantId = message.participantId;
if (group.host === ws) {
// host发送answer，转发给所有participants（通常host不发送answer）
group.participants.forEach(participantWs => {
participantWs.send(JSON.stringify({ from: connectionId, to: "", type: "answer", data: newAnswer }));
});
// host发送answer给特定participant
if (targetParticipantId) {
group.participants.forEach(participantWs => {
if ((participantWs as any).participantId === targetParticipantId) {
participantWs.send(JSON.stringify({ from: connectionId, to: "", type: "answer", data: newAnswer, participantId: targetParticipantId }));
}
});
} else {
// 兼容：没有targetParticipantId时广播给所有participants
group.participants.forEach(participantWs => {
participantWs.send(JSON.stringify({ from: connectionId, to: "", type: "answer", data: newAnswer }));
});
}
} else {
// participant发送answer，转发给host
group.host.send(JSON.stringify({ from: connectionId, to: "", type: "answer", data: newAnswer }));
// participant发送answer给host，携带自己的participantId
group.host.send(JSON.stringify({ from: connectionId, to: "", type: "answer", data: newAnswer, participantId: senderParticipantId }));
}
}
@@ -298,150 +306,163 @@ function onAnswer(ws: WebSocket, message: any): void {
* @param message 消息数据
*/
function onCandidate(ws: WebSocket, message: any): void {
// 获取连接ID
const connectionId = message.connectionId;
// 创建新的candidate
const candidate = new Candidate(message.candidate, message.sdpMLineIndex, message.sdpMid, Date.now());
const senderParticipantId = (ws as any).participantId;
const targetParticipantId = message.participantId;
// 处理私有模式
if (isPrivate) {
if (connectionGroup.has(connectionId)) {
const group = connectionGroup.get(connectionId);
if (group.host === ws) {
// host发送candidate，转发给所有participants
group.participants.forEach(participantWs => {
participantWs.send(JSON.stringify({ from: connectionId, to: "", type: "candidate", data: candidate }));
});
// host发送candidate给特定participant
if (targetParticipantId) {
group.participants.forEach(participantWs => {
if ((participantWs as any).participantId === targetParticipantId) {
participantWs.send(JSON.stringify({ from: connectionId, to: "", type: "candidate", data: candidate, participantId: targetParticipantId }));
}
});
} else {
group.participants.forEach(participantWs => {
participantWs.send(JSON.stringify({ from: connectionId, to: "", type: "candidate", data: candidate }));
});
}
} else {
// participant发送candidate，转发给host
group.host.send(JSON.stringify({ from: connectionId, to: "", type: "candidate", data: candidate }));
// participant发送candidate给host，携带自己的participantId
group.host.send(JSON.stringify({ from: connectionId, to: "", type: "candidate", data: candidate, participantId: senderParticipantId }));
}
}
return;
}
}
function onCallConnectionId(ws: WebSocket, message: any): void {
// 获取连接ID
const connectionId = message.connectionId;
const clientId = message.clientId;
// 在1对多模式下，通知host有新的呼叫请求
if (connectionGroup.has(connectionId)) {
const group = connectionGroup.get(connectionId);
if (group.host !== ws) {
// participant发起呼叫，通知host
group.host.send(JSON.stringify({ from: connectionId, to: "", type: "call-request", data: connectionId }));
function onCallConnectionId(ws: WebSocket, message: any): void {
// 获取连接ID
const connectionId = message.connectionId;
const clientId = message.clientId;
// 在1对多模式下，通知host有新的呼叫请求
if (connectionGroup.has(connectionId)) {
const group = connectionGroup.get(connectionId);
if (group.host !== ws) {
// participant发起呼叫，通知host
group.host.send(JSON.stringify({ from: connectionId, to: "", type: "call-request", data: connectionId }));
}
} else {
// 兼容旧的广播方式
clients.forEach((_v, k) => {
if (k === ws) {
return;
}
} else {
// 兼容旧的广播方式
clients.forEach((_v, k) => {
if (k === ws) {
return;
}
if (_v == clientId) {
k.send(JSON.stringify({ from: connectionId, to: "", type: "call-request", data: connectionId }));
}
if (_v == clientId) {
k.send(JSON.stringify({ from: connectionId, to: "", type: "call-request", data: connectionId }));
}
});
}
}
/**
* 处理广播消息请求（1对多模式）
* @param ws WebSocket连接实例
* @param message 消息数据
*/
function onBroadcast(ws: WebSocket, message: any): void {
const broadcastMessage = message.message;
const targetConnectionId = message.targetConnectionId;
if (targetConnectionId) {
// 向指定连接组广播
if (connectionGroup.has(targetConnectionId)) {
const group = connectionGroup.get(targetConnectionId);
// 向组内所有成员发送消息
group.host.send(JSON.stringify({
type: "broadcast",
message: broadcastMessage,
from: "server"
}));
group.participants.forEach(participantWs => {
participantWs.send(JSON.stringify({
type: "broadcast",
message: broadcastMessage,
from: "server"
}));
});
}
} else {
// 全局广播：向所有客户端发送消息
clients.forEach((_v, k) => {
k.send(JSON.stringify({
type: "broadcast",
message: broadcastMessage,
from: "server"
}));
});
}
}
function AddHeartbeat(ws: WebSocket, connectionId: string) {
// 初始化心跳检测
(ws as any).lastActivity = Date.now();
/**
* 处理广播消息请求（1对多模式）
* @param ws WebSocket连接实例
* @param message 消息数据
*/
function onBroadcast(ws: WebSocket, message: any): void {
const broadcastMessage = message.message;
const targetConnectionId = message.targetConnectionId;
// 设置心跳检测定时器，每30秒发送一次ping
(ws as any).heartbeatTimer = setInterval(() => {
const now = Date.now();
// 检查上次活动时间，如果超过60秒没有活动，关闭连接
if (now - (ws as any).lastActivity > 10000) {
console.log('WebSocket connection timeout, closing...');
clearInterval((ws as any).heartbeatTimer);
//ws.close();
onDisconnect(ws, connectionId);
} else {
// 发送ping消息
ws.send(JSON.stringify({ from: connectionId, to: "", type: "on-message", data: { type: "ping"} }));
console.log('WebSocket connection heartbeat, lastActivity: ', (ws as any).lastActivity);
}
}, 3000);
}
if (targetConnectionId) {
// 向指定连接组广播
if (connectionGroup.has(targetConnectionId)) {
const group = connectionGroup.get(targetConnectionId);
// 向组内所有成员发送消息
group.host.send(JSON.stringify({
type: "broadcast",
message: broadcastMessage,
from: "server"
}));
group.participants.forEach(participantWs => {
participantWs.send(JSON.stringify({
type: "broadcast",
message: broadcastMessage,
from: "server"
}));
});
}
} else {
// 全局广播：向所有客户端发送消息
clients.forEach((_v, k) => {
k.send(JSON.stringify({
type: "broadcast",
message: broadcastMessage,
from: "server"
}));
});
}
}
function AddHeartbeat(ws: WebSocket, connectionId: string) {
// 初始化心跳检测
(ws as any).lastActivity = Date.now();
function RemoveHeartbeat(ws: WebSocket) {
// 清除心跳检测定时器
if ((ws as any).heartbeatTimer) {
clearInterval((ws as any).heartbeatTimer);
}
}
// 设置心跳检测定时器，每30秒发送一次ping
(ws as any).heartbeatTimer = setInterval(() => {
const now = Date.now();
// 检查上次活动时间，如果超过60秒没有活动，关闭连接
if (now - (ws as any).lastActivity > 10000) {
console.log('WebSocket connection timeout, closing...');
clearInterval((ws as any).heartbeatTimer);
//ws.close();
onDisconnect(ws, connectionId);
} else {
// 发送ping消息
ws.send(JSON.stringify({ from: connectionId, to: "", type: "on-message", data: { type: "ping"} }));
console.log('WebSocket connection heartbeat, lastActivity: ', (ws as any).lastActivity);
}
}, 3000);
/**
* 处理获取所有连接ID的请求
* @param ws WebSocket连接实例
*/
function onGetAllConnectionIds(): string[] {
// 获取所有connectionId
const connectionIds = Array.from(connectionGroup.keys());
return connectionIds;
}
/**
* 处理chat-message信令（1对多模式）
* host的消息转发给所有participants，participant的消息转发给host
* @param ws WebSocket连接实例
* @param message 消息数据
*/
function onMessage(ws: WebSocket, message: any): void {
// 获取连接ID
const connectionId = message.connectionId;
const chatMessage = message.message;
if (connectionGroup.has(connectionId)) {
const group = connectionGroup.get(connectionId);
if (group.host === ws) {
// host发送消息，转发给所有participants
group.participants.forEach(participantWs => {
participantWs.send(JSON.stringify({ from: connectionId, to: "", type: "on-message", data: chatMessage }));
});
} else {
// participant发送消息，转发给host
group.host.send(JSON.stringify({ from: connectionId, to: "", type: "on-message", data: chatMessage }));
}
function RemoveHeartbeat(ws: WebSocket) {
// 清除心跳检测定时器
if ((ws as any).heartbeatTimer) {
clearInterval((ws as any).heartbeatTimer);
}
}
/**
* 处理获取所有连接ID的请求
* @param ws WebSocket连接实例
*/
function onGetAllConnectionIds(): string[] {
// 获取所有connectionId
const connectionIds = Array.from(connectionGroup.keys());
return connectionIds;
}
/**
* 处理chat-message信令（1对多模式）
* host的消息转发给所有participants，participant的消息转发给host
* @param ws WebSocket连接实例
* @param message 消息数据
*/
function onMessage(ws: WebSocket, message: any): void {
// 获取连接ID
const connectionId = message.connectionId;
const chatMessage = message.message;
if (connectionGroup.has(connectionId)) {
const group = connectionGroup.get(connectionId);
if (group.host === ws) {
// host发送消息，转发给所有participants
group.participants.forEach(participantWs => {
participantWs.send(JSON.stringify({ from: connectionId, to: "", type: "on-message", data: chatMessage }));
});
} else {
// participant发送消息，转发给host
group.host.send(JSON.stringify({ from: connectionId, to: "", type: "on-message", data: chatMessage }));
}
}
}
/**
* 导出WebSocket处理器函数
*/
export { reset, add, remove, onConnect, onDisconnect, onOffer, onAnswer, onCandidate, onCallConnectionId, onBroadcast, onGetAllConnectionIds, AddHeartbeat, RemoveHeartbeat, onMessage, isHost, broadcastToGroup, connectionGroup };
}
}
/**
* 导出WebSocket处理器函数
*/
export { reset, add, remove, onConnect, onDisconnect, onOffer, onAnswer, onCandidate, onCallConnectionId, onBroadcast, onGetAllConnectionIds, AddHeartbeat, RemoveHeartbeat, onMessage, isHost, broadcastToGroup, connectionGroup };