Pixel Stream 源码分析

MIT License

Copyright (c) 2021 XOSG

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

node_modules/
.vscode/
package.json
package-lock.json
.DS_Store
jsconfig.json
test/index_*

# 3D 像素流: 虚幻引擎 WebRTC 核心组件

和 EpicGames 官方的像素流 SDK 相比，我们开发出了更轻量的像素流 SDK，包含 2 个文件：前端组件（WebComponents API）外加信令服务器（NodeJS）。

- signal.js：信令服务器，5KB
- peer-stream.js：前端组件，20KB

## 启动信令服务器

首先从 npm 安装 WebSocket 依赖，然后启动 signal.js。

```
npm install ws@8.2.3
node signal.js {key}={value}
```

启动选项:

| 选项   | 默认值  | 作用                 |
| ------ | ------- | -------------------- |
| player | 88      | 浏览器（玩家）端口   |
| engine | 8888    | UE4 端口             |
| token  | insigma | 信令密码（url 末端） |
| limit  | 4       | 玩家数量上限         |

## 启动 UE4

首先开启像素流插件，然后在独立启动模式的设置中，或者打包后的文件中输入启动选项。

```
Plugins > Built-In > Graphics > Pixel Streaming > Enabled
Editor Preferences > Level Editor > Play > Additional Launch Parameters
start packaged.exe -{key}={value}
```

常用的启动选项：

```
 -ForceRes
 -ResX=1920
 -ResY=1080
 -AudioMixer
 -RenderOffScreen
 -graphicsadapter=0
 -AllowPixelStreamingCommands
 -PixelStreamingEncoderRateControl=VBR
 -PixelStreamingURL="ws://localhost:8888"
```

## 前端的 2 种调用方法

JavaScript:

```
import "peer-stream.js";
const ps = document.createElement("video", { is: "peer-stream" });
ps.setAttribute("signal", "ws://127.0.0.1:88/insigma");
document.body.append(ps);
```

or HTML:

```
<script src="peer-stream.js"></script>
<video is="peer-stream" signal="ws://127.0.0.1:88/insigma"></video>
```

## 常用的调试命令

信令服务器可以通过 eval 函数解释执行任意的 NodeJS 代码，使用时需要注意安全。

```
ps.debug('PLAYER.clients.size')   // 显示玩家数量
ps.debug('PLAYER.clients.forEach(p=>p.playerId!==playerId&&p.close(1011,"Infinity"));limit=1;')  // 移除其他玩家
ps.debug('[...PLAYER.clients].map(x=>x.req.socket.remoteAddress)')  // 每个玩家的IP地址
ps.debug('playerId')  // 我的ID
ps.onmouseenter=_=>{ps.focus();ps.requestPointerLock()})    // 鼠标锁
ps.style.pointerEvents='none'   // 只读的<video>
```

## 常见排错方法和技巧（前后端、测试组、三维组遇到的各种坑汇总）

- nginx 代理时，心跳超时问题。
- video 标签被遮挡（等 UI 和样式问题）。
- video 标签是否存在、是否在 DOM 中（window 和 parent 上都挂有 ps）。
- 其他 WebSocket 请求堵塞单线程，导致信令被挂起。
- 所有依赖升级到最新版，包括浏览器、NodeJS、UE4、像素流。
- 网络问题：是否能 ping 通，是否开了防火墙（可用 test/unreal.html 测试）。
- 高频的 WebRTC 重连导致 UE4 崩溃。
- 通过 ps.ws 检查信令服务，通过 ps.pc 检查 WebRTC。
- 网络带宽过低（至少 10m 才能跑一路视频，启动 VBR 以节省带宽）。
- 前端意外打包 peer-stream.js 导致文件出错。
- 检查当前人数是否已满（limit）。
- UE4 跑了几天几夜后需要重启，否则画面撕裂。
- CPU、GPU 超负荷导致视频卡顿。
- 检查信令密码（token）。
- 浏览器 console 中可以看到各种日志，其中 verbose 一栏可查看周期性日志。
- UE4 还未启动完全的时候，不要发请求。
- 使用 ps.debug 在信令服务器上执行任意的 NodeJS 代码并返回结果至前端。
- UE4 是否成功地启用了像素流插件。
- 信令服务器和 UE4 一一对应，与玩家（浏览器）一对多，多余的玩家和多余 UE4 无法连接到信令。
- 前端 Vue 框架集成 peer-stream.js 静态文件的问题（如路径问题）。
- UE 端通过检查启动命令行来判断像素流的相关信息。
- 不需要像素流的时候只要把 video 移出 DOM 即可，不用手动关闭 WebRTC。
- 访问外网时，需要添加 stun。
- 修改 signal、ip、port、token 属性会触发重连。
- 默认不接收音频，需要的话得手动开启。
- 使用 test/index.html 进行前端测试，可以监控 WebRTC。
- 像素流 2 个 js 文件的版本号和虚幻引擎的版本号同步。
- 在任务管理器中通过“命令行”一列获悉 UE4 程序的启动参数。
- 窗口模式下（非后台）运行时，最小化窗口会导致视频流暂停。

