fix node leave and join

liuxiaolong

2019-10-24 ca2652d724c91bcc0849c749d10f2b9af8129a09

 transport.go

@@ -2,6 +2,10 @@

import (
   "basic.com/valib/logger.git"
   "bufio"
   "bytes"
   "encoding/binary"
   "io"
   "net"
   "strconv"
)
@@ -16,7 +20,12 @@
   defer conn.Close()

   //发送
   _, err = conn.Write(sendBuf)
   sizeBuf := make([]byte,4)
   var buf bytes.Buffer
   binary.BigEndian.PutUint32(sizeBuf,uint32(len(sendBuf)))
   buf.Write(sizeBuf)
   buf.Write(sendBuf)
   _, err = conn.Write(buf.Bytes())
   if err != nil {
      logger.Debug("conn.Write err", err)
      return err
@@ -27,14 +36,17 @@
}

func RawReceiveTcpMsg() {
   var tcpAddr *net.TCPAddr
   tcpAddr,_ = net.ResolveTCPAddr("tcp","127.0.0.1:"+strconv.Itoa(TcpTransportPort))

   listener,_ := net.ListenTCP("tcp",tcpAddr)
   tcpAddr := "0.0.0.0:"+strconv.Itoa(TcpTransportPort)
   listener,err := net.Listen("tcp",tcpAddr)
   if err != nil {
      logger.Debug("RawReceive server listen err:",err)
      return
   }
   defer listener.Close()

   for{
      conn,err := listener.AcceptTCP()
      conn,err := listener.Accept()
      if err!=nil {
         logger.Debug("listener.Accept err:", err)
         continue
@@ -44,18 +56,28 @@
   }
}

func readStream(conn *net.TCPConn) {
   data := make([]byte,0)
   buf := make([]byte, 4096)
   for {
      n,err :=conn.Read(buf)
      if n == 0{
         break
func readStream(conn net.Conn) {
   defer conn.Close()

   var data []byte
   var reader io.Reader = bufio.NewReader(conn)
   sizeBuf :=make([]byte,4)
   if _,err := reader.Read(sizeBuf[:]);err !=nil {
      logger.Debug("read tcpStream msg length err:",err)
   } else {
      var msgLen uint32
      binary.Read(bytes.NewBuffer(sizeBuf),binary.BigEndian,&msgLen)
      dataLen := int(msgLen)
      logger.Debug("read tcpStream msg lenth:",dataLen)
      if dataLen >0 {
         data =make([]byte, dataLen)
         n,err := io.ReadAtLeast(reader, data, dataLen)
         if err ==nil {
            logger.Debug("io.ReadAtLeast n:",n)
         } else {
            logger.Debug("io.readAtLeast err:",err)
         }
      }
      if err !=nil {
         return
      }
      data = append(data,buf...)
   }
   QueryTcpResponseChan <- data
}