gorpc之serverHandleConnection

func serverHandleConnection(s *Server, conn io.ReadWriteCloser, clientAddr string, workersCh chan struct{}) {
    defer s.stopWg.Done()

    if s.OnConnect != nil {
        newConn, err := s.OnConnect(clientAddr, conn)
        if err != nil {
            s.LogError("gorpc.Server: [%s]->[%s]. OnConnect error: [%s]", clientAddr, s.Addr, err)
            conn.Close()
            return
        }
        conn = newConn
    }

    var enabledCompression bool
    var err error
    var stopping atomic.Value

    zChan := make(chan bool, 1)
    go func() {
        var buf [1]byte
        if _, err = conn.Read(buf[:]); err != nil {
            if stopping.Load() == nil {
                s.LogError("gorpc.Server: [%s]->[%s]. Error when reading handshake from client: [%s]", clientAddr, s.Addr, err)
            }
        }
        zChan <- (buf[0] != 0)
    }()
    select {
    case enabledCompression = <-zChan: // 第一个标志位 代表被压缩
        if err != nil {
            conn.Close()
            return
        }
    case <-s.serverStopChan:
        stopping.Store(true)
        conn.Close()
        return
    case <-time.After(10 * time.Second):
        s.LogError("gorpc.Server: [%s]->[%s]. Cannot obtain handshake from client during 10s", clientAddr, s.Addr)
        conn.Close()
        return
    }

    responsesChan := make(chan *serverMessage, s.PendingResponses)
    stopChan := make(chan struct{})

    readerDone := make(chan struct{})
    go serverReader(s, conn, clientAddr, responsesChan, stopChan, readerDone, enabledCompression, workersCh)

    writerDone := make(chan struct{})
    go serverWriter(s, conn, clientAddr, responsesChan, stopChan, writerDone, enabledCompression)

    select {
    case <-readerDone:
        close(stopChan)
        conn.Close()
        <-writerDone
    case <-writerDone:
        close(stopChan)
        conn.Close()
        <-readerDone
    case <-s.serverStopChan:
        close(stopChan)
        conn.Close()
        <-readerDone
        <-writerDone
    }
}

• OnConnect，像中间件一样对原本的clientAddr和conn进行处理，可以记录日志，甚至可以封装。
• zChan用于通知消息是否被压缩。
• 启动读协程和写协程去读取消息。读写协程之间的交互靠responsesChan，停止靠stopCh通信同时通知两个协程。
• 调用函数被 读写协程readDone和writerDone通知，如果读写协程崩溃则通知调用函数。进而同时通知stopChan关闭另一个
• 如果serverStopChan（上层停止调用）通知，则直接close（stopChan）通知读写协程退出。

写的很巧妙，像诗一样