Ma/file-transfer-go
1
0
Fork 0
mirror of https://github.com/MatrixSeven/file-transfer-go.git synced 2026-03-04 14:07:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat:shareConnect拆分|处理effect竞争引发的Bug

Browse Source
This commit is contained in:
MatrixSeven
2025-08-28 15:31:21 +08:00
parent 63e6e956e4
commit bc01224c11
17 changed files with 1878 additions and 417 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
chuan-next/src/hooks/useConnectionStatus.ts Normal file
View File

@@ -0,0 +1 @@

227
chuan-next/src/hooks/webrtc/core/DataChannelManager.ts Normal file
View File

@@ -0,0 +1,227 @@
import { EventEmitter } from 'events';
import { WebRTCError, WebRTCMessage, ConnectionEvent, EventHandler, MessageHandler, DataHandler } from './types';
interface DataChannelManagerConfig {
channelName: string;
onMessage?: MessageHandler;
onData?: DataHandler;
ordered?: boolean;
maxRetransmits?: number;
}
export class DataChannelManager extends EventEmitter {
private dataChannel: RTCDataChannel | null = null;
private config: DataChannelManagerConfig;
private messageQueue: WebRTCMessage[] = [];
private dataQueue: ArrayBuffer[] = [];
private isReady = false;
constructor(config: DataChannelManagerConfig) {
super();
this.config = {
ordered: true,
maxRetransmits: 3,
...config
};
}
initializeDataChannel(dataChannel: RTCDataChannel): void {
this.dataChannel = dataChannel;
this.setupEventHandlers();
}
createDataChannel(pc: RTCPeerConnection): RTCDataChannel {
if (this.dataChannel) {
throw new WebRTCError('DC_ALREADY_EXISTS', '数据通道已存在', false);
}
try {
this.dataChannel = pc.createDataChannel(this.config.channelName, {
ordered: this.config.ordered,
maxRetransmits: this.config.maxRetransmits,
});
this.setupEventHandlers();
return this.dataChannel;
} catch (error) {
throw new WebRTCError('DC_CREATE_FAILED', '创建数据通道失败', false, error as Error);
}
}
private setupEventHandlers(): void {
if (!this.dataChannel) return;
this.dataChannel.onopen = () => {
console.log(`[DataChannel] 数据通道已打开: ${this.config.channelName}`);
this.isReady = true;
this.flushQueues();
this.emit('state-change', { type: 'state-change', state: { isPeerConnected: true, error: null } });
};
this.dataChannel.onmessage = (event) => {
if (typeof event.data === 'string') {
try {
const message = JSON.parse(event.data) as WebRTCMessage;
console.log(`[DataChannel] 收到消息: ${message.type}, 通道: ${message.channel || this.config.channelName}`);
if (this.config.onMessage) {
this.config.onMessage(message);
}
} catch (error) {
console.error('[DataChannel] 解析消息失败:', error);
this.emit('error', {
type: 'error',
error: new WebRTCError('DC_MESSAGE_PARSE_ERROR', '消息解析失败', false, error as Error)
});
}
} else if (event.data instanceof ArrayBuffer) {
console.log(`[DataChannel] 收到数据: ${event.data.byteLength} bytes`);
if (this.config.onData) {
this.config.onData(event.data);
}
}
};
this.dataChannel.onerror = (error) => {
console.error(`[DataChannel] 数据通道错误: ${this.config.channelName}`, error);
const errorMessage = this.getDetailedErrorMessage();
this.emit('error', {
type: 'error',
error: new WebRTCError('DC_ERROR', errorMessage, true)
});
};
this.dataChannel.onclose = () => {
console.log(`[DataChannel] 数据通道已关闭: ${this.config.channelName}`);
this.isReady = false;
this.emit('disconnected', {
type: 'disconnected',
reason: `数据通道关闭: ${this.config.channelName}`
});
};
}
private getDetailedErrorMessage(): string {
if (!this.dataChannel) return '数据通道不可用';
switch (this.dataChannel.readyState) {
case 'connecting':
return '数据通道正在连接中，请稍候...';
case 'closing':
return '数据通道正在关闭，连接即将断开';
case 'closed':
return '数据通道已关闭，P2P连接失败';
default:
return '数据通道连接失败，可能是网络环境受限';
}
}
sendMessage(message: WebRTCMessage): boolean {
if (!this.isReady || !this.dataChannel || this.dataChannel.readyState !== 'open') {
this.messageQueue.push(message);
return false;
}
try {
this.dataChannel.send(JSON.stringify(message));
console.log(`[DataChannel] 发送消息: ${message.type}, 通道: ${message.channel || this.config.channelName}`);
return true;
} catch (error) {
console.error('[DataChannel] 发送消息失败:', error);
this.emit('error', {
type: 'error',
error: new WebRTCError('DC_SEND_ERROR', '发送消息失败', true, error as Error)
});
return false;
}
}
sendData(data: ArrayBuffer): boolean {
if (!this.isReady || !this.dataChannel || this.dataChannel.readyState !== 'open') {
this.dataQueue.push(data);
return false;
}
try {
this.dataChannel.send(data);
console.log(`[DataChannel] 发送数据: ${data.byteLength} bytes`);
return true;
} catch (error) {
console.error('[DataChannel] 发送数据失败:', error);
this.emit('error', {
type: 'error',
error: new WebRTCError('DC_SEND_DATA_ERROR', '发送数据失败', true, error as Error)
});
return false;
}
}
private flushQueues(): void {
// 发送排队的消息
while (this.messageQueue.length > 0) {
const message = this.messageQueue.shift();
if (message) {
this.sendMessage(message);
}
}
// 发送排队的数据
while (this.dataQueue.length > 0) {
const data = this.dataQueue.shift();
if (data) {
this.sendData(data);
}
}
}
getState(): RTCDataChannelState {
return this.dataChannel?.readyState || 'closed';
}
isChannelReady(): boolean {
return this.isReady && this.dataChannel?.readyState === 'open';
}
getBufferedAmount(): number {
return this.dataChannel?.bufferedAmount || 0;
}
getBufferedAmountLowThreshold(): number {
return this.dataChannel?.bufferedAmountLowThreshold || 0;
}
setBufferedAmountLowThreshold(threshold: number): void {
if (this.dataChannel) {
this.dataChannel.bufferedAmountLowThreshold = threshold;
}
}
onBufferedAmountLow(handler: () => void): void {
if (this.dataChannel) {
this.dataChannel.onbufferedamountlow = handler;
}
}
close(): void {
if (this.dataChannel) {
this.dataChannel.close();
this.dataChannel = null;
}
this.isReady = false;
this.messageQueue = [];
this.dataQueue = [];
this.emit('disconnected', { type: 'disconnected', reason: '数据通道已关闭' });
}
on(event: string, handler: EventHandler<ConnectionEvent>): this {
return super.on(event, handler);
}
emit(eventName: string, event?: ConnectionEvent): boolean {
return super.emit(eventName, event);
}
}

197
chuan-next/src/hooks/webrtc/core/MessageRouter.ts Normal file
View File

