update

src/views/Student/CustomWebsocketProvider.ts

@@ -0,0 +1,255 @@
+import type * as Y from 'yjs'
+import * as syncProtocol from 'y-protocols/sync'
+import * as awarenessProtocol from 'y-protocols/awareness'
+import * as encoding from 'lib0/encoding'
+import * as decoding from 'lib0/decoding'
+import { Observable } from 'lib0/observable'
+
+// y-redis 使用的消息类型常量（与 y-websocket 一致）
+const messageSync = 0
+const messageAwareness = 1
+const messageAuth = 2
+const messageQueryAwareness = 3
+
+export class YRedisWebsocketProvider extends Observable<string> {
+  public doc: Y.Doc
+  public awareness: awarenessProtocol.Awareness
+  public ws: WebSocket | null = null
+  public wsconnected = false
+  public synced = false
+  public shouldConnect = true
+
+  private url: string
+  private roomname: string
+  private params: Record<string, string>
+  private _updateHandler: (update: Uint8Array, origin: any) => void
+  private _awarenessUpdateHandler: (...args: any[]) => void
+  private messageHandlers: Array<(
+    encoder: encoding.Encoder,
+    decoder: decoding.Decoder,
+    provider: YRedisWebsocketProvider,
+    emitSynced: boolean,
+    messageType: number
+  ) => void>
+
+  constructor(
+    serverUrl: string,
+    roomname: string,
+    doc: Y.Doc,
+    opts: { params?: Record<string, string>; awareness?: awarenessProtocol.Awareness } = {}
+  ) {
+    super()
+    this.doc = doc
+    this.roomname = roomname
+    this.params = opts.params || {}
+    this.awareness = opts.awareness || new awarenessProtocol.Awareness(doc)
+
+    // 构建带认证参数的 URL
+    const queryParams = new URLSearchParams(this.params)
+    this.url = `${serverUrl}/${roomname}${queryParams.toString() ? '?' + queryParams.toString() : ''}`
+
+    this.messageHandlers = []
+
+    // messageSync (0): 处理文档同步
+    this.messageHandlers[messageSync] = (encoder, decoder, provider, emitSynced) => {
+      encoding.writeVarUint(encoder, messageSync)
+      const syncMessageType = syncProtocol.readSyncMessage(decoder, encoder, doc, provider)
+      if (emitSynced && syncMessageType === syncProtocol.messageYjsSyncStep2 && !provider.synced) {
+        provider.synced = true
+        provider.emit('sync', [true])
+      }
+    }
+
+    // messageAwareness (1): 处理光标/在线状态
+    this.messageHandlers[messageAwareness] = (encoder, decoder, provider) => {
+      awarenessProtocol.applyAwarenessUpdate(
+        provider.awareness,
+        decoding.readVarUint8Array(decoder),
+        provider
+      )
+    }
+
+    // ⭐ y-redis 关键：messageAuth (2) — 处理认证结果
+    this.messageHandlers[messageAuth] = (encoder, decoder, provider) => {
+      const authType = decoding.readVarUint(decoder)
+      if (authType === 0) {
+        // 认证失败：权限被拒绝
+        provider.emit('auth-message', ['permission-denied'])
+        console.error('[YRedisProvider] 认证失败，权限被拒绝')
+        provider.disconnect()
+      }
+      else if (authType === 1) {
+        // 认证成功（y-redis 某些版本会返回）
+        provider.emit('auth-message', ['authenticated'])
+      }
+    }
+
+    // messageQueryAwareness (3): 请求 awareness 信息
+    this.messageHandlers[messageQueryAwareness] = (encoder, decoder, provider) => {
+      encoding.writeVarUint(encoder, messageAwareness)
+      encoding.writeVarUint8Array(
+        encoder,
+        awarenessProtocol.encodeAwarenessUpdate(
+          provider.awareness,
+          Array.from(provider.awareness.getStates().keys())
+        )
+      )
+    }
+
+    // 本地文档更新 → 广播
+    this._updateHandler = (update, origin) => {
+      if (origin !== this) {
+        const encoder = encoding.createEncoder()
+        encoding.writeVarUint(encoder, messageSync)
+        syncProtocol.writeUpdate(encoder, update)
+        this.broadcastMessage(encoding.toUint8Array(encoder))
+        console.log(encoding.toUint8Array(encoder))
+      }
+    }
+    this.doc.on('update', this._updateHandler)
+
+    // 本地 awareness 更新 → 广播
+    this._awarenessUpdateHandler = ({ added, updated, removed }, origin) => {
+      const changedClients = added.concat(updated, removed)
+      if (origin !== this) {
+        const encoder = encoding.createEncoder()
+        encoding.writeVarUint(encoder, messageAwareness)
+        encoding.writeVarUint8Array(
+          encoder,
+          awarenessProtocol.encodeAwarenessUpdate(this.awareness, changedClients)
+        )
+        this.broadcastMessage(encoding.toUint8Array(encoder))
+      }
+    }
+    this.awareness.on('update', this._awarenessUpdateHandler)
+
+    this.connect()
+  }
+
+  /**
+   * ⭐ 核心修复：循环解析整个消息缓冲区
+   * y-redis 可能将多条更新合并后一次性推送，必须全部处理
+   */
+  private readMessage(data: Uint8Array): void {
+    let offset = 0
+    while (offset < data.length) {
+      // 基于剩余数据创建解码器，decoder.pos 是相对于当前消息起点的
+      const decoder = decoding.createDecoder(data.subarray(offset))
+      try {
+        const messageType = decoding.readVarUint(decoder)
+        const encoder = encoding.createEncoder()
+        const handler = this.messageHandlers[messageType]
+
+        if (handler) {
+          handler(encoder, decoder, this, true, messageType)
+
+          // 如果 handler 生成了回复，立即发送
+          if (encoding.length(encoder) > 0) {
+            if (this.ws?.readyState === WebSocket.OPEN) {
+              this.ws.send(encoding.toUint8Array(encoder))
+            }
+          }
+        }
+
+        // ⭐ 关键修正：使用处理完消息后的 decoder.pos 来推进 offset
+        const consumed = decoder.pos
+        if (consumed === 0) {
+          // 防止死循环：理论上不会发生，因为至少读取了消息类型
+          console.warn('[YRedisProvider] decoder 未消耗任何字节，终止解析')
+          break
+        }
+
+        offset += consumed
+
+        // 每条消息处理完后触发自定义事件
+        this.emit('message', [{ data: data.slice(offset - consumed, offset), type: messageType }])
+      }
+      catch (err) {
+        console.error('[YRedisProvider] 解析消息失败，offset:', offset, err)
+        // 解析失败时终止本次处理，避免因错位导致更多误报
+        break
+      }
+    }
+  }
+
+  connect(): void {
+    if (this.wsconnected || !this.shouldConnect) return
+    this.emit('status', [{ status: 'connecting' }])
+
+    const ws = new WebSocket(this.url)
+    this.ws = ws
+    ws.binaryType = 'arraybuffer'
+
+    ws.onopen = () => {
+      this.wsconnected = true
+      this.emit('status', [{ status: 'connected' }])
+
+      // 发送 SyncStep1 发起初始同步
+      const encoder = encoding.createEncoder()
+      encoding.writeVarUint(encoder, messageSync)
+      syncProtocol.writeSyncStep1(encoder, this.doc)
+      ws.send(encoding.toUint8Array(encoder))
+
+      // 发送本地 awareness 状态
+      const states = this.awareness.getStates()
+      if (states.size > 0) {
+        const enc = encoding.createEncoder()
+        encoding.writeVarUint(enc, messageAwareness)
+        encoding.writeVarUint8Array(
+          enc,
+          awarenessProtocol.encodeAwarenessUpdate(this.awareness, Array.from(states.keys()))
+        )
+        ws.send(encoding.toUint8Array(enc))
+      }
+    }
+
+    ws.onmessage = (event: MessageEvent) => {
+      if (event.data instanceof ArrayBuffer) {
+        console.log('123', new Uint8Array(event.data))
+        this.readMessage(new Uint8Array(event.data))
+      }
+    }
+
+    ws.onclose = (event: CloseEvent) => {
+      this.wsconnected = false
+      this.synced = false
+      // 清除其他客户端的 awareness
+      awarenessProtocol.removeAwarenessStates(
+        this.awareness,
+        Array.from(this.awareness.getStates().keys()).filter(
+          client => client !== this.doc.clientID
+        ),
+        this
+      )
+      this.emit('status', [{ status: 'disconnected' }])
+      this.emit('connection-close', [event, this])
+      this.ws = null
+      if (this.shouldConnect) {
+        setTimeout(() => this.connect(), 2500)
+      }
+    }
+
+    ws.onerror = (event: Event) => {
+      this.emit('connection-error', [event, this])
+    }
+  }
+
+  disconnect(): void {
+    this.shouldConnect = false
+    if (this.ws) this.ws.close()
+  }
+
+  destroy(): void {
+    this.disconnect()
+    this.doc.off('update', this._updateHandler)
+    this.awareness.off('update', this._awarenessUpdateHandler)
+    this.awareness.destroy()
+    super.destroy()
+  }
+
+  private broadcastMessage(data: Uint8Array): void {
+    if (this.ws?.readyState === WebSocket.OPEN) {
+      this.ws.send(data)
+    }
+  }
+}

