Browse Source

增强 ProcessMessage 异常处理:1) 添加 try-catch 异常记录 2) 将 OperationCanceledException 转换为 WebSocketException 3) 添加输入参数验证检查

feature/x1-web-request
root 1 week ago
parent
commit
d678bf19e5
  1. 92
      src/X1.WebSocket/Middleware/WebSocketMiddleware.cs

92
src/X1.WebSocket/Middleware/WebSocketMiddleware.cs

@ -388,43 +388,81 @@ public class WebSocketMiddleware
{
var processingStartTime = DateTime.UtcNow;
var success = await _errorHandler.HandleWithRetryAsync(async () =>
try
{
// 为每个消息创建独立的消息缓冲区,避免并发问题
using var messageBuffer = new WebSocketMessageBuffer(_options.MaxMessageSize);
if (!messageBuffer.TryWrite(buffer, 0, receiveResult.Count))
var success = await _errorHandler.HandleWithRetryAsync(async () =>
{
throw new WebSocketException("消息缓冲区溢出");
}
// 为每个消息创建独立的消息缓冲区,避免并发问题
using var messageBuffer = new WebSocketMessageBuffer(_options.MaxMessageSize);
if (receiveResult.EndOfMessage)
{
var message = new WebSocketMessage
// 验证输入参数
if (buffer == null || buffer.Length == 0)
{
ConnectionId = connectionId,
Data = messageBuffer.GetMessage(),
MessageType = receiveResult.MessageType,
IsComplete = true
};
throw new WebSocketException("接收缓冲区为空或无效");
}
await messageChannel.Writer.WriteAsync(message, cancellationToken);
if (receiveResult.Count <= 0 || receiveResult.Count > buffer.Length)
{
throw new WebSocketException($"无效的消息大小:{receiveResult.Count},缓冲区大小:{buffer.Length}");
}
var processingTime = (DateTime.UtcNow - processingStartTime).TotalMilliseconds;
_performanceMonitor.RecordMessage(
connectionId,
message.Data.Length,
(long)processingTime);
if (receiveResult.Count > _options.MaxMessageSize)
{
throw new WebSocketException($"消息大小超过限制:{receiveResult.Count} > {_options.MaxMessageSize}");
}
// 记录 Channel 消息处理统计
_channelManager.RecordMessageProcessed(messageChannel, message.Data.Length);
if (!messageBuffer.TryWrite(buffer, 0, receiveResult.Count))
{
throw new WebSocketException("消息缓冲区溢出");
}
// messageBuffer 会在 using 语句中自动释放
}
});
if (receiveResult.EndOfMessage)
{
var message = new WebSocketMessage
{
ConnectionId = connectionId,
Data = messageBuffer.GetMessage(),
MessageType = receiveResult.MessageType,
IsComplete = true
};
if (!success)
await messageChannel.Writer.WriteAsync(message, cancellationToken);
var processingTime = (DateTime.UtcNow - processingStartTime).TotalMilliseconds;
_performanceMonitor.RecordMessage(
connectionId,
message.Data.Length,
(long)processingTime);
// 记录 Channel 消息处理统计
_channelManager.RecordMessageProcessed(messageChannel, message.Data.Length);
// messageBuffer 会在 using 语句中自动释放
}
});
if (!success)
{
await HandleMessageProcessingFailure(webSocket, connectionId, cancellationToken);
}
}
catch (WebSocketException ex)
{
_logger.LogError(ex, "WebSocket 消息处理异常,连接ID:{ConnectionId},消息类型:{MessageType},消息大小:{MessageSize}字节,错误:{ErrorMessage}",
connectionId, receiveResult.MessageType, receiveResult.Count, ex.Message);
await HandleMessageProcessingFailure(webSocket, connectionId, cancellationToken);
}
catch (OperationCanceledException ex)
{
_logger.LogWarning(ex, "消息处理被取消,连接ID:{ConnectionId},消息类型:{MessageType},消息大小:{MessageSize}字节",
connectionId, receiveResult.MessageType, receiveResult.Count);
// 取消操作转换为 WebSocket 异常
throw new WebSocketException("消息处理操作被取消");
}
catch (Exception ex)
{
_logger.LogError(ex, "消息处理发生未知异常,连接ID:{ConnectionId},消息类型:{MessageType},消息大小:{MessageSize}字节,错误:{ErrorMessage}",
connectionId, receiveResult.MessageType, receiveResult.Count, ex.Message);
await HandleMessageProcessingFailure(webSocket, connectionId, cancellationToken);
}
}

Loading…
Cancel
Save