@@ -0,0 +1,197 @@
import { WebRTCMessage, MessageHandler, DataHandler } from './types';
interface ChannelHandlers {
messageHandlers: Set<MessageHandler>;
dataHandlers: Set<DataHandler>;
}
export class MessageRouter {
private channels = new Map<string, ChannelHandlers>();
private defaultChannelHandlers: ChannelHandlers | null = null;
constructor() {
this.createDefaultChannel();
}
private createDefaultChannel(): void {
this.defaultChannelHandlers = {
messageHandlers: new Set(),
dataHandlers: new Set(),
};
}
registerMessageHandler(channel: string, handler: MessageHandler): () => void {
let channelHandlers = this.channels.get(channel);
if (!channelHandlers) {
channelHandlers = {
messageHandlers: new Set(),
dataHandlers: new Set(),
};
this.channels.set(channel, channelHandlers);
}
channelHandlers.messageHandlers.add(handler);
// 返回取消注册函数
return () => {
channelHandlers!.messageHandlers.delete(handler);
// 如果通道没有处理器了，删除通道
if (channelHandlers!.messageHandlers.size === 0 && channelHandlers!.dataHandlers.size === 0) {
this.channels.delete(channel);
}
};
}
registerDataHandler(channel: string, handler: DataHandler): () => void {
let channelHandlers = this.channels.get(channel);
if (!channelHandlers) {
channelHandlers = {
messageHandlers: new Set(),
dataHandlers: new Set(),
};
this.channels.set(channel, channelHandlers);
}
channelHandlers.dataHandlers.add(handler);
// 返回取消注册函数
return () => {
channelHandlers!.dataHandlers.delete(handler);
// 如果通道没有处理器了，删除通道
if (channelHandlers!.messageHandlers.size === 0 && channelHandlers!.dataHandlers.size === 0) {
this.channels.delete(channel);
}
};
}
registerDefaultMessageHandler(handler: MessageHandler): () => void {
if (!this.defaultChannelHandlers) {
this.createDefaultChannel();
}
this.defaultChannelHandlers?.messageHandlers.add(handler);
return () => {
this.defaultChannelHandlers?.messageHandlers.delete(handler);
};
}
registerDefaultDataHandler(handler: DataHandler): () => void {
if (!this.defaultChannelHandlers) {
this.createDefaultChannel();
}
this.defaultChannelHandlers?.dataHandlers.add(handler);
return () => {
this.defaultChannelHandlers?.dataHandlers.delete(handler);
};
}
routeMessage(message: WebRTCMessage): void {
const channel = message.channel;
if (channel) {
// 路由到特定通道
const channelHandlers = this.channels.get(channel);
if (channelHandlers && channelHandlers.messageHandlers.size > 0) {
channelHandlers.messageHandlers.forEach(handler => {
try {
handler(message);
} catch (error) {
console.error(`消息处理器错误 (通道: ${channel}):`, error);
}
});
return;
}
}
// 回退到默认处理器
if (this.defaultChannelHandlers?.messageHandlers.size) {
this.defaultChannelHandlers.messageHandlers.forEach(handler => {
try {
handler(message);
} catch (error) {
console.error('默认消息处理器错误:', error);
}
});
} else {
console.warn('没有找到消息处理器:', message.type, channel || 'default');
}
}
routeData(data: ArrayBuffer, channel?: string): void {
if (channel) {
// 路由到特定通道
const channelHandlers = this.channels.get(channel);
if (channelHandlers && channelHandlers.dataHandlers.size > 0) {
channelHandlers.dataHandlers.forEach(handler => {
try {
handler(data);
} catch (error) {
console.error(`数据处理器错误 (通道: ${channel}):`, error);
}
});
return;
}
}
// 回退到默认处理器
if (this.defaultChannelHandlers?.dataHandlers.size) {
this.defaultChannelHandlers.dataHandlers.forEach(handler => {
try {
handler(data);
} catch (error) {
console.error('默认数据处理器错误:', error);
}
});
} else {
console.warn('没有找到数据处理器，数据大小:', data.byteLength, 'bytes');
}
}
hasHandlers(channel?: string): boolean {
if (channel) {
const channelHandlers = this.channels.get(channel);
return channelHandlers ?
(channelHandlers.messageHandlers.size > 0 || channelHandlers.dataHandlers.size > 0) :
false;
}
return this.defaultChannelHandlers ?
(this.defaultChannelHandlers.messageHandlers.size > 0 || this.defaultChannelHandlers.dataHandlers.size > 0) :
false;
}
getChannelList(): string[] {
return Array.from(this.channels.keys());
}
getHandlerCount(channel?: string): { message: number; data: number } {
if (channel) {
const channelHandlers = this.channels.get(channel);
return channelHandlers ? {
message: channelHandlers.messageHandlers.size,
data: channelHandlers.dataHandlers.size,
} : { message: 0, data: 0 };
}
return this.defaultChannelHandlers ? {
message: this.defaultChannelHandlers.messageHandlers.size,
data: this.defaultChannelHandlers.dataHandlers.size,
} : { message: 0, data: 0 };
}
clear(): void {
this.channels.clear();
this.createDefaultChannel();
}
clearChannel(channel: string): void {
this.channels.delete(channel);
}
}

293
chuan-next/src/hooks/webrtc/core/PeerConnectionManager.ts Normal file
View File