## 丑化 JS 代码

为了屏蔽我们的开发环境（虚幻引擎），需要对 JS 文件进行丑化，删除关键字，替换变量名。

```
npm install uglify-js -g
uglifyjs peer-stream.js > ps.min.js
uglifyjs signal.js > signal.min.js
```

## 软件要求

- Google Chrome 88+
- Unreal Engine 4.27+
- NodeJS 14+
- npm/ws 8.0+

# Pixel Streamer: Unreal + WebRTC

Compared to EpicGame's heavily-designed SDK for Pixel Streaming, PixelStreamer is a lightweight WebRTC library with 0 dependency, containing a frontend component (using WebComponents API), along with a signaling server (using NodeJS).

- peer-stream.js (20KB): https://xosg.github.io/PixelStreamer/peer-stream.js
- signal.js (5KB): https://xosg.github.io/PixelStreamer/signal.js
- WebSocket for NodeJS: https://www.npmjs.com/package/ws
- EpicGame's SDK: https://github.com/EpicGames/UnrealEngine/tree/release/Samples/PixelStreaming/WebServers/SignallingWebServer
- Pixel Streaming Plugin: https://github.com/EpicGames/UnrealEngine/tree/release/Engine/Plugins/Media/PixelStreaming

## Signaling Server

install WebSocket dependency:

```
npm install ws@8.2.3
node signal.js {key}={value}
```

startup options:

| key    | default | usage                    |
| ------ | ------- | ------------------------ |
| player | 88      | browser port             |
| engine | 8888    | unreal engine port       |
| token  | insigma | password appended to URL |
| limit  | 4       | max number of players    |

## Unreal Engine 4

enable the plugin:

```
Plugins > Built-In > Graphics > Pixel Streaming > Enabled
Editor Preferences > Level Editor > Play > Additional Launch Parameters
start myPackagedGame.exe -{key}={value}
```

common startup options:

```
 -ForceRes
 -ResX=1920
 -ResY=1080
 -AudioMixer
 -RenderOffScreen
 -graphicsadapter=0
 -AllowPixelStreamingCommands
 -PixelStreamingEncoderRateControl=VBR
 -PixelStreamingURL="ws://localhost:8888"
```

## Browser

JavaScript:

```
import "peer-stream.js";
const ps = document.createElement("video", { is: "peer-stream" });
ps.setAttribute("signal", "ws://127.0.0.1:88/insigma");
document.body.append(ps);
```

or HTML:

```
<script src="peer-stream.js"></script>
<video is="peer-stream" signal="ws://127.0.0.1:88/insigma"></video>
```

## APIs

lifecycle:

```
ps.addEventListener("open", e => {});
ps.addEventListener("message", e => {});
ps.addEventListener("close", e => {});
```

Mouse, Keyboard, Touch events:

```
ps.registerTouchEvents()
ps.registerKeyboardEvents()
ps.registerFakeMouseEvents()
ps.registerMouseHoverEvents()
ps.registerPointerlockEvents()
```

## Common Commands

```
ps.debug('PLAYER.clients.size')   // show players count
ps.debug('PLAYER.clients.forEach(p=>p.playerId!==playerId&&p.close(1011,"Infinity"));limit=1;')  // kick other players
ps.debug('[...PLAYER.clients].map(x=>x.req.socket.remoteAddress)')  // every player's IP
ps.debug('playerId')  // show my id
ps.pc.getReceivers().find(x=>x.track.kind==='video').transport.iceTransport.getSelectedCandidatePair().remote    // show selected candidate
ps.addEventListener('mouseenter',_=>{ps.focus();ps.requestPointerLock()})    // pointer lock
ps.style.pointerEvents='none'   // read only <video>
```

## Requirement

- Google Chrome 88+
- Unreal Engine 4.27+
- NodeJS 14+
- npm/ws 8.0+

## License

[MIT License](./LICENSE)

Copyright (c) 2021 XOSG

