using Microsoft.AspNetCore.SignalR.Client;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using RemoteExec.Client.Exceptions;
using RemoteExec.Shared;
using System.Collections.Concurrent;
using System.Reflection;
using System.Text.Json;
namespace RemoteExec.Client;
///
/// Manages remote execution of static methods across one or more server connections with load balancing and fault tolerance.
///
public partial class RemoteExecutor : IAsyncDisposable
{
private readonly BlockingCollection globalQueue;
private readonly List servers = [];
private readonly ConcurrentDictionary> pendingResults = new();
private readonly ConcurrentDictionary> serverAssignedTasks = new();
private readonly AsyncManualResetEvent serverAvailableSignal = new(false);
private CancellationTokenSource distributorCts = new();
private Task? distributorTask;
private readonly RemoteExecutorOptions options;
private readonly ILogger logger;
private bool disposedValue;
///
/// Occurs when server metrics are updated.
///
public event EventHandler? MetricsUpdated;
///
/// Initializes a new instance of the class with a single server URL.
///
/// The URL of the remote server.
/// An action to configure the executor options.
public RemoteExecutor(string url, Action configure) : this([url], NullLogger.Instance, configure)
{
}
///
/// Initializes a new instance of the class with a single server URL and logger.
///
/// The URL of the remote server.
/// The logger instance.
/// An action to configure the executor options.
public RemoteExecutor(string url, ILogger logger, Action configure) : this([url], logger, configure)
{
}
///
/// Initializes a new instance of the class with multiple server URLs.
///
/// The URLs of the remote servers.
/// An action to configure the executor options.
public RemoteExecutor(string[] urls, Action configure) : this(urls, NullLogger.Instance, configure)
{
}
///
/// Initializes a new instance of the class with multiple server URLs and logger.
///
/// The URLs of the remote servers.
/// The logger instance.
/// An action to configure the executor options.
public RemoteExecutor(string[] urls, ILogger logger, Action configure)
{
this.logger = logger;
options = new RemoteExecutorOptions();
globalQueue = new BlockingCollection(options.GlobalQueueCapacity);
configure(options);
InitializeServers(urls);
}
///
/// Initializes a new instance of the class with preconfigured options.
///
/// The URLs of the remote servers.
/// The executor options.
/// The logger instance.
public RemoteExecutor(string[] urls, RemoteExecutorOptions options, ILogger logger)
{
this.logger = logger;
this.options = options;
globalQueue = new BlockingCollection(this.options.GlobalQueueCapacity);
InitializeServers(urls);
}
private void InitializeServers(string[] urls)
{
if (string.IsNullOrWhiteSpace(options.ApiKey))
{
throw new InvalidOperationException("API Key is required. Configure it in RemoteExecutorOptions.");
}
foreach (string url in urls)
{
Uri baseUri = new(url);
Uri signalRUri = new(baseUri, "/remote");
HubConnection connection = new HubConnectionBuilder()
.WithUrl(signalRUri, httpOptions =>
{
httpOptions.Headers["X-API-Key"] = options.ApiKey;
})
.ConfigureLogging(logging =>
{
_ = logging.AddProvider(new RemoteExecLoggerProvider(logger));
})
.Build();
HttpClient httpClient = new()
{
BaseAddress = baseUri
};
httpClient.DefaultRequestHeaders.Add("X-API-Key", options.ApiKey);
ServerConnection serverConnection = new ServerConnection(connection, httpClient);
servers.Add(serverConnection);
serverAssignedTasks[serverConnection] = new ConcurrentDictionary();
}
}
///
/// Gets the current metrics for all connected servers.
///
/// A dictionary mapping server IDs to their metrics.
public Dictionary GetCurrentServerMetrics()
{
return servers
.Select(server => server.Metrics)
.Where(metrics => metrics != null)
.ToDictionary(metrics => metrics!.ServerId, metrics => metrics!);
}
///
/// Executes a delegate remotely and returns the strongly-typed result.
///
/// The delegate type.
/// The return type.
/// The delegate to execute.
/// A cancellation token to cancel the operation.
/// The arguments to pass to the method.
/// The result of the remote execution.
public async Task ExecuteAsync(TDelegate @delegate, CancellationToken cancellationToken, params object[] args) where TDelegate : Delegate
{
object? execResult = await ExecuteAsync(@delegate, cancellationToken, args);
if (execResult is TResult typedResult)
{
return typedResult;
}
else if (execResult is JsonElement jsonElement)
{
return jsonElement.Deserialize()!;
}
else
{
throw new InvalidCastException("The result cannot be cast to the specified type.");
}
}
///
/// Executes a delegate remotely and returns the strongly-typed result.
///
/// The delegate type.
/// The return type.
/// The delegate to execute.
/// The arguments to pass to the method.
/// The result of the remote execution.
public async Task ExecuteAsync(TDelegate @delegate, params object[] args) where TDelegate : Delegate
{
return await ExecuteAsync(@delegate, CancellationToken.None, args);
}
///
/// Executes a delegate remotely and returns the result.
///
/// The delegate type.
/// The delegate to execute. Must be a static method.
/// A cancellation token to cancel the operation.
/// The arguments to pass to the method.
/// The result of the remote execution.
/// Thrown when the delegate is not a static method or uses a dynamic assembly.
/// Thrown when the remote execution fails.
public async Task