问Kubernetes中的Socket.IO集群，多个节点没有发送到所有客户端

EN

const cluster = require("cluster");
const http = require("http");
const { Server } = require("socket.io");
const numCPUs = require("os").cpus().length;
const { setupMaster, setupWorker } = require("@socket.io/sticky");
const { createAdapter, setupPrimary } = require("@socket.io/cluster-adapter");

if (cluster.isMaster) {
    console.log(`Master ${process.pid} is running`);

    const httpServer = http.createServer();

    setupMaster(httpServer, {
        loadBalancingMethod: "least-connection",
    });

    setupPrimary();

    cluster.setupMaster({
        serialization: "advanced",
    });

    httpServer.listen(80);

    console.log(`Starting ${numCPUs} workers...`);

    for (let i = 0; i < numCPUs; i++) {
        var worker = cluster.fork();
        console.log(`Worker ${worker.process.pid} started.`);
    }

    console.log(`Started ${numCPUs} workers.`);

    cluster.on("exit", (worker) => {
        console.log(`Worker ${worker.process.pid} died`);
        cluster.fork();
    });
} else {
    console.log(`Worker ${process.pid} started`);

    const httpServer = http.createServer();
    const io = new Server(httpServer, {
        transports: ["websocket"],
        cors: {
            origin: "*",
            methods: ["GET", "POST"]
        }
    });

    io.adapter(createAdapter());
    setupWorker(io);

    io.on('connection', (socket) => {
        socket.emit("MESSAGE", "Welcome to Stream Socket.");

        socket.on('disconnect', () => { });

        socket.on('SUBSCRIBE', (msg) => {
            try {
                console.log(msg);
                
                var obj = JSON.parse(msg);
                socket.join(obj.requestedStream);

                if (obj.requestedStream.startsWith("OLD_MESSAGES")) {
                    // This here is only emitting to clients connected to same server.
                    io.to("OLD_MESSAGES").emit("___OLD MESSAGES HERE____");
                }

            } catch (e) { console.log(e); }
        });

        socket.on('UNSUBSCRIBE', (msg) => {
            var obj = JSON.parse(msg);
            socket.leave(obj.requestedStream);
        });
    });
}

nginx.ingress.kubernetes.io/affinity: "cookie"
nginx.ingress.kubernetes.io/affinity-mode: "persistent"
nginx.ingress.kubernetes.io/affinity-canary-behavior: "sticky"
nginx.ingress.kubernetes.io/session-cookie-name: "route"
nginx.ingress.kubernetes.io/session-cookie-hash: "sha1"
nginx.ingress.kubernetes.io/session-cookie-expires: "10800"
nginx.ingress.kubernetes.io/session-cookie-max-age: "10800"
nginx.ingress.kubernetes.io/proxy-connect-timeout: "3600" 
nginx.ingress.kubernetes.io/proxy-send-timeout: "3600"
nginx.ingress.kubernetes.io/proxy-read-timeout: "1800"
nginx.ingress.kubernetes.io/send-timeout: "3600"
nginx.ingress.kubernetes.io/proxy-ssl-server-name: "on"

    var server = document.getElementById('address').value;

    socket = io(server, {
        transports: ["websocket"],
        'reconnection': false,
        
    });

    socket.on('MESSAGE', (msg) => {
        logResponse(msg);
    });

    socket.on('MESSAGE', (msg) => {
        logResponse(msg);
    });

FROM node:14
EXPOSE 80
EXPOSE 443

WORKDIR /usr/src/app
COPY package*.json ./
RUN apt-get update && \
    apt-get install -y software-properties-common && \
    rm -rf /var/lib/apt/lists/*
RUN sed -i "/^# deb.*multiverse/ s/^# //" /etc/apt/sources.list
RUN sed -i "/^# deb.*universe/ s/^# //" /etc/apt/sources.list
RUN npm install
RUN npm ci --only=production
COPY . .
CMD ["node", "clustered-socket.js"]