@@ -0,0 +1,293 @@
import { EventEmitter } from 'events';
import { WebRTCError, WebRTCConfig, ConnectionEvent, EventHandler } from './types';
interface PeerConnectionManagerConfig extends WebRTCConfig {
onSignalingMessage: (message: any) => void;
onTrack?: (event: RTCTrackEvent) => void;
}
interface NegotiationOptions {
offerToReceiveAudio?: boolean;
offerToReceiveVideo?: boolean;
}
export class PeerConnectionManager extends EventEmitter {
private pc: RTCPeerConnection | null = null;
private config: PeerConnectionManagerConfig;
private isNegotiating = false;
private negotiationQueue: Array<() => Promise<void>> = [];
private localCandidates: RTCIceCandidate[] = [];
private remoteCandidates: RTCIceCandidate[] = [];
constructor(config: PeerConnectionManagerConfig) {
super();
this.config = config;
}
async createPeerConnection(): Promise<RTCPeerConnection> {
if (this.pc) {
this.destroyPeerConnection();
}
try {
this.pc = new RTCPeerConnection({
iceServers: this.config.iceServers,
iceCandidatePoolSize: this.config.iceCandidatePoolSize,
});
this.setupEventHandlers();
this.emit('state-change', { type: 'state-change', state: { isPeerConnected: false } });
return this.pc;
} catch (error) {
throw new WebRTCError('PC_CREATE_FAILED', '创建PeerConnection失败', true, error as Error);
}
}
private setupEventHandlers(): void {
if (!this.pc) return;
this.pc.onicecandidate = (event) => {
if (event.candidate) {
this.localCandidates.push(event.candidate);
this.config.onSignalingMessage({
type: 'ice-candidate',
payload: event.candidate
});
} else {
console.log('[PeerConnection] ICE收集完成');
}
};
this.pc.oniceconnectionstatechange = () => {
console.log('[PeerConnection] ICE连接状态:', this.pc!.iceConnectionState);
switch (this.pc!.iceConnectionState) {
case 'connected':
case 'completed':
this.emit('state-change', { type: 'state-change', state: { isPeerConnected: true, error: null } });
break;
case 'failed':
this.emit('error', {
type: 'error',
error: new WebRTCError('ICE_FAILED', 'ICE连接失败', true)
});
break;
case 'disconnected':
this.emit('state-change', { type: 'state-change', state: { isPeerConnected: false } });
break;
}
};
this.pc.onconnectionstatechange = () => {
console.log('[PeerConnection] 连接状态:', this.pc!.connectionState);
switch (this.pc!.connectionState) {
case 'connected':
this.emit('state-change', { type: 'state-change', state: { isPeerConnected: true, error: null } });
break;
case 'failed':
this.emit('error', {
type: 'error',
error: new WebRTCError('CONNECTION_FAILED', 'WebRTC连接失败', true)
});
break;
case 'disconnected':
this.emit('state-change', { type: 'state-change', state: { isPeerConnected: false } });
break;
}
};
this.pc.ontrack = (event) => {
console.log('[PeerConnection] 收到轨道:', event.track.kind);
if (this.config.onTrack) {
this.config.onTrack(event);
}
};
this.pc.onsignalingstatechange = () => {
console.log('[PeerConnection] 信令状态:', this.pc!.signalingState);
if (this.pc!.signalingState === 'stable') {
this.isNegotiating = false;
this.processNegotiationQueue();
}
};
}
async createOffer(options: NegotiationOptions = {}): Promise<RTCSessionDescriptionInit> {
if (!this.pc) {
throw new WebRTCError('PC_NOT_READY', 'PeerConnection未准备就绪', false);
}
try {
const offerOptions: RTCOfferOptions = {
offerToReceiveAudio: options.offerToReceiveAudio ?? true,
offerToReceiveVideo: options.offerToReceiveVideo ?? true,
};
const offer = await this.pc.createOffer(offerOptions);
await this.pc.setLocalDescription(offer);
return offer;
} catch (error) {
throw new WebRTCError('OFFER_FAILED', '创建Offer失败', true, error as Error);
}
}
async createAnswer(): Promise<RTCSessionDescriptionInit> {
if (!this.pc) {
throw new WebRTCError('PC_NOT_READY', 'PeerConnection未准备就绪', false);
}
try {
const answer = await this.pc.createAnswer();
await this.pc.setLocalDescription(answer);
return answer;
} catch (error) {
throw new WebRTCError('ANSWER_FAILED', '创建Answer失败', true, error as Error);
}
}
async setRemoteDescription(description: RTCSessionDescriptionInit): Promise<void> {
if (!this.pc) {
throw new WebRTCError('PC_NOT_READY', 'PeerConnection未准备就绪', false);
}
try {
await this.pc.setRemoteDescription(description);
// 添加缓存的远程候选
for (const candidate of this.remoteCandidates) {
await this.pc.addIceCandidate(candidate);
}
this.remoteCandidates = [];
} catch (error) {
throw new WebRTCError('REMOTE_DESC_FAILED', '设置远程描述失败', false, error as Error);
}
}
async addIceCandidate(candidate: RTCIceCandidateInit): Promise<void> {
if (!this.pc) {
this.remoteCandidates.push(new RTCIceCandidate(candidate));
return;
}
try {
await this.pc.addIceCandidate(candidate);
} catch (error) {
console.warn('添加ICE候选失败:', error);
}
}
addTrack(track: MediaStreamTrack, stream: MediaStream): RTCRtpSender | null {
if (!this.pc) {
throw new WebRTCError('PC_NOT_READY', 'PeerConnection未准备就绪', false);
}
try {
return this.pc.addTrack(track, stream);
} catch (error) {
throw new WebRTCError('ADD_TRACK_FAILED', '添加轨道失败', false, error as Error);
}
}
removeTrack(sender: RTCRtpSender): void {
if (!this.pc) {
throw new WebRTCError('PC_NOT_READY', 'PeerConnection未准备就绪', false);
}
try {
this.pc.removeTrack(sender);
} catch (error) {
throw new WebRTCError('REMOVE_TRACK_FAILED', '移除轨道失败', false, error as Error);
}
}
createDataChannel(label: string, options?: RTCDataChannelInit): RTCDataChannel {
if (!this.pc) {
throw new WebRTCError('PC_NOT_READY', 'PeerConnection未准备就绪', false);
}
return this.pc.createDataChannel(label, options);
}
async renegotiate(options: NegotiationOptions = {}): Promise<void> {
if (!this.pc || this.isNegotiating) {
this.negotiationQueue.push(() => this.doRenegotiate(options));
return;
}
await this.doRenegotiate(options);
}
private async doRenegotiate(options: NegotiationOptions): Promise<void> {
if (!this.pc || this.isNegotiating) return;
this.isNegotiating = true;
try {
const offer = await this.createOffer(options);
this.config.onSignalingMessage({
type: 'offer',
payload: offer
});
} catch (error) {
this.isNegotiating = false;
throw error;
}
}
private processNegotiationQueue(): void {
if (this.negotiationQueue.length === 0) return;
const nextNegotiation = this.negotiationQueue.shift();
if (nextNegotiation) {
nextNegotiation().catch(error => {
console.error('处理协商队列失败:', error);
});
}
}
getStats(): Promise<RTCStatsReport> {
if (!this.pc) {
throw new WebRTCError('PC_NOT_READY', 'PeerConnection未准备就绪', false);
}
return this.pc.getStats();
}
getConnectionState(): RTCPeerConnectionState {
return this.pc?.connectionState || 'closed';
}
getIceConnectionState(): RTCIceConnectionState {
return this.pc?.iceConnectionState || 'closed';
}
getSignalingState(): RTCSignalingState {
return this.pc?.signalingState || 'stable';
}
destroyPeerConnection(): void {
if (this.pc) {
this.pc.close();
this.pc = null;
}
this.isNegotiating = false;
this.negotiationQueue = [];
this.localCandidates = [];
this.remoteCandidates = [];
this.emit('state-change', { type: 'state-change', state: { isPeerConnected: false } });
}
on(event: string, handler: EventHandler<ConnectionEvent>): this {
return super.on(event, handler);
}
emit(eventName: string, event?: ConnectionEvent): boolean {
return super.emit(eventName, event);
}
}

455
chuan-next/src/hooks/webrtc/core/WebRTCManager.ts Normal file
View File

