You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

722 lines
28 KiB

using Microsoft.AspNetCore.Mvc;
using LTEMvcApp.Models;
using LTEMvcApp.Services;
using Newtonsoft.Json.Linq;
using Microsoft.Extensions.Logging;
using System.Threading.Tasks;
using System.Linq;
using System.Threading;
namespace LTEMvcApp.Controllers
{
/// <summary>
/// WebSocket控制器 - 提供LTE客户端WebSocket管理API
/// </summary>
[ApiController]
[Route("api/[controller]")]
public class WebSocketController : ControllerBase
{
private readonly WebSocketManagerService _webSocketManager;
private readonly ILogger<WebSocketController> _logger;
public WebSocketController(WebSocketManagerService webSocketManager, ILogger<WebSocketController> logger)
{
_webSocketManager = webSocketManager;
_logger = logger;
}
/// <summary>
/// 获取所有客户端状态
/// </summary>
/// <returns>客户端状态列表</returns>
[HttpGet("clients")]
public ActionResult<Dictionary<string, ClientState>> GetClientStates()
{
_logger.LogInformation("获取所有客户端状态");
var states = _webSocketManager.GetAllClientStates();
return Ok(states);
}
/// <summary>
/// 获取客户端配置
/// </summary>
/// <param name="clientName">客户端名称</param>
/// <returns>客户端配置</returns>
[HttpGet("clients/{clientName}/config")]
public ActionResult<ClientConfig?> GetClientConfig(string clientName)
{
var config = _webSocketManager.GetClientConfig(clientName);
if (config == null)
return NotFound($"客户端 '{clientName}' 不存在");
return Ok(config);
}
/// <summary>
/// 获取所有客户端配置
/// </summary>
/// <returns>客户端配置列表</returns>
[HttpGet("configs")]
public ActionResult<List<ClientConfig>> GetAllConfigs()
{
var configs = _webSocketManager.GetAllClientConfigs();
return Ok(configs);
}
/// <summary>
/// 添加客户端配置
/// </summary>
/// <param name="config">客户端配置</param>
/// <returns>操作结果</returns>
[HttpPost("configs")]
public ActionResult AddClientConfig([FromBody] ClientConfig config)
{
if (string.IsNullOrEmpty(config.Name))
return BadRequest("客户端名称不能为空");
var success = _webSocketManager.AddClientConfig(config);
if (success)
return Ok(new { message = $"客户端 '{config.Name}' 配置已添加" });
else
return BadRequest("添加客户端配置失败");
}
/// <summary>
/// 启动客户端
/// </summary>
/// <param name="clientName">客户端名称</param>
/// <returns>操作结果</returns>
[HttpPost("clients/{clientName}/start")]
public ActionResult StartClient(string clientName)
{
_logger.LogInformation($"API请求: 启动客户端 {clientName}");
var success = _webSocketManager.StartClient(clientName);
if (success)
{
_logger.LogInformation($"客户端 {clientName} 启动成功");
return Ok(new { message = $"客户端 '{clientName}' 已启动" });
}
else
{
_logger.LogWarning($"客户端 {clientName} 启动失败");
return BadRequest($"启动客户端 '{clientName}' 失败");
}
}
/// <summary>
/// 停止客户端
/// </summary>
/// <param name="clientName">客户端名称</param>
/// <returns>操作结果</returns>
[HttpPost("clients/{clientName}/stop")]
public ActionResult StopClient(string clientName)
{
var success = _webSocketManager.StopClient(clientName);
if (success)
return Ok(new { message = $"客户端 '{clientName}' 已停止" });
else
return BadRequest($"停止客户端 '{clientName}' 失败");
}
/// <summary>
/// 播放/暂停客户端
/// </summary>
/// <param name="clientName">客户端名称</param>
/// <returns>操作结果</returns>
[HttpPost("clients/{clientName}/playpause")]
public ActionResult PlayPauseClient(string clientName)
{
var success = _webSocketManager.PlayPauseClient(clientName);
if (success)
return Ok(new { message = $"客户端 '{clientName}' 播放/暂停状态已切换" });
else
return BadRequest($"切换客户端 '{clientName}' 播放/暂停状态失败");
}
/// <summary>
/// 重置客户端日志
/// </summary>
/// <param name="clientName">客户端名称</param>
/// <returns>操作结果</returns>
[HttpPost("clients/{clientName}/reset-logs")]
public ActionResult ResetClientLogs(string clientName)
{
var success = _webSocketManager.ResetClientLogs(clientName);
if (success)
return Ok(new { message = $"客户端 '{clientName}' 日志已重置" });
else
return BadRequest($"重置客户端 '{clientName}' 日志失败");
}
/// <summary>
/// 获取客户端日志
/// </summary>
/// <param name="clientName">客户端名称</param>
/// <param name="limit">日志数量限制</param>
/// <returns>日志列表</returns>
[HttpGet("clients/{clientName}/logs")]
public ActionResult<List<LTELog>?> GetClientLogs(string clientName, [FromQuery] int limit = 100)
{
var logs = _webSocketManager.GetClientLogs(clientName);
if (logs == null)
return NotFound($"客户端 '{clientName}' 不存在或未连接");
// 限制返回的日志数量
var limitedLogs = logs.TakeLast(limit).ToList();
return Ok(limitedLogs);
}
/// <summary>
/// 设置客户端日志配置
/// </summary>
/// <param name="clientName">客户端名称</param>
/// <param name="request">日志配置请求</param>
/// <returns>操作结果</returns>
[HttpPost("clients/{clientName}/logs-config")]
public ActionResult SetClientLogsConfig(string clientName, [FromBody] LogsConfigRequest request)
{
var success = _webSocketManager.SetClientLogsConfig(clientName, request.Config, request.Save);
if (success)
return Ok(new { message = $"客户端 '{clientName}' 日志配置已更新" });
else
return BadRequest($"更新客户端 '{clientName}' 日志配置失败");
}
/// <summary>
/// 发送消息到客户端
/// </summary>
/// <param name="clientName">客户端名称</param>
/// <param name="message">消息内容</param>
/// <returns>操作结果</returns>
[HttpPost("clients/{clientName}/send-message")]
public ActionResult SendMessage(string clientName, [FromBody] JObject message)
{
var messageId = _webSocketManager.SendMessageToClient(clientName, message);
if (messageId >= 0)
return Ok(new { messageId, message = $"消息已发送到客户端 '{clientName}'" });
else
return BadRequest($"发送消息到客户端 '{clientName}' 失败");
}
/// <summary>
/// 获取连接统计信息
/// </summary>
/// <returns>统计信息</returns>
[HttpGet("statistics")]
public ActionResult<ConnectionStatistics> GetStatistics()
{
var stats = _webSocketManager.GetConnectionStatistics();
return Ok(stats);
}
/// <summary>
/// 启动所有已配置的客户端
/// </summary>
/// <returns>操作结果</returns>
[HttpPost("start-all")]
public ActionResult StartAllClients()
{
_webSocketManager.StartAllConfiguredClients();
return Ok(new { message = "所有已配置的客户端已启动" });
}
/// <summary>
/// 停止所有客户端
/// </summary>
/// <returns>操作结果</returns>
[HttpPost("stop-all")]
public ActionResult StopAllClients()
{
_webSocketManager.StopAllClients();
return Ok(new { message = "所有客户端已停止" });
}
/// <summary>
/// 移除客户端配置
/// </summary>
/// <param name="clientName">客户端名称</param>
/// <returns>操作结果</returns>
[HttpDelete("configs/{clientName}")]
public ActionResult RemoveClientConfig(string clientName)
{
var success = _webSocketManager.RemoveClientConfig(clientName);
if (success)
return Ok(new { message = $"客户端 '{clientName}' 配置已移除" });
else
return BadRequest($"移除客户端 '{clientName}' 配置失败");
}
/// <summary>
/// 使用 Server-Sent Events (SSE) 实时推送客户端消息
/// </summary>
/// <param name="clientName">客户端名称</param>
[HttpGet("clients/{clientName}/messages/stream")]
public async Task StreamClientMessages(string clientName)
{
try
{
Response.ContentType = "text/event-stream";
Response.Headers.Append("Cache-Control", "no-cache");
Response.Headers.Append("Connection", "keep-alive");
Response.Headers.Append("Access-Control-Allow-Origin", "*");
var client = _webSocketManager.GetClientInstance(clientName);
if (client == null)
{
// 发送一个错误事件然后关闭
await SendSseEvent("error", new { message = "客户端未连接或不存在", clientName });
return;
}
// 发送一个连接成功事件
await SendSseEvent("open", new {
message = "成功连接到服务器事件流",
clientName,
timestamp = DateTime.UtcNow
});
await Response.Body.FlushAsync(HttpContext.RequestAborted);
int lastSentCount = 0;
int lastReceivedCount = 0;
var cancellationToken = HttpContext.RequestAborted;
while (!cancellationToken.IsCancellationRequested)
{
try
{
bool hasNewMessages = false;
// 检查并高效地发送新的"已发送"消息
var currentSentCount = client.SentMessagesCount;
if (currentSentCount > lastSentCount)
{
var sentMessages = client.SentMessages?.ToList() ?? new List<string>();
if (sentMessages.Count > lastSentCount)
{
var newMessages = sentMessages.Skip(lastSentCount).ToList();
if (newMessages.Any())
{
await SendSseEvent("update", new {
type = "sent",
messages = newMessages,
totalCount = currentSentCount,
newCount = newMessages.Count
});
lastSentCount = currentSentCount;
hasNewMessages = true;
}
}
}
// 检查并高效地发送新的"已接收"消息
var currentReceivedCount = client.ReceivedMessagesCount;
if (currentReceivedCount > lastReceivedCount)
{
var receivedMessages = client.ReceivedMessages?.ToList() ?? new List<string>();
if (receivedMessages.Count > lastReceivedCount)
{
var newMessages = receivedMessages.Skip(lastReceivedCount).ToList();
if (newMessages.Any())
{
await SendSseEvent("update", new {
type = "received",
messages = newMessages,
totalCount = currentReceivedCount,
newCount = newMessages.Count
});
lastReceivedCount = currentReceivedCount;
hasNewMessages = true;
}
}
}
if (hasNewMessages)
{
await Response.Body.FlushAsync(cancellationToken);
}
await Task.Delay(250, cancellationToken); // 每250毫秒检查一次新消息
}
catch (OperationCanceledException)
{
// 正常的取消操作,退出循环
break;
}
catch (Exception ex)
{
_logger.LogError(ex, "StreamClientMessages 循环中发生错误,客户端: {ClientName}", clientName);
await SendSseEvent("error", new {
message = "处理消息流时发生错误",
error = ex.Message,
clientName,
timestamp = DateTime.UtcNow
});
await Response.Body.FlushAsync(cancellationToken);
// 等待一段时间后继续,避免频繁错误
await Task.Delay(1000, cancellationToken);
}
}
// 发送断开连接事件
await SendSseEvent("disconnected", new {
message = "客户端消息流连接已断开",
clientName,
timestamp = DateTime.UtcNow
});
await Response.Body.FlushAsync(cancellationToken);
}
catch (OperationCanceledException)
{
_logger.LogInformation("StreamClientMessages 连接被客户端取消,客户端: {ClientName}", clientName);
}
catch (Exception ex)
{
_logger.LogError(ex, "StreamClientMessages 方法执行时发生未处理的异常,客户端: {ClientName}", clientName);
try
{
await SendSseEvent("fatal_error", new {
message = "服务器内部错误",
error = ex.Message,
clientName,
timestamp = DateTime.UtcNow
});
await Response.Body.FlushAsync();
}
catch
{
// 忽略发送错误事件时的异常
}
}
}
private async Task SendSseEvent(string eventName, object data)
{
try
{
if (string.IsNullOrEmpty(eventName))
{
_logger.LogWarning("尝试发送空事件名称的SSE事件");
return;
}
if (data == null)
{
_logger.LogWarning("尝试发送空数据的SSE事件: {EventName}", eventName);
return;
}
var json = Newtonsoft.Json.JsonConvert.SerializeObject(data);
var eventData = $"event: {eventName}\ndata: {json}\n\n";
await Response.WriteAsync(eventData);
}
catch (Exception ex)
{
_logger.LogError(ex, "发送SSE事件时发生错误: {EventName}", eventName);
// 不重新抛出异常,避免影响整个流
}
}
/// <summary>
/// 获取测试客户端配置
/// </summary>
/// <returns>测试客户端配置</returns>
[HttpGet("test-client-config")]
public ActionResult<ClientConfig> GetTestClientConfig()
{
var testConfig = _webSocketManager.GetTestClientConfig();
return Ok(testConfig);
}
/// <summary>
/// 设置测试客户端配置
/// </summary>
/// <param name="config">测试客户端配置</param>
/// <returns>操作结果</returns>
[HttpPost("test-client-config")]
public ActionResult SetTestClientConfig([FromBody] ClientConfig config)
{
if (string.IsNullOrEmpty(config.Name))
return BadRequest("客户端名称不能为空");
var success = _webSocketManager.SetTestClientConfig(config);
if (success)
return Ok(new { message = "测试客户端配置已更新" });
else
return BadRequest("更新测试客户端配置失败");
}
/// <summary>
/// 启动测试客户端
/// </summary>
/// <returns>操作结果</returns>
[HttpPost("test-client/start")]
public ActionResult StartTestClient()
{
var success = _webSocketManager.StartTestClient();
if (success)
return Ok(new { message = "测试客户端已启动" });
else
return BadRequest("启动测试客户端失败");
}
/// <summary>
/// 停止测试客户端
/// </summary>
/// <returns>操作结果</returns>
[HttpPost("test-client/stop")]
public ActionResult StopTestClient()
{
_logger.LogInformation("API 请求: 停止测试客户端");
var success = _webSocketManager.StopTestClient();
if (success)
return Ok(new { message = "测试客户端停止成功" });
else
return BadRequest("停止测试客户端失败");
}
/// <summary>
/// 获取日志缓存统计信息
/// </summary>
/// <returns>统计信息</returns>
[HttpGet("logs/stats")]
public ActionResult<object> GetLogCacheStats()
{
try
{
var stats = new
{
totalLogs = _webSocketManager.GetLogCacheCount(),
cacheSize = 10000, // LogCacheSize
timestamp = DateTime.UtcNow
};
return Ok(stats);
}
catch (Exception ex)
{
_logger.LogError(ex, "获取日志缓存统计信息时发生错误");
return StatusCode(500, new { message = "获取统计信息失败", error = ex.Message });
}
}
/// <summary>
/// 清空全局日志缓存
/// </summary>
/// <returns>操作结果</returns>
[HttpPost("logs/clear")]
public ActionResult ClearLogCache()
{
try
{
_webSocketManager.ClearLogCache();
return Ok(new { message = "日志缓存已清空" });
}
catch (Exception ex)
{
_logger.LogError(ex, "清空日志缓存时发生错误");
return StatusCode(500, new { message = "清空日志缓存失败", error = ex.Message });
}
}
/// <summary>
/// 重置全局日志缓存
/// </summary>
/// <returns>操作结果</returns>
[HttpPost("logs/reset")]
public ActionResult ResetLogCache()
{
try
{
_webSocketManager.ResetLogCache();
return Ok(new { message = "日志缓存已重置" });
}
catch (Exception ex)
{
_logger.LogError(ex, "重置日志缓存时发生错误");
return StatusCode(500, new { message = "重置日志缓存失败", error = ex.Message });
}
}
/// <summary>
/// 添加测试日志数据
/// </summary>
/// <returns>操作结果</returns>
[HttpPost("logs/add-test-data")]
public ActionResult AddTestLogData()
{
try
{
var testLogs = new List<LTELog>
{
};
// 手动添加到日志缓存
_webSocketManager.AddLogsToCache(testLogs);
return Ok(new { message = $"已添加 {testLogs.Count} 条测试日志" });
}
catch (Exception ex)
{
_logger.LogError(ex, "添加测试日志数据时发生错误");
return StatusCode(500, new { message = "添加测试日志失败", error = ex.Message });
}
}
/// <summary>
/// 使用 Server-Sent Events (SSE) 实时推送全局日志
/// </summary>
[HttpGet("logs/stream")]
public async Task StreamLogs(CancellationToken cancellationToken)
{
try
{
Response.ContentType = "text/event-stream";
Response.Headers.Append("Cache-Control", "no-cache");
Response.Headers.Append("Connection", "keep-alive");
Response.Headers.Append("Access-Control-Allow-Origin", "*");
// 发送连接成功事件
await SendSseEvent("connected", new { message = "日志流连接已建立", timestamp = DateTime.UtcNow });
await Response.Body.FlushAsync(cancellationToken);
int lastLogCount = 0;
var lastLogs = new List<LTELog>();
// 首先,一次性推送所有已缓存的日志
try
{
var initialLogs = _webSocketManager.GetLogCache()?.ToList() ?? new List<LTELog>();
_logger.LogInformation("StreamLogs: 获取到初始日志 {Count} 条", initialLogs.Count);
if (initialLogs.Any())
{
_logger.LogInformation("StreamLogs: 发送历史日志事件,日志数量: {Count}", initialLogs.Count);
await SendSseEvent("history", new { logs = initialLogs, totalCount = initialLogs.Count });
await Response.Body.FlushAsync(cancellationToken);
lastLogCount = initialLogs.Count;
lastLogs = initialLogs.ToList(); // 保存副本用于比较
}
else
{
_logger.LogInformation("StreamLogs: 没有历史日志数据");
}
}
catch (Exception ex)
{
_logger.LogError(ex, "获取初始日志时发生错误");
await SendSseEvent("error", new { message = "获取初始日志失败", error = ex.Message });
await Response.Body.FlushAsync(cancellationToken);
}
while (!cancellationToken.IsCancellationRequested)
{
try
{
var currentLogs = _webSocketManager.GetLogCache()?.ToList() ?? new List<LTELog>();
// 检查是否有新日志
if (currentLogs.Count > lastLogCount)
{
// 计算新增的日志
var newLogs = currentLogs.Skip(lastLogCount).ToList();
if (newLogs.Any())
{
_logger.LogInformation("StreamLogs: 发送新日志事件,新增日志数量: {NewCount}, 总日志数量: {TotalCount}", newLogs.Count, currentLogs.Count);
await SendSseEvent("new_logs", new {
logs = newLogs,
totalCount = currentLogs.Count,
newCount = newLogs.Count
});
await Response.Body.FlushAsync(cancellationToken);
// 更新索引和缓存
lastLogCount = currentLogs.Count;
lastLogs = currentLogs.ToList();
}
}
else if (currentLogs.Count < lastLogCount)
{
// 日志被清空或重置的情况
_logger.LogInformation("检测到日志缓存被重置,重新同步");
await SendSseEvent("reset", new {
message = "日志缓存已重置",
totalCount = currentLogs.Count
});
await Response.Body.FlushAsync(cancellationToken);
lastLogCount = currentLogs.Count;
lastLogs = currentLogs.ToList();
}
await Task.Delay(250, cancellationToken);
}
catch (OperationCanceledException)
{
// 正常的取消操作,退出循环
break;
}
catch (Exception ex)
{
_logger.LogError(ex, "StreamLogs 循环中发生错误");
await SendSseEvent("error", new {
message = "处理日志流时发生错误",
error = ex.Message,
timestamp = DateTime.UtcNow
});
await Response.Body.FlushAsync(cancellationToken);
// 等待一段时间后继续,避免频繁错误
await Task.Delay(1000, cancellationToken);
}
}
// 发送断开连接事件
await SendSseEvent("disconnected", new {
message = "日志流连接已断开",
timestamp = DateTime.UtcNow
});
await Response.Body.FlushAsync(cancellationToken);
}
catch (OperationCanceledException)
{
_logger.LogInformation("StreamLogs 连接被客户端取消");
}
catch (Exception ex)
{
_logger.LogError(ex, "StreamLogs 方法执行时发生未处理的异常");
try
{
await SendSseEvent("fatal_error", new {
message = "服务器内部错误",
error = ex.Message,
timestamp = DateTime.UtcNow
});
await Response.Body.FlushAsync();
}
catch
{
// 忽略发送错误事件时的异常
}
}
}
}
/// <summary>
/// 日志配置请求
/// </summary>
public class LogsConfigRequest
{
/// <summary>
/// 日志配置
/// </summary>
public Dictionary<string, object> Config { get; set; } = new();
/// <summary>
/// 是否保存配置
/// </summary>
public bool Save { get; set; } = false;
}
}