完善ws-flv协议代码

youxiuchen · Sep 10, 2023 · a8d3829 · a8d3829
1 parent 4ec539e
commit a8d3829
Show file tree

Hide file tree

Showing 7 changed files with 85 additions and 51 deletions.
diff --git a/.vscode/launch.json b/.vscode/launch.json
@@ -0,0 +1,15 @@
+{
+    // 使用 IntelliSense 了解相关属性。 
+    // 悬停以查看现有属性的描述。
+    // 欲了解更多信息，请访问: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Launch Chrome",
+            "request": "launch",
+            "type": "chrome",
+            "url": "http://127.0.0.1:8081/fixtures/flv/index.html",
+            "webRoot": "${workspaceFolder}"
+        }
+    ]
+}
diff --git a/README.md b/README.md
@@ -23,6 +23,9 @@
 
 
 ### 魔改说明
-因业务需要特基于西瓜播放器源代码基础上进行魔改，使其支持ws-flv协议。后续考虑在此基础上再添加其他解码功能。使用安装与官方保持一致。 [flv视频演示地址](http://h5player.bytedance.com/)
+因业务需要特基于西瓜播放器源代码基础上进行魔改，使其支持ws-flv协议。后续考虑在此基础上再添加其他解码功能。使用方法与官方保持一致。
+flv插件自行打包安转自己项目中
+ [flv视频演示地址](http://h5player.bytedance.com/)
+
 
 
diff --git a/fixtures/flv/index.js b/fixtures/flv/index.js
@@ -22,7 +22,7 @@ var cachedOpt = localStorage.getItem('xg:test:flv:opt')
 try { cachedOpt = JSON.parse(cachedOpt) } catch (error) { cachedOpt = undefined }
 var opts = Object.assign({
   // url: 'https://1011.hlsplay.aodianyun.com/demo/game.flv',
-  url: 'https://pull-flv-l1.douyincdn.com/stage/stream-399911386870710302_ld.flv?keeptime=00093a80&wsSecret=84c8c84e064fb6c6aaad6ec54c5c8247&wsTime=63315a10&abr_pts=1950715',
+  url: 'http://114.116.122.72:8000/flv/live/34020000001320000108_34020000001320000108_0200000108.flv',
 }, defaultOpt(), cachedOpt)
 var testPoint = Number(localStorage.getItem('xg:test:flv:point'))
 

diff --git a/packages/xgplayer-flv/src/flv/index.js b/packages/xgplayer-flv/src/flv/index.js
@@ -276,11 +276,13 @@ export class Flv extends EventEmitter {
         this._mediaLoader?.cancel()
         return
       }
-      const headers = response.headers
-      this.emit(EVENT.TTFB, { url: this._opts.url, responseUrl: response.url, elapsed: st ? (firstByteTime - st) : (endTime - startTime) })
-      this.emit(EVENT.LOAD_RESPONSE_HEADERS, { headers })
-      this._acceptRanges = !!headers?.get('Accept-Ranges') || !!headers?.get('Content-Range')
-      this._firstProgressEmit = true
+      if (!!response?.headers){
+        const headers = response.headers
+        this.emit(EVENT.TTFB, { url: this._opts.url, responseUrl: response.url, elapsed: st ? (firstByteTime - st) : (endTime - startTime) })
+        this.emit(EVENT.LOAD_RESPONSE_HEADERS, { headers })
+        this._acceptRanges = !!headers?.get('Accept-Ranges') || !!headers?.get('Content-Range')
+        this._firstProgressEmit = true
+      }
     }
 
     if (!this._bufferService) return

diff --git a/packages/xgplayer-streaming-shared/src/net/fetch.js b/packages/xgplayer-streaming-shared/src/net/fetch.js
@@ -220,6 +220,7 @@ export class FetchLoader extends EventEmitter {
       startTime = Date.now()
       try {
         data = await reader.read()
+        console.log(data)
         endTime = Date.now()
       } catch (e) {
         // request aborted
@@ -269,6 +270,7 @@ export class FetchLoader extends EventEmitter {
         retData = data.value
       }
       if (retData && retData.byteLength > 0 || data.done) {
+        console.log(retData,'fetch')
         onProgress(retData, data.done, {
           range: [this._range[0] + this._receivedLength - (retData ? retData.byteLength : 0), this._range[0] + this._receivedLength],
           vid: this._vid,

diff --git a/packages/xgplayer-streaming-shared/src/net/index.js b/packages/xgplayer-streaming-shared/src/net/index.js
@@ -27,14 +27,11 @@ export class NetLoader extends EventEmitter {
   constructor (cfg) {
     super(cfg)
     this._config = getConfig(cfg)
-    const _isWebSocketURL = !!this.isWebSocketURL(cfg.url)
     if (
       this._config.loaderType === LoaderType.XHR ||
       (!FetchLoader.isSupported() && !WsLoader.isSupported())
     ) {
       this.type = LoaderType.XHR
-    } else if ((this._config.loaderType === LoaderType.Ws || _isWebSocketURL) && WsLoader.isSupported()){
-      this.type = LoaderType.WS
     }
     this.log = cfg.logger
   }
@@ -60,12 +57,18 @@ export class NetLoader extends EventEmitter {
   }
 
   load (url, config = {}) {
+    let _isWebSocketURL = false
     if (typeof url === 'string' || !url) {
       config.url = url || config.url || this._config.url
+      _isWebSocketURL = !!this.isWebSocketURL(config.url)
     } else {
       config = url
+      _isWebSocketURL = !!this.isWebSocketURL(config.url)
+    }
+    if ((this._config.loaderType === LoaderType.WS || _isWebSocketURL ) && WsLoader.isSupported()){
+      this.type = LoaderType.WS
+      this._config.loaderType = config.loaderType = LoaderType.WS
     }
-
     config = Object.assign({}, this._config, config)
     if (config.params) config.params = Object.assign({}, config.params)
     if (config.headers && isPlainObject(config.headers)) config.headers = Object.assign({}, config.headers)

diff --git a/packages/xgplayer-streaming-shared/src/net/ws.js b/packages/xgplayer-streaming-shared/src/net/ws.js
@@ -1,5 +1,5 @@
 import { NetError } from './error'
-import { createResponse, getRangeValue, setUrlParams, calculateSpeed } from './helper'
+import { createResponse, setUrlParams, calculateSpeed } from './helper'
 import { ResponseType } from './types'
 import { EVENT } from '../event'
 import EventEmitter from 'eventemitter3'
@@ -20,6 +20,8 @@ export class WsLoader extends EventEmitter {
   _onProcessMinLen = 0
   _onCancel = null
   _priOptions = null // 比较私有化的参数传递，回调时候透传
+  _startTime = null
+  _endTime = null
 
   constructor () {
     super()
@@ -78,21 +80,23 @@ export class WsLoader extends EventEmitter {
     }
 
     const startTime = Date.now()
-    this._logger.debug('[websocket load start], index,', index, ',range,', range)
 
     return new Promise((resolve, reject) => {
       this._socket = new WebSocket(url)
-
+      this._socket.binaryType = responseType
+      this._logger.debug('[websocket load start], index,', index, ',ws,', this._socket)
       this._socket.onopen = () => {
         clearTimeout(this._timeoutTimer)
         if (this._aborted || !this._running) return
-        this._logger.debug('[websocket connected]')
+        this._logger.debug('[websocket connected],ws',this._socket)
 
+        this._startTime = Date.now()
         const firstByteTime = Date.now()
 
-        this._socket.onmessage = async (event) => {
+        this._socket.onmessage = (event) => {
+          this._startTime = this._endTime || this._startTime
+          this._endTime = Date.now()
           if (this._aborted || !this._running) return
-
           let data
           if (responseType === ResponseType.TEXT) {
             data = event.data
@@ -104,7 +108,7 @@ export class WsLoader extends EventEmitter {
             if (onProgress) {
               this.resolve = resolve
               this.reject = reject
-              this._loadChunk(event.data, onProgress, startTime, firstByteTime)
+              this._loadChunk(new Uint8Array(event.data), onProgress,startTime,firstByteTime)
               return
             } else {
               data = new Uint8Array(event.data)
@@ -132,8 +136,12 @@ export class WsLoader extends EventEmitter {
           ))
         }
       }
-
+      this._socket.onclose = (e)=>{
+        this._endTime = null
+        this._startTime = null
+      }
       this._socket.onerror = (error) => {
+        this._endTime = Date.now()
         clearTimeout(this._timeoutTimer)
         this._running = false
         if (this._aborted && !isTimeout) return
@@ -166,36 +174,35 @@ export class WsLoader extends EventEmitter {
     }
   }
 
-  _loadChunk (data, onProgress, st, firstByteTime) {
+  _loadChunk (data, onProgress,st, firstByteTime,intervalTime) {
     if (this._onProcessMinLen > 0) {
       this._cache = new Uint8Array(CACHESIZE)
       this._writeIdx = 0
     }
-
-    let startTime
-    let endTime
     const startRange = this._range?.length > 0 ? this._range[0] : 0
     const startByte = startRange + this._receivedLength
-
     if (this._aborted) {
       this._running = false
-      onProgress(undefined, false, { range: [startByte, startByte], vid: this._vid, index: this._index, startTime, endTime, st, firstByteTime, priOptions: this._priOptions })
+      onProgress(undefined, false, { range: [startByte, startByte], vid: this._vid, index: this._index, startTime:this._startTime, endTime: this._endTime, st, firstByteTime, priOptions: this._priOptions })
       return
     }
 
     const curLen = data.byteLength
     this._receivedLength += curLen
 
-    this._logger.debug('【WsLoader,onProgress call】,task,', this._range, ', start,', startByte, ', end,', startRange + this._receivedLength, ', done,', true)
+    const _done = [2,3].indexOf(this._socket.readyState) > 0
+    this._logger.debug('【WsLoader,onProgress call】,task,', this._range, ', start,', startByte, ', end,', startRange + this._receivedLength, ', done,', _done)
 
     let retData
+
+
     if (this._onProcessMinLen > 0) {
-      if (this._writeIdx + curLen >= this._onProcessMinLen || true) {
+      if (this._writeIdx + curLen >= this._onProcessMinLen || _done) {
         retData = new Uint8Array(this._writeIdx + curLen)
         retData.set(this._cache.slice(0, this._writeIdx), 0)
         curLen > 0 && retData.set(data, this._writeIdx)
         this._writeIdx = 0
-        this._logger.debug('【WsLoader,onProgress enough】,done,', true, ',len,', retData.byteLength, ', writeIdx,', this._writeIdx)
+        this._logger.debug('【WsLoader,onProgress enough】,done,', _done, ',len,', retData.byteLength, ', writeIdx,', this._writeIdx)
       } else {
         if (curLen > 0 && this._writeIdx + curLen < CACHESIZE) {
           this._cache.set(data, this._writeIdx)
@@ -216,39 +223,41 @@ export class WsLoader extends EventEmitter {
       retData = data
     }
 
-    if (retData && retData.byteLength > 0 || true) {
-      onProgress(retData, true, {
+    if (retData && retData.byteLength > 0 || _done) {
+      onProgress(retData, _done, {
         range: [this._range[0] + this._receivedLength - (retData ? retData.byteLength : 0), this._range[0] + this._receivedLength],
         vid: this._vid,
         index: this._index,
-        startTime,
-        endTime,
+        startTime:this._startTime,
+        endTime:this._endTime,
         st,
         firstByteTime,
         priOptions: this._priOptions
       }, null)
     }
 
-    const costTime = Date.now() - st
-    const speed = calculateSpeed(this._receivedLength, costTime)
-    this.emit(EVENT.REAL_TIME_SPEED, { speed, len: this._receivedLength, time: costTime, vid: this._vid, index: this._index, range: this._range, priOptions: this._priOptions })
-
-    this._running = false
-    this._logger.debug('[WsLoader onProgress end],task,', this._range, ',done,', true)
+    if (_done){
+      const costTime = Date.now() - st
+      const speed = calculateSpeed(this._receivedLength, costTime)
+      this.emit(EVENT.REAL_TIME_SPEED, { speed, len: this._receivedLength, time: costTime, vid: this._vid, index: this._index, range: this._range, priOptions: this._priOptions })
 
-    this.resolve(createResponse(
-      retData,
-      true,
-      null,
-      null,
-      null,
-      st,
-      firstByteTime,
-      this._index,
-      this._range,
-      this._vid,
-      this._priOptions
-    ))
+      this._running = false
+      this._logger.debug('[WsLoader onProgress end],task,', this._range, ',done,', true)
+
+      this.resolve(createResponse(
+        retData,
+        true,
+        null,
+        null,
+        null,
+        st,
+        firstByteTime,
+        this._index,
+        this._range,
+        this._vid,
+        this._priOptions
+      ))
+    }
   }
 
   get receiveLen () {