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 Casters { get; } = new ConcurrentDictionary(); public LiveProcessor(ILoggerFactory loggerFactory) { _logger = loggerFactory.CreateLogger(); } public async Task 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(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>(); bool endOfInput = false; sBuffer = new ArraySegment(buffer,0,received.Count); fsInputQueue.Enqueue(sBuffer); var taskWritingToFs = liveHandler.ReceiveUserFile(user, _logger, destDir, fsInputQueue, fileName, () => endOfInput); Stack ToClose = new Stack(); 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(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"); } } }