src/views/Student/index.vue

@@ -443,6 +443,7 @@ import AIWorkModal from './components/AIWorkModal.vue'
 import DialoguePanel from './components/DialoguePanel.vue'
 import ChoiceStatistics from './components/ChoiceStatistics.vue'
 import * as Y from 'yjs'
+import { YRedisWebsocketProvider } from './CustomWebsocketProvider'
 import { WebsocketProvider } from 'y-websocket'
 import { Refresh } from '@icon-park/vue-next'
 import answerTheResult from './components/answerTheResult.vue'
@@ -700,7 +701,8 @@ const yTimerMessages = ref<any | null>(null)
 const yLaserMessages = ref<any | null>(null)
 const yWritingBoardMessages = ref<any | null>(null)
 const yIsResultArrayMessages = ref<any | null>(null)
-const providerSocket = ref<WebsocketProvider | null>(null)
+// const providerSocket = ref<WebsocketProvider | null>(null)
+const providerSocket = ref<YRedisWebsocketProvider | null>(null)
 // 学生端画图同步数据
 const writingBoardSyncDataURL = ref<string | null>(null)
 const writingBoardSyncBlackboard = ref<boolean | null>(null)
@@ -3115,8 +3117,13 @@ const checkWorkArrayChanged = (oldArray: WorkItem[], newArray: WorkItem[]): bool
   return false
 }
 