@@ -0,0 +1,455 @@
import { EventEmitter } from 'events';
import { WebSocketManager } from './WebSocketManager';
import { PeerConnectionManager } from './PeerConnectionManager';
import { DataChannelManager } from './DataChannelManager';
import { MessageRouter } from './MessageRouter';
import {
WebRTCConnectionState,
WebRTCMessage,
WebRTCConfig,
WebRTCError,
MessageHandler,
DataHandler
} from './types';
import { getWsUrl } from '@/lib/config';
interface WebRTCManagerConfig extends Partial<WebRTCConfig> {
dataChannelName?: string;
enableLogging?: boolean;
}
interface SignalingMessage {
type: string;
payload: any;
}
export class WebRTCManager extends EventEmitter {
private wsManager: WebSocketManager;
private pcManager: PeerConnectionManager;
private dcManager: DataChannelManager;
private messageRouter: MessageRouter;
private config: WebRTCManagerConfig;
private state: WebRTCConnectionState;
private currentRoom: { code: string; role: 'sender' | 'receiver' } | null = null;
private isUserDisconnecting = false;
private abortController = new AbortController();
constructor(config: WebRTCManagerConfig = {}) {
super();
this.config = {
iceServers: [
{ urls: 'stun:stun.l.google.com:19302' },
{ urls: 'stun:stun1.l.google.com:19302' },
{ urls: 'stun:stun2.l.google.com:19302' },
{ urls: 'stun:global.stun.twilio.com:3478' },
],
iceCandidatePoolSize: 10,
chunkSize: 256 * 1024,
maxRetries: 5,
retryDelay: 1000,
ackTimeout: 5000,
dataChannelName: 'shared-channel',
enableLogging: true,
...config
};
this.state = {
isConnected: false,
isConnecting: false,
isWebSocketConnected: false,
isPeerConnected: false,
error: null,
canRetry: false,
currentRoom: null,
};
// 初始化各个管理器
this.wsManager = new WebSocketManager({
url: '',
reconnectAttempts: 5,
reconnectDelay: 1000,
timeout: 10000,
});
this.pcManager = new PeerConnectionManager({
iceServers: this.config.iceServers!,
iceCandidatePoolSize: this.config.iceCandidatePoolSize!,
chunkSize: this.config.chunkSize!,
maxRetries: this.config.maxRetries!,
retryDelay: this.config.retryDelay!,
ackTimeout: this.config.ackTimeout!,
onSignalingMessage: this.handleSignalingMessage.bind(this),
onTrack: this.handleTrack.bind(this),
});
this.dcManager = new DataChannelManager({
channelName: this.config.dataChannelName!,
onMessage: this.handleDataChannelMessage.bind(this),
onData: this.handleDataChannelData.bind(this),
});
this.messageRouter = new MessageRouter();
this.setupEventHandlers();
}
private setupEventHandlers(): void {
// WebSocket 事件处理
this.wsManager.on('connecting', (event) => {
this.updateState({ isConnecting: true, error: null });
});
this.wsManager.on('connected', (event) => {
this.updateState({
isWebSocketConnected: true,
isConnecting: false,
isConnected: true
});
});
this.wsManager.on('disconnected', (event: any) => {
this.updateState({ isWebSocketConnected: false });
if (!this.isUserDisconnecting) {
this.updateState({
error: event.reason,
canRetry: true
});
}
});
this.wsManager.on('error', (event: any) => {
this.updateState({
error: event.error.message,
canRetry: event.error.retryable
});
});
this.wsManager.on('message', (message: any) => {
this.handleWebSocketMessage(message);
});
// PeerConnection 事件处理
this.pcManager.on('state-change', (event: any) => {
this.updateState(event.state);
});
this.pcManager.on('error', (event: any) => {
this.updateState({
error: event.error.message,
canRetry: event.error.retryable
});
});
// DataChannel 事件处理
this.dcManager.on('state-change', (event: any) => {
this.updateState(event.state);
});
this.dcManager.on('error', (event: any) => {
this.updateState({
error: event.error.message,
canRetry: event.error.retryable
});
});
}
private updateState(updates: Partial<WebRTCConnectionState>): void {
this.state = { ...this.state, ...updates };
this.emit('state-change', { type: 'state-change', state: updates });
}
private handleWebSocketMessage(message: SignalingMessage): void {
if (this.config.enableLogging) {
console.log('[WebRTCManager] 收到信令消息:', message.type);
}
switch (message.type) {
case 'peer-joined':
this.handlePeerJoined(message.payload);
break;
case 'offer':
this.handleOffer(message.payload);
break;
case 'answer':
this.handleAnswer(message.payload);
break;
case 'ice-candidate':
this.handleIceCandidate(message.payload);
break;
case 'error':
this.handleError(message);
break;
case 'disconnection':
this.handleDisconnection(message);
break;
default:
if (this.config.enableLogging) {
console.warn('[WebRTCManager] 未知消息类型:', message.type);
}
}
}
private handleSignalingMessage(message: SignalingMessage): void {
this.wsManager.send(message);
}
private handleTrack(event: RTCTrackEvent): void {
if (this.config.enableLogging) {
console.log('[WebRTCManager] 收到媒体轨道:', event.track.kind);
}
// 这里可以添加轨道处理逻辑，或者通过事件传递给业务层
}
private handleDataChannelMessage(message: WebRTCMessage): void {
this.messageRouter.routeMessage(message);
}
private handleDataChannelData(data: ArrayBuffer): void {
// 默认路由到文件传输通道
this.messageRouter.routeData(data, 'file-transfer');
}
private async handlePeerJoined(payload: any): Promise<void> {
if (!this.currentRoom) return;
const { role } = payload;
const { role: currentRole } = this.currentRoom;
if (this.config.enableLogging) {
console.log('[WebRTCManager] 对方加入房间:', role);
}
if (currentRole === 'sender' && role === 'receiver') {
this.updateState({ isPeerConnected: true });
try {
await this.pcManager.createOffer();
} catch (error) {
console.error('[WebRTCManager] 创建Offer失败:', error);
}
} else if (currentRole === 'receiver' && role === 'sender') {
this.updateState({ isPeerConnected: true });
}
}
private async handleOffer(payload: RTCSessionDescriptionInit): Promise<void> {
try {
await this.pcManager.setRemoteDescription(payload);
const answer = await this.pcManager.createAnswer();
this.handleSignalingMessage({ type: 'answer', payload: answer });
} catch (error) {
console.error('[WebRTCManager] 处理Offer失败:', error);
this.updateState({
error: '处理连接请求失败',
canRetry: true
});
}
}
private async handleAnswer(payload: RTCSessionDescriptionInit): Promise<void> {
try {
await this.pcManager.setRemoteDescription(payload);
} catch (error) {
console.error('[WebRTCManager] 处理Answer失败:', error);
this.updateState({
error: '处理连接响应失败',
canRetry: true
});
}
}
private async handleIceCandidate(payload: RTCIceCandidateInit): Promise<void> {
try {
await this.pcManager.addIceCandidate(payload);
} catch (error) {
console.warn('[WebRTCManager] 添加ICE候选失败:', error);
}
}
private handleError(message: any): void {
this.updateState({
error: message.error || '信令服务器错误',
canRetry: true
});
}
private handleDisconnection(message: any): void {
this.updateState({
isPeerConnected: false,
error: '对方已离开房间',
canRetry: true
});
// 清理P2P连接但保持WebSocket连接
this.pcManager.destroyPeerConnection();
this.dcManager.close();
}
async connect(roomCode: string, role: 'sender' | 'receiver'): Promise<void> {
if (this.state.isConnecting) {
console.warn('[WebRTCManager] 正在连接中，跳过重复连接请求');
return;
}
this.isUserDisconnecting = false;
this.abortController = new AbortController();
try {
this.currentRoom = { code: roomCode, role };
this.updateState({
isConnecting: true,
error: null,
currentRoom: { code: roomCode, role }
});
// 创建PeerConnection
const pc = await this.pcManager.createPeerConnection();
// 如果是发送方，创建数据通道
if (role === 'sender') {
this.dcManager.createDataChannel(pc);
}
// 连接WebSocket
const baseWsUrl = getWsUrl();
if (!baseWsUrl) {
throw new WebRTCError('WS_URL_NOT_CONFIGURED', 'WebSocket URL未配置', false);
}
const wsUrl = baseWsUrl.replace('/ws/p2p', `/ws/webrtc?code=${roomCode}&role=${role}&channel=shared`);
this.wsManager = new WebSocketManager({
url: wsUrl,
reconnectAttempts: 5,
reconnectDelay: 1000,
timeout: 10000,
});
this.setupEventHandlers();
await this.wsManager.connect();
} catch (error) {
console.error('[WebRTCManager] 连接失败:', error);
this.updateState({
error: error instanceof WebRTCError ? error.message : '连接失败',
isConnecting: false,
canRetry: true
});
throw error;
}
}
disconnect(): void {
if (this.config.enableLogging) {
console.log('[WebRTCManager] 主动断开连接');
}
this.isUserDisconnecting = true;
this.abortController.abort();
// 通知对方断开连接
this.wsManager.send({
type: 'disconnection',
payload: { reason: '用户主动断开' }
});
// 清理所有连接
this.dcManager.close();
this.pcManager.destroyPeerConnection();
this.wsManager.disconnect();
this.messageRouter.clear();
// 重置状态
this.currentRoom = null;
this.updateState({
isConnected: false,
isConnecting: false,
isWebSocketConnected: false,
isPeerConnected: false,
error: null,
canRetry: false,
currentRoom: null,
});
}
async retry(): Promise<void> {
if (!this.currentRoom) {
throw new WebRTCError('NO_ROOM_INFO', '没有房间信息，无法重试', false);
}
if (this.config.enableLogging) {
console.log('[WebRTCManager] 重试连接:', this.currentRoom);
}
this.disconnect();
await this.connect(this.currentRoom.code, this.currentRoom.role);
}
sendMessage(message: WebRTCMessage, channel?: string): boolean {
const messageWithChannel = channel ? { ...message, channel } : message;
return this.dcManager.sendMessage(messageWithChannel);
}
sendData(data: ArrayBuffer): boolean {
return this.dcManager.sendData(data);
}
registerMessageHandler(channel: string, handler: MessageHandler): () => void {
return this.messageRouter.registerMessageHandler(channel, handler);
}
registerDataHandler(channel: string, handler: DataHandler): () => void {
return this.messageRouter.registerDataHandler(channel, handler);
}
addTrack(track: MediaStreamTrack, stream: MediaStream): RTCRtpSender | null {
return this.pcManager.addTrack(track, stream);
}
removeTrack(sender: RTCRtpSender): void {
this.pcManager.removeTrack(sender);
}
onTrack(handler: (event: RTCTrackEvent) => void): void {
// 简化实现，直接设置处理器
this.pcManager.on('track', (event) => {
if (event.type === 'state-change' && 'onTrack' in this.config) {
// 这里需要适配事件类型
}
});
}
getPeerConnection(): RTCPeerConnection | null {
// 返回内部 PeerConnection 的引用
return (this.pcManager as any).pc;
}
async createOfferNow(): Promise<boolean> {
try {
await this.pcManager.createOffer();
return true;
} catch (error) {
console.error('[WebRTCManager] 创建Offer失败:', error);
return false;
}
}
getChannelState(): RTCDataChannelState {
return this.dcManager.getState();
}
isConnectedToRoom(roomCode: string, role: 'sender' | 'receiver'): boolean {
return this.currentRoom?.code === roomCode &&
this.currentRoom?.role === role &&
this.state.isConnected;
}
getState(): WebRTCConnectionState {
return { ...this.state };
}
getConfig(): WebRTCManagerConfig {
return { ...this.config };
}
}

