refactor SDK to stateful architecture and implement reactive websocket handling
This commit is contained in:
@@ -16,6 +16,9 @@ namespace Chtn.CSharpSDK.Core
|
||||
private ClientWebSocket _webSocket;
|
||||
private string _connectionId;
|
||||
|
||||
public event Action<string> OnNewConnectionId;
|
||||
public event Action<string, string> OnPayloadReceived;
|
||||
|
||||
public static WebSocketHandler GetInstance()
|
||||
{
|
||||
if (_instance == null)
|
||||
@@ -25,6 +28,8 @@ namespace Chtn.CSharpSDK.Core
|
||||
return _instance;
|
||||
}
|
||||
|
||||
public string ConnId => _connectionId;
|
||||
|
||||
public async Task ConnectAsync(string userId, string token)
|
||||
{
|
||||
var client = HttpClientFactory.CreateClient();
|
||||
@@ -44,10 +49,58 @@ namespace Chtn.CSharpSDK.Core
|
||||
await _webSocket.ConnectAsync(new Uri(wsUrl), CancellationToken.None);
|
||||
|
||||
Console.WriteLine("Connected to websocket successfully");
|
||||
|
||||
_ = ReceiveLoop();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Console.WriteLine($"Websocket hiba: {ex.Message}");
|
||||
await Task.Delay(5000);
|
||||
await ConnectAsync(userId, token);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ReceiveLoop()
|
||||
{
|
||||
var buffer = new byte[1024 * 4];
|
||||
|
||||
while (_webSocket.State == WebSocketState.Open)
|
||||
{
|
||||
var result = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
|
||||
|
||||
if (result.MessageType == WebSocketMessageType.Close)
|
||||
{
|
||||
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None);
|
||||
}
|
||||
else
|
||||
{
|
||||
var message = Encoding.UTF8.GetString(buffer, 0, result.Count);
|
||||
HandleIncomingMessage(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void HandleIncomingMessage(string json)
|
||||
{
|
||||
try
|
||||
{
|
||||
var payload = JsonConvert.DeserializeObject<dynamic>(json);
|
||||
string action = payload.action;
|
||||
string data = JsonConvert.SerializeObject(payload.data);
|
||||
|
||||
if (action == "connectionId")
|
||||
{
|
||||
_connectionId = payload.data.connId;
|
||||
OnNewConnectionId?.Invoke(_connectionId);
|
||||
}
|
||||
else
|
||||
{
|
||||
OnPayloadReceived?.Invoke(action, data);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Console.WriteLine($"Payload feldolgozási hiba: {ex.Message}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
using System.Threading.Tasks;
|
||||
using Chtn.CSharp.SDK.Core;
|
||||
using Chtn.CSharp.SDK.Models.Call;
|
||||
using Chtn.CSharpSDK.Core;
|
||||
|
||||
namespace Chtn.CSharp.SDK.Services
|
||||
{
|
||||
@@ -15,10 +16,28 @@ namespace Chtn.CSharp.SDK.Services
|
||||
public class BroadcastServiceProvider : IBroadcastService
|
||||
{
|
||||
private readonly ApiClient _apiClient;
|
||||
private readonly string _channelId;
|
||||
private readonly string _categoryId;
|
||||
private readonly string _networkId;
|
||||
private readonly string _userId;
|
||||
private string _connId;
|
||||
|
||||
public BroadcastServiceProvider(ApiClient apiClient)
|
||||
public BroadcastServiceProvider(
|
||||
ApiClient apiClient,
|
||||
string channelId,
|
||||
string categoryId,
|
||||
string networkId,
|
||||
string userId)
|
||||
{
|
||||
_apiClient = apiClient ?? throw new ArgumentNullException(nameof(apiClient));
|
||||
_channelId = channelId;
|
||||
_categoryId = categoryId;
|
||||
_networkId = networkId;
|
||||
_userId = userId;
|
||||
|
||||
var ws = WebSocketHandler.GetInstance();
|
||||
this._connId = ws.ConnId;
|
||||
ws.OnNewConnectionId += (newConnId) => this._connId = newConnId;
|
||||
}
|
||||
|
||||
public async Task<StreamRegistry> GetData(GetRtmpDataReq req) =>
|
||||
Reference in New Issue
Block a user