SmartMeeting/speakr/static/js/wsconnecter.js

/**
 * Realtime ASR websocket client.
 * Browser audio is sent only to the Speakr backend proxy.
 */

function getBackendAsrWebSocketUrl(mode) {
  const protocol = window.location.protocol === "https:" ? "wss:" : "ws:";
  const basePath = "/tool/speakr/ws/asr/live";
  return `${protocol}//${window.location.host}${basePath}?mode=${encodeURIComponent(mode || "online")}`;
}

function WebSocketConnectMethod(config) {
  //定义socket连接方法类

  var speechSokt;
  var connKeeperID;

  var msgHandle = config.msgHandle;
  var stateHandle = config.stateHandle;
  let modeType = "offline";

  // 发送队列和流量控制
  var sendQueue = [];
  var isSending = false;
  var maxQueueSize = 1000; // 最大队列长度，防止内存溢出
  var queueWarningThreshold = 500; // 队列警告阈值
  var lastDisconnWarnTime = 0; // WebSocket未连接警告节流
  var gracefulStopResolver = null;
  var gracefulStopTimer = null;
  function trimQueueIfNeeded() {
    if (sendQueue.length <= maxQueueSize) {
      return false;
    }

    const dropCount = sendQueue.length - maxQueueSize;
    console.warn(
      `⚠️ 发送队列超过最大限制 (${maxQueueSize})，丢弃最旧的 ${dropCount} 个数据包`
    );
    sendQueue = sendQueue.slice(-maxQueueSize);
    return true;
  }

  function enqueueData(data) {
    sendQueue.push(data);
    trimQueueIfNeeded();

    if (sendQueue.length > queueWarningThreshold) {
      console.warn(`⚠️ WebSocket发送队列堆积: ${sendQueue.length} 个数据包待发送`);
    }
  }

  this.wsStart = function (e) {
    if (speechSokt) {
      speechSokt.close();
    }
    if (e === "offline" || (e && e !== "online")) {
      modeType = "offline";
    } else {
      modeType = "online";
    }
    var Uri = getBackendAsrWebSocketUrl(modeType);
    if (Uri.match(/wss:\S*|ws:\S*/)) {
      console.log("Uri" + Uri);
    } else {
      alert("请检查wss地址正确性");
      return 0;
    }

    if ("WebSocket" in window) {
      speechSokt = new WebSocket(Uri); // 定义socket连接对象
      speechSokt.onopen = function (e) {
        onOpen(e);
      }; // 定义响应函数
      speechSokt.onclose = function (e) {
        console.log("onclose ws!");
        onClose(e);
      };
      speechSokt.onmessage = function (e) {
        onMessage(e);
      };
      speechSokt.onerror = function (e) {
        onError(e);
      };
      return 1;
    } else {
      alert("当前浏览器不支持 WebSocket");
      return 0;
    }
  };

  // 定义停止与发送函数
  function resolveGracefulStop() {
    if (gracefulStopTimer) {
      clearTimeout(gracefulStopTimer);
      gracefulStopTimer = null;
    }
    if (gracefulStopResolver) {
      gracefulStopResolver();
      gracefulStopResolver = null;
    }
  }

  function closeSocketNow() {
    if (speechSokt != undefined) {
      try {
        speechSokt.close();
      } catch (e) {
        console.warn("Failed to close WebSocket", e);
      }
      speechSokt = undefined;
    }
    sendQueue = [];
    isSending = false;
  }

  this.wsStop = function () {
    if (speechSokt != undefined) {
      console.log("stop ws!");
      try {
        if (speechSokt.readyState === 1) {
          speechSokt.send(JSON.stringify({ type: "stop", is_speaking: false }));
        }
      } catch (e) {
        console.warn("Failed to send WebSocket stop frame", e);
      }
    }
    resolveGracefulStop();
    closeSocketNow();
  };

  this.wsFinish = function (timeoutMs = 3000) {
    if (speechSokt == undefined || speechSokt.readyState >= 2) {
      resolveGracefulStop();
      return Promise.resolve();
    }

    sendQueue = [];
    isSending = false;

    return new Promise((resolve) => {
      gracefulStopResolver = resolve;
      gracefulStopTimer = setTimeout(() => {
        // Give the server ownership of the close. Timeout only unblocks UI.
        resolveGracefulStop();
      }, timeoutMs);

      try {
        if (speechSokt.readyState === 1) {
          speechSokt.send(JSON.stringify({ type: "stop", is_speaking: false }));
        }
      } catch (e) {
        console.warn("Failed to send WebSocket finish frame", e);
        resolveGracefulStop();
      }
    });
  };

  // 处理发送队列
  function processQueue() {
    if (
      isSending ||
      sendQueue.length === 0 ||
      !speechSokt ||
      speechSokt.readyState !== 1
    ) {
      return;
    }

    isSending = true;
    const data = sendQueue.shift();

    try {
      speechSokt.send(data);
      isSending = false;

      // 队列警告
      if (sendQueue.length > queueWarningThreshold) {
        console.warn(`⚠️ WebSocket发送队列堆积: ${sendQueue.length} 个数据包待发送`);
      }

      // 继续处理队列
      if (sendQueue.length > 0) {
        // 使用 setTimeout 避免阻塞
        setTimeout(() => processQueue(), 0);
      }
    } catch (e) {
      console.error("❌ WebSocket发送失败，数据已放回队列:", e);
      sendQueue.unshift(data); // 放回队列头部
      isSending = false;
      trimQueueIfNeeded();
    }
  }

  this.wsSend = function (oneData) {
    if (speechSokt == undefined) {
      // WebSocket未初始化，缓存数据
      enqueueData(oneData);
      return true;
    }

    if (speechSokt.readyState === 1) {
      // WebSocket已打开
      if (sendQueue.length > 0) {
        // 队列中有数据，先加入队列
        enqueueData(oneData);
        processQueue();
        return true;
      } else {
        // 队列为空，尝试直接发送
        try {
          speechSokt.send(oneData);
          return true;
        } catch (e) {
          console.error("❌ WebSocket直接发送失败，加入队列:", e);
          enqueueData(oneData);
          processQueue();
          return true;
        }
      }
    } else if (speechSokt.readyState === 0) {
      // WebSocket正在连接，缓存数据
      enqueueData(oneData);
      return true;
    } else {
      // WebSocket已关闭或关闭中，丢弃数据并警告（节流 5 秒）
      var now = Date.now();
      if (now - lastDisconnWarnTime > 5000) {
        console.warn("⚠️ WebSocket未连接，数据无法发送 (state:", speechSokt.readyState, ")");
        lastDisconnWarnTime = now;
      }
      return false;
    }
  };

  // 获取队列状态（用于调试）
  this.getQueueStatus = function () {
    return {
      length: sendQueue.length,
      isSending: isSending,
      readyState: speechSokt ? speechSokt.readyState : -1
    };
  };


  // SOCEKT连接中的消息与状态响应
  function onOpen(e) {
    console.log(e, "e");
    console.log("WebSocket connected to backend ASR proxy");
    stateHandle(0);

    if (sendQueue.length > 0) {
      console.log(`Processing ${sendQueue.length} queued audio chunks`);
      processQueue();
    }
  }

  function onClose(e) {
    resolveGracefulStop();
    speechSokt = undefined;
    stateHandle(1);
  }

  function onMessage(e) {
    msgHandle(e);
    try {
      const payload = JSON.parse(e.data);
      if (payload && payload.type === "state" && payload.state === "closed") {
        resolveGracefulStop();
      }
    } catch (_) {}
  }

  function onError(e) {
    // info_div.innerHTML="连接"+e;
    console.log(e);
    stateHandle(2);
  }
}