218
chuan-next/src/hooks/webrtc/core/WebSocketManager.ts Normal file
View File

@@ -0,0 +1,218 @@
import { EventEmitter } from 'events';
import { WebRTCError, WebRTCMessage, ConnectionEvent, EventHandler } from './types';
interface WebSocketManagerConfig {
url: string;
reconnectAttempts?: number;
reconnectDelay?: number;
timeout?: number;
}
interface WebSocketMessage {
type: string;
payload: any;
}
export class WebSocketManager extends EventEmitter {
private ws: WebSocket | null = null;
private config: WebSocketManagerConfig;
private reconnectCount = 0;
private isConnecting = false;
private isUserDisconnecting = false;
private messageQueue: WebSocketMessage[] = [];
private reconnectTimer: NodeJS.Timeout | null = null;
constructor(config: WebSocketManagerConfig) {
super();
this.config = {
reconnectAttempts: 5,
reconnectDelay: 1000,
timeout: 10000,
...config
};
}
async connect(): Promise<void> {
if (this.isConnecting || this.isConnected()) {
return;
}
this.isConnecting = true;
this.isUserDisconnecting = false;
this.emit('connecting', { type: 'connecting' });
try {
const ws = new WebSocket(this.config.url);
// 设置超时
const timeout = setTimeout(() => {
if (ws.readyState === WebSocket.CONNECTING) {
ws.close();
throw new WebRTCError('WS_TIMEOUT', 'WebSocket连接超时', true);
}
}, this.config.timeout);
await new Promise<void>((resolve, reject) => {
ws.onopen = () => {
clearTimeout(timeout);
this.isConnecting = false;
this.reconnectCount = 0;
this.ws = ws;
this.setupEventHandlers();
this.flushMessageQueue();
this.emit('connected', { type: 'connected' });
resolve();
};
ws.onerror = (errorEvent) => {
clearTimeout(timeout);
this.isConnecting = false;
const wsError = new WebRTCError('WS_ERROR', 'WebSocket连接错误', true, new Error('WebSocket连接错误'));
this.emit('error', { type: 'error', error: wsError });
reject(wsError);
};
ws.onclose = (closeEvent) => {
clearTimeout(timeout);
this.isConnecting = false;
this.ws = null;
if (!this.isUserDisconnecting) {
this.handleReconnect();
}
this.emit('disconnected', {
type: 'disconnected',
reason: `WebSocket关闭: ${closeEvent.code} - ${closeEvent.reason}`
});
};
});
} catch (error) {
this.isConnecting = false;
if (error instanceof WebRTCError) {
this.emit('error', { type: 'error', error });
throw error;
}
throw new WebRTCError('WS_CONNECTION_FAILED', 'WebSocket连接失败', true, error as Error);
}
}
private setupEventHandlers(): void {
if (!this.ws) return;
this.ws.onmessage = (event) => {
try {
const message = JSON.parse(event.data);
this.emit('message', message);
} catch (error) {
console.error('解析WebSocket消息失败:', error);
this.emit('error', {
type: 'error',
error: new WebRTCError('WS_MESSAGE_PARSE_ERROR', '消息解析失败', false, error as Error)
});
}
};
this.ws.onerror = (errorEvent) => {
this.emit('error', {
type: 'error',
error: new WebRTCError('WS_ERROR', 'WebSocket错误', true, new Error('WebSocket错误'))
});
};
this.ws.onclose = (event) => {
this.ws = null;
if (!this.isUserDisconnecting) {
this.handleReconnect();
}
this.emit('disconnected', {
type: 'disconnected',
reason: `WebSocket关闭: ${event.code} - ${event.reason}`
});
};
}
private handleReconnect(): void {
if (this.reconnectCount >= this.config.reconnectAttempts!) {
this.emit('error', {
type: 'error',
error: new WebRTCError('WS_RECONNECT_FAILED', '重连失败，已达最大重试次数', false)
});
return;
}
const delay = this.config.reconnectDelay! * Math.pow(2, this.reconnectCount);
this.reconnectCount++;
console.log(`WebSocket重连中... (${this.reconnectCount}/${this.config.reconnectAttempts})，延迟: ${delay}ms`);
this.reconnectTimer = setTimeout(() => {
this.emit('retry', { type: 'retry' });
this.connect().catch(error => {
console.error('WebSocket重连失败:', error);
});
}, delay);
}
send(message: WebSocketMessage): boolean {
if (!this.isConnected()) {
this.messageQueue.push(message);
return false;
}
try {
this.ws!.send(JSON.stringify(message));
return true;
} catch (error) {
console.error('发送WebSocket消息失败:', error);
this.emit('error', {
type: 'error',
error: new WebRTCError('WS_SEND_ERROR', '发送消息失败', true, error as Error)
});
return false;
}
}
private flushMessageQueue(): void {
while (this.messageQueue.length > 0) {
const message = this.messageQueue.shift();
if (message) {
this.send(message);
}
}
}
disconnect(): void {
this.isUserDisconnecting = true;
if (this.reconnectTimer) {
clearTimeout(this.reconnectTimer);
this.reconnectTimer = null;
}
if (this.ws) {
this.ws.close(1000, '用户主动断开');
this.ws = null;
}
this.messageQueue = [];
this.isConnecting = false;
}
isConnected(): boolean {
return this.ws !== null && this.ws.readyState === WebSocket.OPEN;
}
isConnectingState(): boolean {
return this.isConnecting;
}
on(event: string, handler: EventHandler<ConnectionEvent>): this {
return super.on(event, handler);
}
emit(eventName: string, event?: ConnectionEvent): boolean {
return super.emit(eventName, event);
}
}

