Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions protocol/czar/engine.go
Original file line number Diff line number Diff line change
Expand Up @@ -46,12 +46,17 @@ import (
var Conf = struct {
// The newly created stream has a higher write priority.
FastWriteDuration time.Duration
// If no new data is received for a period of time, close the conn.
IdleProbeDuration time.Duration
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,
}

Expand Down
22 changes: 21 additions & 1 deletion protocol/czar/mux.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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()
}
Expand Down