entanglement

index.js

@@ -70,6 +70,51 @@ app.ws('/webrtc/:channel', (ws, req) => {
     })
 })
 
+// Entanglement
+app.ws('/entanglement', (ws, req) => {
+    ws.id = req.headers['sec-websocket-key']
+    ws.channel = req.params.channel
+    ws.name = req.query.name
+    console.log('ws.name:', ws.name)
+    // 设备离开频道时广播给所有在线设备
+    ws.on('close', () => {
+        console.log(ws.id, '设备离开频道:', ws.channel, wsInstance.getWss().clients.size)
+        wsInstance.getWss().clients.forEach(client => {
+            if (client !== ws && client.readyState === 1 && client.channel === ws.channel) {
+                client.send(JSON.stringify({ type: 'pull', id: ws.id, channel: ws.channel }))
+            }
+        })
+    })
+    // 设备发生错误时广播给所有在线设备
+    ws.on('error', () => {
+        console.log(ws.id, '设备发生错误:', ws.channel, wsInstance.getWss().clients.size)
+        wsInstance.getWss().clients.forEach(client => {
+            if (client !== ws && client.readyState === 1 && client.channel === ws.channel) {
+                client.send(JSON.stringify({ type: 'error', id: ws.id, channel: ws.channel }))
+            }
+        })
+    })
+    // 设备发送信令时转发给指定在线设备
+    ws.on('message', message => {
+        console.log(ws.id, '设备发送信令:', ws.channel, wsInstance.getWss().clients.size)
+        const data = JSON.parse(message)
+        wsInstance.getWss().clients.forEach(client => {
+            if (client !== ws && client.readyState === 1 && client.channel === ws.channel && client.id === data.id) {
+                client.send(JSON.stringify({ ...data, id: ws.id, name: ws.name }))
+            }
+        })
+    })
+    // 设备加入频道时广播给所有在线设备(也获取所有在线设备)
+    console.log(ws.id, '设备加入频道:', ws.channel, wsInstance.getWss().clients.size)
+    wsInstance.getWss().clients.forEach(client => {
+        if (client !== ws && client.readyState === 1 && client.channel === ws.channel) {
+            console.log(ws.name, '广播给在线设备:', client.name)
+            client.send(JSON.stringify({ type: 'push', id: ws.id, name: ws.name, channel: ws.channel }))
+            ws.send(JSON.stringify({ type: 'list', id: client.id, name: client.name, channel: client.channel }))
+        }
+    })
+})
+
 // WEBHOOK 处理 GitHub 事件
 app.post('/webhook', (req, res) => {
     console.log('WEBHOOK:', req.body)

public/client.js

@@ -96,8 +96,6 @@ export default class ClientList {
                             }))
                         }
                     }