58
chuan-next/src/hooks/webrtc/core/types.ts Normal file
View File

@@ -0,0 +1,58 @@
// WebRTC 核心类型定义
// 基础连接状态
export interface WebRTCConnectionState {
isConnected: boolean;
isConnecting: boolean;
isWebSocketConnected: boolean;
isPeerConnected: boolean;
error: string | null;
canRetry: boolean;
currentRoom: { code: string; role: 'sender' | 'receiver' } | null;
}
// 消息类型
export interface WebRTCMessage<T = any> {
type: string;
payload: T;
channel?: string;
}
// 消息处理器类型
export type MessageHandler = (message: WebRTCMessage) => void;
export type DataHandler = (data: ArrayBuffer) => void;
// WebRTC 配置
export interface WebRTCConfig {
iceServers: RTCIceServer[];
iceCandidatePoolSize: number;
chunkSize: number;
maxRetries: number;
retryDelay: number;
ackTimeout: number;
}
// 错误类型
export class WebRTCError extends Error {
constructor(
public code: string,
message: string,
public retryable: boolean = false,
public cause?: Error
) {
super(message);
this.name = 'WebRTCError';
}
}
// 连接事件
export type ConnectionEvent =
| { type: 'connecting' }
| { type: 'connected' }
| { type: 'disconnected'; reason?: string }
| { type: 'error'; error: WebRTCError }
| { type: 'retry' }
| { type: 'state-change'; state: Partial<WebRTCConnectionState> };
// 事件处理器
export type EventHandler<T extends ConnectionEvent> = (event: T) => void;

109
chuan-next/src/hooks/webrtc/useFileTransferBusiness.ts
View File

