paul/yavsc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

a first file sent using packets at max size

Browse Source
This commit is contained in:
Paul Schneider
2019-06-30 02:41:48 +01:00
parent d0f1e9ca22
commit 59617ffe3f
13 changed files with 217 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/Yavsc.Server/Constants.cs
View File

@ -37,9 +37,8 @@ namespace Yavsc
AnonAvatar = "/images/Users/icon_anon_user.png",
YavscConnectionStringEnvName = "YAVSC_DB_CONNECTION";
public const int WebSocketsMaxBufLen = 4*1024;
// at the end, let 4*4 bytes in peace
public const int WebSocketsMaxBufLen = 4*1020;
public static readonly long DefaultFSQ = 1024*1024*500;

1
src/Yavsc.Server/Models/Messaging/LiveFlow.cs
View File

@ -1,3 +1,4 @@
using System;
using System.ComponentModel.DataAnnotations;
using System.ComponentModel.DataAnnotations.Schema;
using Yavsc.Abstract.Streaming;

37
src/Yavsc/Helpers/FileSystemHelpers.cs
View File

@ -11,6 +11,7 @@ using System.Threading;
using System.Threading.Tasks;
using System.Web;
using Microsoft.AspNet.Http;
using Microsoft.Extensions.Logging;
using Yavsc.Abstract.FileSystem;
using Yavsc.Exceptions;
using Yavsc.Models;
@ -115,43 +116,7 @@ public static FileRecievedInfo ReceiveProSignature(this ClaimsPrincipal user, st
return ReceiveUserFile(user, root, f.OpenReadStream(), destFileName ?? f.ContentDisposition, f.ContentType, CancellationToken.None);
}
public static FileRecievedInfo ReceiveUserFile(this ApplicationUser user, string root, Queue<ArraySegment<byte>> queue, string destFileName, string contentType, CancellationToken token)
{
// TODO lock user's disk usage for this scope,
// this process is not safe at concurrent access.
long usage = user.DiskUsage;
var item = new FileRecievedInfo();
item.FileName = Yavsc.Abstract.FileSystem.AbstractFileSystemHelpers.FilterFileName (destFileName);
item.MimeType = contentType;
item.DestDir = root;
var fi = new FileInfo(Path.Combine(root, item.FileName));
if (fi.Exists)
{
item.Overriden = true;
usage -= fi.Length;
}
using (var dest = fi.OpenWrite())
{
while (!token.IsCancellationRequested)
{
if (queue.Count==0) Task.Delay(300);
else {
var buffer = queue.Dequeue();
dest.Write(buffer.Array,buffer.Offset, buffer.Count);
usage += buffer.Count;
}
if (usage >= user.DiskQuota) break;
}
user.DiskUsage = usage;
dest.Close();
}
if (usage >= user.DiskQuota) {
item.QuotaOffensed = true;
}
user.DiskUsage = usage;
return item;
}
public static FileRecievedInfo ReceiveUserFile(this ApplicationUser user, string root, Stream inputStream, string destFileName, string contentType, CancellationToken token)
{
// TODO lock user's disk usage for this scope,

127
src/Yavsc/Services/LiveProcessor.cs
View File

@ -52,56 +52,71 @@ namespace Yavsc.Services
return false;
}
_logger.LogInformation("flow : "+flow.Title+" for "+uname);
LiveCastHandler meta = null;
LiveCastHandler liveHandler = null;
if (Casters.ContainsKey(uname))
{
_logger.LogWarning($"Casters.ContainsKey({uname})");
meta = Casters[uname];
if (meta.Socket.State == WebSocketState.Open || meta.Socket.State == WebSocketState.Connecting )
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 meta.Socket.CloseAsync( WebSocketCloseStatus.EndpointUnavailable, "one by user", CancellationToken.None);
await liveHandler.Socket.CloseAsync( WebSocketCloseStatus.EndpointUnavailable, "one by user", CancellationToken.None);
}
if (!meta.TokenSource.IsCancellationRequested) {
meta.TokenSource.Cancel();
if (!liveHandler.TokenSource.IsCancellationRequested) {
liveHandler.TokenSource.Cancel();
}
meta.Socket.Dispose();
meta.Socket = await context.WebSockets.AcceptWebSocketAsync();
meta.TokenSource = new CancellationTokenSource();
liveHandler.Socket.Dispose();
liveHandler.Socket = await context.WebSockets.AcceptWebSocketAsync();
liveHandler.TokenSource = new CancellationTokenSource();
}
else
{
_logger.LogInformation($"new caster");
// Accept the socket
meta = new LiveCastHandler { Socket = await context.WebSockets.AcceptWebSocketAsync() };
liveHandler = new LiveCastHandler { Socket = await context.WebSockets.AcceptWebSocketAsync() };
}
_logger.LogInformation("Accepted web socket");
// Dispatch the flow
try
{
if (meta.Socket != null && meta.Socket.State == WebSocketState.Open)
if (liveHandler.Socket != null && liveHandler.Socket.State == WebSocketState.Open)
{
Casters[uname] = meta;
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 buffer = new byte[Constants.WebSocketsMaxBufLen+16];
var sBuffer = new ArraySegment<byte>(buffer);
_logger.LogInformation("Receiving bytes...");
WebSocketReceiveResult received = await meta.Socket.ReceiveAsync(sBuffer, meta.TokenSource.Token);
_logger.LogInformation($"Received bytes : {received.Count}");
WebSocketReceiveResult received = await liveHandler.Socket.ReceiveAsync(sBuffer, liveHandler.TokenSource.Token);
int count = (received.Count<4)? 0 : buffer[0]*256*1024 +buffer[1]*1024+buffer[2]*256 + buffer[3];
_logger.LogInformation($"Received bytes : {count}");
_logger.LogInformation($"Is the end : {received.EndOfMessage}");
const string livePath = "live";
string destDir = context.User.InitPostToFileSystem(livePath);
_logger.LogInformation($"Saving flow to {destDir}");
string fileName = flow.GetFileName();
FileInfo destFileInfo = new FileInfo(Path.Combine(destDir, fileName));
// this should end :-)
while (destFileInfo.Exists) {
flow.SequenceNumber++;
fileName = flow.GetFileName();
destFileInfo = new FileInfo(Path.Combine(destDir, fileName));
}
var fsInputQueue = new Queue<ArraySegment<byte>>();
var taskWritingToFs = Task<FileRecievedInfo>.Run( ()=> user.ReceiveUserFile(destDir, fsInputQueue, fileName, flow.MediaType, meta.TokenSource.Token));
bool endOfInput=false;
fsInputQueue.Enqueue(sBuffer);
var taskWritingToFs = liveHandler.ReceiveUserFile(user, _logger, destDir, fsInputQueue, fileName, flow.MediaType, ()=> endOfInput);
var hubContext = GlobalHost.ConnectionManager.GetHubContext<ChatHub>();
hubContext.Clients.All.addPublicStream(new PublicStreamInfo
@ -117,80 +132,70 @@ namespace Yavsc.Services
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
foreach (var cliItem in meta.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, meta.TokenSource.Token);
sBuffer, received.MessageType, received.EndOfMessage, liveHandler.TokenSource.Token);
}
else
if (listenningSocket.State == WebSocketState.CloseReceived || listenningSocket.State == WebSocketState.CloseSent)
else if (listenningSocket.State == WebSocketState.CloseReceived || listenningSocket.State == WebSocketState.CloseSent)
{
ToClose.Push(cliItem.Key);
}
}
fsInputQueue.Enqueue(sBuffer);
// logger.LogInformation("replying...");
while (!received.CloseStatus.HasValue)
{
// reply echo
// await meta.Socket.SendAsync(new ArraySegment<byte>(buffer), received.MessageType, received.EndOfMessage, meta.TokenSource.Token);
_logger.LogInformation("Receiving new bytes...");
buffer = new byte[Constants.WebSocketsMaxBufLen];
buffer = new byte[Constants.WebSocketsMaxBufLen+16];
sBuffer = new ArraySegment<byte>(buffer);
received = await liveHandler.Socket.ReceiveAsync(sBuffer, liveHandler.TokenSource.Token);
received = await meta.Socket.ReceiveAsync(sBuffer, meta.TokenSource.Token);
foreach (var cliItem in meta.Listeners)
{
var listenningSocket = cliItem.Value;
if (listenningSocket.State == WebSocketState.Open) {
await listenningSocket.SendAsync(
sBuffer, received.MessageType, received.EndOfMessage, meta.TokenSource.Token);
}
else
if (listenningSocket.State == WebSocketState.CloseReceived || listenningSocket.State == WebSocketState.CloseSent)
{
ToClose.Push(cliItem.Key);
}
}
fsInputQueue.Enqueue(sBuffer);
_logger.LogInformation($"Received new bytes : {received.Count}");
count = (received.Count<4)? 0 : buffer[0]*256*1024 +buffer[1]*1024+buffer[2]*256 + buffer[3];
_logger.LogInformation($"Received bytes : {count}");
_logger.LogInformation($"Is the end : {received.EndOfMessage}");
if (received.Count<=4 || count > Constants.WebSocketsMaxBufLen) {
if (received.CloseStatus.HasValue) {
_logger.LogInformation($"received a close status: {received.CloseStatus.Value.ToString()}: {received.CloseStatusDescription}");
}
else {
_logger.LogError("Wrong packet size: "+count.ToString());
_logger.LogError(JsonConvert.SerializeObject(received));
}
}
else fsInputQueue.Enqueue(sBuffer);
while (ToClose.Count >0)
{
string no = ToClose.Pop();
_logger.LogInformation("Closing follower connection");
WebSocket listenningSocket;
if (meta.Listeners.TryRemove(no, out listenningSocket)) {
if (liveHandler.Listeners.TryRemove(no, out listenningSocket)) {
await listenningSocket.CloseAsync(WebSocketCloseStatus.EndpointUnavailable,
"State != WebSocketState.Open", CancellationToken.None);
listenningSocket.Dispose();
}
}
}
while (!received.CloseStatus.HasValue);
_logger.LogInformation("Closing connection");
await meta.Socket.CloseAsync(received.CloseStatus.Value, received.CloseStatusDescription, CancellationToken.None);
meta.Socket.Dispose();
meta.TokenSource.Cancel();
endOfInput=true;
taskWritingToFs.Wait();
_logger.LogInformation("Resulting file : " +JsonConvert.SerializeObject(taskWritingToFs.Result));
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);
await meta.Socket.CloseAsync(received.CloseStatus.Value, "exception occured", CancellationToken.None);
meta.Socket.Dispose();
meta.TokenSource.Cancel();
liveHandler.TokenSource.Cancel();
throw;
}
taskWritingToFs.Dispose();
}
@ -198,16 +203,15 @@ namespace Yavsc.Services
{
// Socket was not accepted open ...
// not (meta.Socket != null && meta.Socket.State == WebSocketState.Open)
if (meta.Socket != null)
if (liveHandler.Socket != null)
{
_logger.LogError($"meta.Socket.State not Open: {meta.Socket.State.ToString()} ");
meta.Socket.Dispose();
_logger.LogError($"meta.Socket.State not Open: {liveHandler.Socket.State.ToString()} ");
liveHandler.Socket.Dispose();
}
else
_logger.LogError("socket object is null");
}
RemoveLiveInfo(uname);
}
catch (IOException ex)
@ -219,10 +223,9 @@ namespace Yavsc.Services
else
{
_logger.LogError($"Really unexpected end of stream");
await meta.Socket?.CloseAsync(WebSocketCloseStatus.EndpointUnavailable, ex.Message, CancellationToken.None);
await liveHandler.Socket?.CloseAsync(WebSocketCloseStatus.EndpointUnavailable, ex.Message, CancellationToken.None);
}
meta.Socket?.Dispose();
liveHandler.Socket?.Dispose();
RemoveLiveInfo(uname);
}
return true;

