From 6da3415d5a5a51972a7b133c886ac9fa437bd5e3 Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Mon, 25 Nov 2024 13:28:04 +0100 Subject: [PATCH] taking shape --- src/CoreIpc.sln | 8 +- src/Playground/Program.cs | 43 ++- ...ath.CoreIpc.Extensions.Abstractions.csproj | 17 ++ src/UiPath.CoreIpc.Http/BidiHttpListener.cs | 285 ------------------ .../BidiHttpServerTransport.cs | 278 +++++++++++++++++ .../UiPath.CoreIpc.Http.csproj | 2 +- src/UiPath.CoreIpc/Client/ServiceClient.cs | 32 +- src/UiPath.CoreIpc/Config/ClientConfig.cs | 43 +-- ...erviceClientConfig.cs => IClientConfig.cs} | 6 +- src/UiPath.CoreIpc/Config/IClientState.cs | 9 + src/UiPath.CoreIpc/Config/IpcClient.cs | 50 ++- src/UiPath.CoreIpc/Config/IpcServer.cs | 26 +- src/UiPath.CoreIpc/Config/Peer.cs | 10 +- src/UiPath.CoreIpc/Config/ServerTransport.cs | 5 +- src/UiPath.CoreIpc/Helpers/Router.cs | 2 +- src/UiPath.CoreIpc/Server/EndpointSettings.cs | 2 +- src/UiPath.CoreIpc/Server/ServerConnection.cs | 12 +- ...{TcpTransport.cs => TcpClientTransport.cs} | 4 +- ...ansport.cs => WebSocketClientTransport.cs} | 4 +- src/UiPath.CoreIpc/UiPath.CoreIpc.csproj | 8 + src/UiPath.Ipc.Tests/ComputingTests.cs | 51 +--- .../ComputingTestsOverNamedPipes.cs | 2 +- .../Config/OverrideConfigAttribute.cs | 4 +- src/UiPath.Ipc.Tests/Helpers/IpcHelpers.cs | 28 +- src/UiPath.Ipc.Tests/NamedPipeSmokeTests.cs | 13 +- src/UiPath.Ipc.Tests/Program.cs | 4 +- src/UiPath.Ipc.Tests/RobotTests.cs | 23 +- .../RobotTestsOverNamedPipes.cs | 43 +-- src/UiPath.Ipc.Tests/SpyTestBase.cs | 20 ++ src/UiPath.Ipc.Tests/SyncOverAsyncTests.cs | 10 +- src/UiPath.Ipc.Tests/SystemTests.cs | 38 +-- .../SystemTestsOverNamedPipes.cs | 2 +- src/UiPath.Ipc.Tests/SystemTestsOverTcp.cs | 4 +- .../SystemTestsOverWebSockets.cs | 4 +- src/UiPath.Ipc.Tests/TestBase.cs | 104 ++++--- 35 files changed, 594 insertions(+), 602 deletions(-) create mode 100644 src/UiPath.CoreIpc.Extensions.Abstractions/UiPath.CoreIpc.Extensions.Abstractions.csproj delete mode 100644 src/UiPath.CoreIpc.Http/BidiHttpListener.cs create mode 100644 src/UiPath.CoreIpc.Http/BidiHttpServerTransport.cs rename src/UiPath.CoreIpc/Config/{IServiceClientConfig.cs => IClientConfig.cs} (58%) create mode 100644 src/UiPath.CoreIpc/Config/IClientState.cs rename src/UiPath.CoreIpc/Transport/Tcp/{TcpTransport.cs => TcpClientTransport.cs} (89%) rename src/UiPath.CoreIpc/Transport/WebSocket/{WebSocketTransport.cs => WebSocketClientTransport.cs} (84%) create mode 100644 src/UiPath.Ipc.Tests/SpyTestBase.cs diff --git a/src/CoreIpc.sln b/src/CoreIpc.sln index 12d4a07d..74779ed7 100644 --- a/src/CoreIpc.sln +++ b/src/CoreIpc.sln @@ -1,6 +1,6 @@  Microsoft Visual Studio Solution File, Format Version 12.00 -# 17 +# Visual Studio Version 17 VisualStudioVersion = 17.0.31919.166 MinimumVisualStudioVersion = 10.0.40219.1 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "UiPath.CoreIpc", "UiPath.CoreIpc\UiPath.CoreIpc.csproj", "{58200319-1F71-4E22-894D-7E69E0CD0B57}" @@ -17,6 +17,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "UiPath.CoreIpc.Http", "UiPa EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "UiPath.Ipc.Tests", "UiPath.Ipc.Tests\UiPath.Ipc.Tests.csproj", "{E238E183-92CF-48A6-890F-C422853D6656}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "UiPath.CoreIpc.Extensions.Abstractions", "UiPath.CoreIpc.Extensions.Abstractions\UiPath.CoreIpc.Extensions.Abstractions.csproj", "{F519AE2B-88A6-482E-A6E2-B525F71F566D}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -39,6 +41,10 @@ Global {E238E183-92CF-48A6-890F-C422853D6656}.Debug|Any CPU.Build.0 = Debug|Any CPU {E238E183-92CF-48A6-890F-C422853D6656}.Release|Any CPU.ActiveCfg = Release|Any CPU {E238E183-92CF-48A6-890F-C422853D6656}.Release|Any CPU.Build.0 = Release|Any CPU + {F519AE2B-88A6-482E-A6E2-B525F71F566D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {F519AE2B-88A6-482E-A6E2-B525F71F566D}.Debug|Any CPU.Build.0 = Debug|Any CPU + {F519AE2B-88A6-482E-A6E2-B525F71F566D}.Release|Any CPU.ActiveCfg = Release|Any CPU + {F519AE2B-88A6-482E-A6E2-B525F71F566D}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE diff --git a/src/Playground/Program.cs b/src/Playground/Program.cs index 613e8d5a..e373bd18 100644 --- a/src/Playground/Program.cs +++ b/src/Playground/Program.cs @@ -44,7 +44,7 @@ private static async Task Main(string[] args) typeof(Contracts.IServerOperations), // DEVINE new EndpointSettings(typeof(Contracts.IServerOperations)) // ASTALALT { - BeforeCall = async (callInfo, _) => + BeforeIncommingCall = async (callInfo, _) => { Console.WriteLine($"Server: {callInfo.Method.Name}"); } @@ -76,36 +76,30 @@ private static async Task Main(string[] args) var c1 = new IpcClient() { - Config = new() + Callbacks = new() { - Callbacks = new() - { - typeof(Contracts.IClientOperations), - { typeof(Contracts.IClientOperations2), new Impl.Client2() }, - }, - ServiceProvider = clientSP, - Scheduler = clientScheduler, + typeof(Contracts.IClientOperations), + { typeof(Contracts.IClientOperations2), new Impl.Client2() }, }, + ServiceProvider = clientSP, + Scheduler = clientScheduler, Transport = new NamedPipeClientTransport() { PipeName = Contracts.PipeName, ServerName = ".", AllowImpersonation = false, - }, + } }; var c2 = new IpcClient() { - Config = new() + ServiceProvider = clientSP, + Callbacks = new() { - ServiceProvider = clientSP, - Callbacks = new() - { - typeof(Contracts.IClientOperations), - { typeof(Contracts.IClientOperations2), new Impl.Client2() }, - }, - Scheduler = clientScheduler, + typeof(Contracts.IClientOperations), + { typeof(Contracts.IClientOperations2), new Impl.Client2() }, }, + Scheduler = clientScheduler, Transport = new NamedPipeClientTransport() { PipeName = Contracts.PipeName, @@ -116,16 +110,13 @@ private static async Task Main(string[] args) var proxy1 = new IpcClient() { - Config = new() + ServiceProvider = clientSP, + Callbacks = new() { - ServiceProvider = clientSP, - Callbacks = new() - { - typeof(Contracts.IClientOperations), - { typeof(Contracts.IClientOperations2), new Impl.Client2() }, - }, - Scheduler = clientScheduler, + typeof(Contracts.IClientOperations), + { typeof(Contracts.IClientOperations2), new Impl.Client2() }, }, + Scheduler = clientScheduler, Transport = new NamedPipeClientTransport() { PipeName = Contracts.PipeName, diff --git a/src/UiPath.CoreIpc.Extensions.Abstractions/UiPath.CoreIpc.Extensions.Abstractions.csproj b/src/UiPath.CoreIpc.Extensions.Abstractions/UiPath.CoreIpc.Extensions.Abstractions.csproj new file mode 100644 index 00000000..2186c679 --- /dev/null +++ b/src/UiPath.CoreIpc.Extensions.Abstractions/UiPath.CoreIpc.Extensions.Abstractions.csproj @@ -0,0 +1,17 @@ + + + + net6.0;net461;net6.0-windows + enable + enable + preview + true + enable + true + + + + + + + diff --git a/src/UiPath.CoreIpc.Http/BidiHttpListener.cs b/src/UiPath.CoreIpc.Http/BidiHttpListener.cs deleted file mode 100644 index 06076a21..00000000 --- a/src/UiPath.CoreIpc.Http/BidiHttpListener.cs +++ /dev/null @@ -1,285 +0,0 @@ -using Nito.AsyncEx; -using System.Buffers; -using System.Collections.Concurrent; -using System.Diagnostics.CodeAnalysis; -using System.IO.Pipelines; -using System.Net; -using System.Net.Http; -using System.Threading.Channels; - -namespace UiPath.Ipc.Http; - -using static Constants; -using IBidiHttpListenerConfig = IListenerConfig; - -public sealed partial record BidiHttpListener : ServerTransport, IBidiHttpListenerConfig -{ - public required Uri Uri { get; init; } - - BidiHttpListenerState IBidiHttpListenerConfig.CreateListenerState(IpcServer server) - => new(server, this); - - BidiHttpServerConnectionState IBidiHttpListenerConfig.CreateConnectionState(IpcServer server, BidiHttpListenerState listenerState) - => new(server, listenerState); - - async ValueTask IBidiHttpListenerConfig.AwaitConnection(BidiHttpListenerState listenerState, BidiHttpServerConnectionState connectionState, CancellationToken ct) - { - await connectionState.WaitForConnection(ct); - return connectionState; - } - - public IEnumerable Validate() - { - throw new NotImplementedException(); - } -} - -internal sealed class BidiHttpListenerState : IAsyncDisposable -{ - private readonly IpcServer _ipcServer; - private readonly CancellationTokenSource _cts = new(); - private readonly HttpListener _httpListener; - private readonly Task _processing; - private readonly Lazy _disposing; - - private readonly ConcurrentDictionary> _connections = new(); - private readonly Channel<(Guid connectionId, Uri reverseUri)> _newConnections = Channel.CreateUnbounded<(Guid connectionId, Uri reverseUri)>(); - - public ChannelReader<(Guid connectionId, Uri reverseUri)> NewConnections => _newConnections.Reader; - public ChannelReader GetConnectionChannel(Guid connectionId) => _connections[connectionId]; - - public BidiHttpListenerState(IpcServer ipcServer, BidiHttpListener listener) - { - _ipcServer = ipcServer; - _httpListener = new HttpListener() - { - Prefixes = - { - listener.Uri.ToString() - } - }; - _processing = ProcessContexts(); - _disposing = new(DisposeCore); - } - - public ValueTask DisposeAsync() => new(_disposing.Value); - - private async Task DisposeCore() - { - _cts.Cancel(); - try - { - await _processing; - } - catch (OperationCanceledException ex) when (ex.CancellationToken == _cts.Token) - { - } - - foreach (var pair in _connections) - { - pair.Value.Writer.Complete(); - } - _cts.Dispose(); - } - - private async Task ProcessContexts() - { - await foreach (var (context, connectionId, reverseUri) in AwaitContexts()) - { - var connectionChannel = _connections.GetOrAdd(connectionId, _ => - { - _newConnections.Writer.TryWrite((connectionId, reverseUri)); - return Channel.CreateUnbounded(); - }); - - await connectionChannel.Writer.WriteAsync(context, _cts.Token); - } - - async IAsyncEnumerable<(HttpListenerContext context, Guid connectionId, Uri reverseUri)> AwaitContexts() - { - while (!_cts.Token.IsCancellationRequested) - { - var context = await _httpListener.GetContextAsync(); - - if (!TryAcceptContext(context, out var connectionId, out var reverseUri)) - { - context.Response.StatusCode = 400; - context.Response.Close(); - continue; - } - - yield return (context, connectionId, reverseUri); - } - } - - bool TryAcceptContext(HttpListenerContext context, out Guid connectionId, [NotNullWhen(returnValue: true)] out Uri? reverseUri) - { - if (!Guid.TryParse(context.Request.Headers[ConnectionIdHeader], out connectionId) || - !Uri.TryCreate(context.Request.Headers[ReverseUriHeader], UriKind.Absolute, out reverseUri)) - { - connectionId = Guid.Empty; - reverseUri = null; - return false; - } - - return true; - } - } -} - -internal sealed class BidiHttpServerConnectionState : Stream, IAsyncDisposable -{ - private readonly Pipe _pipe = new(); - - private readonly IpcServer _server; - private readonly BidiHttpListenerState _listenerState; - - private readonly CancellationTokenSource _cts = new(); - private readonly AsyncLock _lock = new(); - private (Guid connectionId, Uri reverseUri)? _connection = null; - private HttpClient? _client; - private Task? _processing = null; - private readonly Lazy _disposing; - - public BidiHttpServerConnectionState(IpcServer server, BidiHttpListenerState listenerState) - { - _server = server; - _listenerState = listenerState; - _disposing = new(DisposeCore); - } - - public -#if !NET461 - override -#endif - ValueTask DisposeAsync() => new(_disposing.Value); - - private async Task DisposeCore() - { - _cts.Cancel(); - - _client?.Dispose(); - - try - { - await (_processing ?? Task.CompletedTask); - } - catch (OperationCanceledException ex) when (ex.CancellationToken == _cts.Token) - { - // ignored - } - - _cts.Dispose(); - } - - public async Task WaitForConnection(CancellationToken ct) - { - using (await _lock.LockAsync(ct)) - { - if (_connection is not null) - { - throw new InvalidOperationException(); - } - - _connection = await _listenerState.NewConnections.ReadAsync(ct); - - _client = new() - { - BaseAddress = _connection.Value.reverseUri, - DefaultRequestHeaders = - { - { ConnectionIdHeader, _connection.Value.connectionId.ToString() } - } - }; - - _processing = ProcessContexts(_cts.Token); - } - } - - private async Task ProcessContexts(CancellationToken ct) - { - var reader = _listenerState.GetConnectionChannel(_connection!.Value.connectionId); - - while (await reader.WaitToReadAsync(ct)) - { - if (!reader.TryRead(out var context)) - { - continue; - } - await ProcessContext(context); - } - - async Task ProcessContext(HttpListenerContext context) - { - try - { - while (true) - { - var memory = _pipe.Writer.GetMemory(); - var cbRead = await context.Request.InputStream.ReadAsync(memory, ct); - if (cbRead is 0) - { - break; - } - _pipe.Writer.Advance(cbRead); - var flushResult = await _pipe.Writer.FlushAsync(ct); - if (flushResult.IsCompleted) - { - break; - } - } - } - finally - { - context.Response.StatusCode = 200; - context.Response.Close(); - } - } - } - - public override bool CanRead => true; - public override bool CanSeek => false; - public override bool CanWrite => true; - - public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken ct) - { - var memory = new Memory(buffer, offset, count); - var readResult = await _pipe.Reader.ReadAsync(ct); - - var take = (int)Math.Min(readResult.Buffer.Length, memory.Length); - - readResult.Buffer.Slice(start: 0, length: take).CopyTo(memory.Span); - _pipe.Reader.AdvanceTo(readResult.Buffer.GetPosition(take)); - - return take; - } - - public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken ct) - { - var memory = new ReadOnlyMemory(buffer, offset, count); - if (_client is null) - { - throw new InvalidOperationException(); - } - - HttpContent content = -#if NET461 - new ByteArrayContent(memory.ToArray()); -#else - new ReadOnlyMemoryContent(memory); -#endif - - await _client.PostAsync(requestUri: "", content, ct); - } - - public override Task FlushAsync(CancellationToken cancellationToken) - => Task.CompletedTask; - - public override void Flush() => throw new NotImplementedException(); - public override long Seek(long offset, SeekOrigin origin) => throw new NotImplementedException(); - public override void SetLength(long value) => throw new NotImplementedException(); - public override int Read(byte[] buffer, int offset, int count) => throw new NotImplementedException(); - public override void Write(byte[] buffer, int offset, int count) => throw new NotImplementedException(); - public override long Length => throw new NotImplementedException(); - public override long Position { get => throw new NotImplementedException(); set => throw new NotImplementedException(); } -} diff --git a/src/UiPath.CoreIpc.Http/BidiHttpServerTransport.cs b/src/UiPath.CoreIpc.Http/BidiHttpServerTransport.cs new file mode 100644 index 00000000..4e43ce58 --- /dev/null +++ b/src/UiPath.CoreIpc.Http/BidiHttpServerTransport.cs @@ -0,0 +1,278 @@ +using Nito.AsyncEx; +using System.Buffers; +using System.Collections.Concurrent; +using System.Diagnostics.CodeAnalysis; +using System.IO.Pipelines; +using System.Net; +using System.Net.Http; +using System.Threading.Channels; + +namespace UiPath.Ipc.Http; + +using static Constants; + +public sealed partial class BidiHttpServerTransport : ServerTransport +{ + public required Uri Uri { get; init; } + + protected override IServerState CreateServerState() + => new BidiHttpServerState(this); + + protected override IEnumerable ValidateCore() => []; + + private sealed class BidiHttpServerState : IServerState + { + private readonly CancellationTokenSource _cts = new(); + private readonly HttpListener _httpListener; + private readonly Task _processing; + private readonly Lazy _disposing; + + private readonly ConcurrentDictionary> _connections = new(); + private readonly Channel<(Guid connectionId, Uri reverseUri)> _newConnections = Channel.CreateUnbounded<(Guid connectionId, Uri reverseUri)>(); + + public ChannelReader<(Guid connectionId, Uri reverseUri)> NewConnections => _newConnections.Reader; + public ChannelReader GetConnectionChannel(Guid connectionId) => _connections[connectionId]; + + public BidiHttpServerState(BidiHttpServerTransport transport) + { + _httpListener = new HttpListener() + { + Prefixes = + { + transport.Uri.ToString() + } + }; + _processing = ProcessContexts(); + _disposing = new(DisposeCore); + } + + public ValueTask DisposeAsync() => new(_disposing.Value); + + private async Task DisposeCore() + { + _cts.Cancel(); + try + { + await _processing; + } + catch (OperationCanceledException ex) when (ex.CancellationToken == _cts.Token) + { + } + + foreach (var pair in _connections) + { + pair.Value.Writer.Complete(); + } + _cts.Dispose(); + } + + private async Task ProcessContexts() + { + await foreach (var (context, connectionId, reverseUri) in AwaitContexts()) + { + var connectionChannel = _connections.GetOrAdd(connectionId, _ => + { + _newConnections.Writer.TryWrite((connectionId, reverseUri)); + return Channel.CreateUnbounded(); + }); + + await connectionChannel.Writer.WriteAsync(context, _cts.Token); + } + + async IAsyncEnumerable<(HttpListenerContext context, Guid connectionId, Uri reverseUri)> AwaitContexts() + { + while (!_cts.Token.IsCancellationRequested) + { + var context = await _httpListener.GetContextAsync(); + + if (!TryAcceptContext(context, out var connectionId, out var reverseUri)) + { + context.Response.StatusCode = 400; + context.Response.Close(); + continue; + } + + yield return (context, connectionId, reverseUri); + } + } + + bool TryAcceptContext(HttpListenerContext context, out Guid connectionId, [NotNullWhen(returnValue: true)] out Uri? reverseUri) + { + if (!Guid.TryParse(context.Request.Headers[ConnectionIdHeader], out connectionId) || + !Uri.TryCreate(context.Request.Headers[ReverseUriHeader], UriKind.Absolute, out reverseUri)) + { + connectionId = Guid.Empty; + reverseUri = null; + return false; + } + + return true; + } + } + + IServerConnectionSlot IServerState.CreateConnectionSlot() => new BidiHttpServerConnectionSlot(this); + } + + private sealed class BidiHttpServerConnectionSlot : Stream, IServerConnectionSlot, IAsyncDisposable + { + private readonly Pipe _pipe = new(); + + private readonly BidiHttpServerState _listenerState; + + private readonly CancellationTokenSource _cts = new(); + private readonly AsyncLock _lock = new(); + private (Guid connectionId, Uri reverseUri)? _connection = null; + private HttpClient? _client; + private Task? _processing = null; + private readonly Lazy _disposing; + + public BidiHttpServerConnectionSlot(BidiHttpServerState serverState) + { + _listenerState = serverState; + _disposing = new(DisposeCore); + } + + public +#if !NET461 + override +#endif + ValueTask DisposeAsync() => new(_disposing.Value); + + private async Task DisposeCore() + { + _cts.Cancel(); + + _client?.Dispose(); + + try + { + await (_processing ?? Task.CompletedTask); + } + catch (OperationCanceledException ex) when (ex.CancellationToken == _cts.Token) + { + // ignored + } + + _cts.Dispose(); + } + + public async Task WaitForConnection(CancellationToken ct) + { + using (await _lock.LockAsync(ct)) + { + if (_connection is not null) + { + throw new InvalidOperationException(); + } + + _connection = await _listenerState.NewConnections.ReadAsync(ct); + + _client = new() + { + BaseAddress = _connection.Value.reverseUri, + DefaultRequestHeaders = + { + { ConnectionIdHeader, _connection.Value.connectionId.ToString() } + } + }; + + _processing = ProcessContexts(_cts.Token); + } + } + + private async Task ProcessContexts(CancellationToken ct) + { + var reader = _listenerState.GetConnectionChannel(_connection!.Value.connectionId); + + while (await reader.WaitToReadAsync(ct)) + { + if (!reader.TryRead(out var context)) + { + continue; + } + await ProcessContext(context); + } + + async Task ProcessContext(HttpListenerContext context) + { + try + { + while (true) + { + var memory = _pipe.Writer.GetMemory(); + var cbRead = await context.Request.InputStream.ReadAsync(memory, ct); + if (cbRead is 0) + { + break; + } + _pipe.Writer.Advance(cbRead); + var flushResult = await _pipe.Writer.FlushAsync(ct); + if (flushResult.IsCompleted) + { + break; + } + } + } + finally + { + context.Response.StatusCode = 200; + context.Response.Close(); + } + } + } + + ValueTask IServerConnectionSlot.AwaitConnection(CancellationToken ct) + { + throw new NotImplementedException(); + } + + + + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => true; + + public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken ct) + { + var memory = new Memory(buffer, offset, count); + var readResult = await _pipe.Reader.ReadAsync(ct); + + var take = (int)Math.Min(readResult.Buffer.Length, memory.Length); + + readResult.Buffer.Slice(start: 0, length: take).CopyTo(memory.Span); + _pipe.Reader.AdvanceTo(readResult.Buffer.GetPosition(take)); + + return take; + } + + public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken ct) + { + var memory = new ReadOnlyMemory(buffer, offset, count); + if (_client is null) + { + throw new InvalidOperationException(); + } + + HttpContent content = +#if NET461 + new ByteArrayContent(memory.ToArray()); +#else + new ReadOnlyMemoryContent(memory); +#endif + + await _client.PostAsync(requestUri: "", content, ct); + } + + public override Task FlushAsync(CancellationToken cancellationToken) + => Task.CompletedTask; + + public override void Flush() => throw new NotImplementedException(); + public override long Seek(long offset, SeekOrigin origin) => throw new NotImplementedException(); + public override void SetLength(long value) => throw new NotImplementedException(); + public override int Read(byte[] buffer, int offset, int count) => throw new NotImplementedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotImplementedException(); + + public override long Length => throw new NotImplementedException(); + public override long Position { get => throw new NotImplementedException(); set => throw new NotImplementedException(); } + } +} diff --git a/src/UiPath.CoreIpc.Http/UiPath.CoreIpc.Http.csproj b/src/UiPath.CoreIpc.Http/UiPath.CoreIpc.Http.csproj index 524ffbc1..83f0f21e 100644 --- a/src/UiPath.CoreIpc.Http/UiPath.CoreIpc.Http.csproj +++ b/src/UiPath.CoreIpc.Http/UiPath.CoreIpc.Http.csproj @@ -21,7 +21,7 @@ - + diff --git a/src/UiPath.CoreIpc/Client/ServiceClient.cs b/src/UiPath.CoreIpc/Client/ServiceClient.cs index 4c0edd2f..95ef34c3 100644 --- a/src/UiPath.CoreIpc/Client/ServiceClient.cs +++ b/src/UiPath.CoreIpc/Client/ServiceClient.cs @@ -11,7 +11,7 @@ private static IpcProxy CreateProxy(ServiceClient serviceClient) where T : cl return proxy; } - protected abstract IServiceClientConfig Config { get; } + protected abstract IClientConfig Config { get; } public abstract Stream? Network { get; } public event EventHandler? ConnectionClosed; @@ -66,11 +66,7 @@ async Task Invoke() var (connection, newConnection) = await EnsureConnection(ct); - if (Config.BeforeCall is not null) - { - var callInfo = new CallInfo(newConnection, method, args); - await Config.BeforeCall(callInfo, ct); - } + await (Config.BeforeOutgoingCall?.Invoke(new CallInfo(newConnection, method, args), ct) ?? Task.CompletedTask); var requestId = connection.NewRequestId(); var request = new Request(_interfaceType.Name, requestId, methodName, serializedArguments, messageTimeout.TotalSeconds) @@ -78,18 +74,18 @@ async Task Invoke() UploadStream = uploadStream }; - Config.Logger?.ServiceClient_Calling(methodName, requestId, Config.DebugName); + Config.Logger?.ServiceClient_Calling(methodName, requestId, Config.GetComputedDebugName()); Response response; try { response = await connection.RemoteCall(request, ct); // returns user errors instead of throwing them (could throw for system bugs) - Config.Logger?.ServiceClient_CalledSuccessfully(request.MethodName, requestId, Config.DebugName); + Config.Logger?.ServiceClient_CalledSuccessfully(request.MethodName, requestId, Config.GetComputedDebugName()); } catch (Exception ex) { - Config.Logger?.ServiceClient_FailedToCall(request.MethodName, requestId, Config.DebugName, ex); + Config.Logger?.ServiceClient_FailedToCall(request.MethodName, requestId, Config.GetComputedDebugName(), ex); throw; } @@ -133,7 +129,7 @@ string[] SerializeArguments() public abstract void Dispose(); - public override string ToString() => Config.DebugName; + public override string ToString() => Config.GetComputedDebugName(); #region Generic adapter cache private static readonly MethodInfo GenericDefOf_Invoke = ((Func>)Invoke).Method.GetGenericMethodDefinition(); @@ -227,9 +223,9 @@ public override async ValueTask CloseConnection() var network = await Connect(ct); - LatestConnection = new Connection(network, Config.DebugName, Config.Logger); - var router = new Router(_client.Config.CreateCallbackRouterConfig(), _client.Config.ServiceProvider); - _latestServer = new Server(router, _client.Config.RequestTimeout, LatestConnection); + LatestConnection = new Connection(network, Config.GetComputedDebugName(), Config.Logger); + var router = new Router(_client.CreateCallbackRouterConfig(), _client.ServiceProvider); + _latestServer = new Server(router, _client.RequestTimeout, LatestConnection); _ = Pal(); return (LatestConnection, newlyConnected: true); @@ -242,7 +238,7 @@ async Task Pal() } catch (Exception ex) { - Config.Logger.LogException(ex, Config.DebugName); + Config.Logger.LogException(ex, Config.GetComputedDebugName()); } } } @@ -260,17 +256,17 @@ private async Task Connect(CancellationToken ct) return network; } - protected override IServiceClientConfig Config => _client.Config; + protected override IClientConfig Config => _client; } internal sealed class ServiceClientForCallback : ServiceClient where TInterface : class { private readonly Connection _connection; - private readonly IServiceClientConfig _config; + private readonly IClientConfig _config; public override Stream? Network => _connection.Network; - public ServiceClientForCallback(Connection connection, IServiceClientConfig config) : base(typeof(TInterface)) + public ServiceClientForCallback(Connection connection, IClientConfig config) : base(typeof(TInterface)) { _connection = connection; _config = config; @@ -286,5 +282,5 @@ public override void Dispose() protected override Task<(Connection connection, bool newlyConnected)> EnsureConnection(CancellationToken ct) => Task.FromResult((_connection, newlyConnected: false)); - protected override IServiceClientConfig Config => _config; + protected override IClientConfig Config => _config; } diff --git a/src/UiPath.CoreIpc/Config/ClientConfig.cs b/src/UiPath.CoreIpc/Config/ClientConfig.cs index c6ac4bdb..6c05c192 100644 --- a/src/UiPath.CoreIpc/Config/ClientConfig.cs +++ b/src/UiPath.CoreIpc/Config/ClientConfig.cs @@ -1,18 +1,9 @@ -using System.ComponentModel; - -namespace UiPath.Ipc; +namespace UiPath.Ipc; public sealed class ClientConfig : Peer, IServiceClientConfig { - public EndpointCollection? Callbacks { get; init; } - - public ILogger? Logger { get; init; } - public BeforeConnectHandler? BeforeConnect { get; init; } public BeforeCallHandler? BeforeCall { get; init; } - [EditorBrowsable(EditorBrowsableState.Never)] - public string DebugName { get; set; } = null!; - internal void Validate() { var haveDeferredInjectedCallbacks = Callbacks?.Any(x => x.Service.MaybeGetServiceProvider() is null && x.Service.MaybeGetInstance() is null) ?? false; @@ -22,36 +13,4 @@ internal void Validate() throw new InvalidOperationException("ServiceProvider is required when you register injectable callbacks. Consider registering a callback instance."); } } - - internal ILogger? GetLogger(string name) - { - if (Logger is not null) - { - return Logger; - } - - if (ServiceProvider?.GetService() is not { } loggerFactory) - { - return null; - } - - return loggerFactory.CreateLogger(name); - } - - internal override RouterConfig CreateCallbackRouterConfig() - => RouterConfig.From( - Callbacks.OrDefault(), - endpoint => endpoint with - { - BeforeCall = null, // callbacks don't support BeforeCall - Scheduler = endpoint.Scheduler ?? Scheduler - }); } - -public interface IClientState : IDisposable -{ - Stream? Network { get; } - - bool IsConnected(); - ValueTask Connect(IpcClient client, CancellationToken ct); -} \ No newline at end of file diff --git a/src/UiPath.CoreIpc/Config/IServiceClientConfig.cs b/src/UiPath.CoreIpc/Config/IClientConfig.cs similarity index 58% rename from src/UiPath.CoreIpc/Config/IServiceClientConfig.cs rename to src/UiPath.CoreIpc/Config/IClientConfig.cs index 19ed49fd..9e4741ab 100644 --- a/src/UiPath.CoreIpc/Config/IServiceClientConfig.cs +++ b/src/UiPath.CoreIpc/Config/IClientConfig.cs @@ -1,11 +1,11 @@ namespace UiPath.Ipc; // Maybe decommission -internal interface IServiceClientConfig +internal interface IClientConfig { TimeSpan RequestTimeout { get; } BeforeConnectHandler? BeforeConnect { get; } - BeforeCallHandler? BeforeCall { get; } + BeforeCallHandler? BeforeOutgoingCall { get; } ILogger? Logger { get; } - string DebugName { get; } + string GetComputedDebugName(); } diff --git a/src/UiPath.CoreIpc/Config/IClientState.cs b/src/UiPath.CoreIpc/Config/IClientState.cs new file mode 100644 index 00000000..74ff640c --- /dev/null +++ b/src/UiPath.CoreIpc/Config/IClientState.cs @@ -0,0 +1,9 @@ +namespace UiPath.Ipc; + +public interface IClientState : IDisposable +{ + Stream? Network { get; } + + bool IsConnected(); + ValueTask Connect(IpcClient client, CancellationToken ct); +} diff --git a/src/UiPath.CoreIpc/Config/IpcClient.cs b/src/UiPath.CoreIpc/Config/IpcClient.cs index 24104b16..7bed1c05 100644 --- a/src/UiPath.CoreIpc/Config/IpcClient.cs +++ b/src/UiPath.CoreIpc/Config/IpcClient.cs @@ -1,10 +1,22 @@ -namespace UiPath.Ipc; +using System.ComponentModel; -public sealed class IpcClient : Peer +namespace UiPath.Ipc; + +public sealed class IpcClient : Peer, IClientConfig { - public required ClientConfig Config { get; init; } + public EndpointCollection? Callbacks { get; set; } + + public ILogger? Logger { get; init; } + public BeforeConnectHandler? BeforeConnect { get; set; } + public BeforeCallHandler? BeforeOutgoingCall { get; set; } + + [EditorBrowsable(EditorBrowsableState.Never)] + public string DebugName { get; set; } = null!; + public required ClientTransport Transport { get; init; } + string IClientConfig.GetComputedDebugName() => DebugName ?? Transport.ToString(); + private readonly ConcurrentDictionary _clients = new(); private ServiceClient GetServiceClient(Type proxyType) { @@ -16,18 +28,42 @@ private ServiceClient GetServiceClient(Type proxyType) internal void Validate() { - if (Config is null) + var haveDeferredInjectedCallbacks = Callbacks?.Any(x => x.Service.MaybeGetServiceProvider() is null && x.Service.MaybeGetInstance() is null) ?? false; + + if (haveDeferredInjectedCallbacks && ServiceProvider is null) { - throw new InvalidOperationException($"{Config} is required."); + throw new InvalidOperationException("ServiceProvider is required when you register injectable callbacks. Consider registering a callback instance."); } + if (Transport is null) { throw new InvalidOperationException($"{Transport} is required."); } - Config.Validate(); Transport.Validate(); + } - Config.DebugName ??= Transport.ToString(); + internal ILogger? GetLogger(string name) + { + if (Logger is not null) + { + return Logger; + } + + if (ServiceProvider?.GetService() is not { } loggerFactory) + { + return null; + } + + return loggerFactory.CreateLogger(name); } + + internal RouterConfig CreateCallbackRouterConfig() + => RouterConfig.From( + Callbacks.OrDefault(), + endpoint => endpoint with + { + BeforeIncommingCall = null, // callbacks don't support BeforeCall + Scheduler = endpoint.Scheduler ?? Scheduler + }); } diff --git a/src/UiPath.CoreIpc/Config/IpcServer.cs b/src/UiPath.CoreIpc/Config/IpcServer.cs index d0f58020..e9c3d14d 100644 --- a/src/UiPath.CoreIpc/Config/IpcServer.cs +++ b/src/UiPath.CoreIpc/Config/IpcServer.cs @@ -14,8 +14,16 @@ public sealed class IpcServer : Peer, IAsyncDisposable private bool _disposeStarted; private Accepter? _accepter; + private Lazy _dispose; - public async ValueTask DisposeAsync() + public IpcServer() + { + _dispose = new(DisposeCore); + } + + public ValueTask DisposeAsync() => new(_dispose.Value); + + private async Task DisposeCore() { Accepter? accepter = null; lock (_lock) @@ -77,6 +85,13 @@ private void OnNewConnectionError(Exception ex) _stopped.TrySetException(ex); } + internal RouterConfig CreateRouterConfig(IpcServer server) => RouterConfig.From( + server.Endpoints, + endpoint => endpoint with + { + Scheduler = endpoint.Scheduler ?? server.Scheduler + }); + private sealed class ObserverAdapter : IObserver { public required Action OnNext { get; init; } @@ -95,6 +110,7 @@ private sealed class Accepter : IAsyncDisposable private readonly Task _running; private readonly IObserver _newConnection; private readonly TaskCompletionSource _tcsStartedAccepting = new(); + private readonly Lazy _dispose; public Task StartedAccepting => _tcsStartedAccepting.Task; @@ -103,15 +119,19 @@ public Accepter(ServerTransport transport, IObserver connected) _serverState = transport.CreateServerState(); _newConnection = connected; _running = RunOnThreadPool(LoopAccept, parallelCount: transport.ConcurrentAccepts, _cts.Token); + _dispose = new(DisposeCore); } - public async ValueTask DisposeAsync() - { + public ValueTask DisposeAsync() => new(_dispose.Value); + + private async Task DisposeCore() + { _cts.Cancel(); await _running; _cts.Dispose(); } + private async Task LoopAccept(CancellationToken ct) { try diff --git a/src/UiPath.CoreIpc/Config/Peer.cs b/src/UiPath.CoreIpc/Config/Peer.cs index c7e9ba8a..4d4f7121 100644 --- a/src/UiPath.CoreIpc/Config/Peer.cs +++ b/src/UiPath.CoreIpc/Config/Peer.cs @@ -2,11 +2,7 @@ public abstract class Peer { - public TimeSpan RequestTimeout { get; init; } = Timeout.InfiniteTimeSpan; - public IServiceProvider? ServiceProvider { get; init; } - public TaskScheduler? Scheduler { get; init; } - - internal virtual RouterConfig CreateRouterConfig(IpcServer server) => throw new NotSupportedException(); - - internal virtual RouterConfig CreateCallbackRouterConfig() => throw new NotSupportedException(); + public TimeSpan RequestTimeout { get; set; } = Timeout.InfiniteTimeSpan; + public IServiceProvider? ServiceProvider { get; set; } + public TaskScheduler? Scheduler { get; set; } } diff --git a/src/UiPath.CoreIpc/Config/ServerTransport.cs b/src/UiPath.CoreIpc/Config/ServerTransport.cs index ed49685e..c63bc155 100644 --- a/src/UiPath.CoreIpc/Config/ServerTransport.cs +++ b/src/UiPath.CoreIpc/Config/ServerTransport.cs @@ -5,8 +5,9 @@ namespace UiPath.Ipc; public abstract class ServerTransport { - public int ConcurrentAccepts { get; init; } = 5; - public byte MaxReceivedMessageSizeInMegabytes { get; init; } = 2; + public int ConcurrentAccepts { get; set; } = 5; + public byte MaxReceivedMessageSizeInMegabytes { get; set; } = 2; + public X509Certificate? Certificate { get; init; } internal int MaxMessageSize => MaxReceivedMessageSizeInMegabytes * 1024 * 1024; diff --git a/src/UiPath.CoreIpc/Helpers/Router.cs b/src/UiPath.CoreIpc/Helpers/Router.cs index 45252243..26684819 100644 --- a/src/UiPath.CoreIpc/Helpers/Router.cs +++ b/src/UiPath.CoreIpc/Helpers/Router.cs @@ -131,7 +131,7 @@ public static Route From(IServiceProvider? serviceProvider, EndpointSettings end => new Route() { Service = endpointSettings.Service.WithProvider(serviceProvider), - BeforeCall = endpointSettings.BeforeCall, + BeforeCall = endpointSettings.BeforeIncommingCall, Scheduler = endpointSettings.Scheduler.OrDefault(), LoggerFactory = serviceProvider.MaybeCreateServiceFactory(), }; diff --git a/src/UiPath.CoreIpc/Server/EndpointSettings.cs b/src/UiPath.CoreIpc/Server/EndpointSettings.cs index 26a39d6b..3b00df73 100644 --- a/src/UiPath.CoreIpc/Server/EndpointSettings.cs +++ b/src/UiPath.CoreIpc/Server/EndpointSettings.cs @@ -5,7 +5,7 @@ public record EndpointSettings { public TaskScheduler? Scheduler { get; set; } - public BeforeCallHandler? BeforeCall { get; set; } + public BeforeCallHandler? BeforeIncommingCall { get; set; } public Type ContractType => Service.Type; public object? ServiceInstance => Service.MaybeGetInstance(); public IServiceProvider? ServiceProvider => Service.MaybeGetServiceProvider(); diff --git a/src/UiPath.CoreIpc/Server/ServerConnection.cs b/src/UiPath.CoreIpc/Server/ServerConnection.cs index cde20546..276985b1 100644 --- a/src/UiPath.CoreIpc/Server/ServerConnection.cs +++ b/src/UiPath.CoreIpc/Server/ServerConnection.cs @@ -2,7 +2,7 @@ namespace UiPath.Ipc; -internal sealed class ServerConnection : IClient, IDisposable, IServiceClientConfig +internal sealed class ServerConnection : IClient, IDisposable, IClientConfig { public static void CreateAndListen(IpcServer server, Stream network, CancellationToken ct) { @@ -71,10 +71,10 @@ void IClient.Impersonate(Action action) } #region IServiceClientConfig - TimeSpan IServiceClientConfig.RequestTimeout => _ipcServer.RequestTimeout; - BeforeConnectHandler? IServiceClientConfig.BeforeConnect => null; - BeforeCallHandler? IServiceClientConfig.BeforeCall => null; - ILogger? IServiceClientConfig.Logger => _logger; - string IServiceClientConfig.DebugName => _debugName; + TimeSpan IClientConfig.RequestTimeout => _ipcServer.RequestTimeout; + BeforeConnectHandler? IClientConfig.BeforeConnect => null; + BeforeCallHandler? IClientConfig.BeforeOutgoingCall => null; + ILogger? IClientConfig.Logger => _logger; + string IClientConfig.GetComputedDebugName() => _debugName; #endregion } diff --git a/src/UiPath.CoreIpc/Transport/Tcp/TcpTransport.cs b/src/UiPath.CoreIpc/Transport/Tcp/TcpClientTransport.cs similarity index 89% rename from src/UiPath.CoreIpc/Transport/Tcp/TcpTransport.cs rename to src/UiPath.CoreIpc/Transport/Tcp/TcpClientTransport.cs index c114f4f6..dd7cf5a3 100644 --- a/src/UiPath.CoreIpc/Transport/Tcp/TcpTransport.cs +++ b/src/UiPath.CoreIpc/Transport/Tcp/TcpClientTransport.cs @@ -2,7 +2,7 @@ namespace UiPath.Ipc.Transport.Tcp; -public sealed record TcpTransport : ClientTransport +public sealed record TcpClientTransport : ClientTransport { public required IPEndPoint EndPoint { get; init; } @@ -32,7 +32,7 @@ public bool IsConnected() public async ValueTask Connect(IpcClient client, CancellationToken ct) { - var transport = client.Transport as TcpTransport ?? throw new InvalidOperationException(); + var transport = client.Transport as TcpClientTransport ?? throw new InvalidOperationException(); _tcpClient = new System.Net.Sockets.TcpClient(); #if NET461 diff --git a/src/UiPath.CoreIpc/Transport/WebSocket/WebSocketTransport.cs b/src/UiPath.CoreIpc/Transport/WebSocket/WebSocketClientTransport.cs similarity index 84% rename from src/UiPath.CoreIpc/Transport/WebSocket/WebSocketTransport.cs rename to src/UiPath.CoreIpc/Transport/WebSocket/WebSocketClientTransport.cs index 6558b2c7..e5b36472 100644 --- a/src/UiPath.CoreIpc/Transport/WebSocket/WebSocketTransport.cs +++ b/src/UiPath.CoreIpc/Transport/WebSocket/WebSocketClientTransport.cs @@ -2,7 +2,7 @@ namespace UiPath.Ipc.Transport.WebSocket; -public sealed record WebSocketTransport : ClientTransport +public sealed record WebSocketClientTransport : ClientTransport { public required Uri Uri { get; init; } public override string ToString() => $"WebSocketClient={Uri}"; @@ -28,7 +28,7 @@ internal sealed class WebSocketClientState : IClientState public async ValueTask Connect(IpcClient client, CancellationToken ct) { - var transport = client.Transport as WebSocketTransport ?? throw new InvalidOperationException(); + var transport = client.Transport as WebSocketClientTransport ?? throw new InvalidOperationException(); _clientWebSocket = new(); await _clientWebSocket.ConnectAsync(transport.Uri, ct); diff --git a/src/UiPath.CoreIpc/UiPath.CoreIpc.csproj b/src/UiPath.CoreIpc/UiPath.CoreIpc.csproj index af37d5f6..d2bb74a6 100644 --- a/src/UiPath.CoreIpc/UiPath.CoreIpc.csproj +++ b/src/UiPath.CoreIpc/UiPath.CoreIpc.csproj @@ -27,6 +27,13 @@ + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + @@ -35,6 +42,7 @@ + diff --git a/src/UiPath.Ipc.Tests/ComputingTests.cs b/src/UiPath.Ipc.Tests/ComputingTests.cs index e6858e6b..aeecd313 100644 --- a/src/UiPath.Ipc.Tests/ComputingTests.cs +++ b/src/UiPath.Ipc.Tests/ComputingTests.cs @@ -2,7 +2,6 @@ using Nito.AsyncEx; using Nito.Disposables; using NSubstitute; -using System.Collections.Concurrent; using System.Runtime.InteropServices; using System.Text; using UiPath.Ipc.Transport.NamedPipe; @@ -12,7 +11,7 @@ namespace UiPath.Ipc.Tests; -public abstract class ComputingTests : TestBase +public abstract class ComputingTests : SpyTestBase { #region " Setup " protected readonly ComputingCallback _computingCallback = new(); @@ -25,8 +24,10 @@ public abstract class ComputingTests : TestBase protected sealed override IpcProxy? IpcProxy => Proxy as IpcProxy; protected sealed override Type ContractType => typeof(IComputingService); - - protected readonly ConcurrentBag _clientBeforeCalls = new(); + protected override EndpointCollection? Callbacks => new() + { + { typeof(IComputingCallback), _computingCallback } + }; protected ComputingTests(ITestOutputHelper outputHelper) : base(outputHelper) { @@ -42,24 +43,7 @@ protected override void ConfigureSpecificServices(IServiceCollection services) .AddSingletonAlias() ; - protected override ServerTransport ConfigTransportAgnostic(ServerTransport listener) - => listener with - { - ConcurrentAccepts = 10, - RequestTimeout = Timeouts.DefaultRequest, - MaxReceivedMessageSizeInMegabytes = 1, - }; - protected override ClientConfig CreateClientConfig(EndpointCollection? callbacks = null) - => new() - { - RequestTimeout = Timeouts.DefaultRequest, - Scheduler = GuiScheduler, - Callbacks = callbacks ?? new() - { - { typeof(IComputingCallback), _computingCallback } - }, - BeforeCall = async (callInfo, _) => _clientBeforeCalls.Add(callInfo), - }; + protected override TimeSpan ServerRequestTimeout => Timeouts.DefaultRequest; #endregion [Theory, IpcAutoData] @@ -231,16 +215,13 @@ public async Task BeforeConnect_ShouldStartExternalServerJIT() }); var proxy = new IpcClient { - Config = new() + Scheduler = GuiScheduler, + BeforeConnect = async (_) => { - Scheduler = GuiScheduler, - BeforeConnect = async (_) => - { - serverProcess.Start(); - var time = TimeSpan.FromSeconds(1); - _outputHelper.WriteLine($"Server started. Waiting {time}. PID={serverProcess.Id}"); - await Task.Delay(time); - }, + serverProcess.Start(); + var time = TimeSpan.FromSeconds(1); + _outputHelper.WriteLine($"Server started. Waiting {time}. PID={serverProcess.Id}"); + await Task.Delay(time); }, Transport = externalServerParams.CreateClientTransport() }.GetProxy(); @@ -260,7 +241,7 @@ await Enumerable.Range(1, CParallelism) var mockCallback = Substitute.For(); mockCallback.AddInts(0, 1).Returns(1); - var proxy = CreateClient(callbacks: new() + var proxy = CreateIpcClient(callbacks: new() { { typeof(IComputingCallback), mockCallback } })!.GetProxy(); @@ -310,8 +291,8 @@ public readonly record struct ExternalServerParams(ServerKind Kind, string? Pipe public ClientTransport CreateClientTransport() => Kind switch { ServerKind.NamedPipes => new NamedPipeClientTransport() { PipeName = PipeName! }, - ServerKind.Tcp => new TcpTransport() { EndPoint = new(System.Net.IPAddress.Loopback, Port) }, - ServerKind.WebSockets => new WebSocketTransport() { Uri = new($"ws://localhost:{Port}") }, + ServerKind.Tcp => new TcpClientTransport() { EndPoint = new(System.Net.IPAddress.Loopback, Port) }, + ServerKind.WebSockets => new WebSocketClientTransport() { Uri = new($"ws://localhost:{Port}") }, _ => throw new NotSupportedException($"Kind not supported. Kind was {Kind}") }; } @@ -319,7 +300,7 @@ public enum ServerKind { NamedPipes, Tcp, WebSockets } private sealed class DisableInProcClientServer : OverrideConfig { - public override async Task Override(Func> listener) => null; + public override async Task Override(Func> ipcServerFactory) => null; public override IpcClient? Override(Func client) => null; } } diff --git a/src/UiPath.Ipc.Tests/ComputingTestsOverNamedPipes.cs b/src/UiPath.Ipc.Tests/ComputingTestsOverNamedPipes.cs index 067b5ba7..88ef854f 100644 --- a/src/UiPath.Ipc.Tests/ComputingTestsOverNamedPipes.cs +++ b/src/UiPath.Ipc.Tests/ComputingTestsOverNamedPipes.cs @@ -9,7 +9,7 @@ public sealed class ComputingTestsOverNamedPipes : ComputingTests public ComputingTestsOverNamedPipes(ITestOutputHelper outputHelper) : base(outputHelper) { } - protected override async Task CreateListener() => new NamedPipeServerTransport + protected override async Task CreateServerTransport() => new NamedPipeServerTransport { PipeName = PipeName }; diff --git a/src/UiPath.Ipc.Tests/Config/OverrideConfigAttribute.cs b/src/UiPath.Ipc.Tests/Config/OverrideConfigAttribute.cs index 237c1e40..cecdedde 100644 --- a/src/UiPath.Ipc.Tests/Config/OverrideConfigAttribute.cs +++ b/src/UiPath.Ipc.Tests/Config/OverrideConfigAttribute.cs @@ -34,6 +34,6 @@ public OverrideConfigAttribute(Type overrideConfigType) public abstract class OverrideConfig { - public virtual Task Override(Func> listener) => listener()!; - public virtual IpcClient? Override(Func client) => client(); + public virtual Task Override(Func> ipcServer) => ipcServer()!; + public virtual IpcClient? Override(Func ipcClientFactory) => ipcClientFactory(); } \ No newline at end of file diff --git a/src/UiPath.Ipc.Tests/Helpers/IpcHelpers.cs b/src/UiPath.Ipc.Tests/Helpers/IpcHelpers.cs index 0296354b..3ccc33c9 100644 --- a/src/UiPath.Ipc.Tests/Helpers/IpcHelpers.cs +++ b/src/UiPath.Ipc.Tests/Helpers/IpcHelpers.cs @@ -35,23 +35,27 @@ public static IServiceProvider GetRequired(this IServiceProvider serviceProvi internal static class IpcClientExtensions { public static IpcClient WithRequestTimeout(this IpcClient ipcClient, TimeSpan requestTimeout) - => new() { - Config = ipcClient.Config with { RequestTimeout = requestTimeout }, - Transport = ipcClient.Transport, - }; + ipcClient.RequestTimeout = requestTimeout; + return ipcClient; + } + public static IpcServer WithRequestTimeout(this IpcServer ipcServer, TimeSpan requestTimeout) + { + ipcServer.RequestTimeout = requestTimeout; + return ipcServer; + } + public static async Task WithRequestTimeout(this Task ipcServerTask, TimeSpan requestTimeout) + => (await ipcServerTask).WithRequestTimeout(requestTimeout); public static IpcClient WithCallbacks(this IpcClient ipcClient, EndpointCollection callbacks) - => new() { - Config = ipcClient.Config with { Callbacks = callbacks }, - Transport = ipcClient.Transport, - }; + ipcClient.Callbacks = callbacks; + return ipcClient; + } public static IpcClient WithBeforeConnect(this IpcClient ipcClient, BeforeConnectHandler beforeConnect) - => new() { - Config = ipcClient.Config with { BeforeConnect = beforeConnect }, - Transport = ipcClient.Transport, - }; + ipcClient.BeforeConnect = beforeConnect; + return ipcClient; + } } \ No newline at end of file diff --git a/src/UiPath.Ipc.Tests/NamedPipeSmokeTests.cs b/src/UiPath.Ipc.Tests/NamedPipeSmokeTests.cs index 71fa5d83..e43eee21 100644 --- a/src/UiPath.Ipc.Tests/NamedPipeSmokeTests.cs +++ b/src/UiPath.Ipc.Tests/NamedPipeSmokeTests.cs @@ -26,12 +26,10 @@ public async Task NamedPipesShoulNotLeak() private static IpcServer CreateServer(string pipeName) => new IpcServer { - Transport = [ - new NamedPipeServerTransport - { - PipeName = pipeName, - } - ], + Transport = new NamedPipeServerTransport + { + PipeName = pipeName, + }, Endpoints = new() { typeof(IComputingService) @@ -45,8 +43,7 @@ private static IpcServer CreateServer(string pipeName) private static IpcClient CreateClient(string pipeName) => new() { - Transport = new NamedPipeClientTransport { PipeName = pipeName }, - Config = new() + Transport = new NamedPipeClientTransport { PipeName = pipeName } }; private static Task ListPipes(string pattern) diff --git a/src/UiPath.Ipc.Tests/Program.cs b/src/UiPath.Ipc.Tests/Program.cs index 084096ce..feb7a67e 100644 --- a/src/UiPath.Ipc.Tests/Program.cs +++ b/src/UiPath.Ipc.Tests/Program.cs @@ -10,7 +10,7 @@ return 1; } var externalServerParams = JsonConvert.DeserializeObject(Encoding.UTF8.GetString(Convert.FromBase64String(base64))); -await using var asyncDisposable = externalServerParams.CreateListenerConfig(out var listener); +await using var asyncDisposable = externalServerParams.CreateListenerConfig(out var serverTransport); await using var serviceProvider = new ServiceCollection() .AddLogging(builder => builder.AddConsole()) @@ -25,7 +25,7 @@ { { typeof(IComputingService) }, }, - Transport = [listener], + Transport = serverTransport, }; ipcServer.Start(); await ipcServer.WaitForStop(); diff --git a/src/UiPath.Ipc.Tests/RobotTests.cs b/src/UiPath.Ipc.Tests/RobotTests.cs index c07902d8..627bb566 100644 --- a/src/UiPath.Ipc.Tests/RobotTests.cs +++ b/src/UiPath.Ipc.Tests/RobotTests.cs @@ -17,6 +17,10 @@ public abstract class RobotTests : TestBase protected sealed override IpcProxy? IpcProxy => Proxy as IpcProxy; protected sealed override Type ContractType => typeof(IStudioOperations); + protected override EndpointCollection? Callbacks => new() + { + { typeof(IStudioEvents), _studioEvents } + }; protected readonly ConcurrentBag _clientBeforeCalls = new(); @@ -31,24 +35,7 @@ protected override void ConfigureSpecificServices(IServiceCollection services) .AddSingleton() .AddSingletonAlias(); - protected override ServerTransport ConfigTransportAgnostic(ServerTransport listener) - => listener with - { - ConcurrentAccepts = 10, - RequestTimeout = Timeouts.DefaultRequest, - MaxReceivedMessageSizeInMegabytes = 1, - }; - protected override ClientConfig CreateClientConfig(EndpointCollection? callbacks = null) - => new() - { - RequestTimeout = Timeouts.DefaultRequest, - Scheduler = GuiScheduler, - Callbacks = callbacks ?? new() - { - { typeof(IStudioEvents), _studioEvents } - }, - BeforeCall = async (callInfo, _) => _clientBeforeCalls.Add(callInfo), - }; + protected override TimeSpan ServerRequestTimeout => Timeouts.DefaultRequest; #endregion [Fact] diff --git a/src/UiPath.Ipc.Tests/RobotTestsOverNamedPipes.cs b/src/UiPath.Ipc.Tests/RobotTestsOverNamedPipes.cs index fed8d637..5e9b11eb 100644 --- a/src/UiPath.Ipc.Tests/RobotTestsOverNamedPipes.cs +++ b/src/UiPath.Ipc.Tests/RobotTestsOverNamedPipes.cs @@ -12,7 +12,7 @@ public sealed class RobotTestsOverNamedPipes : RobotTests public RobotTestsOverNamedPipes(ITestOutputHelper outputHelper) : base(outputHelper) { } - protected override async Task CreateListener() => new NamedPipeServerTransport + protected override async Task CreateServerTransport() => new NamedPipeServerTransport { PipeName = PipeName }; @@ -62,13 +62,8 @@ public static void OnConnectingToUserService() public TContract CreateUserServiceProxy(string pipeName) where TContract : class => RobotIpcHelpers.CreateProxy( - listener: new NamedPipeServerTransport() - { - PipeName = pipeName, - AccessControl = security => security.AllowCurrentUser(), - MaxReceivedMessageSizeInMegabytes = 10, - RequestTimeout = TimeSpan.FromSeconds(40), - }, + pipeName, + requestTimeout: TimeSpan.FromSeconds(40), callbacks: new EndpointCollection() { { typeof(TCallback), Instance } @@ -81,16 +76,6 @@ internal static partial class RobotIpcHelpers { private static readonly ConcurrentDictionary PipeClients = new(); - public static TContract CreateProxy( - NamedPipeServerTransport listener, - EndpointCollection? callbacks = null, - IServiceProvider? provider = null, - Action? beforeConnect = null, - BeforeCallHandler? beforeCall = null, - bool allowImpersonation = false, - TaskScheduler? scheduler = null) where TContract : class - => CreateProxy(listener.PipeName, listener.RequestTimeout, callbacks, provider, beforeConnect, beforeCall, allowImpersonation, scheduler); - public static TContract CreateProxy( string pipeName, TimeSpan? requestTimeout = null, @@ -101,6 +86,7 @@ public static TContract CreateProxy( bool allowImpersonation = false, TaskScheduler? scheduler = null) where TContract : class { + // TODO: Fix this // Dirty hack (temporary): different callback sets will result in different connections // Hopefully, the different sets are also disjunctive. @@ -139,20 +125,17 @@ private static ClientAndParams CreateClient(CreateProxyRequest request) => new( Client: new() { - Config = new() + RequestTimeout = request.Params.RequestTimeout ?? Timeout.InfiniteTimeSpan, + ServiceProvider = request.Params.Provider, + Logger = request.Params.Provider?.GetService>(), + Callbacks = request.Callbacks, + BeforeConnect = request.Params.BeforeConnect is null ? null : _ => { - RequestTimeout = request.Params.RequestTimeout ?? Timeout.InfiniteTimeSpan, - ServiceProvider = request.Params.Provider, - Logger = request.Params.Provider?.GetService>(), - Callbacks = request.Callbacks, - BeforeConnect = request.Params.BeforeConnect is null ? null : _ => - { - request.Params.BeforeConnect(); - return Task.CompletedTask; - }, - BeforeCall = request.Params.BeforeCall, - Scheduler = request.Params.Scheduler, + request.Params.BeforeConnect(); + return Task.CompletedTask; }, + BeforeOutgoingCall = request.Params.BeforeCall, + Scheduler = request.Params.Scheduler, Transport = new NamedPipeClientTransport { PipeName = request.ActualKey.Name, diff --git a/src/UiPath.Ipc.Tests/SpyTestBase.cs b/src/UiPath.Ipc.Tests/SpyTestBase.cs new file mode 100644 index 00000000..2fc866f3 --- /dev/null +++ b/src/UiPath.Ipc.Tests/SpyTestBase.cs @@ -0,0 +1,20 @@ +using System.Collections.Concurrent; +using Xunit.Abstractions; + +namespace UiPath.Ipc.Tests; + +public abstract class SpyTestBase : TestBase +{ + protected readonly ConcurrentBag _clientBeforeCalls = new(); + + protected SpyTestBase(ITestOutputHelper outputHelper) : base(outputHelper) + { + } + + protected override void ConfigureClient(IpcClient ipcClient) + { + base.ConfigureClient(ipcClient); + + ipcClient.BeforeOutgoingCall = async (callInfo, _) => _clientBeforeCalls.Add(callInfo); + } +} diff --git a/src/UiPath.Ipc.Tests/SyncOverAsyncTests.cs b/src/UiPath.Ipc.Tests/SyncOverAsyncTests.cs index 01d4ef79..6992eeeb 100644 --- a/src/UiPath.Ipc.Tests/SyncOverAsyncTests.cs +++ b/src/UiPath.Ipc.Tests/SyncOverAsyncTests.cs @@ -47,12 +47,7 @@ public async Task RemoteCallingSyncOverAsync_IpcShouldBeResilient(ScenarioId sce private static IpcServer CreateServer(string pipeName) => new IpcServer { - Transport = [ - new NamedPipeServerTransport - { - PipeName = pipeName, - } - ], + Transport = new NamedPipeServerTransport { PipeName = pipeName }, Endpoints = new() { typeof(IComputingService) @@ -66,8 +61,7 @@ private static IpcServer CreateServer(string pipeName) private static IpcClient CreateClient(string pipeName) => new() { - Transport = new NamedPipeClientTransport { PipeName = pipeName }, - Config = new() + Transport = new NamedPipeClientTransport { PipeName = pipeName } }; diff --git a/src/UiPath.Ipc.Tests/SystemTests.cs b/src/UiPath.Ipc.Tests/SystemTests.cs index 73dc59a3..70db74df 100644 --- a/src/UiPath.Ipc.Tests/SystemTests.cs +++ b/src/UiPath.Ipc.Tests/SystemTests.cs @@ -30,20 +30,7 @@ protected override void ConfigureSpecificServices(IServiceCollection services) .AddSingleton() .AddSingletonAlias(); - protected override ServerTransport ConfigTransportAgnostic(ServerTransport listener) - => listener with - { - ConcurrentAccepts = 10, - RequestTimeout = Timeouts.DefaultRequest, - MaxReceivedMessageSizeInMegabytes = 1, - }; - protected override ClientConfig CreateClientConfig(EndpointCollection? callbacks = null) - => new() - { - RequestTimeout = Timeouts.DefaultRequest, - ServiceProvider = ServiceProvider, - Callbacks = callbacks - }; + protected override TimeSpan ServerRequestTimeout => Timeouts.DefaultRequest; #endregion [Theory, IpcAutoData] @@ -76,7 +63,7 @@ public async Task NotPassingAnOptionalMessage_ShouldWork() .ShouldBeAsync(true); [Fact] - [OverrideConfig(typeof(ServerExecutingTooLongACall_ShouldThrowTimeout_Config))] + [OverrideConfig(typeof(ShortServerLongClientTimeout))] public async Task ServerExecutingTooLongACall_ShouldThrowTimeout() => await Proxy.EchoGuidAfter(Guid.Empty, Timeout.InfiniteTimeSpan) // method takes forever but we have a server side RequestTimeout configured .ShouldThrowAsync() @@ -92,23 +79,24 @@ public async Task ClientWaitingForTooLongACall_ShouldThrowTimeout() => await Proxy.EchoGuidAfter(Guid.Empty, Timeout.InfiniteTimeSpan) // method takes forever but we have a server side RequestTimeout configured .ShouldThrowAsync(); - private sealed class ServerExecutingTooLongACall_ShouldThrowTimeout_Config : OverrideConfig + private sealed class ShortServerLongClientTimeout : OverrideConfig { - public override async Task Override(Func> listener) => await listener() with { RequestTimeout = Timeouts.Short }; - public override IpcClient? Override(Func client) - => client().WithRequestTimeout(Timeout.InfiniteTimeSpan); + public override async Task Override(Func> ipcServerFactory) + { + var ipcServer = await ipcServerFactory(); + ipcServer.RequestTimeout = Timeouts.Short; + return ipcServer; + } + + public override IpcClient? Override(Func client) => client().WithRequestTimeout(Timeout.InfiniteTimeSpan); } private sealed class ClientWaitingForTooLongACall_ShouldThrowTimeout_Config : OverrideConfig { - public override async Task Override(Func> listener) => await listener() with { RequestTimeout = Timeout.InfiniteTimeSpan }; - public override IpcClient? Override(Func client) - => client().WithRequestTimeout(Timeouts.IpcRoundtrip); + public override Task Override(Func> ipcServerFactory) => ipcServerFactory().WithRequestTimeout(Timeout.InfiniteTimeSpan)!; + public override IpcClient? Override(Func client) => client().WithRequestTimeout(Timeouts.IpcRoundtrip); } - private ServerTransport ShortClientTimeout(ServerTransport listener) => listener with { RequestTimeout = TimeSpan.FromMilliseconds(100) }; - private ServerTransport InfiniteServerTimeout(ServerTransport listener) => listener with { RequestTimeout = Timeout.InfiniteTimeSpan }; - [Fact] public async Task FireAndForget_ShouldWork() { diff --git a/src/UiPath.Ipc.Tests/SystemTestsOverNamedPipes.cs b/src/UiPath.Ipc.Tests/SystemTestsOverNamedPipes.cs index d0430161..27f83f70 100644 --- a/src/UiPath.Ipc.Tests/SystemTestsOverNamedPipes.cs +++ b/src/UiPath.Ipc.Tests/SystemTestsOverNamedPipes.cs @@ -9,7 +9,7 @@ public sealed class SystemTestsOverNamedPipes : SystemTests public SystemTestsOverNamedPipes(ITestOutputHelper outputHelper) : base(outputHelper) { } - protected sealed override async Task CreateListener() => new NamedPipeServerTransport + protected sealed override async Task CreateServerTransport() => new NamedPipeServerTransport { PipeName = PipeName }; diff --git a/src/UiPath.Ipc.Tests/SystemTestsOverTcp.cs b/src/UiPath.Ipc.Tests/SystemTestsOverTcp.cs index 265a9134..d9878b51 100644 --- a/src/UiPath.Ipc.Tests/SystemTestsOverTcp.cs +++ b/src/UiPath.Ipc.Tests/SystemTestsOverTcp.cs @@ -10,12 +10,12 @@ public sealed class SystemTestsOverTcp : SystemTests public SystemTestsOverTcp(ITestOutputHelper outputHelper) : base(outputHelper) { } - protected sealed override async Task CreateListener() + protected sealed override async Task CreateServerTransport() => new TcpServerTransport { EndPoint = _endPoint, }; protected override ClientTransport CreateClientTransport() - => new TcpTransport() { EndPoint = _endPoint }; + => new TcpClientTransport() { EndPoint = _endPoint }; } diff --git a/src/UiPath.Ipc.Tests/SystemTestsOverWebSockets.cs b/src/UiPath.Ipc.Tests/SystemTestsOverWebSockets.cs index b8795dd2..bc17f6f6 100644 --- a/src/UiPath.Ipc.Tests/SystemTestsOverWebSockets.cs +++ b/src/UiPath.Ipc.Tests/SystemTestsOverWebSockets.cs @@ -15,7 +15,7 @@ protected override async Task DisposeAsync() await base.DisposeAsync(); } - protected override async Task CreateListener() + protected override async Task CreateServerTransport() { var listener = new WebSocketServerTransport { @@ -27,5 +27,5 @@ protected override async Task CreateListener() } protected override ClientTransport CreateClientTransport() - => new WebSocketTransport() { Uri = _webSocketContext.ClientUri }; + => new WebSocketClientTransport() { Uri = _webSocketContext.ClientUri }; } diff --git a/src/UiPath.Ipc.Tests/TestBase.cs b/src/UiPath.Ipc.Tests/TestBase.cs index 3086c9b3..5652726f 100644 --- a/src/UiPath.Ipc.Tests/TestBase.cs +++ b/src/UiPath.Ipc.Tests/TestBase.cs @@ -45,8 +45,8 @@ public TestBase(ITestOutputHelper outputHelper) _guiThread.SynchronizationContext.Send(() => Thread.CurrentThread.Name = Names.GuiThreadName); _serviceProvider = IpcHelpers.ConfigureServices(_outputHelper, ConfigureSpecificServices); - _ipcServer = new(CreateServer); - _ipcClient = new(() => CreateClient()); + _ipcServer = new(CreateIpcServer); + _ipcClient = new(() => CreateIpcClient()); OverrideConfig? GetOverrideConfig() { @@ -67,78 +67,84 @@ public TestBase(ITestOutputHelper outputHelper) protected abstract void ConfigureSpecificServices(IServiceCollection services); - private Task CreateListenerAndConfigure() - { - var factory = async () => - { - _outputHelper.WriteLine("Creating listener..."); - var listener = await CreateListener(); - listener = ConfigTransportAgnostic(listener); - return listener; - }; + protected virtual EndpointCollection? Callbacks => []; - if (_overrideConfig is null) + private Task CreateIpcServer() + { + if (_overrideConfig is not null) { - return factory(); + return _overrideConfig.Override(Core); } - return _overrideConfig.Override(factory); - } - - protected async Task CreateServer() - { - if (await CreateListenerAndConfigure() is not { } listener) return null; + return Core()!; - return new() + async Task Core() { - Endpoints = new() { - new EndpointSettings(ContractType) + _outputHelper.WriteLine($"Creating {nameof(ServerTransport)}..."); + + var serverTransport = await CreateServerTransport(); + ConfigTransportBase(serverTransport); + + var endpointSettings = new EndpointSettings(ContractType) + { + BeforeIncommingCall = (callInfo, ct) => { - BeforeCall = (callInfo, ct) => - { - _serverBeforeCalls.Add(callInfo); - return _tailBeforeCall?.Invoke(callInfo, ct) ?? Task.CompletedTask; - } + _serverBeforeCalls.Add(callInfo); + return _tailBeforeCall?.Invoke(callInfo, ct) ?? Task.CompletedTask; } - }, - Transport = [listener], - ServiceProvider = _serviceProvider, - Scheduler = GuiScheduler - }; - } + }; - protected IpcClient? CreateClient(EndpointCollection? callbacks = null) + return new() + { + Endpoints = new() { endpointSettings }, + Transport = serverTransport, + ServiceProvider = _serviceProvider, + Scheduler = GuiScheduler, + RequestTimeout = ServerRequestTimeout + }; + } + } + protected IpcClient? CreateIpcClient(EndpointCollection? callbacks = null) { - var factory = () => + if (_overrideConfig is null) + { + return CreateDefaultClient(); + } + + return _overrideConfig.Override(CreateDefaultClient); + + IpcClient CreateDefaultClient() { - var config = CreateClientConfig(callbacks); - var transport = CreateClientTransport(); var client = new IpcClient { - Config = config, - Transport = transport + Callbacks = callbacks ?? Callbacks, + Transport = CreateClientTransport() }; + ConfigureClient(client); return client; - }; - - if (_overrideConfig is null) - { - return factory(); } - - return _overrideConfig.Override(factory); } + protected TContract? GetProxy() where TContract : class => _ipcClient.Value?.GetProxy(); protected void CreateLazyProxy(out Lazy lazy) where TContract : class => lazy = new(GetProxy); - protected abstract Task CreateListener(); + protected abstract Task CreateServerTransport(); + protected abstract TimeSpan ServerRequestTimeout { get; } - protected abstract ClientConfig CreateClientConfig(EndpointCollection? callbacks = null); + protected virtual void ConfigureClient(IpcClient ipcClient) + { + ipcClient.RequestTimeout = Timeouts.DefaultRequest; + ipcClient.Scheduler = GuiScheduler; + } protected abstract ClientTransport CreateClientTransport(); - protected abstract ServerTransport ConfigTransportAgnostic(ServerTransport listener); + protected virtual void ConfigTransportBase(ServerTransport serverTransport) + { + serverTransport.ConcurrentAccepts = 10; + serverTransport.MaxReceivedMessageSizeInMegabytes = 1; + } protected virtual async Task DisposeAsync() {