@@ -87,14 +87,14 @@ const ACK_TIMEOUT = 5000; // 确认超时（毫秒）
function calculateChecksum(data: ArrayBuffer): string {
const buffer = new Uint8Array(data);
let crc = 0xFFFFFFFF;
for (let i = 0; i < buffer.length; i++) {
crc ^= buffer[i];
for (let j = 0; j < 8; j++) {
crc = crc & 1 ? (crc >>> 1) ^ 0xEDB88320 : crc >>> 1;
}
}
return (crc ^ 0xFFFFFFFF).toString(16).padStart(8, '0');
}
@@ -104,11 +104,11 @@ function calculateChecksum(data: ArrayBuffer): string {
function simpleChecksum(data: ArrayBuffer): string {
const buffer = new Uint8Array(data);
let sum = 0;
for (let i = 0; i < Math.min(buffer.length, 1000); i++) {
sum += buffer[i];
}
return sum.toString(16);
}
@@ -161,12 +161,12 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
// 消息处理器
const handleMessage = useCallback((message: any) => {
if (!message.type.startsWith('file-')) return;
console.log('文件传输收到消息:', message.type, message); switch (message.type) {
console.log('文件传输收到消息:', message.type, message); switch (message.type) {
case 'file-metadata':
const metadata: FileMetadata = message.payload;
console.log('开始接收文件:', metadata.name);
receivingFiles.current.set(metadata.id, {
metadata,
chunks: [],
@@ -182,7 +182,7 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
totalChunks,
progress: 0
});
// 设置当前活跃的接收文件
activeReceiveFile.current = metadata.id;
updateState({ isTransferring: true, progress: 0 });
@@ -196,16 +196,16 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
case 'file-complete':
const { fileId } = message.payload;
const fileInfo = receivingFiles.current.get(fileId);
if (fileInfo) {
// 组装文件
const blob = new Blob(fileInfo.chunks, { type: fileInfo.metadata.type });
const file = new File([blob], fileInfo.metadata.name, {
type: fileInfo.metadata.type
const file = new File([blob], fileInfo.metadata.name, {
type: fileInfo.metadata.type
});
console.log('文件接收完成:', file.name);
setState(prev => ({
...prev,
receivedFiles: [...prev.receivedFiles, { id: fileId, file }],
@@ -216,7 +216,7 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
fileReceivedCallbacks.current.forEach(cb => cb({ id: fileId, file }));
receivingFiles.current.delete(fileId);
receiveProgress.current.delete(fileId);
// 清除活跃文件
if (activeReceiveFile.current === fileId) {
activeReceiveFile.current = null;
@@ -238,7 +238,7 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
case 'file-chunk-ack':
const ack: ChunkAck = message.payload;
console.log('收到块确认:', ack);
// 清除超时定时器
const chunkKey = `${ack.fileId}-${ack.chunkIndex}`;
const timeout = pendingChunks.current.get(chunkKey);
@@ -277,15 +277,15 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
const { fileId, chunkIndex, totalChunks, checksum: expectedChecksum } = expectedChunk.current;
const fileInfo = receivingFiles.current.get(fileId);
if (fileInfo) {
// 验证数据完整性
const actualChecksum = calculateChecksum(data);
const isValid = !expectedChecksum || actualChecksum === expectedChecksum;
if (!isValid) {
console.warn(`文件块校验失败: 期望 ${expectedChecksum}, 实际 ${actualChecksum}`);
// 发送失败确认
connection.sendMessage({
type: 'file-chunk-ack',
@@ -296,7 +296,7 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
checksum: actualChecksum
}
}, CHANNEL_NAME);
expectedChunk.current = null;
return;
}
@@ -309,14 +309,14 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
const progressInfo = receiveProgress.current.get(fileId);
if (progressInfo) {
progressInfo.receivedChunks++;
progressInfo.progress = progressInfo.totalChunks > 0 ?
progressInfo.progress = progressInfo.totalChunks > 0 ?
(progressInfo.receivedChunks / progressInfo.totalChunks) * 100 : 0;
// 只有当这个文件是当前活跃文件时才更新全局进度
if (activeReceiveFile.current === fileId) {
updateState({ progress: progressInfo.progress });
}
// 触发进度回调
fileProgressCallbacks.current.forEach(cb => cb({
fileId: fileId,
@@ -326,7 +326,7 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
console.log(`文件 ${progressInfo.fileName} 接收进度: ${progressInfo.progress.toFixed(1)}%`);
}
// 发送成功确认
connection.sendMessage({
type: 'file-chunk-ack',
@@ -337,22 +337,26 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
checksum: actualChecksum
}
}, CHANNEL_NAME);
expectedChunk.current = null;
}
}, [updateState, connection]);
// 设置处理器 - 使用稳定的引用避免反复注册
const connectionRef = useRef(connection);
useEffect(() => {
connectionRef.current = connection;
}, [connection]);
useEffect(() => {
// 使用共享连接的注册方式
const unregisterMessage = connection.registerMessageHandler(CHANNEL_NAME, handleMessage);
const unregisterData = connection.registerDataHandler(CHANNEL_NAME, handleData);
const unregisterMessage = connectionRef.current.registerMessageHandler(CHANNEL_NAME, handleMessage);
const unregisterData = connectionRef.current.registerDataHandler(CHANNEL_NAME, handleData);
return () => {
unregisterMessage();
unregisterData();
};
}, [connection]); // 只依赖 connection 对象，不依赖处理函数
}, []); // 只依赖 connection 对象，不依赖处理函数
// 监听连接状态变化 (直接使用 connection 的状态)
useEffect(() => {
@@ -379,8 +383,21 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
retryCount = 0
): Promise<boolean> => {
return new Promise((resolve) => {
// 主要检查数据通道状态，因为数据通道是文件传输的实际通道
const channelState = connection.getChannelState();
if (channelState === 'closed') {
console.warn(`数据通道已关闭，停止发送文件块 ${chunkIndex}`);
resolve(false);
return;
}
// 如果连接暂时断开但数据通道可用，仍然可以尝试发送
if (!connection.isConnected && channelState === 'connecting') {
console.warn(`WebRTC 连接暂时断开，但数据通道正在连接，继续尝试发送文件块 ${chunkIndex}`);
}
const chunkKey = `${fileId}-${chunkIndex}`;
// 设置确认回调
const ackCallback = (ack: ChunkAck) => {
if (ack.success) {
@@ -468,6 +485,18 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
let retryCount = 0;
while (!success && retryCount <= MAX_RETRIES) {
// 检查数据通道状态，这是文件传输的实际通道
const channelState = connection.getChannelState();
if (channelState === 'closed') {
console.warn(`数据通道已关闭，停止文件传输`);
throw new Error('数据通道已关闭');
}
// 如果连接暂时断开但数据通道可用，仍然可以尝试发送
if (!connection.isConnected && channelState === 'connecting') {
console.warn(`WebRTC 连接暂时断开，但数据通道正在连接，继续尝试发送文件块 ${chunkIndex}`);
}
const start = chunkIndex * CHUNK_SIZE;
const end = Math.min(start + CHUNK_SIZE, file.size);
const chunk = file.slice(start, end);
@@ -483,7 +512,7 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
status.sentChunks.add(chunkIndex);
status.acknowledgedChunks.add(chunkIndex);
status.failedChunks.delete(chunkIndex);
// 计算传输速度
const now = Date.now();
const timeDiff = (now - status.lastChunkTime) / 1000; // 秒
@@ -495,12 +524,12 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
} else {
retryCount++;
status.retryCount.set(chunkIndex, retryCount);
if (retryCount > MAX_RETRIES) {
status.failedChunks.add(chunkIndex);
throw new Error(`文件块 ${chunkIndex} 发送失败，超过最大重试次数`);
}
// 指数退避
const delay = Math.min(RETRY_DELAY * Math.pow(2, retryCount - 1), 10000);
console.log(`等待 ${delay}ms 后重试文件块 ${chunkIndex}`);
@@ -511,7 +540,7 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
// 更新进度
const progress = (status.acknowledgedChunks.size / totalChunks) * 100;
updateState({ progress });
fileProgressCallbacks.current.forEach(cb => cb({
fileId: actualFileId,
fileName: file.name,
@@ -524,7 +553,7 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
const expectedTime = (chunkSize / 1024) / status.averageSpeed;
const actualTime = Date.now() - status.lastChunkTime;
const delay = Math.max(0, expectedTime - actualTime);
if (delay > 10) {
await new Promise(resolve => setTimeout(resolve, Math.min(delay, 100)));
}
@@ -548,9 +577,9 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
} catch (error) {
console.error('安全发送文件失败:', error);
updateState({
updateState({
error: error instanceof Error ? error.message : '发送失败',
isTransferring: false
isTransferring: false
});
transferStatus.current.delete(actualFileId);
}
@@ -567,17 +596,17 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
// 检查连接状态 - 优先检查数据通道状态，因为 P2P 连接可能已经建立但状态未及时更新
const channelState = connection.getChannelState();
const peerConnected = connection.isPeerConnected;
console.log('发送文件列表检查:', {
channelState,
peerConnected,
fileListLength: fileList.length
});
// 如果数据通道已打开或者 P2P 已连接，就可以发送文件列表
if (channelState === 'open' || peerConnected) {
console.log('发送文件列表:', fileList);
connection.sendMessage({
type: 'file-list',
payload: fileList
@@ -595,7 +624,7 @@ export function useFileTransferBusiness(connection: WebRTCConnection) {
}
console.log('请求文件:', fileName, fileId);
connection.sendMessage({
type: 'file-request',
payload: { fileId, fileName }

23
chuan-next/src/hooks/webrtc/useSharedWebRTCManager.ts
View File

@@ -762,15 +762,32 @@ export function useSharedWebRTCManager(): WebRTCConnection {
const pc = pcRef.current;
if (!pc) {
console.warn('[SharedWebRTC] PeerConnection 尚未准备就绪，将在连接建立后设置onTrack');
// 检查WebSocket连接状态，只有连接后才尝试设置
if (!webrtcStore.isWebSocketConnected) {
console.log('[SharedWebRTC] WebSocket未连接，等待连接建立...');
return;
}
// 延迟设置，等待PeerConnection准备就绪
let retryCount = 0;
const maxRetries = 30; // 最多重试30次，即3秒
const checkAndSetTrackHandler = () => {
const currentPc = pcRef.current;
if (currentPc) {
console.log('[SharedWebRTC] ✅ PeerConnection 已准备就绪，设置onTrack处理器');
currentPc.ontrack = handler;
} else {
console.log('[SharedWebRTC] ⏳ 等待PeerConnection准备就绪...');
setTimeout(checkAndSetTrackHandler, 100);
retryCount++;
if (retryCount < maxRetries) {
// 只在偶数次重试时输出日志，减少日志数量
if (retryCount % 2 === 0) {
console.log(`[SharedWebRTC] ⏳ 等待PeerConnection准备就绪... (尝试: ${retryCount}/${maxRetries})`);
}
setTimeout(checkAndSetTrackHandler, 100);
} else {
console.error('[SharedWebRTC] ❌ PeerConnection 长时间未准备就绪，停止重试');
}
}
};
checkAndSetTrackHandler();
@@ -779,7 +796,7 @@ export function useSharedWebRTCManager(): WebRTCConnection {
console.log('[SharedWebRTC] ✅ 立即设置onTrack处理器');
pc.ontrack = handler;
}, []);
}, [webrtcStore.isWebSocketConnected]);
// 获取PeerConnection实例
const getPeerConnection = useCallback(() => {

195
chuan-next/src/hooks/webrtc/useWebRTCManager.ts Normal file
View File

@@ -0,0 +1,195 @@
import { useState, useEffect, useCallback, useRef } from 'react';
import { WebRTCManager } from './core/WebRTCManager';
import { WebRTCConnectionState, WebRTCMessage, MessageHandler, DataHandler } from './core/types';
import { WebRTCConnection } from './useSharedWebRTCManager';
interface WebRTCManagerConfig {
dataChannelName?: string;
enableLogging?: boolean;
iceServers?: RTCIceServer[];
iceCandidatePoolSize?: number;
chunkSize?: number;
maxRetries?: number;
retryDelay?: number;
ackTimeout?: number;
}
/**
* 新的 WebRTC 管理器 Hook
* 替代原有的 useSharedWebRTCManager，提供更好的架构和错误处理
*/
export function useWebRTCManager(config: WebRTCManagerConfig = {}): WebRTCConnection {
const managerRef = useRef<WebRTCManager | null>(null);
const [state, setState] = useState<WebRTCConnectionState>({
isConnected: false,
isConnecting: false,
isWebSocketConnected: false,
isPeerConnected: false,
error: null,
canRetry: false,
currentRoom: null,
});
// 初始化管理器
useEffect(() => {
if (!managerRef.current) {
managerRef.current = new WebRTCManager(config);
// 监听状态变化
managerRef.current.on('state-change', (event: any) => {
setState(prev => ({ ...prev, ...event.state }));
});
}
return () => {
if (managerRef.current) {
managerRef.current.disconnect();
managerRef.current = null;
}
};
}, [config]);
// 连接
const connect = useCallback(async (roomCode: string, role: 'sender' | 'receiver') => {
if (!managerRef.current) {
throw new Error('WebRTC 管理器未初始化');
}
return managerRef.current.connect(roomCode, role);
}, []);
// 断开连接
const disconnect = useCallback(() => {
if (!managerRef.current) return;
managerRef.current.disconnect();
}, []);
// 重试连接
const retry = useCallback(async () => {
if (!managerRef.current) {
throw new Error('WebRTC 管理器未初始化');
}
return managerRef.current.retry();
}, []);
// 发送消息
const sendMessage = useCallback((message: WebRTCMessage, channel?: string) => {
if (!managerRef.current) return false;
return managerRef.current.sendMessage(message, channel);
}, []);
// 发送数据
const sendData = useCallback((data: ArrayBuffer) => {
if (!managerRef.current) return false;
return managerRef.current.sendData(data);
}, []);
// 注册消息处理器
const registerMessageHandler = useCallback((channel: string, handler: MessageHandler) => {
if (!managerRef.current) return () => {};
return managerRef.current.registerMessageHandler(channel, handler);
}, []);
// 注册数据处理器
const registerDataHandler = useCallback((channel: string, handler: DataHandler) => {
if (!managerRef.current) return () => {};
return managerRef.current.registerDataHandler(channel, handler);
}, []);
// 添加媒体轨道
const addTrack = useCallback((track: MediaStreamTrack, stream: MediaStream) => {
if (!managerRef.current) return null;
return managerRef.current.addTrack(track, stream);
}, []);
// 移除媒体轨道
const removeTrack = useCallback((sender: RTCRtpSender) => {
if (!managerRef.current) return;
managerRef.current.removeTrack(sender);
}, []);
// 设置轨道处理器
const onTrack = useCallback((handler: (event: RTCTrackEvent) => void) => {
if (!managerRef.current) return;
managerRef.current.onTrack(handler);
}, []);
// 获取 PeerConnection
const getPeerConnection = useCallback(() => {
if (!managerRef.current) return null;
return managerRef.current.getPeerConnection();
}, []);
// 立即创建 offer
const createOfferNow = useCallback(async () => {
if (!managerRef.current) return false;
return managerRef.current.createOfferNow();
}, []);
// 获取数据通道状态
const getChannelState = useCallback(() => {
if (!managerRef.current) return 'closed';
return managerRef.current.getChannelState();
}, []);
// 检查是否已连接到指定房间
const isConnectedToRoom = useCallback((roomCode: string, role: 'sender' | 'receiver') => {
if (!managerRef.current) return false;
return managerRef.current.isConnectedToRoom(roomCode, role);
}, []);
return {
// 状态
isConnected: state.isConnected,
isConnecting: state.isConnecting,
isWebSocketConnected: state.isWebSocketConnected,
isPeerConnected: state.isPeerConnected,
error: state.error,
canRetry: state.canRetry,
// 操作方法
connect,
disconnect,
retry,
sendMessage,
sendData,
// 处理器注册
registerMessageHandler,
registerDataHandler,
// 工具方法
getChannelState,
isConnectedToRoom,
// 媒体轨道方法
addTrack,
removeTrack,
onTrack,
getPeerConnection,
createOfferNow,
// 当前房间信息
currentRoom: state.currentRoom,
};
}
/**
* 迁移辅助 Hook - 提供向后兼容性
* 可以逐步将现有代码迁移到新的架构
*/
export function useWebRTCMigration() {
const newManager = useWebRTCManager();
// 注意：这里需要先创建一个包装器来兼容旧的接口
// 暂时注释掉，避免循环依赖
// const oldManager = useSharedWebRTCManager();
return {
newManager,
// oldManager, // 暂时禁用
// 可以添加迁移工具函数
migrateState: () => {
// 将旧状态迁移到新状态
console.log('状态迁移功能待实现');
},
};
}