diff --git a/Example.Backend/Program.cs b/Example.Backend/Program.cs index 9763db9..fdf34c8 100644 --- a/Example.Backend/Program.cs +++ b/Example.Backend/Program.cs @@ -24,7 +24,7 @@ class Program builder.Services.AddOptions(); var listening = new IPEndPoint(IPAddress.Any, 4567); builder.Services.Configure(options => options.Endpoint = listening); - builder.Services.Configure(o => o.DistributionType = EDistributionType.Channeled); + builder.Services.Configure(o => o.DistributionType = EDistributionType.ExtractorFirst); builder.Services.AddSingleton(); builder.Services.AddSingleton(); diff --git a/mROA/Implementation/Backend/HubRequestExtractor.cs b/mROA/Implementation/Backend/HubRequestExtractor.cs index 362db29..4a3c48a 100644 --- a/mROA/Implementation/Backend/HubRequestExtractor.cs +++ b/mROA/Implementation/Backend/HubRequestExtractor.cs @@ -1,3 +1,4 @@ +using System.Collections.Generic; using Microsoft.Extensions.Options; using mROA.Abstract; using mROA.Implementation.Frontend; @@ -10,6 +11,7 @@ namespace mROA.Implementation.Backend private readonly IInstanceRepository _remoteContextRepository; private readonly IExecuteModule _executeModule; private readonly DistributionOptions _mode; + private Dictionary _producedExtractors = new(); public HubRequestExtractor(IRealStoreInstanceRepository contextRepository, IInstanceRepository remoteContextRepository, IExecuteModule executeModule, @@ -21,6 +23,8 @@ namespace mROA.Implementation.Backend _mode = mode.Value; } + public IRequestExtractor this[int id] => _producedExtractors[id]; + public IRequestExtractor HubOnOnConnected(IRepresentationModule interaction) { var extractor = CreateExtractor(interaction); @@ -29,6 +33,7 @@ namespace mROA.Implementation.Backend extractor.StartExtraction().ContinueWith(_ => OnDisconnected(interaction)); } + _producedExtractors[interaction.Id] = extractor; return extractor; } diff --git a/mROA/Implementation/Backend/NetworkGatewayModule.cs b/mROA/Implementation/Backend/NetworkGatewayModule.cs index a51f4a6..ecdc97b 100644 --- a/mROA/Implementation/Backend/NetworkGatewayModule.cs +++ b/mROA/Implementation/Backend/NetworkGatewayModule.cs @@ -19,7 +19,10 @@ namespace mROA.Implementation.Backend private readonly Dictionary _extractorsTokenSources = new(); private readonly ICallIndexProvider _callIndexProvider; private readonly IIdentityGenerator _identityGenerator; - public NetworkGatewayModule(IOptions options, IIdentityGenerator identityGenerator, IContextualSerializationToolKit serialization, ICallIndexProvider callIndexProvider, IConnectionHub hub, IOptions distribution, HubRequestExtractor hre) + + public NetworkGatewayModule(IOptions options, IIdentityGenerator identityGenerator, + IContextualSerializationToolKit serialization, ICallIndexProvider callIndexProvider, IConnectionHub hub, + IOptions distribution, HubRequestExtractor hre) { _tcpListener = new(options.Value.Endpoint); _identityGenerator = identityGenerator; @@ -57,7 +60,7 @@ namespace mROA.Implementation.Backend { Console.WriteLine($"Client connected from {client.Client.RemoteEndPoint}"); var interaction = new ChannelInteractionModule(_serialization, _identityGenerator); - + var context = new EndPointContext(null, null) { CallIndexProvider = _callIndexProvider @@ -100,17 +103,44 @@ namespace mROA.Implementation.Backend interaction.PostMessageAsync(new NetworkMessageHeader(_serialization, new IdAssignment { Id = interaction.ConnectionId }, null)); _extractorsTokenSources[interaction.ConnectionId] = cts; - - _hub.RegisterInteraction(interaction); - _hre.HubOnOnConnected(new RepresentationModule(interaction, _serialization)); + _hub.RegisterInteraction(interaction); + var requestExtractor = _hre.HubOnOnConnected(new RepresentationModule(interaction, _serialization)); + if (_distribution.DistributionType != EDistributionType.Channeled) { - + BindRequestFirstDistribution(context, interaction, streamExtractor, requestExtractor); } } - private void RecoverDisconnectedClient(NetworkMessageHeader connectionRequest, ChannelInteractionModule.StreamExtractor streamExtractor, + private void BindRequestFirstDistribution(IEndPointContext context, IChannelInteractionModule interaction, + ChannelInteractionModule.StreamExtractor streamExtractor, IRequestExtractor requestExtractor) + { + var converters = requestExtractor.Converters; + streamExtractor.MessageReceived = message => + { + if (requestExtractor.Rule(message)) + { + for (int i = 0; i < converters.Length; i++) + { + var func = converters[i]; + if (func(message) is { } t) + { + var deserialized = _serialization.Deserialize(message.Data, t, context); + requestExtractor.PushMessage(deserialized, message.MessageType); + break; + } + } + + return; + } + + interaction.ReceiveChanel.Writer.WriteAsync(message).ConfigureAwait(false); + }; + } + + private void RecoverDisconnectedClient(NetworkMessageHeader connectionRequest, + ChannelInteractionModule.StreamExtractor streamExtractor, CancellationTokenSource cts) { var recoveryRequest = _serialization.Deserialize(connectionRequest.Data, null); @@ -125,6 +155,11 @@ namespace mROA.Implementation.Backend }; _ = streamExtractor.SendFromChannel(recoveryInteraction.TrustedPostChanel, cts.Token); + if (_distribution.DistributionType == EDistributionType.ExtractorFirst) + { + BindRequestFirstDistribution(recoveryInteraction.Context, recoveryInteraction, streamExtractor, _hre[recoveryInteraction.ConnectionId]); + } + Task.Run(async () => await streamExtractor.LoopedReceive(cts.Token).ConfigureAwait(false)); recoveryInteraction.Restart(false); diff --git a/mROA/Implementation/ChannelInteractionModule.cs b/mROA/Implementation/ChannelInteractionModule.cs index 53bffc8..d7b85bf 100644 --- a/mROA/Implementation/ChannelInteractionModule.cs +++ b/mROA/Implementation/ChannelInteractionModule.cs @@ -165,7 +165,7 @@ namespace mROA.Implementation await _ioStream.ReadAsync(_lenBuffer); var len = BitConverter.ToUInt16(_lenBuffer); - + return len; }