Surprisingly, it works on the channels

This commit is contained in:
2025-05-02 20:15:04 +03:00
parent 557a52f24c
commit 23684d51de
9 changed files with 32 additions and 19 deletions
+1 -1
View File
@@ -45,7 +45,7 @@ class Program
Console.WriteLine(TransmissionConfig.OwnershipRepository.GetOwnershipId()); Console.WriteLine(TransmissionConfig.OwnershipRepository.GetOwnershipId());
var context = builder.GetModule<RemoteContextRepository>(); var context = builder.GetModule<RemoteContextRepository>();
var factory = context.GetSingleObject(typeof(IPrinterFactory), 0) as IPrinterFactory; var factory = context.GetSingleObject(typeof(IPrinterFactory), -TransmissionConfig.OwnershipRepository.GetHostOwnershipId()) as IPrinterFactory;
using (var disposingPrinter = factory.Create("Test")) using (var disposingPrinter = factory.Create("Test"))
{ {
@@ -13,10 +13,8 @@ namespace mROA.Implementation.Backend
return _ownerships.GetValueOrDefault(Environment.CurrentManagedThreadId, 0); return _ownerships.GetValueOrDefault(Environment.CurrentManagedThreadId, 0);
} }
public int GetHostOwnershipId() public int GetHostOwnershipId() => 0;
{
return 0;
}
public void RegisterOwnership(int ownershipId) public void RegisterOwnership(int ownershipId)
{ {
@@ -78,7 +78,10 @@ namespace mROA.Implementation.Backend
var streamExtractor = new StreamExtractor(client.GetStream(), _serialization); var streamExtractor = new StreamExtractor(client.GetStream(), _serialization);
interaction.IsConnected = () => streamExtractor.IsConnected; interaction.IsConnected = () => streamExtractor.IsConnected;
streamExtractor.MessageReceived = message => interaction.ReceiveChanel.Writer.WriteAsync(message); streamExtractor.MessageReceived = message =>
{
interaction.ReceiveChanel.Writer.WriteAsync(message);
};
streamExtractor.SingleReceive(); streamExtractor.SingleReceive();
var connectionRequest = interaction.GetNextMessageReceiving(false) var connectionRequest = interaction.GetNextMessageReceiving(false)
.GetAwaiter().GetResult()!; .GetAwaiter().GetResult()!;
@@ -98,7 +101,10 @@ namespace mROA.Implementation.Backend
var recoveryRequest = _serialization!.Deserialize<ClientRecovery>(connectionRequest.Data)!; var recoveryRequest = _serialization!.Deserialize<ClientRecovery>(connectionRequest.Data)!;
var recoveryInteraction = _hub.GetInteraction(recoveryRequest.Id); var recoveryInteraction = _hub.GetInteraction(recoveryRequest.Id);
streamExtractor.MessageReceived = message => recoveryInteraction.ReceiveChanel.Writer.WriteAsync(message); streamExtractor.MessageReceived = message =>
{
recoveryInteraction.ReceiveChanel.Writer.WriteAsync(message);
};
recoveryInteraction.Restart(false); recoveryInteraction.Restart(false);
Console.WriteLine("Connection recovery for client {0} finished", recoveryRequest.Id); Console.WriteLine("Connection recovery for client {0} finished", recoveryRequest.Id);
@@ -28,14 +28,14 @@ namespace mROA.Implementation
{ {
SingleReader = false, SingleReader = false,
SingleWriter = false, SingleWriter = false,
AllowSynchronousContinuations = true // AllowSynchronousContinuations = true
}); });
_receiveReader = ReceiveChanel.Reader; _receiveReader = ReceiveChanel.Reader;
_outputTrustedChannel = Channel.CreateBounded<NetworkMessageHeader>(new BoundedChannelOptions(1) _outputTrustedChannel = Channel.CreateBounded<NetworkMessageHeader>(new BoundedChannelOptions(1)
{ {
SingleReader = true, SingleReader = true,
SingleWriter = true, SingleWriter = true,
AllowSynchronousContinuations = true, // AllowSynchronousContinuations = true,
}); });
_trustedWriter = _outputTrustedChannel.Writer; _trustedWriter = _outputTrustedChannel.Writer;
@@ -43,7 +43,7 @@ namespace mROA.Implementation
{ {
SingleReader = true, SingleReader = true,
SingleWriter = true, SingleWriter = true,
AllowSynchronousContinuations = true // AllowSynchronousContinuations = true
}); });
_untrustedWriter = _outputUntrustedChannel.Writer; _untrustedWriter = _outputUntrustedChannel.Writer;
} }
@@ -49,7 +49,7 @@ namespace mROA.Implementation.Frontend
_tcpClient.Connect(_serverEndPoint); _tcpClient.Connect(_serverEndPoint);
PrepareExtractor(); PrepareExtractor();
_interactionModule.IsConnected = () => _currentExtractor.IsConnected; _interactionModule.IsConnected = () => _currentExtractor.IsConnected;
_interactionModule.OnDisconnected += id => { Reconnect(); }; _interactionModule.OnDisconnected += id => { Reconnect(); };
_interactionModule.PostMessageAsync(new NetworkMessageHeader(_serialization, new ClientConnect())).Wait(); _interactionModule.PostMessageAsync(new NetworkMessageHeader(_serialization, new ClientConnect())).Wait();
@@ -78,7 +78,10 @@ namespace mROA.Implementation.Frontend
_ = _currentExtractor.SendFromChannel(_interactionModule.TrustedPostChanel, _ = _currentExtractor.SendFromChannel(_interactionModule.TrustedPostChanel,
_rawExtractorCancellation.Token); _rawExtractorCancellation.Token);
_currentExtractor.MessageReceived = message => _interactionModule.ReceiveChanel.Writer.WriteAsync(message); _currentExtractor.MessageReceived = message =>
{
_interactionModule.ReceiveChanel.Writer.WriteAsync(message);
};
} }
private async Task Reconnect() private async Task Reconnect()
@@ -48,11 +48,14 @@ namespace mROA.Implementation.Frontend
{ {
ThrowIfNotInjected(); ThrowIfNotInjected();
TransmissionConfig.OwnershipRepository = new StaticOwnershipRepository(_representationModule.Id);
var multiClientOwnershipRepository = var multiClientOwnershipRepository =
TransmissionConfig.OwnershipRepository as MultiClientOwnershipRepository; TransmissionConfig.OwnershipRepository as MultiClientOwnershipRepository;
multiClientOwnershipRepository?.RegisterOwnership(_representationModule!.Id); multiClientOwnershipRepository?.RegisterOwnership(_representationModule!.Id);
if (multiClientOwnershipRepository is not null)
{
TransmissionConfig.OwnershipRepository = new StaticOwnershipRepository(_representationModule.Id);
}
try try
{ {
@@ -48,7 +48,7 @@ namespace mROA.Implementation
throw new NullReferenceException("representation producer is not initialized"); throw new NullReferenceException("representation producer is not initialized");
var representationModule = var representationModule =
_representationProducer.Produce(TransmissionConfig.OwnershipRepository.GetOwnershipId()); _representationProducer.Produce(ownerId);
_producedRemoteEndpoints.Add((Activator.CreateInstance(RemoteTypes[type], -1, _producedRemoteEndpoints.Add((Activator.CreateInstance(RemoteTypes[type], -1,
representationModule) as RemoteObjectBase)!); representationModule) as RemoteObjectBase)!);
+4 -1
View File
@@ -35,7 +35,10 @@ namespace mROA.Implementation
CancellationToken token = default, params Func<NetworkMessageHeader, Type?>[] converter) CancellationToken token = default, params Func<NetworkMessageHeader, Type?>[] converter)
{ {
var writer = _interaction.ReceiveChanel.Writer; var writer = _interaction.ReceiveChanel.Writer;
await foreach (var message in _interaction.ReceiveChanel.Reader.ReadAllAsync(token)) var reader = _interaction.ReceiveChanel.Reader;
await foreach (var message in reader.ReadAllAsync(token))
{ {
if (!rule(message)) if (!rule(message))
{ {
+1 -1
View File
@@ -21,7 +21,7 @@
</PropertyGroup> </PropertyGroup>
<PropertyGroup Condition=" '$(Configuration)' == 'Debug' "> <PropertyGroup Condition=" '$(Configuration)' == 'Debug' ">
<DefineConstants>TRACE;</DefineConstants> <DefineConstants></DefineConstants>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>