+const isSelectCourseSLook = ref(false)
 // 查询课程跟随状态
 const selectCourseSLook = async (type = 2) => {
+  if (isSelectCourseSLook.value) {
+    return
+  }
+  isSelectCourseSLook.value = true
   const res = await api.selectCourseSLook(props.courseid as string)
   console.log('selectCourseSLook', res)
   if (res[0][0].follow == 2) {
@@ -3142,7 +3149,7 @@ const selectCourseSLook = async (type = 2) => {
   if (props.type == '2') {
     message.success(isFollowModeActive.value ? lang.ssFollowOnTip : lang.ssFreeOnTip)
   }
-  
+  isSelectCourseSLook.value = false
   checkParentMode()
 }
 
@@ -3680,7 +3687,7 @@ const getMessages = (msgObj: any) => {
   console.log('message', msgObj)
   
   // 处理幻灯片切换消息
-  if (props.type == '2' && msgObj.type === 'slideIndex') {
+  if (props.type == '2' && msgObj.type === 'slideIndex' && isFollowModeActive.value) {
     goToSlide(msgObj.slideIndex)
   }
   
@@ -4392,94 +4399,19 @@ const createWebSocketConnection = async (type = 1) => {
 
     docSocket.value = new Y.Doc()
     docSocket.value.gc = true
-    providerSocket.value = new WebsocketProvider(
+    providerSocket.value = new YRedisWebsocketProvider(
       api.yweb_socket,
       'PPT' + props.courseid,
       docSocket.value,
       {
         params: { yauth: authToken.value },
         // resyncInterval: 5000, // 每 5 秒用标准协议请求服务器状态（代替畸形心跳）
-        maxBackoffTime: 5000, // 重连最长退避 5 秒
+        // maxBackoffTime: 5000, // 重连最长退避 5 秒
       }
     )
     providerSocket.value.awareness.setLocalState(null)
     providerSocket.value.awareness.destroy() // 完全禁用 awareness
 
-    // ★ 三层日志之一：raw websocket 消息（确认数据到达浏览器）
-    const _ws = (providerSocket.value as any).ws
-    if (_ws && !_ws.__logged) {
-      _ws.__logged = true // 防止重复绑定
-      // 包一层 onmessage，抓 provider 处理异常
-      const _origOnMsg = _ws.onmessage
-      if (typeof _origOnMsg === 'function') {
-        _ws.onmessage = function(ev: MessageEvent) {
-          try {
-            _origOnMsg.call(_ws, ev)
-          }
-          catch (e: any) {
-            console.error('❌ provider onmessage 异常:', e?.message || e, e?.stack || '')
-          }
-        }
-      }
-      _ws.addEventListener('message', (ev: MessageEvent) => {
-        try {
-          const data = new Uint8Array(ev.data)
-          const msgType = data[0]
-          const typeNames: any = { 0: 'sync', 1: 'awareness', 2: 'queryAwareness' }
-          let detail = ''
-          if (msgType === 0) {
-            const sub = data[1]
-            detail = ({ 0: 'step1(请求缺失)', 1: 'step2(服务器推送)', 2: 'update' } as any)[sub] || 'unknown'
-          }
-          console.log(`📥 ws raw: ${typeNames[msgType] || msgType} ${detail} (${data.length} bytes)`)
-
-          // ★ 诊断：sync update 到达时手动解码 + applyUpdate，看是不是重复数据
-          if (msgType === 0 && data[1] === 2) {
-            const doc = docSocket.value
-            if (doc) {
-              // 解码 varint 提取 update bytes
-              let pos = 2 // 跳过 msgType + subType
-              let updateLen = 0
-              let shift = 0
-              while (pos < data.length) {
-                const b = data[pos++]
-                updateLen |= (b & 0x7f) << shift
-                if (b < 0x80) break
-                shift += 7
-              }
-              const updateBytes = data.slice(pos, pos + updateLen)
-
-              // ★ 用 Y.decodeUpdate 解码 update（不创建 temp doc，不触发 applyUpdate 的 DELETE）
-              try {
-                const { structs } = Y.decodeUpdate(updateBytes)
-                console.log(`📦 解码: updateBytes=${updateBytes.length}, structs=${structs.length}`)
-                for (const struct of structs) {
-                  const arr = struct?.content?.arr
-                  console.log(`📦 struct: type=${struct?.content?.constructor?.name}, parentSub=${struct?.parentSub}, arr=${arr ? JSON.stringify(arr)?.slice(0, 150) : '无'}`)
-                  if (arr && Array.isArray(arr)) {
-                    for (const val of arr) {
-                      if (val && typeof val === 'object' && val.type && val.mId) {
-                        console.log(`📦 ✅ 从 update 提取消息: type=${val.type}, mId=${val.mId}, 当前 mId=${mId.value}`)
-                        if (val.mId !== mId.value) {
-                          getMessages(val)
-                        }
-                      }
-                    }
-                  }
-                }
-              }
-              catch (e) {
-                console.log('📦 update 解码失败:', e)
-              }
-            }
-          }
-        }
-        catch (e) {
-          console.log('📥 ws raw: parse error', e)
-        }
-      })
-    }
-
     // ★ 三层日志之二：Y.Doc update 事件 — 列出所有根类型的 size，看更新到底改了哪个
     docSocket.value.on('update', (update: Uint8Array, origin: any) => {
       // origin === providerSocket.value 表示来自网络
@@ -4575,6 +4507,11 @@ const createWebSocketConnection = async (type = 1) => {
     providerSocket.value.on('connection-close', (event: any) => {
       console.log('👉 WebSocket 连接关闭（将自动重连）:', event)
     })
+    
+    // 每条消息到达时触发（用于调试批量消息问题）
+    providerSocket.value.on('message', ({ data, type }) => {
+      console.log(`收到消息 type=${type} len=${data.length}`)
+    })
 
     // ★ synced 事件：在服务器状态同步完成（sync step 2 已应用到 Y.Doc）后触发
     // 'connected' 在 onopen 时就触发，此时 sync 尚未完成，messageInit 的 catch-up 读不到数据