192 lines
8.0 KiB
JavaScript
192 lines
8.0 KiB
JavaScript
export default class Entanglement {
|
|
constructor({ options }) {
|
|
this.event = {}
|
|
this.options = options
|
|
this.store = {}
|
|
this.users = []
|
|
this.channels = [{ name: 'json' }]
|
|
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, '发生错误')
|
|
}
|
|
}
|
|
// 创建数据通道
|
|
user.channels = this.channels.map(item => {
|
|
const channel = pc.createDataChannel(item.name, { 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 { channel, ...item }
|
|
})
|
|
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.channels.find(item => item.name === channel_name).channel
|
|
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]
|
|
}
|
|
}
|