export default class Entanglement { constructor({ options, server }) { this.event = {} this.options = options this.server = server this.channels = {} this.client = {} this.store = {} this.users = [] this.account = {} this.ws = this.__create_websocket() } // 当创建一个对象时,会调用这个方法 // 当创建一个对象时,所有连接在一起的终端都会同步这个对象 // 并不是所有的终端都会监听全局变化, 因此需要在这里注册监听(频道) async __create_webrtc(user) { const pc = new RTCPeerConnection(this.options) // 当有新的媒体流加入时触发 pc.onicecandidate = (event) => { if (event.candidate) { console.debug(user.name, '发出 candidate 候选通道') this.ws.send(JSON.stringify({ type: 'candidate', user, candidate: event.candidate })) } } // 当连接状态发生改变时触发 pc.oniceconnectionstatechange = (event) => { if (pc.iceConnectionState === 'disconnected' || pc.iceConnectionState === 'failed') { console.error(user.name, '需要添加新的 candidate') } else if (pc.iceConnectionState === 'connected' || pc.iceConnectionState === 'completed') { console.debug(user.name, 'WebRTC 连接已经建立成功') } } // 协商新的会话, 建立初始连接或在网络条件发生变化后重新协商连接 pc.onnegotiationneeded = async (event) => { console.log('onnegotiationneeded', event) //const offer = await pc.createOffer() //await pc.setLocalDescription(offer) //this.ws.send(JSON.stringify({ type: 'offer', user, offer })) } // 当有新的媒体流加入时触发 pc.ontrack = (event) => { console.log('ontrack', event) } // 当有新的数据通道加入时触发 pc.ondatachannel = event => { console.log(user.name, '建立', event.channel.label, '通道') event.channel.onmessage = event => { console.log(user.name, '收到', event.target.label, '通道消息', event.data) } event.channel.onopen = () => { console.log(user.name, '打开', event.channel.label, '通道') event.channel.send('hello') } event.channel.onclose = () => { console.log(user.name, '关闭', event.channel.label, '通道') } event.channel.onerror = () => { console.log(user.name, '通道', event.channel.label, '发生错误') } } // 创建数据通道 const channel = pc.createDataChannel('json', { reliable: true }) channel.onopen = () => { console.log('打开数据通道') channel.send(JSON.stringify({ name: 'sato', hello: 'world' })) } channel.onmessage = event => { console.log('收到数据通道消息', event.data) } channel.onclose = () => { console.log('关闭数据通道') } channel.onerror = () => { console.log('数据通道发生错误') } return pc } async __create_websocket() { const host = window.location.host const protocol = window.location.protocol === 'https:' ? 'wss' : 'ws' const ws = new WebSocket(`${protocol}://${host}/entanglement?name=sato&channel=chat`) ws.onopen = async () => { console.log('websocket 连接成功') if (this.ws instanceof Promise) { this.ws = await this.ws } } ws.onclose = async () => { console.log('websocket 连接关闭, 3s后尝试重新连接...') await new Promise(resolve => setTimeout(resolve, 3000)) this.ws = await this.__create_websocket() } ws.onerror = async () => { console.log('websocket 连接错误, 3s后尝试重新连接...') await new Promise(resolve => setTimeout(resolve, 3000)) this.ws = await this.__create_websocket() } ws.onmessage = async event => { const data = JSON.parse(event.data) if (data.type === 'list') { console.debug('收到在线列表', data.list) this.users = await Promise.all(data.list.map(async user => { console.debug('发送给', user.name, 'offer') const pc = await this.__create_webrtc(user) const offer = await pc.createOffer() await pc.setLocalDescription(offer) this.ws.send(JSON.stringify({ type: 'offer', user, offer })) return { ...user, webrtc: pc } })) return } if (data.type === 'push') { console.debug(data.user.name, '上线', data) return } if (data.type === 'pull') { console.debug(data.user.name, '下线', data) return } if (data.type === 'offer') { console.debug(data.user.name, '发来 offer') const pc = await this.__create_webrtc(data.user) await pc.setRemoteDescription(data.offer) const answer = await pc.createAnswer() await pc.setLocalDescription(answer) this.ws.send(JSON.stringify({ type: 'answer', user: data.user, answer })) this.users.push({ ...data.user, webrtc: pc }) return } if (data.type === 'answer') { console.debug(data.user.name, '发来 answer') const pc = this.users.find(user => user.id === data.user.id).webrtc await pc.setRemoteDescription(data.answer) return } if (data.type === 'candidate') { console.debug(data.user.name, '发来 candidate 候选通道') const pc = this.users.find(user => user.id === data.user.id).webrtc await pc.addIceCandidate(data.candidate) return } console.error('收到未知数据:', data) } return ws } // 向所有在线的用户广播消息(webrtc) send_all(channel_name, data) { console.log('向', channel_name, '频道广播消息:', data) console.log('在线用户:', this.users) this.users.filter(user => { const status = user.webrtc.iceConnectionState console.log('用户', user.name, '连接状态:', status) return status === 'connected' || status === 'completed' }).forEach(user => { console.log('向', user.name, '发送', channel_name, '频道消息:', data) const channel = user.webrtc.createDataChannel(channel_name, { reliable: true }) channel.onopen = () => channel.send(JSON.stringify(data)) }) } // 数据被修改时触发 set(name, data) { // 递归创建代理对象 const createDeepProxy = (obj, path = []) => { const proxy = new Proxy(obj, { set: (target, key, value) => { if (!Array.isArray(target) || key !== 'length') { console.log('对象被修改', [...path, key], value) this.send_all(name, { key: [...path, key], value }) // 向所有在线的用户广播消息 } return Reflect.set(target, key, value) }, get: (target, key) => { return target[key] } }) Object.keys(obj).forEach(key => { if (typeof obj[key] === 'object') { obj[key] = createDeepProxy(obj[key], [...path, key]) } }) return proxy } return Reflect.set(this.store, name, createDeepProxy(data)) } // 读取一个通道 get(name) { return this.store[name] } }