7
src/Yavsc/Startup/Startup.WebSockets.cs
View File

@ -16,10 +16,9 @@ namespace Yavsc
{
var webSocketOptions = new WebSocketOptions()
{
KeepAliveInterval = TimeSpan.FromSeconds(320),
ReceiveBufferSize = Constants.WebSocketsMaxBufLen,
ReplaceFeature = false
KeepAliveInterval = TimeSpan.FromSeconds(30),
ReceiveBufferSize = Constants.WebSocketsMaxBufLen+4*sizeof(int),
ReplaceFeature = true
};
app.UseWebSockets(webSocketOptions);

4
src/Yavsc/Startup/Startup.cs
View File

@ -178,9 +178,7 @@ namespace Yavsc
// Add framework services.
services.AddEntityFramework()
.AddNpgsql()
.AddDbContext<ApplicationDbContext>(
db => db.UseNpgsql(ConnectionString)
);
.AddDbContext<ApplicationDbContext>();
ConfigureOAuthServices(services);

96
src/Yavsc/ViewModels/Streaming/LiveCastHandler.cs Normal file
View File

@ -0,0 +1,96 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Yavsc.Models;
using Yavsc.Models.FileSystem;
namespace Yavsc.ViewModels.Streaming
{
public class LiveCastClient {
public string UserName { get; set; }
public WebSocket Socket { get; set; }
}
public class LiveEntryViewModel {
public string UserName { get; set; }
public string FlowId { get; set; }
}
public class LiveCastHandler : IDisposable
{
public WebSocket Socket { get; set; }
public ConcurrentDictionary<string, WebSocket> Listeners { get; set; } = new ConcurrentDictionary<string, WebSocket>();
public CancellationTokenSource TokenSource { get; set; } = new CancellationTokenSource();
public void Dispose()
{
}
public async Task<FileRecievedInfo> ReceiveUserFile(ApplicationUser user, ILogger logger, string root, Queue<ArraySegment<byte>> queue, string destFileName, string contentType, Func<bool> isEndOfInput)
{
// TODO lock user's disk usage for this scope,
// this process is not safe at concurrent access.
long usage = user.DiskUsage;
var item = new FileRecievedInfo();
item.FileName = Yavsc.Abstract.FileSystem.AbstractFileSystemHelpers.FilterFileName (destFileName);
item.MimeType = contentType;
item.DestDir = root;
var fi = new FileInfo(Path.Combine(root, item.FileName));
if (fi.Exists)
{
item.Overriden = true;
usage -= fi.Length;
}
logger.LogInformation("Opening the file");
using (var dest = fi.OpenWrite())
{
logger.LogInformation("Appening to file");
while (queue.Count>0)
{
while (queue.Count>0)
{
var buffer = queue.Dequeue();
int count = buffer.Array[0]*256*1024 +buffer.Array[1]*1024+buffer.Array[2]*256 + buffer.Array[3];
if (count >0 && count <= Constants.WebSocketsMaxBufLen
&& buffer.Array.Length >= count+4) {
logger.LogInformation($"writing {count} bytes from {buffer.Array.Length}.");
await dest.WriteAsync(buffer.Array, 4, count);
logger.LogInformation($"wrote {count} bytes.");
usage += count;
}
else {
var packetInfo = JsonConvert.SerializeObject(buffer);
logger.LogError($"didn´t wrote {count} bytes from {buffer.Array.Length}!\n{packetInfo}");
}
if (usage >= user.DiskQuota) break;
}
if (isEndOfInput()) break;
if (usage >= user.DiskQuota) break;
logger.LogInformation($"Waitting 200ms.");
await Task.Delay(100);
logger.LogInformation($"Done waiting");
}
user.DiskUsage = usage;
dest.Close();
}
if (usage >= user.DiskQuota) {
item.QuotaOffensed = true;
}
user.DiskUsage = usage;
return item;
}
}
}

25
src/Yavsc/ViewModels/Streaming/LiveCastMeta.cs
View File

@ -1,25 +0,0 @@
using System.Collections.Concurrent;
using System.Net.WebSockets;
using System.Threading;
namespace Yavsc.ViewModels.Streaming
{
public class LiveCastClient {
public string UserName { get; set; }
public WebSocket Socket { get; set; }
}
public class LiveEntryViewModel {
public string UserName { get; set; }
public string FlowId { get; set; }
}
public class LiveCastHandler
{
public WebSocket Socket { get; set; }
public ConcurrentDictionary<string, WebSocket> Listeners { get; set; } = new ConcurrentDictionary<string, WebSocket>();
public CancellationTokenSource TokenSource { get; set; } = new CancellationTokenSource();
}
}

2
src/Yavsc/project.json
View File

@ -100,7 +100,6 @@
"Microsoft.AspNet.SignalR.JS": "2.2.1",
"Microsoft.AspNet.StaticFiles": "1.0.0-rc1-*",
"Microsoft.AspNet.Tooling.Razor": "1.0.0-rc1-*",
"Microsoft.AspNet.WebSockets.Server": "1.0.0-rc1-*",
"Microsoft.AspNet.Session": "1.0.0-rc1-final",
"Microsoft.AspNet.Web.Optimization": "1.1.3",
"Microsoft.AspNet.Http.Extensions": "1.0.0-rc1-final",
@ -108,6 +107,7 @@
"Microsoft.AspNet.DataProtection.SystemWeb": "1.0.0-rc1-final",
"Microsoft.AspNet.Authentication.OAuth": "1.0.0-rc1-final",
"Microsoft.AspNet.Mvc.Formatters.Json": "6.0.0-rc1-final",
"Microsoft.AspNet.WebSockets.Server": "1.0.0-rc1-final",
"Microsoft.AspNet.OWin": "1.0.0-rc1-final",
"Microsoft.Framework.ConfigurationModel.Json": "1.0.0-beta4",
"Microsoft.Framework.Configuration.Abstractions": "1.0.0-beta8",

25
src/cli/Commands/Streamer.cs
View File

@ -83,7 +83,7 @@ namespace cli {
await _client.ConnectAsync(new Uri(url), _tokenSource.Token);
_logger.LogInformation("Connected");
const int bufLen = Constants.WebSocketsMaxBufLen;
byte [] buffer = new byte[bufLen];
byte [] buffer = new byte[bufLen+4*sizeof(int)];
const int offset=0;
int read = 0;
/*
@ -99,13 +99,24 @@ namespace cli {
} ); */
do {
read = await stream.ReadAsync(buffer, offset, bufLen);
var segment = new ArraySegment<byte>(buffer, offset, read);
bool end = read < bufLen;
await _client.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Binary, end, _tokenSource.Token);
_logger.LogInformation($"sent {read} bytes end:{end} ");
read = await stream.ReadAsync(buffer, offset + sizeof(int), bufLen);
if (read>0) {
// assert sizeof(int)==4
buffer[3]= (byte) (read % 256);
var left = read / 256;
buffer[2]= (byte) (left % 256);
left = left / 256;
buffer[1] = (byte) (left % 256);
left = left /256;
buffer[0]=(byte) (byte) (left % 256);
var segment = new ArraySegment<byte>(buffer, offset, read+4);
} while (read>0 && stream.CanRead );
await _client.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Binary, false, _tokenSource.Token);
_logger.LogInformation($"sent {segment.Count} ");
}
} while (read>0);
// reciving.Wait();
await _client.CloseAsync(WebSocketCloseStatus.NormalClosure, "EOF", _tokenSource.Token);
}