-                    webrtc.getReceivers
-                    webrtc.getTransceivers
                     webrtc.oniceconnectionstatechange = async event => {
                         if (webrtc.iceConnectionState === 'disconnected' || webrtc.iceConnectionState === 'failed') {
                             console.error(data.name, '需要添加新的 candidate')

public/entanglement.js

@@ -0,0 +1,165 @@
+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_websocket() {
+        const host = window.location.host
+        const protocol = window.location.protocol === 'https:' ? 'wss' : 'ws'
+        const ws = new WebSocket(`${protocol}://${host}/entanglement`)
+        ws.onopen = () => console.log('websocket 连接成功')
+        ws.onclose = async () => {
+            console.log('websocket 连接关闭, 3s后尝试重新连接...')
+            await new Promise(resolve => setTimeout(resolve, 3000))
+            this.ws = this.__create_websocket()
+        }
+        ws.onerror = async () => {
+            console.log('websocket 连接错误, 3s后尝试重新连接...')
+            await new Promise(resolve => setTimeout(resolve, 3000))
+            this.ws = this.__create_websocket()
+        }
+        ws.onmessage = async event => {
+            const data = JSON.parse(event.data)
+            //console.log('收到消息', data)
+            if (data.type === 'list') {
+                console.debug('收到已在线对端列表', data.list)
+                this.users = Promise.all(data.list.map(async user => {
+                    console.debug('发送给', user.name, 'offer')
+                    const pc = this.__create_webrtc()
+                    const offer = await pc.createOffer()
+                    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 = this.__create_webrtc()
+                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 === 'icecandidate') {
+                console.debug(data.user.name, '发来 icecandidate 候选通道')
+                const pc = this.users.find(user => user.id === data.user.id).webrtc
+                await pc.addIceCandidate(data.icecandidate)
+                return
+            }
+            console.error('收到未知数据:', data)
+        }
+        return ws
+    }
+
+    __create_webrtc() {
+        const pc = new RTCPeerConnection(this.options)
+        pc.onicecandidate = (event) => {
+            if (event.candidate) {
+                this.ws.send(JSON.stringify({ type: 'icecandidate', user: this.account, icecandidate: event.candidate }))
+            }
+        }
+        pc.oniceconnectionstatechange = (event) => {
+            if (webrtc.iceConnectionState === 'disconnected' || webrtc.iceConnectionState === 'failed') {
+                console.error(data.name, '需要添加新的 candidate')
+            } else if (webrtc.iceConnectionState === 'connected' || webrtc.iceConnectionState === 'completed') {
+                console.debug(data.name, 'WebRTC 连接已经建立成功')
+            }
+        }
+        pc.onnegotiationneeded = (event) => {
+            console.log('onnegotiationneeded', event)
+        }
+        pc.ontrack = (event) => {
+            console.log('ontrack', event)
+        }
+        pc.ondatachannel = event => {
+            console.log(data.user.name, '建立', event.channel.label, '通道')
+            event.channel.onmessage = event => {
+                console.log(data.user.name, '收到', event.channel.label, '通道消息', event.data)
+            }
+            event.channel.onopen = () => {
+                console.log(data.user.name, '打开', event.channel.label, '通道')
+                event.channel.send('hello')
+            }
+            event.channel.onclose = () => {
+                console.log(data.user.name, '关闭', event.channel.label, '通道')
+            }
+            event.channel.onerror = () => {
+                console.log(data.user.name, '通道', event.channel.label, '发生错误')
+            }
+        }
+        return pc
+    }
+
+    // 向所有在线的用户广播消息(webrtc)
+    send_all(channel_name, data) {
+        console.log('向', channel_name, '频道广播消息:', data)
+        this.users.filter(user => {
+            const status = user.webrtc.iceConnectionState
+            return status === 'connected' || status === 'completed'
+        }).forEach(user => {
+            console.log('向', user.name, '发送', channel_name, '频道消息:', data)
+            const channel = user.webrtc.createDataChannel(channel_name)
+            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, this.store[name]) // 向所有在线的用户广播消息
+                    }
+                    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]
+    }
+}

public/music.js

@@ -95,7 +95,7 @@ export default class MusicList {
                 Span({
                     textContent: `${item.name} - ${bytesToSize(item.size)}`,
                     onclick: event => {
-                        event.stopPropagation()
+                        event.stopPropagation()                // !如果使用async则此处不能阻止冒泡传递
                         const li = event.target.parentElement  // ListItem
                         const ul = li.parentElement            // List
                         const list = Array.from(ul.children)   // ListItems
@@ -112,7 +112,7 @@ export default class MusicList {
                 Button({
                     textContent: item.save ? '移除' : '缓存',
                     onclick: event => {
-                        event.stopPropagation()
+                        event.stopPropagation()                // !如果使用async则此处不能阻止冒泡传递
                         if (item.save) {
                             event.target.textContent = '缓存'
                             this.ul.querySelector(`#${item.id}`).classList.remove('cache')

public/test.html

@@ -0,0 +1,28 @@
+<!DOCTYPE html>
+<html lang="zh">
+
+<head>
+    <meta charset="UTF-8">
+    <title>webRTC</title>
+</head>
+
+<body>
+    <div>
+        <h1>Entanglement</h1>
+        <p>同步</p>
+    </div>
+    <script type="module">
+        import Entanglement from './entanglement.js'
+        const entanglement = new Entanglement({ options: {}, server: {} })
+        entanglement.set('json', {
+            users: { name: 'users', list: [{ name: 'satori' }] },
+            music: { name: 'music', list: [{ name: 'satori' }] },
+            image: { name: 'image', list: [{ name: 'satori' }] }
+        })
+        const store = entanglement.get('json')
+        store.users.name = 'koishi'
+        //users.list.push({ name: 'koishi' })
+    </script>
+</body>
+
+</html>