从 webscoket 客户端获取渠道价值

Question

我正在运行一个 websocket 客户端，并希望将响应从客户端传递到我可以在我的主文件中使用的通道。目前，通道只返回一次 nil 值，然后就没有别的了。将值传递给频道时，我似乎遇到了问题。有什么帮助吗？这是我到目前为止所做的package mainimport (    "context"    "fmt"    "kraken_client/stored_data"    "kraken_client/ws_client"    "os"    "os/signal"    "strings"    "sync"    "syscall")func main() {    // check if in production or testing mode & find base curency    var testing bool = true    args := os.Args    isTesting(args, &testing, &stored_data.Base_currency)    // go routine handler    comms := make(chan os.Signal, 1)    signal.Notify(comms, os.Interrupt, syscall.SIGTERM)    ctx := context.Background()    ctx, cancel := context.WithCancel(ctx)    var wg sync.WaitGroup    // set ohlc interval and pairs    OHLCinterval := 5    pairs := []string{"BTC/" + stored_data.Base_currency, "EOS/" + stored_data.Base_currency}    // create ws connections    pubSocket, err := ws_client.ConnectToServer("public", testing)    if err != nil {        fmt.Println(err)        os.Exit(1)    }    // listen to websocket connections    ch := make(chan interface{})    wg.Add(1)    go pubSocket.PubListen(ctx, &wg, ch, testing)    // subscribe to a stream    pubSocket.SubscribeToOHLC(pairs, OHLCinterval)    go func() {        for c := range ch {            fmt.Println(c)        }    }()    <-comms    cancel()    wg.Wait()    defer close(ch)}下面是 PubListen 函数的工作原理func (socket *Socket) PubListen(ctx context.Context, wg *sync.WaitGroup, ch chan interface{}, testing bool) {    defer wg.Done()    defer socket.Close()    var res interface{}    socket.OnTextMessage = func(message string, socket Socket) {        //log.Println(message)        res = pubJsonDecoder(message, testing) // this function decodes the message and returns an interface        log.Println(res) // this is printing the correctly decoded value.    }    ch <- res    log.Println(res) // does not print a value    log.Println(ch) // does not print a value    <-ctx.Done()    log.Println("closing public socket")    return}我究竟做错了什么？

Answer 1

问题中的代码在由 OnTextMessage 函数设置之前执行ch <- res一次语句。PubListenres

要ch在每条消息上发送一个值，请将行ch <- res移至 OnTextMessage 函数。该函数为每条消息调用一次。

func (socket *Socket) PubListen(ctx context.Context, wg *sync.WaitGroup, ch chan interface{}, testing bool) {

    defer wg.Done()

    defer socket.Close()

    socket.OnTextMessage = func(message string, socket Socket) {

        res := pubJsonDecoder(message, testing)

        ch <- res

        log.Println(res)

    }

    <-ctx.Done()

    log.Println("closing public socket")

    return

}