You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
Zero/ZeroLevel/Services/Network/FileTransfer/FileSender.cs

130 lines
4.8 KiB

using MemoryPools;
using System;
using System.Collections.Concurrent;
using System.IO;
using System.Threading;
using ZeroLevel.Models;
namespace ZeroLevel.Network.FileTransfer
{
public sealed class FileSender
{
private BlockingCollection<FileTransferTask> _tasks = new BlockingCollection<FileTransferTask>();
private DefaultObjectPool<FileTransferTask> _taskPool = new DefaultObjectPool<FileTransferTask>(new DefaultPooledObjectPolicy<FileTransferTask>());
private readonly Thread _uploadFileThread;
private bool _resendWhenServerError = false;
private bool _resendWhenClientError = false;
public void ResendWhenServerError(bool resend = true) => _resendWhenServerError = resend;
public void ResendWhenClientError(bool resend = true) => _resendWhenClientError = resend;
public FileSender()
{
_uploadFileThread = new Thread(UploadFileProcessing);
_uploadFileThread.IsBackground = true;
_uploadFileThread.Start();
}
public void Send(IClient client, string fileName, Action<string> completeHandler = null!, Action<string, string> errorHandler = null!)
{
if (client == null!) return;
PushTransferTask(client, fileName, completeHandler, errorHandler);
}
private void PushTransferTask(IClient client, string filePath, Action<string> completeHandler = null!, Action<string, string> errorHandler = null!)
{
if (client == null!)
{
throw new ArgumentNullException(nameof(client));
}
if (string.IsNullOrWhiteSpace(filePath))
{
throw new ArgumentNullException(nameof(filePath));
}
if (false == File.Exists(filePath))
{
throw new FileNotFoundException(filePath);
}
var task = _taskPool.Get();
task.CompletedHandler = completeHandler;
task.ErrorHandler = errorHandler;
task.FilePath = filePath;
task.Client = client;
_tasks.Add(task);
}
private void UploadFileProcessing()
{
while (true)
{
var task = _tasks.Take();
try
{
ExecuteSendFile(new FileReader(task.FilePath), task);
task.CompletedHandler?.Invoke(task.FilePath);
}
catch (Exception ex)
{
task.ErrorHandler?.Invoke(task.FilePath, ex.ToString());
}
finally
{
_taskPool.Return(task);
}
}
}
public bool Connected(IClient client, TimeSpan timeout)
{
bool connected = false;
using (var waiter = new ManualResetEvent(false))
{
client.Request<bool>("__file_transfer_ping__", (response) => { connected = response; waiter.Set(); });
waiter.WaitOne(timeout);
}
return connected;
}
private static bool Send<T>(IClient client, string inbox, T frame,
bool resendWhenConnectionError, bool resendWhenServerError)
{
bool sended = false;
var handle = new Action<InvokeResult>(ir =>
{
if (ir.Success == false && resendWhenServerError)
{
Send<T>(client, inbox, frame, resendWhenConnectionError, false);
}
});
sended = client.Request<T, InvokeResult>(inbox, frame, handle);
if (sended == false && resendWhenConnectionError)
{
sended = client.Request<T, InvokeResult>(inbox, frame, handle);
}
return sended;
}
internal void ExecuteSendFile(FileReader reader, FileTransferTask task)
{
Log.Info($"Start upload file {reader.Path}");
var startinfo = reader.GetStartInfo();
if (!Send(task.Client, "__file_transfer_start_transfer__", startinfo, _resendWhenClientError, _resendWhenServerError))
{
Log.Debug($"Upload file {reader.Path} interrupted");
return;
}
foreach (var chunk in reader.Read())
{
if (!Send(task.Client, "__file_transfer_frame__", chunk, _resendWhenClientError, _resendWhenServerError))
{
Log.Debug($"Upload file {reader.Path} interrupted");
return;
}
}
Send(task.Client, "__file_transfer_complete_transfer__", reader.GetCompleteInfo(), _resendWhenClientError, _resendWhenServerError);
GC.Collect();
}
}
}

Powered by TurnKey Linux.