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)通知读写协程退出。
写的很巧妙,像诗一样