CODEWITHSUNDEEP
gowebsocket
Max Gaffney
Max Gaffney

Reputation: 33

Websockets in go routine: error previous message not read to completion

I've recently begun exploring Go and am really enjoying it. I've run into an issue when attempting to detect a timeout in a websocket connection. I am listening to the websocket connection indefinitely and when I don't get a response in X seconds I attempt to reconnect. To accomplish this I've had to revise my for loop to include a select. I then created a type and a channel to listen for websocket responses. However this has led to errors in my websocket connection saying failed to get reader: previous message not read to completion.

I replaced code taken from project to be stand alone. Below is full script with both loops (working and non working available)

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
        }
    }
}

Logs are printing:

{"e":"trade","E":1577140149102,"s":"BTCUSDT","t":220054947,"p":"7304.40000000","q":"0.07153400","b":933798088,"a":933798124,"T":1577140149099,"m":true,"M":true}

failed to get reader: previous message not read to completion

{"e":"trade","E":1577140149107,"s":"BTCUSDT","t":220054948,"p":"7304.95000000","q":"0.28826900","b":933798126,"a":933798125,"T":1577140149104,"m":false,"M":true}

failed to get reader: previous message not read to completion

So its working but its still returning errors. The reader function source is here. https://github.com/nhooyr/websocket/blob/master/conn.go#L390. Suppose I could just raise an issue there.

Upvotes: 0

Views: 1181

Answers (1)

Thundercat
Thundercat

Reputation: 120960

As the error implies, a message must be read fully before the next message can be read. Use the first version of the code or change the second version to slurp up the message to a []byte and send that []byte to the channel.

Assuming that you are using the nhooyr.io/websocket package, the second version will look something like this:

for {
    // Read returns the entire message as a []byte
    msgType, msg, err := c.Read(ctx)

    // bytes.NewReader creates an io.Reader on a []byte
    res := wsResponse{Msg: bytes.NewReader(msg), Err: err, MsgType: msgType}
    wsChan <- res
    if res.Err {
        // Always exit the loop on error. Otherwise, the goroutine will run forever.
        return
    }
}

Upvotes: 1

Related Questions