Add application layer keepalive probe for czar

Author: mohanson

protocol/czar/engine.go

@@ -41,17 +41,29 @@ import (
 // +-----+-----+-----+-----+
 // | Sid |  2  | 0/1 | Rsv |
 // +-----+-----+-----+-----+
+//
+// Keep alive probe and reply.
+//
+// +-----+-----+-----+-----+
+// | 0x0 |  3  | 0/1 | Rsv |
+// +-----+-----+-----+-----+
 
 // Conf is acting as package level configuration.
 var Conf = struct {
 	// The newly created stream has a higher write priority.
 	FastWriteDuration time.Duration
+	// The duration a connection needs to be idle before mux begins sending out keep-alive probe.
+	IdleProbeDuration time.Duration
+	// If no data is read for more than this time, the connection is closed.
+	IdleReplyDuration time.Duration
 	// Packet size. Since the size of the packet header is 4, this value must be greater than 4. If the value is too
 	// small, the transmission efficiency will be reduced, and if it is too large, the concurrency capability of mux
 	// will be reduced.
 	PacketSize int
 }{
 	FastWriteDuration: time.Second * 8,
+	IdleProbeDuration: time.Second * 120,
+	IdleReplyDuration: time.Second * 128,
 	PacketSize:        2048,
 }
 

protocol/czar/mux.go

@@ -204,9 +204,21 @@ func (m *Mux) Recv() {
 		idx uint8
 		msg []byte
 		old *Stream
+		prb = time.AfterFunc(Conf.IdleProbeDuration, func() {
+			if m.pri.Pri(0, func() error {
+				return doa.Err(m.con.Write([]byte{0x00, 0x03, 0x00, 0x00}))
+			}) != nil {
+				m.Close()
+			}
+		})
+		rst = time.AfterFunc(Conf.IdleReplyDuration, func() {
+			m.Close()
+		})
 		stm *Stream
 	)
 	for {
+		prb.Reset(Conf.IdleProbeDuration)
+		rst.Reset(Conf.IdleReplyDuration)
 		_, err = io.ReadFull(m.con, buf[:4])
 		if err != nil {
 			m.rer.Put(err)
@@ -247,7 +259,15 @@ func (m *Mux) Recv() {
 			stm.Esolc()
 			old = NewWither(idx, m)
 			m.usb[idx] = old
-		case cmd >= 0x03:
+		case cmd == 0x03:
+			switch buf[2] {
+			case 0x00:
+				m.pri.Pri(0, func() error {
+					return doa.Err(m.con.Write([]byte{0x00, 0x03, 0x01, 0x00}))
+				})
+			case 0x01:
+			}
+		case cmd >= 0x04:
 			// Packet format error, connection closed.
 			m.con.Close()
 		}