1
id: 42
2
event: price
3
retry: 3000
4
data: {"symbol":"BTC","price":101234}

1
import http from "node:http";
2

3
const clients = new Set();
4
const history = []; // 演示用；生产环境换成 Redis Stream / 数据库
5
let nextId = 0;
6

7
const server = http.createServer((req, res) => {
8
  const url = new URL(req.url, `http://${req.headers.host}`);
9
  if (url.pathname !== "/events") {
10
    res.writeHead(404).end();
11
    return;
12
  }
13

14
  res.writeHead(200, {
15
    "Content-Type": "text/event-stream; charset=utf-8",
16
    "Cache-Control": "no-cache, no-transform",
17
    Connection: "keep-alive",
18
    "X-Accel-Buffering": "no", // 告诉 Nginx 不要缓冲这条响应
19
  });
20

21
  // 浏览器重连时会自动带上 Last-Event-ID
22
  const lastId = Number(req.headers["last-event-id"] ?? -1);
23
  for (const event of history) {
24
    if (event.id > lastId) writeEvent(res, event);
25
  }
26

27
  clients.add(res);
28
  req.on("close", () => clients.delete(res));
29
});
30

31
function writeEvent(res, { id, type, data }) {
32
  res.write(`id: ${id}\n`);
33
  if (type) res.write(`event: ${type}\n`);
34
  res.write(`data: ${JSON.stringify(data)}\n\n`);
35
}
36

37
// 心跳：防止代理或负载均衡器把空闲连接掐掉
38
setInterval(() => {
39
  for (const res of clients) res.write(": heartbeat\n\n");
40
}, 15_000);
41

42
// 模拟业务事件
43
setInterval(() => {
44
  const event = {
45
    id: nextId++,
46
    type: "price",
47
    data: { symbol: "BTC", price: 100_000 + Math.random() * 1_000 },
48
  };
49
  history.push(event);
50
  if (history.length > 100) history.shift();
51
  for (const res of clients) writeEvent(res, event);
52
}, 1_000);
53

54
server.listen(3000);

1
const source = new EventSource("/events");
2

3
source.addEventListener("price", (event) => {
4
  console.log(JSON.parse(event.data));
5
});
6

7
source.onerror = () => {
8
  // 浏览器会自动重连；只有 readyState 变成 CLOSED 才是彻底放弃
9
  if (source.readyState === EventSource.CLOSED) {
10
    console.error("连接已关闭，需要重新创建 EventSource");
11
  }
12
};

1
import http from "node:http";
2
import { WebSocketServer } from "ws";
3

4
const server = http.createServer();
5
const wss = new WebSocketServer({ server, path: "/ws" });
6

7
wss.on("connection", (socket) => {
8
  socket.isAlive = true;
9
  socket.on("pong", () => {
10
    socket.isAlive = true;
11
  });
12

13
  socket.on("message", (raw) => {
14
    const message = JSON.parse(raw.toString());
15
    for (const client of wss.clients) {
16
      if (client.readyState === client.OPEN) {
17
        client.send(JSON.stringify({ type: "chat", text: message.text }));
18
      }
19
    }
20
  });
21
});
22

23
// 每 30 秒探活一次，清掉已经死掉但没收到 close 的连接
24
const timer = setInterval(() => {
25
  for (const socket of wss.clients) {
26
    if (!socket.isAlive) {
27
      socket.terminate();
28
      continue;
29
    }
30
    socket.isAlive = false;
31
    socket.ping();
32
  }
33
}, 30_000);
34

35
wss.on("close", () => clearInterval(timer));
36
server.listen(3000);

1
function connect(url) {
2
  const socket = new WebSocket(url);
3
  let retry = 0;
4
  let closedByUser = false;
5

6
  socket.addEventListener("open", () => {
7
    retry = 0;
8
    console.log("connected");
9
  });
10

11
  socket.addEventListener("message", (event) => {
12
    console.log(JSON.parse(event.data));
13
  });
14

15
  socket.addEventListener("close", (event) => {
16
    if (closedByUser) return;
17
    const base = Math.min(30_000, 500 * 2 ** retry++);
18
    const delay = base * (0.5 + Math.random()); // 0.5x ~ 1.5x 抖动
19
    console.warn(`closed (${event.code}), retry in ${Math.round(delay)}ms`);
20
    setTimeout(() => connect(url), delay);
21
  });
22

23
  return {
24
    send: (data) => socket.send(JSON.stringify(data)),
25
    close: () => {
26
      closedByUser = true;
27
      socket.close(1000, "bye");
28
    },
29
  };
30
}

1
const config = {
2
  iceServers: [
3
    // 替换成你自己的 STUN 服务
4
    { urls: "stun:stun.example.com:3478" },
5
    // 生产环境还需要 TURN：
6
    // { urls: "turn:turn.example.com:3478", username: "u", credential: "p" },
7
  ],
8
};
9

10
// 假设 signaling 是一个已经连好的 WebSocket，send / onmessage 由你实现
11
async function startCaller(signaling) {
12
  const pc = new RTCPeerConnection(config);
13
  const channel = pc.createDataChannel("chat", { ordered: true });
14

15
  channel.onopen = () => channel.send("hello from caller");
16
  channel.onmessage = (event) => console.log("caller recv:", event.data);
17

18
  pc.onicecandidate = ({ candidate }) => {
19
    if (candidate) signaling.send({ type: "candidate", candidate });
20
  };
21

22
  await pc.setLocalDescription(await pc.createOffer());
23
  signaling.send({ type: "offer", sdp: pc.localDescription.sdp });
24
  return pc;
25
}
26

27
async function startCallee(signaling) {
28
  const pc = new RTCPeerConnection(config);
29

30
  // 必须在 setRemoteDescription(offer) 之前注册
31
  pc.ondatachannel = ({ channel }) => {
32
    channel.onmessage = (event) => console.log("callee recv:", event.data);
33
  };
34

35
  pc.onicecandidate = ({ candidate }) => {
36
    if (candidate) signaling.send({ type: "candidate", candidate });
37
  };
38

39
  return pc;
40
}