go 例程中的 Websockets：错误上一条消息未读完整

Question

我最近开始探索 Go 并且非常喜欢它。我在尝试检测 WebSocket 连接超时时遇到了问题。我无限期地监听 websocket 连接，当我在 X 秒内没有得到响应时，我尝试重新连接。为了实现这一点，我必须修改我的 for 循环以包含选择。然后我创建了一个类型和一个通道来监听 websocket 响应。然而，这导致我的 websocket 连接出现错误，提示 failed to get reader: previous message not read to Complete。

我将从项目中获取的代码替换为独立的。下面是包含两个循环的完整脚本（工作和非工作可用）

package main

import (

    "bytes"

    "context"

    "fmt"

    "io"

    "time"

    "nhooyr.io/websocket"

)

func main() {

    ctx := context.Background()

    c, _, err := websocket.Dial(ctx, "wss://stream.binance.com:9443/ws/btcusdt@trade", nil)

    if err != nil {

        fmt.Println(err)

        return

    }

    type wsResponse struct {

        Msg     io.Reader

        Err     error

        MsgType websocket.MessageType

    }

    // THIS LOOP WORKS

    // for {

    //  _, msg, err := c.Reader(ctx)

    //  buf := new(bytes.Buffer)

    //  buf.ReadFrom(msg)

    //  fmt.Println(buf.String())

    //  if err != nil {

    //      fmt.Println(err)

    //      return

    //  }

    // }

    // The following goroutine and loop produces errors

    wsChan := make(chan wsResponse)

    go func() {

        for {

            msgType, msg, err := c.Reader(ctx)

            res := wsResponse{Msg: msg, Err: err, MsgType: msgType}

            //fmt.Printf("%+v\n", res)

            wsChan <- res

        }

    }()

    ticker := time.NewTicker(30 * time.Second)

    for {

        select {

        case res := <-wsChan:

            ticker.Stop()

            if res.Err != nil {

                fmt.Println(res.Err)

                break

            }

            buf := new(bytes.Buffer)

            buf.ReadFrom(res.Msg)

            s := buf.String()

            fmt.Println(s)

            ticker = time.NewTicker(5 * time.Second)

        case <-ticker.C:

            fmt.Println("timeout error")

            break

        }

    }

}

Answer 1

正如该错误所暗示的那样，必须先完全读取一条消息，然后才能读取下一条消息。使用代码的第一个版本或更改第二个版本以将消息吸收为 [] 字节并将该 [] 字节发送到通道。假设您正在使用 nhooyr.io/websocket 包，第二个版本将如下所示：for {&nbsp; &nbsp; // Read returns the entire message as a []byte&nbsp; &nbsp; msgType, msg, err := c.Read(ctx)&nbsp; &nbsp; // bytes.NewReader creates an io.Reader on a []byte&nbsp; &nbsp; res := wsResponse{Msg: bytes.NewReader(msg), Err: err, MsgType: msgType}&nbsp; &nbsp; wsChan <- res&nbsp; &nbsp; if res.Err {&nbsp; &nbsp; &nbsp; &nbsp; // Always exit the loop on error. Otherwise, the goroutine will run forever.&nbsp; &nbsp; &nbsp; &nbsp; return&nbsp; &nbsp; }}