<!DOCTYPE html>
<html>
  <head>
    <meta charset="utf-8" />
    <script type="text/javascript" src="../peer-stream.js"></script>
    <!-- <script src="../node_modules/ps.min.js"></script> -->

    <script defer type="text/javascript" src="index.js"></script>
    <style>
      body {
        background-color: black;
        font-weight: bold;
        font-size: large;
      }

      video {
        position: absolute;
      }

      #stats {
        position: fixed;
        bottom: 0;
        pointer-events: none;
        right: 0;
        z-index: 5;
      }

      #logs {
        position: fixed;
        z-index: 5;
        left: 20px;
        bottom: 20px;
        color: red;
      }
</style>
  </head>

  <body>
    <div id="logs"></div>

    <!-- WebRTC 监控 -->
    <pre id="stats">Not connected</pre>

    <video is="peer-stream" ip="10.0.42.15" port="89"></video>
  </body>
</html>

//  document.querySelector("[is=peer-stream]");

ps.addEventListener("message", (e) => {});

ps.addEventListener("playing", (e) => {
  clearInterval(ps.statTimer);
  ps.statTimer = setInterval(aggregateStats, 1000);
});

ps.addEventListener("suspend", (e) => {
  clearInterval(ps.statTimer);
});

const statsWrapper = document.getElementById("stats");
const logsWrapper = document.getElementById("logs");

console.info = (...text) => {
  console.log(...text);
  // show log, disappear after timeout

  const log = document.createElement("pre");
  log.innerHTML = text.join(" ");
  logsWrapper.append(log);
  setTimeout(() => log.remove(), 3000);
};

Number.prototype.format = function () {
  const suffix = ["", "K", "M", "G", "T"];
  let quotient = this;
  while (quotient > 9999) {
    quotient /= 1024;
    suffix.shift();
  }
  return ~~quotient + " " + suffix[0];
};

let lastTransport = {};
async function aggregateStats() {
  const stats = await ps.pc.getStats(null);

  let statsText = "";

  if (ps.VideoEncoderQP > 35) {
    statsWrapper.style.color = "red";
    statsText += `\n Bad Network 😭`;
  } else if (ps.VideoEncoderQP > 26) {
    statsWrapper.style.color = "orange";
    statsText += `\n Spotty Network 😂`;
  } else {
    statsWrapper.style.color = "lime";
  }
  statsText += `\n Video Quantization Parameter: ${ps.VideoEncoderQP}`;

  stats.forEach((stat) => {
    switch (stat.type) {
      case "data-channel": {
        statsText += `\n Data Channel << ${stat.bytesSent.format()}B`;
        statsText += `\n Data Channel >> ${stat.bytesReceived.format()}B`;
        break;
      }
      case "inbound-rtp": {
        if (stat.mediaType === "video")
          statsText += `
      Size: ${stat.frameWidth} x ${stat.frameHeight}
      Frames Decoded: ${stat.framesDecoded.format()}
      Packets Lost: ${stat.packetsLost.format()}
      FPS: ${stat.framesPerSecond} Hz
      Frames Dropped: ${stat.framesDropped?.format()}
      Video >> ${stat.bytesReceived.format()}B`;
        else if (stat.mediaType === "audio")
          statsText += `\n Audio >> ${stat.bytesReceived.format()}B`;
        break;
      }
      case "candidate-pair": {
        if (stat.state === "succeeded")
          statsText += `\n Latency(RTT): ${stat.currentRoundTripTime} s`;
        break;
      }
      case "remote-candidate": {
        statsText += `\n ` + stat.protocol + ":// " + stat.ip + ": " + stat.port;
        break;
      }
      case "transport": {
        const bitrate =
          ((stat.bytesReceived - lastTransport.bytesReceived) /
            (stat.timestamp - lastTransport.timestamp)) *
          (1000 * 8);

        statsText += `\n Bitrate: ${bitrate.format()}bps `;

        lastTransport = stat;
        break;
      }
      default: {
      }
    }
  });

  statsWrapper.innerHTML = statsText;
}

<!DOCTYPE html>
<html>
  <head>
    <!-- networking problems sniffing -->

    <meta charset="utf-8" />
    <title>Unreal Simulator</title>
    <style>
      body {
        background-color: black;
        color: white;
        font-size: large;
      }

      #stats {
        position: fixed;
        top: 0;
        right: 0;
      }

      #setting {
        position: fixed;
        bottom: 0;
        pointer-events: none;
        left: 0;
        z-index: 5;
      }
</style>
  </head>

  <body>
    <h1 id="h1">Unreal Engine Simulator</h1>
    <video autoplay id="video" muted></video>
    <canvas id="canvas" width="500" height="500"></canvas>
    <pre id="stats"></pre>
    <pre id="setting"></pre>

    <script src="unreal.js"></script>
  </body>
