notazof/yavsc
1
1
Fork 0
Code Issues Pull requests Projects 1 Releases 4 Packages Wiki Activity Actions

REORG

Browse source
This commit is contained in:
Paul Schneider 2026-02-14 16:54:27 +00:00
commit 45514010f2
3505 changed files with 154 additions and 523 deletions
Download patch file Download diff file Expand all files Collapse all files

202
src/Server/Services/LiveProcessor.cs Normal file
View file

@ -0,0 +1,202 @@
using System.Collections.Concurrent;
using System.Net.WebSockets;
using Yavsc.Models;
using Yavsc.ViewModels.Streaming;
using Newtonsoft.Json;
using Microsoft.Extensions.Logging;
using Microsoft.AspNetCore.Http;
namespace Yavsc.Services
{
public class LiveProcessor : ILiveProcessor
{
private readonly ILogger _logger;
public ConcurrentDictionary<string, LiveCastHandler> Casters { get; } = new ConcurrentDictionary<string, LiveCastHandler>();
public LiveProcessor(ILoggerFactory loggerFactory)
{
_logger = loggerFactory.CreateLogger<LiveProcessor>();
}
public async Task<bool> AcceptStream(HttpContext context, ApplicationUser user, string destDir, string fileName)
{
// TODO defer request handling
string uname = user.UserName;
LiveCastHandler liveHandler = null;
if (Casters.ContainsKey(uname))
{
_logger.LogWarning($"Casters.ContainsKey({uname})");
liveHandler = Casters[uname];
if (liveHandler.Socket.State == WebSocketState.Open || liveHandler.Socket.State == WebSocketState.Connecting)
{
_logger.LogWarning($"Closing cx");
// FIXME loosed connexion should be detected & disposed else where
await liveHandler.Socket.CloseAsync(WebSocketCloseStatus.EndpointUnavailable, "one by user", CancellationToken.None);
}
if (!liveHandler.TokenSource.IsCancellationRequested)
{
liveHandler.TokenSource.Cancel();
}
liveHandler.Socket.Dispose();
liveHandler.Socket = await context.WebSockets.AcceptWebSocketAsync();
liveHandler.TokenSource = new CancellationTokenSource();
}
else
{
_logger.LogInformation($"new caster");
// Accept the socket
liveHandler = new LiveCastHandler { Socket = await context.WebSockets.AcceptWebSocketAsync() };
}
_logger.LogInformation("Accepted web socket");
// Dispatch the flow
try
{
if (liveHandler.Socket != null && liveHandler.Socket.State == WebSocketState.Open)
{
Casters[uname] = liveHandler;
// TODO: Handle the socket here.
// Find receivers: others in the chat room
// send them the flow
var buffer = new byte[Constants.WebSocketsMaxBufLen];
var sBuffer = new ArraySegment<byte>(buffer);
_logger.LogInformation("Receiving bytes...");
WebSocketReceiveResult received = await liveHandler.Socket.ReceiveAsync(sBuffer, liveHandler.TokenSource.Token);
_logger.LogInformation($"Received bytes : {received.Count}");
_logger.LogInformation($"Is the end : {received.EndOfMessage}");
var fsInputQueue = new Queue<ArraySegment<byte>>();
bool endOfInput = false;
sBuffer = new ArraySegment<byte>(buffer,0,received.Count);
fsInputQueue.Enqueue(sBuffer);
var taskWritingToFs = liveHandler.ReceiveUserFile(user, _logger, destDir, fsInputQueue, fileName, () => endOfInput);
Stack<string> ToClose = new Stack<string>();
try
{
do
{
_logger.LogInformation($"Echoing {received.Count} bytes received in a {received.MessageType} message; Fin={received.EndOfMessage}");
// Echo anything we receive
// and send to all listner found
_logger.LogInformation($"{liveHandler.Listeners.Count} listeners");
foreach (var cliItem in liveHandler.Listeners)
{
var listenningSocket = cliItem.Value;
if (listenningSocket.State == WebSocketState.Open)
{
_logger.LogInformation(cliItem.Key);
await listenningSocket.SendAsync(
sBuffer, received.MessageType, received.EndOfMessage, liveHandler.TokenSource.Token);
}
else if (listenningSocket.State == WebSocketState.CloseReceived || listenningSocket.State == WebSocketState.CloseSent)
{
ToClose.Push(cliItem.Key);
}
}
if (!received.CloseStatus.HasValue)
{
_logger.LogInformation("try and receive new bytes");
buffer = new byte[Constants.WebSocketsMaxBufLen];
received = await liveHandler.Socket.ReceiveAsync(sBuffer, liveHandler.TokenSource.Token);
_logger.LogInformation($"Received bytes : {received.Count}");
sBuffer = new ArraySegment<byte>(buffer,0,received.Count);
_logger.LogInformation($"segment : offset: {sBuffer.Offset} count: {sBuffer.Count}");
_logger.LogInformation($"Is the end : {received.EndOfMessage}");
if (received.CloseStatus.HasValue)
{
endOfInput=true;
_logger.LogInformation($"received a close status: {received.CloseStatus.Value}: {received.CloseStatusDescription}");
}
else fsInputQueue.Enqueue(sBuffer);
}
else endOfInput=true;
while (ToClose.Count > 0)
{
string no = ToClose.Pop();
_logger.LogInformation("Closing follower connection");
WebSocket listenningSocket;
if (liveHandler.Listeners.TryRemove(no, out listenningSocket))
{
await listenningSocket.CloseAsync(WebSocketCloseStatus.EndpointUnavailable,
"State != WebSocketState.Open", CancellationToken.None);
listenningSocket.Dispose();
}
}
}
while (liveHandler.Socket.State == WebSocketState.Open);
_logger.LogInformation("Closing connection");
taskWritingToFs.Wait();
await liveHandler.Socket.CloseAsync(WebSocketCloseStatus.NormalClosure, received.CloseStatusDescription, liveHandler.TokenSource.Token);
liveHandler.TokenSource.Cancel();
liveHandler.Dispose();
_logger.LogInformation("Resulting file : " + JsonConvert.SerializeObject(taskWritingToFs.Result));
}
catch (Exception ex)
{
_logger.LogError($"Exception occured : {ex.Message}");
_logger.LogError(ex.StackTrace);
liveHandler.TokenSource.Cancel();
throw;
}
taskWritingToFs.Dispose();
}
else
{
// Socket was not accepted open ...
// not (meta.Socket != null && meta.Socket.State == WebSocketState.Open)
if (liveHandler.Socket != null)
{
_logger.LogError($"meta.Socket.State not Open: {liveHandler.Socket.State} ");
liveHandler.Socket.Dispose();
}
else
_logger.LogError("socket object is null");
}
RemoveLiveInfo(uname);
}
catch (IOException ex)
{
if (ex.Message == "Unexpected end of stream")
{
_logger.LogError($"Unexpected end of stream");
}
else
{
_logger.LogError($"Really unexpected end of stream");
await liveHandler.Socket?.CloseAsync(WebSocketCloseStatus.EndpointUnavailable, ex.Message, CancellationToken.None);
}
liveHandler.Socket?.Dispose();
RemoveLiveInfo(uname);
}
return true;
}
void RemoveLiveInfo(string userName)
{
LiveCastHandler caster;
if (Casters.TryRemove(userName, out caster))
_logger.LogInformation("removed live info");
else
_logger.LogError("could not remove live info");
}
}
}