🔧 fix(api_request): enhance ping keep-alive mechanism with error handling and timeout controls
This commit is contained in:
@@ -109,6 +109,12 @@ func startPingKeepAlive(c *gin.Context, pingInterval time.Duration) context.Canc
|
|||||||
|
|
||||||
gopool.Go(func() {
|
gopool.Go(func() {
|
||||||
defer func() {
|
defer func() {
|
||||||
|
// 增加panic恢复处理
|
||||||
|
if r := recover(); r != nil {
|
||||||
|
if common2.DebugEnabled {
|
||||||
|
println("SSE ping goroutine panic recovered:", fmt.Sprintf("%v", r))
|
||||||
|
}
|
||||||
|
}
|
||||||
if common2.DebugEnabled {
|
if common2.DebugEnabled {
|
||||||
println("SSE ping goroutine stopped.")
|
println("SSE ping goroutine stopped.")
|
||||||
}
|
}
|
||||||
@@ -119,19 +125,32 @@ func startPingKeepAlive(c *gin.Context, pingInterval time.Duration) context.Canc
|
|||||||
}
|
}
|
||||||
|
|
||||||
ticker := time.NewTicker(pingInterval)
|
ticker := time.NewTicker(pingInterval)
|
||||||
// 退出时清理 ticker
|
// 确保在任何情况下都清理ticker
|
||||||
defer ticker.Stop()
|
defer func() {
|
||||||
|
ticker.Stop()
|
||||||
|
if common2.DebugEnabled {
|
||||||
|
println("SSE ping ticker stopped")
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
var pingMutex sync.Mutex
|
var pingMutex sync.Mutex
|
||||||
if common2.DebugEnabled {
|
if common2.DebugEnabled {
|
||||||
println("SSE ping goroutine started")
|
println("SSE ping goroutine started")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 增加超时控制,防止goroutine长时间运行
|
||||||
|
maxPingDuration := 120 * time.Minute // 最大ping持续时间
|
||||||
|
pingTimeout := time.NewTimer(maxPingDuration)
|
||||||
|
defer pingTimeout.Stop()
|
||||||
|
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
// 发送 ping 数据
|
// 发送 ping 数据
|
||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
if err := sendPingData(c, &pingMutex); err != nil {
|
if err := sendPingData(c, &pingMutex); err != nil {
|
||||||
|
if common2.DebugEnabled {
|
||||||
|
println("SSE ping error, stopping goroutine:", err.Error())
|
||||||
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// 收到退出信号
|
// 收到退出信号
|
||||||
@@ -140,6 +159,12 @@ func startPingKeepAlive(c *gin.Context, pingInterval time.Duration) context.Canc
|
|||||||
// request 结束
|
// request 结束
|
||||||
case <-c.Request.Context().Done():
|
case <-c.Request.Context().Done():
|
||||||
return
|
return
|
||||||
|
// 超时保护,防止goroutine无限运行
|
||||||
|
case <-pingTimeout.C:
|
||||||
|
if common2.DebugEnabled {
|
||||||
|
println("SSE ping goroutine timeout, stopping")
|
||||||
|
}
|
||||||
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
@@ -148,19 +173,34 @@ func startPingKeepAlive(c *gin.Context, pingInterval time.Duration) context.Canc
|
|||||||
}
|
}
|
||||||
|
|
||||||
func sendPingData(c *gin.Context, mutex *sync.Mutex) error {
|
func sendPingData(c *gin.Context, mutex *sync.Mutex) error {
|
||||||
mutex.Lock()
|
// 增加超时控制,防止锁死等待
|
||||||
defer mutex.Unlock()
|
done := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
mutex.Lock()
|
||||||
|
defer mutex.Unlock()
|
||||||
|
|
||||||
err := helper.PingData(c)
|
err := helper.PingData(c)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
common2.LogError(c, "SSE ping error: "+err.Error())
|
common2.LogError(c, "SSE ping error: "+err.Error())
|
||||||
|
done <- err
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if common2.DebugEnabled {
|
||||||
|
println("SSE ping data sent.")
|
||||||
|
}
|
||||||
|
done <- nil
|
||||||
|
}()
|
||||||
|
|
||||||
|
// 设置发送ping数据的超时时间
|
||||||
|
select {
|
||||||
|
case err := <-done:
|
||||||
return err
|
return err
|
||||||
|
case <-time.After(10 * time.Second):
|
||||||
|
return errors.New("SSE ping data send timeout")
|
||||||
|
case <-c.Request.Context().Done():
|
||||||
|
return errors.New("request context cancelled during ping")
|
||||||
}
|
}
|
||||||
|
|
||||||
if common2.DebugEnabled {
|
|
||||||
println("SSE ping data sent.")
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func doRequest(c *gin.Context, req *http.Request, info *common.RelayInfo) (*http.Response, error) {
|
func doRequest(c *gin.Context, req *http.Request, info *common.RelayInfo) (*http.Response, error) {
|
||||||
@@ -175,15 +215,23 @@ func doRequest(c *gin.Context, req *http.Request, info *common.RelayInfo) (*http
|
|||||||
client = service.GetHttpClient()
|
client = service.GetHttpClient()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var stopPinger context.CancelFunc
|
||||||
if info.IsStream {
|
if info.IsStream {
|
||||||
helper.SetEventStreamHeaders(c)
|
helper.SetEventStreamHeaders(c)
|
||||||
|
|
||||||
// 处理流式请求的 ping 保活
|
// 处理流式请求的 ping 保活
|
||||||
generalSettings := operation_setting.GetGeneralSetting()
|
generalSettings := operation_setting.GetGeneralSetting()
|
||||||
if generalSettings.PingIntervalEnabled {
|
if generalSettings.PingIntervalEnabled {
|
||||||
pingInterval := time.Duration(generalSettings.PingIntervalSeconds) * time.Second
|
pingInterval := time.Duration(generalSettings.PingIntervalSeconds) * time.Second
|
||||||
stopPinger := startPingKeepAlive(c, pingInterval)
|
stopPinger = startPingKeepAlive(c, pingInterval)
|
||||||
defer stopPinger()
|
// 使用defer确保在任何情况下都能停止ping goroutine
|
||||||
|
defer func() {
|
||||||
|
if stopPinger != nil {
|
||||||
|
stopPinger()
|
||||||
|
if common2.DebugEnabled {
|
||||||
|
println("SSE ping goroutine stopped by defer")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user