Refactoring and configure await false

This commit is contained in:
2025-07-18 13:49:36 +03:00
parent 17a6d3aad6
commit 85cf1bc48a
17 changed files with 87 additions and 133 deletions
+7 -13
View File
@@ -1,5 +1,4 @@
using System; using System;
using System.Linq;
using System.Net; using System.Net;
using Example.Backend; using Example.Backend;
using Example.Shared; using Example.Shared;
@@ -21,13 +20,13 @@ class Program
builder.Services.AddSingleton<IGatewayModule, NetworkGatewayModule>(); builder.Services.AddSingleton<IGatewayModule, NetworkGatewayModule>();
builder.Services.AddSingleton<IUntrustedGateway, UdpGateway>(); builder.Services.AddSingleton<IUntrustedGateway, UdpGateway>();
builder.Services.AddSingleton<IConnectionHub, ConnectionHub>(); builder.Services.AddSingleton<IConnectionHub, ConnectionHub>();
builder.Services.AddOptions(); builder.Services.AddOptions();
var listening = new IPEndPoint(IPAddress.Any, 4567); var listening = new IPEndPoint(IPAddress.Any, 4567);
builder.Services.Configure<GatewayOptions>(options => options.Endpoint = listening); builder.Services.Configure<GatewayOptions>(options => options.Endpoint = listening);
builder.Services.Configure<DistributionOptions>(o => o.DistributionType = EDistributionType.Channeled);
builder.Services.AddSingleton<HubRequestExtractor>(); builder.Services.AddSingleton<HubRequestExtractor>();
builder.Services.AddSingleton<IExecuteModule, BasicExecutionModule>(); builder.Services.AddSingleton<IExecuteModule, BasicExecutionModule>();
builder.Services.AddSingleton<IRepresentationModuleProducer, CreativeRepresentationModuleProducer>(); builder.Services.AddSingleton<IRepresentationModuleProducer, CreativeRepresentationModuleProducer>();
builder.Services.AddSingleton<IInstanceRepository, RemoteInstanceRepository>(); builder.Services.AddSingleton<IInstanceRepository, RemoteInstanceRepository>();
@@ -40,7 +39,6 @@ class Program
return repo; return repo;
})); }));
builder.Services.AddSingleton<IMessageDistributorFactory, ChannelDistributorFactory>();
builder.Services.AddSingleton<IMethodRepository>(p => builder.Services.AddSingleton<IMethodRepository>(p =>
{ {
var methodRepo = new CollectableMethodRepository(); var methodRepo = new CollectableMethodRepository();
@@ -52,17 +50,13 @@ class Program
builder.Services.AddSingleton<ICancellationRepository, CancellationRepository>(); builder.Services.AddSingleton<ICancellationRepository, CancellationRepository>();
var host = builder.Build(); var host = builder.Build();
host.Services.GetService<HubRequestExtractor>();
// //
// builder.Build(); new RemoteTypeBinder();
new RemoteTypeBinder();
// //
// _ = host.Services.GetService<IUntrustedGateway>()!.Start();
_ = host.Services.GetService<IUntrustedGateway>()!.Start(); var gateway = host.Services.GetService<IGatewayModule>();
var gateway = host.Services.GetService<IGatewayModule>(); gateway.Run();
gateway.Run();
Console.ReadLine();
Console.ReadLine();
} }
} }
+3
View File
@@ -39,8 +39,11 @@ class Program
builder.Services.AddSingleton<IRepresentationModule, RepresentationModule>(); builder.Services.AddSingleton<IRepresentationModule, RepresentationModule>();
var serverEndPoint = new IPEndPoint(IPAddress.Loopback, 4567); var serverEndPoint = new IPEndPoint(IPAddress.Loopback, 4567);
builder.Services.AddSingleton<IFrontendBridge, NetworkFrontendBridge>(); builder.Services.AddSingleton<IFrontendBridge, NetworkFrontendBridge>();
builder.Services.AddOptions(); builder.Services.AddOptions();
builder.Services.Configure<GatewayOptions>(options => options.Endpoint = serverEndPoint); builder.Services.Configure<GatewayOptions>(options => options.Endpoint = serverEndPoint);
builder.Services.Configure<DistributionOptions>(o => o.DistributionType = EDistributionType.Channeled);
builder.Services.AddSingleton<IRepresentationModuleProducer, StaticRepresentationModuleProducer>(); builder.Services.AddSingleton<IRepresentationModuleProducer, StaticRepresentationModuleProducer>();
builder.Services.AddSingleton<IRequestExtractor, RequestExtractor>(); builder.Services.AddSingleton<IRequestExtractor, RequestExtractor>();
builder.Services.AddSingleton<IExecuteModule, BasicExecutionModule>(); builder.Services.AddSingleton<IExecuteModule, BasicExecutionModule>();
@@ -1,15 +0,0 @@
using System.Threading.Tasks;
using mROA.Implementation;
namespace mROA.Abstract
{
public interface IPrimaryMessageDistributior
{
Task Distribute(NetworkMessageHeader message);
}
public interface IMessageDistributorFactory
{
IPrimaryMessageDistributior Produce(int clientId);
}
}
@@ -7,18 +7,10 @@ namespace mROA.Implementation.Backend
public class ConnectionHub : IConnectionHub public class ConnectionHub : IConnectionHub
{ {
private readonly Dictionary<int, IChannelInteractionModule> _connections = new(); private readonly Dictionary<int, IChannelInteractionModule> _connections = new();
private readonly IContextualSerializationToolKit _serializationToolkit;
public ConnectionHub(IContextualSerializationToolKit serializationToolkit)
{
_serializationToolkit = serializationToolkit;
}
public void RegisterInteraction(IChannelInteractionModule interaction) public void RegisterInteraction(IChannelInteractionModule interaction)
{ {
_connections.Add(interaction.ConnectionId, interaction); _connections.Add(interaction.ConnectionId, interaction);
var module = new RepresentationModule(interaction, _serializationToolkit);
OnConnected?.Invoke(module);
} }
public IChannelInteractionModule GetInteraction(int id) public IChannelInteractionModule GetInteraction(int id)
@@ -1,3 +1,4 @@
using Microsoft.Extensions.Options;
using mROA.Abstract; using mROA.Abstract;
using mROA.Implementation.Frontend; using mROA.Implementation.Frontend;
@@ -5,28 +6,30 @@ namespace mROA.Implementation.Backend
{ {
public class HubRequestExtractor public class HubRequestExtractor
{ {
private IRealStoreInstanceRepository _contextRepository; private readonly IRealStoreInstanceRepository _contextRepository;
private IInstanceRepository _remoteContextRepository; private readonly IInstanceRepository _remoteContextRepository;
private IMethodRepository _methodRepository; private readonly IExecuteModule _executeModule;
private IContextualSerializationToolKit _serializationToolkit; private readonly DistributionOptions _mode;
private IExecuteModule _executeModule;
public HubRequestExtractor(IConnectionHub hub, IRealStoreInstanceRepository contextRepository, public HubRequestExtractor(IRealStoreInstanceRepository contextRepository,
IInstanceRepository remoteContextRepository, IMethodRepository methodRepository, IInstanceRepository remoteContextRepository, IExecuteModule executeModule,
IContextualSerializationToolKit serializationToolkit, IExecuteModule executeModule) IOptions<DistributionOptions> mode)
{ {
hub.OnConnected += HubOnOnConnected;
_contextRepository = contextRepository; _contextRepository = contextRepository;
_remoteContextRepository = remoteContextRepository; _remoteContextRepository = remoteContextRepository;
_methodRepository = methodRepository;
_serializationToolkit = serializationToolkit;
_executeModule = executeModule; _executeModule = executeModule;
_mode = mode.Value;
} }
private void HubOnOnConnected(IRepresentationModule interaction) public IRequestExtractor HubOnOnConnected(IRepresentationModule interaction)
{ {
var extractor = CreateExtractor(interaction); var extractor = CreateExtractor(interaction);
extractor.StartExtraction().ContinueWith(_ => OnDisconnected(interaction)); if (_mode.DistributionType == EDistributionType.Channeled)
{
extractor.StartExtraction().ContinueWith(_ => OnDisconnected(interaction));
}
return extractor;
} }
private void OnDisconnected(IRepresentationModule representationModule) private void OnDisconnected(IRepresentationModule representationModule)
@@ -11,7 +11,7 @@ namespace mROA.Implementation.Backend
{ {
public static object[] EventBinders = { }; public static object[] EventBinders = { };
private IRepresentationModuleProducer _representationModuleProducer; private readonly IRepresentationModuleProducer _representationModuleProducer;
private Dictionary<int, object?> _singletons = new(); private Dictionary<int, object?> _singletons = new();
private readonly IStorage<object> _storage; private readonly IStorage<object> _storage;
@@ -13,19 +13,21 @@ namespace mROA.Implementation.Backend
{ {
private readonly TcpListener _tcpListener; private readonly TcpListener _tcpListener;
private readonly IConnectionHub _hub; private readonly IConnectionHub _hub;
private readonly HubRequestExtractor _hre;
private readonly DistributionOptions _distribution;
private readonly IContextualSerializationToolKit _serialization; private readonly IContextualSerializationToolKit _serialization;
private readonly Dictionary<int, CancellationTokenSource> _extractorsCTS = new(); private readonly Dictionary<int, CancellationTokenSource> _extractorsCTS = new();
private ICallIndexProvider _callIndexProvider; private readonly ICallIndexProvider _callIndexProvider;
private readonly IIdentityGenerator _identityGenerator; private readonly IIdentityGenerator _identityGenerator;
private readonly IMessageDistributorFactory _distributorFactory; public NetworkGatewayModule(IOptions<GatewayOptions> options, IIdentityGenerator identityGenerator, IContextualSerializationToolKit serialization, ICallIndexProvider callIndexProvider, IConnectionHub hub, IOptions<DistributionOptions> distribution, HubRequestExtractor hre)
public NetworkGatewayModule(IOptions<GatewayOptions> options, IIdentityGenerator identityGenerator, IContextualSerializationToolKit serialization, ICallIndexProvider callIndexProvider, IConnectionHub hub, IMessageDistributorFactory distributorFactory)
{ {
_tcpListener = new(options.Value.Endpoint); _tcpListener = new(options.Value.Endpoint);
_identityGenerator = identityGenerator; _identityGenerator = identityGenerator;
_serialization = serialization; _serialization = serialization;
_callIndexProvider = callIndexProvider; _callIndexProvider = callIndexProvider;
_hub = hub; _hub = hub;
_distributorFactory = distributorFactory; _hre = hre;
_distribution = distribution.Value;
} }
public void Run() public void Run()
@@ -57,7 +59,7 @@ namespace mROA.Implementation.Backend
interaction.IsConnected = () => streamExtractor.IsConnected; interaction.IsConnected = () => streamExtractor.IsConnected;
streamExtractor.MessageReceived = async message => streamExtractor.MessageReceived = async message =>
{ {
await interaction.ReceiveChanel.Writer.WriteAsync(message); await interaction.ReceiveChanel.Writer.WriteAsync(message).ConfigureAwait(false);
}; };
_ = Task.Run(() => streamExtractor.SingleReceive()); _ = Task.Run(() => streamExtractor.SingleReceive());
var connectionRequest = await interaction.ReceiveChanel.Reader.ReadAsync(); var connectionRequest = await interaction.ReceiveChanel.Reader.ReadAsync();
@@ -71,15 +73,21 @@ namespace mROA.Implementation.Backend
interaction.Context = context; interaction.Context = context;
Task.Run(async () => await streamExtractor.LoopedReceive(cts.Token)); Task.Run(async () => await streamExtractor.LoopedReceive(cts.Token));
_ = streamExtractor.SendFromChannel(interaction.TrustedPostChanel, cts.Token); _ = streamExtractor.SendFromChannel(interaction.TrustedPostChanel, cts.Token);
interaction.PostMessageAsync(new NetworkMessageHeader(_serialization!, interaction.PostMessageAsync(new NetworkMessageHeader(_serialization,
new IdAssignment { Id = interaction.ConnectionId }, null)); new IdAssignment { Id = interaction.ConnectionId }, null));
_extractorsCTS[interaction.ConnectionId] = cts; _extractorsCTS[interaction.ConnectionId] = cts;
_hub.RegisterInteraction(interaction); _hub.RegisterInteraction(interaction);
_hre.HubOnOnConnected(new RepresentationModule(interaction, _serialization));
if (_distribution.DistributionType != EDistributionType.Channeled)
{
}
Console.WriteLine("Client registered"); Console.WriteLine("Client registered");
break; break;
case EMessageType.ClientRecovery: case EMessageType.ClientRecovery:
{ {
var recoveryRequest = _serialization!.Deserialize<ClientRecovery>(connectionRequest.Data, null); var recoveryRequest = _serialization.Deserialize<ClientRecovery>(connectionRequest.Data, null);
var recoveryInteraction = _hub.GetInteraction(recoveryRequest.Id); var recoveryInteraction = _hub.GetInteraction(recoveryRequest.Id);
_extractorsCTS[-recoveryRequest.Id].Cancel(); _extractorsCTS[-recoveryRequest.Id].Cancel();
@@ -87,11 +95,11 @@ namespace mROA.Implementation.Backend
recoveryInteraction.IsConnected = () => streamExtractor.IsConnected; recoveryInteraction.IsConnected = () => streamExtractor.IsConnected;
streamExtractor.MessageReceived = message => streamExtractor.MessageReceived = message =>
{ {
recoveryInteraction.ReceiveChanel.Writer.WriteAsync(message); recoveryInteraction.ReceiveChanel.Writer.WriteAsync(message).ConfigureAwait(false);
}; };
_ = streamExtractor.SendFromChannel(recoveryInteraction.TrustedPostChanel, cts.Token); _ = streamExtractor.SendFromChannel(recoveryInteraction.TrustedPostChanel, cts.Token);
Task.Run(async () => await streamExtractor.LoopedReceive(cts.Token)); Task.Run(async () => await streamExtractor.LoopedReceive(cts.Token).ConfigureAwait(false));
recoveryInteraction.Restart(false); recoveryInteraction.Restart(false);
+5 -5
View File
@@ -12,11 +12,11 @@ namespace mROA.Implementation.Backend
{ {
public class UdpGateway : IUntrustedGateway public class UdpGateway : IUntrustedGateway
{ {
private IConnectionHub _hub; private readonly IConnectionHub _hub;
private UdpClient _client; private readonly UdpClient _client;
private Dictionary<IPEndPoint, int> _reservedPorts = new(); private readonly Dictionary<IPEndPoint, int> _reservedPorts = new();
private CancellationTokenSource _tokenSource = new(); private readonly CancellationTokenSource _tokenSource = new();
private IContextualSerializationToolKit _serializationToolkit; private readonly IContextualSerializationToolKit _serializationToolkit;
public UdpGateway(IOptions<GatewayOptions> options, IConnectionHub hub, IContextualSerializationToolKit serializationToolkit) public UdpGateway(IOptions<GatewayOptions> options, IConnectionHub hub, IContextualSerializationToolKit serializationToolkit)
{ {
@@ -1,35 +0,0 @@
using System.Threading.Channels;
using System.Threading.Tasks;
using mROA.Abstract;
namespace mROA.Implementation
{
public class ChannelDistributorFactory : IMessageDistributorFactory
{
private readonly IConnectionHub _hub;
public ChannelDistributorFactory(IConnectionHub hub)
{
_hub = hub;
}
public IPrimaryMessageDistributior Produce(int clientId)
{
return new ChannelMessageDistributor(_hub.GetInteraction(clientId).ReceiveChanel.Writer);
}
}
public class ChannelMessageDistributor : IPrimaryMessageDistributior
{
private readonly ChannelWriter<NetworkMessageHeader> _writer;
public ChannelMessageDistributor(ChannelWriter<NetworkMessageHeader> writer)
{
_writer = writer;
}
public async Task Distribute(NetworkMessageHeader message)
{
await _writer.WriteAsync(message);
}
}
}
@@ -62,11 +62,7 @@ namespace mROA.Implementation
public ValueTask<NetworkMessageHeader> GetNextMessageReceiving(bool infinite = true) public ValueTask<NetworkMessageHeader> GetNextMessageReceiving(bool infinite = true)
{ {
return _receiveReader.ReadAsync(); return _receiveReader.ReadAsync();
// if (_currentReceiving != null) return _currentReceiving;
// _currentReceiving = Task.Run(async () => await GetNextMessage());
// return _currentReceiving;
} }
#pragma warning disable CS8602 // Dereference of a possibly null reference.
private async ValueTask<bool> PostMessageInternal(NetworkMessageHeader messageHeader) private async ValueTask<bool> PostMessageInternal(NetworkMessageHeader messageHeader)
{ {
if (!IsConnected()) if (!IsConnected())
@@ -77,8 +73,6 @@ namespace mROA.Implementation
await _trustedWriter.WriteAsync(messageHeader); await _trustedWriter.WriteAsync(messageHeader);
return true; return true;
} }
#pragma warning restore CS8602 // Dereference of a possibly null reference.
public async Task PostMessageAsync(NetworkMessageHeader messageHeader) public async Task PostMessageAsync(NetworkMessageHeader messageHeader)
{ {
@@ -5,7 +5,7 @@ namespace mROA.Implementation
{ {
public class CollectableMethodRepository : IMethodRepository public class CollectableMethodRepository : IMethodRepository
{ {
private List<IMethodInvoker> _methods = new(); private readonly List<IMethodInvoker> _methods = new();
public void AppendInvokers(IEnumerable<IMethodInvoker> methodInvokers) public void AppendInvokers(IEnumerable<IMethodInvoker> methodInvokers)
{ {
@@ -1,16 +1,13 @@
using System; using mROA.Abstract;
using mROA.Abstract;
namespace mROA.Implementation namespace mROA.Implementation
{ {
public class CreativeRepresentationModuleProducer : IRepresentationModuleProducer public class CreativeRepresentationModuleProducer : IRepresentationModuleProducer
{ {
private IServiceProvider _creationModules; private readonly IConnectionHub _hub;
private IConnectionHub _hub; private readonly IContextualSerializationToolKit _serialization;
private IContextualSerializationToolKit _serialization; public CreativeRepresentationModuleProducer(IConnectionHub hub, IContextualSerializationToolKit serialization)
public CreativeRepresentationModuleProducer(IServiceProvider creationModules, IConnectionHub hub, IContextualSerializationToolKit serialization)
{ {
_creationModules = creationModules;
_hub = hub; _hub = hub;
_serialization = serialization; _serialization = serialization;
} }
@@ -0,0 +1,13 @@
namespace mROA.Implementation
{
public class DistributionOptions
{
public EDistributionType DistributionType { get; set; }
}
public enum EDistributionType
{
Channeled,
ExtractorFirst
}
}
@@ -14,11 +14,11 @@ namespace mROA.Implementation.Frontend
{ {
private readonly IPEndPoint _serverEndPoint; private readonly IPEndPoint _serverEndPoint;
private TcpClient _tcpClient = new(); private TcpClient _tcpClient = new();
private IChannelInteractionModule _interactionModule; private readonly IChannelInteractionModule _interactionModule;
private IContextualSerializationToolKit _serialization; private readonly IContextualSerializationToolKit _serialization;
private ChannelInteractionModule.StreamExtractor? _currentExtractor; private ChannelInteractionModule.StreamExtractor? _currentExtractor;
private CancellationTokenSource _rawExtractorCancellation; private CancellationTokenSource _rawExtractorCancellation;
private IEndPointContext _context; private readonly IEndPointContext _context;
public NetworkFrontendBridge(IOptions<GatewayOptions> options, IEndPointContext context, IContextualSerializationToolKit serialization, IChannelInteractionModule interactionModule) public NetworkFrontendBridge(IOptions<GatewayOptions> options, IEndPointContext context, IContextualSerializationToolKit serialization, IChannelInteractionModule interactionModule)
{ {
@@ -63,7 +63,7 @@ namespace mROA.Implementation.Frontend
_currentExtractor = _currentExtractor =
new ChannelInteractionModule.StreamExtractor(_tcpClient.GetStream(), _serialization, _context); new ChannelInteractionModule.StreamExtractor(_tcpClient.GetStream(), _serialization, _context);
_ = _currentExtractor.SendFromChannel(_interactionModule!.TrustedPostChanel, _ = _currentExtractor.SendFromChannel(_interactionModule.TrustedPostChanel,
_rawExtractorCancellation.Token); _rawExtractorCancellation.Token);
_currentExtractor.MessageReceived = message => _currentExtractor.MessageReceived = message =>
{ {
@@ -93,7 +93,7 @@ namespace mROA.Implementation.Frontend
public void Disconnect() public void Disconnect()
{ {
_ = _interactionModule!.PostMessageAsync(new NetworkMessageHeader(_serialization, new ClientDisconnect(), _ = _interactionModule.PostMessageAsync(new NetworkMessageHeader(_serialization, new ClientDisconnect(),
_context)); _context));
_interactionModule.Dispose(); _interactionModule.Dispose();
_tcpClient.Dispose(); _tcpClient.Dispose();
@@ -10,10 +10,10 @@ namespace mROA.Implementation.Frontend
public class RequestExtractor : IRequestExtractor public class RequestExtractor : IRequestExtractor
{ {
private IExecuteModule _executeModule; private readonly IExecuteModule _executeModule;
private IRepresentationModule _representationModule; private readonly IRepresentationModule _representationModule;
private IEndPointContext _context; private readonly IEndPointContext _context;
public RequestExtractor(IExecuteModule executeModule, IRepresentationModule representationModule, IEndPointContext context) public RequestExtractor(IExecuteModule executeModule, IRepresentationModule representationModule, IEndPointContext context)
{ {
@@ -84,7 +84,7 @@ namespace mROA.Implementation.Frontend
return; return;
} }
_representationModule.PostCallMessage(request.Id, resultType, result, _context); _representationModule.PostCallMessageAsync(request.Id, resultType, result, _context).ConfigureAwait(false);
} }
private void HandleEventRequest(DefaultCallRequest request) private void HandleEventRequest(DefaultCallRequest request)
@@ -7,10 +7,10 @@ namespace mROA.Implementation
{ {
public class RemoteInstanceRepository : IInstanceRepository public class RemoteInstanceRepository : IInstanceRepository
{ {
private List<RemoteObjectBase> _producedProxys = new(); private readonly List<RemoteObjectBase> _producedProxies = new();
private ICallIndexProvider _callIndexProvider; private readonly ICallIndexProvider _callIndexProvider;
private IRepresentationModuleProducer _representationProducer; private readonly IRepresentationModuleProducer _representationProducer;
public RemoteInstanceRepository(ICallIndexProvider callIndexProvider, IRepresentationModuleProducer representationProducer) public RemoteInstanceRepository(ICallIndexProvider callIndexProvider, IRepresentationModuleProducer representationProducer)
{ {
@@ -31,7 +31,7 @@ namespace mROA.Implementation
public T GetObject<T>(ComplexObjectIdentifier id, IEndPointContext context) where T : class public T GetObject<T>(ComplexObjectIdentifier id, IEndPointContext context) where T : class
{ {
var index = _producedProxys.Find(i => i.Identifier.Equals(id)); var index = _producedProxies.Find(i => i.Identifier.Equals(id));
if (index is not null) if (index is not null)
return (T)(index as object); return (T)(index as object);
@@ -42,7 +42,7 @@ namespace mROA.Implementation
var remote = remoteType(id.ContextId, var remote = remoteType(id.ContextId,
representationModule, context, _callIndexProvider.GetIndices(typeof(T))); representationModule, context, _callIndexProvider.GetIndices(typeof(T)));
_producedProxys.Add(remote!); _producedProxies.Add(remote!);
return (remote as T)!; return (remote as T)!;
} }
@@ -62,9 +62,9 @@ namespace mROA.Implementation
var remoteObjectBase = instance; var remoteObjectBase = instance;
_producedProxys.Add(remoteObjectBase); _producedProxies.Add(remoteObjectBase);
return _producedProxys.Last(); return _producedProxies.Last();
} }
public int GetObjectIndex<T>(object o, IEndPointContext context) public int GetObjectIndex<T>(object o, IEndPointContext context)
+2 -2
View File
@@ -12,8 +12,8 @@ namespace mROA.Implementation
{ {
public class RepresentationModule : IRepresentationModule public class RepresentationModule : IRepresentationModule
{ {
private IChannelInteractionModule _interaction; private readonly IChannelInteractionModule _interaction;
private IContextualSerializationToolKit _serialization; private readonly IContextualSerializationToolKit _serialization;
public RepresentationModule(IChannelInteractionModule interaction, IContextualSerializationToolKit serialization) public RepresentationModule(IChannelInteractionModule interaction, IContextualSerializationToolKit serialization)
{ {