从 webscoket 客户端获取渠道价值

Question

我正在运行一个 websocket 客户端，并希望将响应从客户端传递到我可以在我的主文件中使用的通道。目前，通道只返回一次 nil 值，然后就没有别的了。将值传递给频道时，我似乎遇到了问题。有什么帮助吗？这是我到目前为止所做的

package main

import (

    "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) {&nbsp; &nbsp; defer wg.Done()&nbsp; &nbsp; defer socket.Close()&nbsp; &nbsp; socket.OnTextMessage = func(message string, socket Socket) {&nbsp; &nbsp; &nbsp; &nbsp; res := pubJsonDecoder(message, testing)&nbsp; &nbsp; &nbsp; &nbsp; ch <- res&nbsp; &nbsp; &nbsp; &nbsp; log.Println(res)&nbsp; &nbsp; }&nbsp; &nbsp; <-ctx.Done()&nbsp; &nbsp; log.Println("closing public socket")&nbsp; &nbsp; return}