tcp receive data from chan

liuxiaolong

2019-10-08 9e9867a0f667b77648ac4a7881f7a596c363536e

tcp receive data from chan

 agent.go

@@ -26,6 +26,8 @@
    "net"
    "os"
    "strconv"
    "sync"

    //"os"
    "strings"
    "time"
@@ -468,6 +470,7 @@
    From string `json:"from"`
}

var QueryTcpResponseChan = make(chan []byte)
//GetDbFromCluster get the newest database after join cluster
//dbPathWrite the path where to write after got a database,
func (a *Agent) GetTableDataFromCluster(tableNames []string) (*[]string,error) {
@@ -510,25 +513,29 @@

    var dumpSqls []string

    //var wg sync.WaitGroup
    //wg.Add(1)
    //go func() {
    //    defer wg.Done()
    //    respCh := resp.ResponseCh()
    //    for {
    //        select {
    //        case r := <-respCh:
    //            logger.Info("Query response's len:", len(r.Payload))
    //            err := json.Unmarshal(r.Payload, &dumpSqls)
    //            if err ==nil {
    //                logger.Error("dumpSql:",dumpSqls)
    //                logger.Error("data dump success")
    //            }
    //            return
    //        }
    //    }
    //}()
    //wg.Wait()
    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        defer wg.Done()
        respCh := resp.ResponseCh()
        for {
            select {
            case r := <-respCh:
                logger.Info("Query response's len:", len(r.Payload))
                err := json.Unmarshal(r.Payload, &dumpSqls)
                if err ==nil {
                    logger.Error("dumpSql:",dumpSqls)
                    logger.Error("data dump success")
                }
                return
            case msg := <-QueryTcpResponseChan:
                logger.Debug("QueryTcpResponseChan receive msg len:",len(msg))
                return
            }

        }
    }()
    wg.Wait()
    return &dumpSqls,nil

    //r, err = c.Query([]string{query}, false, false)

 userDelegate.go

@@ -23,6 +23,8 @@
func (d *UserDelegate) NotifyMsg(msg []byte) {
    recvMsg := msg
    logger.Info("UserDelegate NotifyMsg.len:",len(recvMsg))
    QueryTcpResponseChan <- msg

    var dumpSqls []string
    err := json.Unmarshal(recvMsg, &dumpSqls)