</html>

window.players = {};
setInterval(() => {
  stats.innerHTML = "";
  for (const id in players) {
    const pc = players[id];
    stats.innerHTML += `\n ${id} ${pc.connectionState}  `;
  }
  setting.textContent = JSON.stringify(window.stream?.track.getSettings(), null, 2);
}, 1000);

// 相机、录屏、画板
navigator.mediaDevices
  .getUserMedia({ video: true, audio: false })
  .then((stream) => (video.srcObject = stream))
  .catch((error) => {
    console.warn("【camera】", error);
    return navigator.mediaDevices.getDisplayMedia({ audio: false });
  })
  .then((stream) => (video.srcObject = stream))
  .catch((error) => {
    console.warn("【screen capture】", error);
    setupCanvas();
  })
  .finally(() => {
    setupSignal();
    window.stream = video.srcObject || canvas.captureStream();
    stream.track = stream.getVideoTracks()[0];
    console.log("Unreal Simulator is running!");
  });

async function onSignalMessage(msg) {
  try {
    msg = JSON.parse(msg.data);
  } catch (err) {
    console.error("cannot JSON.parse message:", msg);
    return;
  }

  const playerId = String(msg.playerId);
  let pc = players[playerId];
  delete msg[playerId];

  if (msg.type === "offer") {
    pc?.close();

    pc = players[playerId] = new RTCPeerConnection();

    pc.onicecandidate = (e) => {
      if (!e.candidate?.candidate) return;
      console.log("sending candidate to", playerId, e.candidate);
      ws.send(
        JSON.stringify({
          type: "iceCandidate",
          candidate: e.candidate,
          playerId,
        })
      );
    };

    const offer = new RTCSessionDescription(msg);
    console.log("Got offer from", playerId, offer);
    await pc.setRemoteDescription(offer);

    // createAnswer之前，否则要重新协商
    pc.addTrack(stream.track, stream);

    const answer = await pc.createAnswer();
    await pc.setLocalDescription(answer);
    console.log("sending answer to", playerId, answer);
    ws.send(JSON.stringify({ playerId, ...answer.toJSON() }));
  } else if (msg.type === "iceCandidate") {
    if (!pc) {
      console.error("player", playerId, "not found");
      return;
    }

    const candidate = new RTCIceCandidate(msg.candidate);
    await pc.addIceCandidate(candidate);
    console.log("Got candidate from", playerId, candidate);
  } else {
    console.log("Got", msg);
  }
}

function setupSignal() {
  window.ws = new WebSocket("ws://localhost:8888");
  ws.onclose = (e) => {
    setTimeout(setupSignal, 1000);
    h1.textContent = "Unreal Engine Simulator";
  };
  ws.onmessage = onSignalMessage;
  ws.onopen = (e) => {
    console.info("connected to", ws.url);
    h1.textContent = ws.url;
  };
  ws.onerror = (e) => {};
}

function setupCanvas() {
  const $ = canvas.getContext("2d");
  const l = Math.min(canvas.width, canvas.height) / 2;

  $.strokeStyle = `hsl(${360 * Math.random()}deg 100% 50%)`;
  $.lineWidth = 6;

  $.beginPath();
  $.arc(l, l, l, 0, Math.PI * 2);
  $.clip();

  (function frame(time) {
    $.clearRect(-l, -l, l * 2, l * 2);

    for (let x = 1; x <= l; x += $.lineWidth + 2) {
      $.strokeRect(-x, -x, x * 2, x * 2);
      const theta = Math.sin(x / l - time / 512) * 60;
      $.setTransform(new DOMMatrix().translate(l, l).rotate(0, 0, theta));
    }

    window.animationFrame = requestAnimationFrame(frame);
  })(0);
}

"4.27.1";

/* eslint-disable */

// node signal.js player=88 engine=8888 token=insigma limit=1

const WebSocket = require("ws");
const http = require("http");

// process.argv[0] == 'path/to/node.exe'
// process.argv[1] === __filename
const args = process.argv.slice(2).reduce((pairs, pair) => {
  let [key, ...value] = pair.split("=");
  value = value.join("") || "true";
  try {
    value = JSON.parse(value);
  } catch {}
  pairs[key] = value;
  return pairs;
}, {});
Object.assign(
  global,
  {
    player: 88,
    engine: 8888,
    token: "insigma",
    limit: 4,
    nextPlayerId: 100,
  },
  args
);