2
src/cli/Makefile
View File

@ -12,7 +12,7 @@ msbuild-restore:
check: run
project.lock.json:
project.lock.json: project.json
dnu restore
@# fixing package id reference case, to System.Xml, from package NJsonSchema.CodeGeneration.CSharp
sed 's/System.XML/System.Xml/' project.lock.json > project.lock.json.new && mv project.lock.json.new project.lock.json

5
src/cli/package.json Normal file
View File

@ -0,0 +1,5 @@
{
"dependencies": {
"ansi-to-html": "^0.6.11"
}
}

15
src/cli/yarn.lock Normal file
View File

@ -0,0 +1,15 @@
# THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY.
# yarn lockfile v1
ansi-to-html@^0.6.11:
version "0.6.11"
resolved "https://registry.yarnpkg.com/ansi-to-html/-/ansi-to-html-0.6.11.tgz#5093fc4962186c0e9343dec572a4f71abdc93439"
integrity sha512-88XZtrcwrfkyn6fGstHnkaF1kl7hGtNCYh4vSmItgEV+6JnQHryDBf7udF4f2RhTRQmYvJvPcTtqgaqrxzc9oA==
dependencies:
entities "^1.1.1"
entities@^1.1.1:
version "1.1.2"
resolved "https://registry.yarnpkg.com/entities/-/entities-1.1.2.tgz#bdfa735299664dfafd34529ed4f8522a275fea56"
integrity sha512-f2LZMYl1Fzu7YSBKg+RoROelpOaNrcGmE9AZubeDfrCEia483oW4MI4VyFd5VNHIgQ/7qm1I0wUHK1eJnn2y2w==