修改节点心跳发送失败逻辑,现在会直接退出;完善数据通道连接超时的错误处理

This commit is contained in:
2025-05-29 15:22:50 +08:00
parent ceb381bc9b
commit f45ab3e89c
5 changed files with 26 additions and 32 deletions

View File

@@ -1,5 +1,9 @@
## TODO
检查连接被提前关闭的问题
访问失败的也返回访问记录
解决节点断开立即重连会导致 status 覆盖的问题
实现一个机制使更新数据的后项能够覆盖前项

View File

@@ -78,30 +78,32 @@ func ctrl(ctx context.Context, id int32, host string) error {
}
// 异步定时发送心跳
var writeErr = make(chan error)
go func() {
ticker := time.NewTicker(time.Duration(core.HeartbeatInterval) * time.Second)
defer ticker.Stop()
ticker := time.Tick(time.Duration(core.HeartbeatInterval) * time.Second)
for {
select {
case <-ctx.Done():
writeErr <- nil
return
case tick := <-ticker.C:
case _ = <-ticker:
err := sendPing(conn)
if err != nil {
slog.Error("发送心跳失败", "time", tick, "err", err)
writeErr <- fmt.Errorf("发送心跳失败: %w", err)
return
}
}
}
}()
// 异步读取节点命令
var errCh = make(chan error)
var readErr = make(chan error)
go func() {
for {
// 读取命令
cmd, err := reader.ReadByte()
if _, err := utils.WarpConnErr(err); err != nil {
errCh <- err
readErr <- err
return
}
switch cmd {
@@ -113,7 +115,7 @@ func ctrl(ctx context.Context, id int32, host string) error {
case 5:
err := onConn(reader, dataAddr)
if err != nil {
errCh <- fmt.Errorf("处理代理命令失败: %w", err)
readErr <- fmt.Errorf("处理代理命令失败: %w", err)
return
}
}
@@ -123,7 +125,7 @@ func ctrl(ctx context.Context, id int32, host string) error {
// 等待建立数据通道
select {
case <-ctx.Done():
case err = <-errCh:
case err = <-readErr:
}
// 发送关闭连接

View File

@@ -60,10 +60,7 @@ func ListenData(ctx context.Context) error {
app.DataConnWg.Add(1)
go func() {
defer app.DataConnWg.Done()
defer func() {
utils.Close(conn)
slog.Debug("关闭数据通道连接")
}()
defer utils.Close(conn)
err := processDataConn(ctx, conn)
if err != nil {
slog.Error("处理数据通道连接失败", "err", err)
@@ -89,12 +86,13 @@ func processDataConn(ctx context.Context, edge net.Conn) error {
// 加载用户连接
user, ok := app.UserConnMap.LoadAndDelete(tag)
if !ok {
if status == 1 {
return fmt.Errorf("用户连接已关闭tag%s", tag)
} else {
return nil
}
defer func() {
utils.Close(user)
slog.Debug("关闭用户连接")
}()
}
defer utils.Close(user)
// 检查状态
if status != 1 {
@@ -119,13 +117,8 @@ func processDataConn(ctx context.Context, edge net.Conn) error {
var waitEdge = make(chan error)
go func() {
_, err := io.Copy(user, reader)
switch {
case errors.Is(err, net.ErrClosed):
slog.Warn("节点连接意外关闭")
case err != nil:
if ok, err := utils.WarpConnErr(err); !ok {
slog.Error("读取节点数据失败", "err", err)
default:
slog.Debug("节点数据读取完成")
}
waitEdge <- err
}()
@@ -134,13 +127,8 @@ func processDataConn(ctx context.Context, edge net.Conn) error {
var waitUser = make(chan error)
go func() {
_, err := io.Copy(edge, teeUser)
switch {
case errors.Is(err, net.ErrClosed):
slog.Warn("用户连接意外关闭")
case err != nil:
if ok, err := utils.WarpConnErr(err); !ok {
slog.Error("读取用户数据失败", "err", err)
default:
slog.Debug("用户数据读取完成")
}
waitUser <- err
}()

View File

@@ -85,7 +85,7 @@ func processUserConn(ctx context.Context, user *core.Conn, ctrl io.Writer) (err
if ok {
utils.Close(user)
if errors.Is(err, context.DeadlineExceeded) {
slog.Error("用户连接超时", "tag", tag, "addr", user.RemoteAddr().String())
slog.Warn("建立数据通道超时", "tag", tag, "addr", user.RemoteAddr().String())
}
}

View File

@@ -31,8 +31,8 @@ func ReadBuffer(reader io.Reader, size int) ([]byte, error) {
// Close 关闭对象,传入值绝对不能为 nil
func Close(v any) {
if v, ok := v.(io.Closer); ok {
err := v.Close()
if c, ok := v.(io.Closer); ok {
err := c.Close()
if err != nil {
var logger = slog.Default()
_, file, line, ok := runtime.Caller(1)