const ENGINE = new WebSocket.Server({ noServer: true });
ENGINE.ws = {}; // Unreal Engine's WebSocket

// browser client
const PLAYER = new WebSocket.Server({
  noServer: true,
  clientTracking: true,
});

http
  .createServer()
  .on("upgrade", (request, socket, head) => {
    try {
      if (request.url.slice(1) !== token) throw "";
      if (PLAYER.clients.size >= limit) throw "";
    } catch (err) {
      socket.destroy();
      return;
    }

    PLAYER.handleUpgrade(request, socket, head, (ws) => PLAYER.emit("connection", ws, request));
  })
  .listen(player, () => console.log("signaling for player:", player));

http
  .createServer()
  .on("upgrade", (request, socket, head) => {
    try {
      // 1个信令服务器只能连1个UE
      if (ENGINE.ws.readyState === WebSocket.OPEN) throw "";
    } catch (err) {
      socket.destroy();
      return;
    }

    ENGINE.handleUpgrade(request, socket, head, (ws) => ENGINE.emit("connection", ws, request));
  })
  .listen(engine, () => console.log("signaling for engine:", engine));

ENGINE.on("connection", (ws, req) => {
  ws.req = req;
  ENGINE.ws = ws;

  console.log("Engine connected:", req.socket.remoteAddress, req.socket.remotePort);

  ws.on("message", (msg) => {
    try {
      msg = JSON.parse(msg);
    } catch (err) {
      console.error("【JSON error】 Engine:", msg);
      return;
    }

    // Convert incoming playerId to a string if it is an integer, if needed. (We support receiving it as an int or string).
    const playerId = String(msg.playerId || "");
    delete msg.playerId; // no need to send it to the player
    console.log("Engine:", msg.type, playerId);

    if (msg.type === "ping") {
      ws.send(JSON.stringify({ type: "pong", time: msg.time }));
      return;
    }

    const p = [...PLAYER.clients].find((x) => x.playerId === playerId);
    if (!p) {
      console.error("cannot find player", playerId);
      return;
    }

    if (["answer", "iceCandidate"].includes(msg.type)) {
      p.send(JSON.stringify(msg));
    } else if (msg.type == "disconnectPlayer") {
      p.send(msg.reason);
      p.close(1011, "Infinity");
    } else {
      console.error("invalid Engine message type:", msg.type);
    }
  });

  ws.on("close", (code, reason) => {
    // reason是buffer？？
    console.log("Engine closed:", String(reason));
    for (const client of PLAYER.clients) {
      client.send(`Engine:${engine} stopped`);
    }
  });

  ws.on("error", (error) => {
    console.error("Engine connection error:", error);
    // A control frame must have a payload length of 125 bytes or less
    // ws.close(1011, error.message.slice(0, 100));
  });

  // sent to Unreal Engine as initial signal
  ws.send(
    JSON.stringify({
      type: "config",
      peerConnectionOptions: {
        // iceServers: [{ urls: ["stun:34.250.222.95:19302"] }],
      },
    })
  );

  for (const client of PLAYER.clients) {
    // reconnect immediately
    client.send(`Engine:${engine} started`);
    client.close(1011, "1");
  }
});

// every player
PLAYER.on("connection", async (ws, req) => {
  const playerId = String(++nextPlayerId);

  console.log("player", playerId, "connected:", req.socket.remoteAddress, req.socket.remotePort);

  ws.req = req;
  ws.playerId = playerId;

  ws.on("message", (msg) => {
    if (ENGINE.ws.readyState !== WebSocket.OPEN) {
      ws.send(`Engine:${engine} not ready`);
      return;
    }

    try {
      msg = JSON.parse(msg);
    } catch (err) {
      console.info("player", playerId, String(msg));
      ws.send("hello " + msg.slice(0, 100));
      return;
    }

    console.log("player", playerId, msg.type);

    msg.playerId = playerId;
    if (["offer", "iceCandidate"].includes(msg.type)) {
      ENGINE.ws.send(JSON.stringify(msg));
    } else if (msg.type === "debug") {
      let debug;
      try {
        debug = String(eval(msg.debug));
      } catch (err) {
        debug = err.message;
      }
      ws.send("【debug】" + debug);
    } else {
      ws.send("hello " + msg.type);
    }
  });

  ws.on("close", (code, reason) => {
    console.log("player", playerId, "closed:", String(reason));
    if (ENGINE.ws.readyState === WebSocket.OPEN)
      ENGINE.ws.send(JSON.stringify({ type: "playerDisconnected", playerId }));
  });

  ws.on("error", (error) => {
    console.error("player", playerId, "connection error:", error);
  });
});