diff --git a/mROA.Cbor/CborSerializationToolkit.cs b/mROA.Cbor/CborSerializationToolkit.cs index c5cd523..f50fe7c 100644 --- a/mROA.Cbor/CborSerializationToolkit.cs +++ b/mROA.Cbor/CborSerializationToolkit.cs @@ -27,7 +27,7 @@ namespace mROA.Cbor private bool FindParser(Type t, out IOrdinaryStructureParser parser) { - if (t == typeof(NetworkMessageHeader)) + if (t == typeof(NetworkMessage)) { parser = _parsers[0]; return true; diff --git a/mROA.Cbor/IOrdinaryStructureParser.cs b/mROA.Cbor/IOrdinaryStructureParser.cs index 8faa205..2d3711a 100644 --- a/mROA.Cbor/IOrdinaryStructureParser.cs +++ b/mROA.Cbor/IOrdinaryStructureParser.cs @@ -16,7 +16,7 @@ namespace mROA.Cbor { public void Write(CborWriter writer, object value, IEndPointContext context, CborSerializationToolkit serialization) { - var v = value as NetworkMessageHeader; + var v = value as NetworkMessage; writer.WriteStartArray(3); writer.WriteByteString(v.Id.ToByteArray()); writer.WriteInt32((int)v.MessageType); @@ -27,7 +27,7 @@ namespace mROA.Cbor public object Read(CborReader reader, IEndPointContext context, CborSerializationToolkit serialization) { reader.ReadStartArray(); - var value = new NetworkMessageHeader + var value = new NetworkMessage { Id = new Guid(reader.ReadByteString()), MessageType = (EMessageType)reader.ReadInt32(), diff --git a/mROA/Abstract/IChannelInteractionModule.cs b/mROA/Abstract/IChannelInteractionModule.cs index e7054c0..0542417 100644 --- a/mROA/Abstract/IChannelInteractionModule.cs +++ b/mROA/Abstract/IChannelInteractionModule.cs @@ -9,13 +9,13 @@ namespace mROA.Abstract { int ConnectionId { get; set; } IEndPointContext Context { get; set; } - Channel ReceiveChanel { get; } - ChannelReader TrustedPostChanel { get; } - ChannelReader UntrustedPostChanel { get; } + Channel ReceiveChanel { get; } + ChannelReader TrustedPostChanel { get; } + ChannelReader UntrustedPostChanel { get; } Func IsConnected { get; set; } - ValueTask GetNextMessageReceiving(); - Task PostMessageAsync(NetworkMessageHeader messageHeader); - Task PostMessageUntrustedAsync(NetworkMessageHeader messageHeader); + ValueTask GetNextMessageReceiving(); + Task PostMessageAsync(NetworkMessage message); + Task PostMessageUntrustedAsync(NetworkMessage message); event Action OnDisconnected; Task Restart(bool sendRecovery); void PassReconnection(); diff --git a/mROA/Abstract/IRepresentationModule.cs b/mROA/Abstract/IRepresentationModule.cs index 7fd4bf5..45d44d3 100644 --- a/mROA/Abstract/IRepresentationModule.cs +++ b/mROA/Abstract/IRepresentationModule.cs @@ -11,13 +11,13 @@ namespace mROA.Abstract int Id { get; } IEndPointContext Context { get; } - Task<(object? Deserialized, EMessageType MessageType)> GetSingle(Predicate rule, + Task<(object? Deserialized, EMessageType MessageType)> GetSingle(Predicate rule, IEndPointContext? context, CancellationToken token = default, - params Func[] converter); + params Func[] converter); - IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream(Predicate rule, + IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream(Predicate rule, IEndPointContext? context, CancellationToken token = default, - params Func[] converter); + params Func[] converter); Task PostCallMessageAsync(Guid id, EMessageType eMessageType, T payload, IEndPointContext? context) where T : notnull; diff --git a/mROA/Abstract/IRequestExtractor.cs b/mROA/Abstract/IRequestExtractor.cs index d4a3c44..6fc3816 100644 --- a/mROA/Abstract/IRequestExtractor.cs +++ b/mROA/Abstract/IRequestExtractor.cs @@ -8,7 +8,7 @@ namespace mROA.Abstract { Task StartExtraction(); void PushMessage(object parced, EMessageType originalType); - Predicate Rule { get; } - Func[] Converters { get; } + Predicate Rule { get; } + Func[] Converters { get; } } } \ No newline at end of file diff --git a/mROA/Implementation/Backend/NetworkGatewayModule.cs b/mROA/Implementation/Backend/NetworkGatewayModule.cs index 1308bb7..6d4bdfb 100644 --- a/mROA/Implementation/Backend/NetworkGatewayModule.cs +++ b/mROA/Implementation/Backend/NetworkGatewayModule.cs @@ -98,14 +98,14 @@ namespace mROA.Implementation.Backend } private void HandleNewClient(EndPointContext context, ChannelInteractionModule interaction, - ChannelInteractionModule.StreamExtractor streamExtractor, CancellationTokenSource cts, NetworkMessageHeader connectionHeader) + ChannelInteractionModule.StreamExtractor streamExtractor, CancellationTokenSource cts, NetworkMessage connection) { context.HostId = 0; context.OwnerId = -interaction.ConnectionId; interaction.Context = context; Task.Run(async () => await streamExtractor.LoopedReceive(cts.Token)); _ = streamExtractor.SendFromChannel(interaction.TrustedPostChanel, cts.Token); - interaction.PostMessageAsync(new NetworkMessageHeader(_serialization, + interaction.PostMessageAsync(new NetworkMessage(_serialization, new IdAssignment { Id = interaction.ConnectionId }, null)); _extractorsTokenSources[interaction.ConnectionId] = cts; @@ -143,7 +143,7 @@ namespace mROA.Implementation.Backend }; } - private void RecoverDisconnectedClient(NetworkMessageHeader connectionRequest, + private void RecoverDisconnectedClient(NetworkMessage connectionRequest, ChannelInteractionModule.StreamExtractor streamExtractor, CancellationTokenSource cts) { diff --git a/mROA/Implementation/Backend/UdpGateway.cs b/mROA/Implementation/Backend/UdpGateway.cs index 12a5297..623858b 100644 --- a/mROA/Implementation/Backend/UdpGateway.cs +++ b/mROA/Implementation/Backend/UdpGateway.cs @@ -41,7 +41,7 @@ namespace mROA.Implementation.Backend while (token.IsCancellationRequested == false) { var incoming = await _client.ReceiveAsync(); - var parsed = _serializationToolkit.Deserialize(incoming.Buffer, null); + var parsed = _serializationToolkit.Deserialize(incoming.Buffer, null); try { int channelId; diff --git a/mROA/Implementation/ChannelInteractionModule.cs b/mROA/Implementation/ChannelInteractionModule.cs index 154c5fa..d084ad4 100644 --- a/mROA/Implementation/ChannelInteractionModule.cs +++ b/mROA/Implementation/ChannelInteractionModule.cs @@ -9,11 +9,11 @@ namespace mROA.Implementation { public class ChannelInteractionModule : IChannelInteractionModule { - private readonly ChannelReader _receiveReader; - private readonly ChannelWriter _trustedWriter; - private readonly ChannelWriter _untrustedWriter; - private readonly Channel _outputTrustedChannel; - private readonly Channel _outputUntrustedChannel; + private readonly ChannelReader _receiveReader; + private readonly ChannelWriter _trustedWriter; + private readonly ChannelWriter _untrustedWriter; + private readonly Channel _outputTrustedChannel; + private readonly Channel _outputUntrustedChannel; private readonly IContextualSerializationToolKit _serialization; private bool _isConnected = true; private bool _isActive = true; @@ -28,19 +28,19 @@ namespace mROA.Implementation public ChannelInteractionModule(IContextualSerializationToolKit serialization) { _serialization = serialization; - ReceiveChanel = Channel.CreateUnbounded(new UnboundedChannelOptions + ReceiveChanel = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = false, SingleWriter = false, }); _receiveReader = ReceiveChanel.Reader; - _outputTrustedChannel = Channel.CreateUnbounded(new UnboundedChannelOptions + _outputTrustedChannel = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true, SingleWriter = true }); _trustedWriter = _outputTrustedChannel.Writer; - _outputUntrustedChannel = Channel.CreateUnbounded(new UnboundedChannelOptions + _outputUntrustedChannel = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true, SingleWriter = true, @@ -52,33 +52,33 @@ namespace mROA.Implementation public int ConnectionId { get; set; } public IEndPointContext Context { get; set; } - public Channel ReceiveChanel { get; } + public Channel ReceiveChanel { get; } - public ChannelReader TrustedPostChanel => _outputTrustedChannel.Reader; - public ChannelReader UntrustedPostChanel => _outputUntrustedChannel.Reader; + public ChannelReader TrustedPostChanel => _outputTrustedChannel.Reader; + public ChannelReader UntrustedPostChanel => _outputUntrustedChannel.Reader; public Func IsConnected { get; set; } = () => false; - public ValueTask GetNextMessageReceiving() + public ValueTask GetNextMessageReceiving() { return _receiveReader.ReadAsync(); } - private async ValueTask PostMessageInternal(NetworkMessageHeader messageHeader) + private async ValueTask PostMessageInternal(NetworkMessage message) { if (!IsConnected()) { return false; } - await _trustedWriter.WriteAsync(messageHeader); + await _trustedWriter.WriteAsync(message); return true; } - public async Task PostMessageAsync(NetworkMessageHeader messageHeader) + public async Task PostMessageAsync(NetworkMessage message) { while (true) { - if (await PostMessageInternal(messageHeader)) + if (await PostMessageInternal(message)) break; if (!_isActive) @@ -91,9 +91,9 @@ namespace mROA.Implementation } } - public async Task PostMessageUntrustedAsync(NetworkMessageHeader messageHeader) + public async Task PostMessageUntrustedAsync(NetworkMessage message) { - await _untrustedWriter.WriteAsync(messageHeader); + await _untrustedWriter.WriteAsync(message); } public event Action? OnDisconnected; @@ -103,12 +103,12 @@ namespace mROA.Implementation if (sendRecovery) { await PostMessageAsync( - new NetworkMessageHeader(_serialization, new ClientRecovery(Math.Abs(ConnectionId)), Context)); + new NetworkMessage(_serialization, new ClientRecovery(Math.Abs(ConnectionId)), Context)); await ReceiveChanel.Reader.ReadAsync(); } else { - await _trustedWriter.WriteAsync(new NetworkMessageHeader()); + await _trustedWriter.WriteAsync(new NetworkMessage()); } PassReconnection(); @@ -141,7 +141,7 @@ namespace mROA.Implementation public class StreamExtractor { - private const int BufferSize = ushort.MaxValue; + private const int BufferSize = ushort.MaxValue + 2; private readonly Stream _ioStream; private readonly IContextualSerializationToolKit _serializationToolkit; @@ -158,25 +158,22 @@ namespace mROA.Implementation _lenBuffer = new byte[2]; } - public Action MessageReceived = _ => { }; - - private async Task ReadMessageLength() - { - await _ioStream.ReadAsync(_lenBuffer); - - var len = BitConverter.ToUInt16(_lenBuffer); - - return len; - } - + public Action MessageReceived = _ => { }; + public async Task SingleReceive(CancellationToken token = default) { - var len = await ReadMessageLength(); - var localSpan = _buffer[..len]; + int firstRead = _ioStream.Read(_buffer.Span); - await _ioStream.ReadExactlyAsync(localSpan, cancellationToken: token); - var message = _serializationToolkit.Deserialize(localSpan, _context); - // _logger.LogTrace("RECV {0}", message.ToString()); + var metadata = new NetworkMessage.NetworkMessageMeta(_buffer.Span); + + var localSpan = _buffer[2..len]; + if (firstRead - 2 != len) + { + localSpan = _buffer[(len + 2)..]; + await _ioStream.ReadExactlyAsync(localSpan, cancellationToken: token); + } + + var message = _serializationToolkit.Deserialize(localSpan, _context); MessageReceived(message); } @@ -188,7 +185,7 @@ namespace mROA.Implementation } } - private async Task Send(NetworkMessageHeader message, CancellationToken token = default) + private async Task Send(NetworkMessage message, CancellationToken token = default) { var bodySpan = _buffer[2..]; var len = _serializationToolkit.Serialize(message, bodySpan.Span, _context); @@ -199,7 +196,7 @@ namespace mROA.Implementation // _logger.LogTrace("SEND {0}", message.ToString()); } - public async Task SendFromChannel(ChannelReader channel, + public async Task SendFromChannel(ChannelReader channel, CancellationToken token = default) { while (token.IsCancellationRequested == false && IsConnected) diff --git a/mROA/Implementation/EMessageType.cs b/mROA/Implementation/EMessageType.cs index 205e85d..c0b5980 100644 --- a/mROA/Implementation/EMessageType.cs +++ b/mROA/Implementation/EMessageType.cs @@ -1,6 +1,6 @@ namespace mROA.Implementation { - public enum EMessageType + public enum EMessageType : byte { Unknown, FinishedCommandExecution, diff --git a/mROA/Implementation/Frontend/NetworkFrontendBridge.cs b/mROA/Implementation/Frontend/NetworkFrontendBridge.cs index 7203285..cfd7f4a 100644 --- a/mROA/Implementation/Frontend/NetworkFrontendBridge.cs +++ b/mROA/Implementation/Frontend/NetworkFrontendBridge.cs @@ -39,7 +39,7 @@ namespace mROA.Implementation.Frontend _interactionModule.IsConnected = () => _currentExtractor.IsConnected; _interactionModule.OnDisconnected += _ => { Reconnect().ConfigureAwait(false); }; - _interactionModule.PostMessageAsync(new NetworkMessageHeader(_serialization, new ClientConnect(), _context)) + _interactionModule.PostMessageAsync(new NetworkMessage(_serialization, new ClientConnect(), _context)) .Wait(); _ = _currentExtractor.SingleReceive().ConfigureAwait(false); @@ -95,7 +95,7 @@ namespace mROA.Implementation.Frontend public void Disconnect() { - _ = _interactionModule.PostMessageAsync(new NetworkMessageHeader(_serialization, new ClientDisconnect(), + _ = _interactionModule.PostMessageAsync(new NetworkMessage(_serialization, new ClientDisconnect(), _context)); _interactionModule.Dispose(); _tcpClient.Dispose(); diff --git a/mROA/Implementation/Frontend/RequestExtractor.cs b/mROA/Implementation/Frontend/RequestExtractor.cs index 1db877d..f13afe1 100644 --- a/mROA/Implementation/Frontend/RequestExtractor.cs +++ b/mROA/Implementation/Frontend/RequestExtractor.cs @@ -56,11 +56,11 @@ namespace mROA.Implementation.Frontend } } - public Predicate Rule { get; } = m => + public Predicate Rule { get; } = m => m.MessageType is EMessageType.CallRequest or EMessageType.CancelRequest or EMessageType.EventRequest or EMessageType.ClientDisconnect; - public Func[] Converters { get; } = + public Func[] Converters { get; } = { m => m.MessageType == EMessageType.CallRequest ? typeof(DefaultCallRequest) : null, m => m.MessageType == EMessageType.CancelRequest ? typeof(CancelRequest) : null, diff --git a/mROA/Implementation/Frontend/UdpUntrustedInteraction.cs b/mROA/Implementation/Frontend/UdpUntrustedInteraction.cs index b663248..13a6bce 100644 --- a/mROA/Implementation/Frontend/UdpUntrustedInteraction.cs +++ b/mROA/Implementation/Frontend/UdpUntrustedInteraction.cs @@ -44,7 +44,7 @@ namespace mROA.Implementation.Frontend while (token.IsCancellationRequested == false) { var message = new Memory((await udpClient.ReceiveAsync()).Buffer); - var parsed = _serializationToolkit.Deserialize(message, _context); + var parsed = _serializationToolkit.Deserialize(message, _context); await writer.WriteAsync(parsed, token); } @@ -52,7 +52,7 @@ namespace mROA.Implementation.Frontend private async Task Posting(UdpClient udpClient, CancellationToken token) { - var initMessage = new NetworkMessageHeader + var initMessage = new NetworkMessage { MessageType = EMessageType.UntrustedConnect, Id = Guid.NewGuid(), Data = BitConverter.GetBytes(_channelInteractionModule.ConnectionId) diff --git a/mROA/Implementation/NetworkMessageHeader.cs b/mROA/Implementation/NetworkMessage.cs similarity index 50% rename from mROA/Implementation/NetworkMessageHeader.cs rename to mROA/Implementation/NetworkMessage.cs index 39715d5..6053b3e 100644 --- a/mROA/Implementation/NetworkMessageHeader.cs +++ b/mROA/Implementation/NetworkMessage.cs @@ -3,9 +3,9 @@ using mROA.Abstract; namespace mROA.Implementation { - public class NetworkMessageHeader + public class NetworkMessage { - private bool Equals(NetworkMessageHeader other) + private bool Equals(NetworkMessage other) { return Id.Equals(other.Id) && MessageType == other.MessageType; } @@ -14,15 +14,15 @@ namespace mROA.Implementation { if (obj is null) return false; if (ReferenceEquals(this, obj)) return true; - return obj.GetType() == GetType() && Equals((NetworkMessageHeader)obj); + return obj.GetType() == GetType() && Equals((NetworkMessage)obj); } - public NetworkMessageHeader() + public NetworkMessage() { Data = Array.Empty(); } - public NetworkMessageHeader(IContextualSerializationToolKit serializationToolkit, + public NetworkMessage(IContextualSerializationToolKit serializationToolkit, INetworkMessage networkMessage, IEndPointContext? context) { MessageType = networkMessage.MessageType; @@ -40,5 +40,30 @@ namespace mROA.Implementation { return $" {Id}:{MessageType} [{Data.Length}]"; } + + public struct NetworkMessageMeta + { + public byte Type; + public Guid Id; + public ushort BodyLength; + + public NetworkMessageMeta(ReadOnlySpan metadata) + { + Type = metadata[0]; + Id = new Guid(metadata[1..17]); + BodyLength = BitConverter.ToUInt16(metadata[17..]); + } + + public NetworkMessage ToMessage(ReadOnlySpan memory) + { + var data = memory[19..][..BodyLength]; + return new NetworkMessage + { + Data = data.ToArray(), + Id = Id, + MessageType = (EMessageType)Type + }; + } + } } } \ No newline at end of file diff --git a/mROA/Implementation/RepresentationModule.cs b/mROA/Implementation/RepresentationModule.cs index ccb20b8..586cb7e 100644 --- a/mROA/Implementation/RepresentationModule.cs +++ b/mROA/Implementation/RepresentationModule.cs @@ -28,8 +28,8 @@ namespace mROA.Implementation public IEndPointContext Context => _interaction.Context; public async Task<(object? Deserialized, EMessageType MessageType)> GetSingle( - Predicate rule, IEndPointContext? context, - CancellationToken token = default, params Func[] converter) + Predicate rule, IEndPointContext? context, + CancellationToken token = default, params Func[] converter) { var writer = _interaction.ReceiveChanel.Writer; var reader = _interaction.ReceiveChanel.Reader; @@ -52,9 +52,9 @@ namespace mROA.Implementation } public async IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream( - Predicate rule, IEndPointContext? context, + Predicate rule, IEndPointContext? context, [EnumeratorCancellation] CancellationToken token = default, - params Func[] converter) + params Func[] converter) { var writer = _interaction.ReceiveChanel.Writer; await foreach (var message in _interaction.ReceiveChanel.Reader.ReadAllAsync(token)) @@ -83,7 +83,7 @@ namespace mROA.Implementation IEndPointContext? context) where T : notnull { var serialized = _serialization.Serialize(payload, context); - await _interaction.PostMessageAsync(new NetworkMessageHeader + await _interaction.PostMessageAsync(new NetworkMessage { Id = id, MessageType = eMessageType, Data = serialized }); } @@ -97,7 +97,7 @@ namespace mROA.Implementation IEndPointContext? context) where T : notnull { var serialized = _serialization.Serialize(payload, context); - await _interaction.PostMessageUntrustedAsync(new NetworkMessageHeader + await _interaction.PostMessageUntrustedAsync(new NetworkMessage { Id = id, MessageType = eMessageType, Data = serialized }); } } diff --git a/mROA/Implementation/RequestId.cs b/mROA/Implementation/RequestId.cs new file mode 100644 index 0000000..e4b49f7 --- /dev/null +++ b/mROA/Implementation/RequestId.cs @@ -0,0 +1,18 @@ +using System; + +namespace mROA.Implementation +{ + public struct RequestId + { + public ulong P0; + public ulong P1; + public RequestId Generate() + { + var guid = Guid.NewGuid(); + var bytes = guid.ToByteArray(); + var high = BitConverter.ToUInt64(bytes, 0); + var low = BitConverter.ToUInt64(bytes, 8); + return new RequestId{ P0 = high, P1 = low}; + } + } +} \ No newline at end of file