단순 이관
This commit is contained in:
237
Bitget_Server/Infrastructure/Bitget/BitgetWebSocketClient.cs
Normal file
237
Bitget_Server/Infrastructure/Bitget/BitgetWebSocketClient.cs
Normal file
@@ -0,0 +1,237 @@
|
||||
using System.Net.WebSockets;
|
||||
using System.Text;
|
||||
using Bitget_Server.Application.Options;
|
||||
using Bitget_Server.Application.State;
|
||||
using Bitget_Server.Domain;
|
||||
using Bitget_Server.Domain.Enums;
|
||||
using Bitget_Server.Domain.Models;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Newtonsoft.Json;
|
||||
using Newtonsoft.Json.Linq;
|
||||
|
||||
namespace Bitget_Server.Infrastructure.Bitget;
|
||||
|
||||
/// <summary>
|
||||
/// 기존 WinForms WebConnection의 WebSocket 로직(BitGetWebSocektStart, closeWebSocket, ContinuePing,
|
||||
/// ValidateJSON, CalculateDataList)을 ASP.NET Core 환경으로 이식.
|
||||
/// 싱글톤으로 등록되어 백그라운드 서비스 + Controller에서 공유됨.
|
||||
/// </summary>
|
||||
public class BitgetWebSocketClient : IAsyncDisposable
|
||||
{
|
||||
private readonly MarketState _state;
|
||||
private readonly ILogger<BitgetWebSocketClient> _logger;
|
||||
private readonly BitgetOptions _options;
|
||||
|
||||
private ClientWebSocket _webSocket;
|
||||
private CancellationTokenSource _cts;
|
||||
private Task _receiveLoopTask;
|
||||
private Task _pingLoopTask;
|
||||
private readonly object _lock = new();
|
||||
private readonly List<Dictionary<string, object>> _webSocketDataList = new();
|
||||
|
||||
public bool IsConnected =>
|
||||
_webSocket != null && _webSocket.State == WebSocketState.Open;
|
||||
|
||||
public BitgetWebSocketClient(
|
||||
MarketState state,
|
||||
IOptions<BitgetOptions> options,
|
||||
ILogger<BitgetWebSocketClient> logger)
|
||||
{
|
||||
_state = state;
|
||||
_options = options.Value;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 기존 BitGetWebSocektStart와 동일 — subscribe 메시지(dic)를 만들어 연결 시작.
|
||||
/// </summary>
|
||||
public async Task StartAsync(BitgetWebSocketType wsType, Dictionary<string, object> subscribeDic, CancellationToken ct = default)
|
||||
{
|
||||
await StopAsync(ct).ConfigureAwait(false);
|
||||
|
||||
_cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||
_webSocket = new ClientWebSocket();
|
||||
|
||||
try
|
||||
{
|
||||
await _webSocket.ConnectAsync(new Uri(BitgetEndpoints.WebSocketPublic), _cts.Token).ConfigureAwait(false);
|
||||
_logger.LogInformation("WebSocket 연결 성공 ({Type})", wsType);
|
||||
|
||||
var jsonData = JsonConvert.SerializeObject(subscribeDic);
|
||||
await SendMessageAsync(jsonData, _cts.Token).ConfigureAwait(false);
|
||||
|
||||
_pingLoopTask = Task.Run(() => PingLoopAsync(_cts.Token));
|
||||
_receiveLoopTask = Task.Run(() => ReceiveLoopAsync(_cts.Token));
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "WebSocket 연결 실패");
|
||||
await StopAsync(CancellationToken.None).ConfigureAwait(false);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ReceiveLoopAsync(CancellationToken ct)
|
||||
{
|
||||
var tempStrData = string.Empty;
|
||||
try
|
||||
{
|
||||
while (!ct.IsCancellationRequested && _webSocket.State == WebSocketState.Open)
|
||||
{
|
||||
byte[] buffer = new byte[1024];
|
||||
var result = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), ct).ConfigureAwait(false);
|
||||
var message = Encoding.UTF8.GetString(buffer, 0, result.Count);
|
||||
|
||||
if (message.Contains("pong"))
|
||||
{
|
||||
var parts = message.Split(new[] { "pong" }, StringSplitOptions.RemoveEmptyEntries);
|
||||
message = string.Concat(parts);
|
||||
}
|
||||
|
||||
// JSON 조각 병합
|
||||
if (tempStrData.Length == 0)
|
||||
{
|
||||
if (!ValidateJSON(message)) { tempStrData = message; continue; }
|
||||
}
|
||||
else
|
||||
{
|
||||
tempStrData += message;
|
||||
if (!ValidateJSON(tempStrData)) continue;
|
||||
message = tempStrData;
|
||||
tempStrData = string.Empty;
|
||||
}
|
||||
|
||||
if (message.Contains("{\"action"))
|
||||
{
|
||||
var splitData = message.Split(new[] { "{\"action" }, StringSplitOptions.RemoveEmptyEntries);
|
||||
foreach (var item in splitData)
|
||||
{
|
||||
var packet = "{\"action" + item;
|
||||
try
|
||||
{
|
||||
var datalist = JsonConvert.DeserializeObject<Dictionary<string, object>>(packet);
|
||||
if (datalist != null && datalist.TryGetValue("action", out var action) && action?.ToString() == "update")
|
||||
{
|
||||
lock (_lock) _webSocketDataList.Add(datalist);
|
||||
}
|
||||
}
|
||||
catch (JsonReaderException) { }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) { }
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "WebSocket 수신 루프 오류");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task PingLoopAsync(CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
while (!ct.IsCancellationRequested)
|
||||
{
|
||||
await Task.Delay(_options.WebSocketPingIntervalMs, ct).ConfigureAwait(false);
|
||||
await SendMessageAsync("ping", ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) { }
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Ping 루프 종료");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task SendMessageAsync(string message, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
byte[] buffer = Encoding.UTF8.GetBytes(message);
|
||||
await _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, ct).ConfigureAwait(false);
|
||||
_logger.LogDebug("WebSocket 메시지 전송: {Message}", message);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "WebSocket 메시지 전송 실패");
|
||||
}
|
||||
}
|
||||
|
||||
private static bool ValidateJSON(string s)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (s.Contains("{\"action"))
|
||||
{
|
||||
var splitData = s.Split(new[] { "{\"action" }, StringSplitOptions.RemoveEmptyEntries);
|
||||
foreach (var item in splitData)
|
||||
{
|
||||
try { JToken.Parse("{\"action" + item); }
|
||||
catch (JsonReaderException) { return false; }
|
||||
}
|
||||
return true;
|
||||
}
|
||||
JToken.Parse(s);
|
||||
return true;
|
||||
}
|
||||
catch (JsonReaderException) { return false; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 기존 CalculateDataList 로직 이식.
|
||||
/// BackgroundService가 1초 주기로 호출하여 WebSocket 누적 데이터를 candleDataLists로 병합.
|
||||
/// </summary>
|
||||
public void CalculateDataList(BitgetWebSocketType bitgetWebSocketType, List<object> intervalList)
|
||||
{
|
||||
try
|
||||
{
|
||||
List<Dictionary<string, object>> cloneList;
|
||||
lock (_lock)
|
||||
{
|
||||
cloneList = _webSocketDataList.Select(d => new Dictionary<string, object>(d)).ToList();
|
||||
_webSocketDataList.Clear();
|
||||
}
|
||||
|
||||
// 주의: 복잡한 캔들 병합 로직은 기존 WinForms 버전 그대로 유지 시
|
||||
// IndicatorProcess와 강하게 결합되므로 별도 서비스에서 처리하도록
|
||||
// 구조만 유지하고 세부 처리는 IndicatorProcessService로 위임한다.
|
||||
// (이 메서드 자체는 기존 WebConnection.CalculateDataList 진입점 역할)
|
||||
foreach (var item in cloneList)
|
||||
{
|
||||
// TODO: 필요 시 기존 시간대별 병합 로직 포팅
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "WebSocket 데이터 처리 중 오류");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken ct)
|
||||
{
|
||||
try { _cts?.Cancel(); } catch { }
|
||||
|
||||
if (_webSocket != null)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (_webSocket.State == WebSocketState.Open)
|
||||
await _webSocket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "WebSocket 연결 종료", ct).ConfigureAwait(false);
|
||||
}
|
||||
catch { }
|
||||
_webSocket.Dispose();
|
||||
_webSocket = null;
|
||||
}
|
||||
|
||||
_pingLoopTask = null;
|
||||
_receiveLoopTask = null;
|
||||
_cts?.Dispose();
|
||||
_cts = null;
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await StopAsync(CancellationToken.None).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user