Websocket等待带有队列的服务器响应

Question

我正在使用 websocket 发送和接收数据（每秒最多 30 条小消息）。我希望客户端发送一个 websocket 有效负载并等待来自服务器的特定消息。

流程：

客户端发送请求

它还将 requestId (163) 存储在 waitingResponse 对象中作为带有 sent 时间戳的新对象

waitingResponse = {
  163: { sent: 1583253453549 }
}

当服务器响应时，另一个函数验证有效负载，然后将结果附加到该请求对象

waitingResponse = {
  163: { sent: 1583253453549, action: "none" }
}

客户端每隔 x 毫秒检查一次该对象的 action 键

我有一个函数sendPayload 发送有效载荷，然后等待来自awaitResponse 的值（下面的函数）。现在这个功能不起作用。我尝试制作 2 个单独的函数，一个是 setTimeout 计时器，另一个是 promise。我还尝试将两者放在同一个函数中，并使用您可以在下面看到的original 参数来确定它是循环还是承诺。现在我在想，即使在循环中，该函数也应该始终返回一个承诺，但我似乎无法使用计时器来完成这项工作，而且我担心彼此之间存在多个承诺的成本。假设我每 5 毫秒检查一次响应，超时时间为 2000 毫秒。这是很多承诺。

public async sendPayload(details) {
    console.log("sendPlayload", details);

    this.waitingResponse[details.requestId] = { sent: +new Date() };

    if (this.socket.readyState === WebSocket.OPEN) {
        this.socket.send(JSON.stringify(details));
    }
    const bindAwaitResponse = this.awaitResponse.bind(this);
    return new Promise(async function (resolve, reject) {
        const result = await bindAwaitResponse(details.requestId, true);
        console.log("RES", result);
        console.info("Time took", (+new Date() - result.sent) / 1000);

        resolve(result);
    });

}

public async awaitResponse(requestId, original) {
    // console.log(requestId, "awaitResponse")
    return new Promise((resolve, reject) => {
        // Is it a valid queued request
        if (this.waitingResponse[requestId]) {
            // Do we have an answer?
            if (this.waitingResponse[requestId].hasOwnProperty("action")) {
                console.log(requestId, "Got a response");
                const tmp = this.waitingResponse[requestId];
                delete this.waitingResponse[requestId]; // Cleanup
                resolve(tmp);
            } else {
                // No answer yet from remote server
                // console.log("no answer: ", JSON.stringify(this.waitingResponse));
                // Check if request took too long
                if (+new Date() - this.waitingResponse[requestId].sent > 5000) { // TODO: Option for time out
                    console.warn(requestId, "Request timed out");

                    // Timed out, result took too long
                    // TODO: Option, default action when timed out
                    delete this.waitingResponse[requestId];  // Cleanup
                    resolve({
                        action: "to" // For now, just sent a timeout action, maybe the default action should be outside of the Network class?
                    })
                } else {
                    // console.log(requestId, "Still waiting for results");
                    console.log(JSON.stringify(this.waitingResponse));
                    // Still waiting, after x ms, recall function
                    return setTimeout(async () => { resolve(await this.awaitResponse(requestId, false)); }, 200);
                }
            }
        }
    });
}

private async processMessage(msg) {
    console.log("WS received Message", JSON.stringify(msg.data));

    console.log("Current: ", JSON.stringify(this.waitingResponse));

    let data = JSON.parse(msg.data);
    // console.log("Received: ", data);


    if (data.hasOwnProperty("requestId") && this.waitingResponse[data.requestId]) {
        // console.log("processMessage ID found");
        this.waitingResponse[data.requestId] = { ...data, ...this.waitingResponse[data.requestId] };

    }
}

注意：我把websocket 标签放在下面是因为我努力寻找它。也许我在没有意识到的情况下遇到了解决方案，但是如果您有更好的标签可以更容易地找到这个问题，请编辑它们:)

Answer 1

是的，您将许多回调样式函数与中间承诺和async/await 混合在一起。等待响应时不要进行轮询！相反，当 writing a queue 时，将 resolve 函数本身放入队列中，以便您可以直接履行/拒绝响应处理程序的相应承诺。

在你的情况下：

public async sendPayload(details) {
    const request = this.waitingResponse[details.requestId] = { sent: +new Date() };
    try {
        if (this.socket.readyState === WebSocket.OPEN) {
           this.socket.send(JSON.stringify(details));
        }
        const result = await new Promise(function(resolve) {
            request.resolve = resolve;

            setTimeout(() => {
                reject(new Error('Timeout')); // or resolve({action: "to"}), or whatever
            }, 5000);
        });
        console.info("Time took", (+new Date() - request.sent) / 1000);
        return result; // or {...request, ...result} if you care
    } finally {
        delete this.waitingResponse[details.requestId];
    }
}


private async processMessage(msg) {
    let data = JSON.parse(msg.data);

    if (data.hasOwnProperty("requestId") {
        const request = this.waitingResponse[data.requestId]
        if (request)
            request.resolve(data)
        else
            console.warn("Got data but found no associated request, already timed out?", data)
    } else {
        console.warn("Got data without request id", data);
    }
}

如果processMessage 函数不需要有关请求的任何详细信息，您甚至可以完全取消request 对象并仅存储resolve 函数本身。

Answer 2

您正在尝试通过 WebSocket 实现基于承诺的通信。由于 WebSockekjust 在该事件上工作。唯一性是一个事件，所以如果我们多次发送相同的事件。 WebSocket 不关心哪个事件针对哪个动作。

为了通过基于 Promise 的解决方案实现这种独特性，我编写了简单的 js lib，它可以在浏览器和 nodejs 中运行。

https://github.com/pathikdevani/linka也许这对你的项目有用。