Merge pull request #34 from jacqueskang/cancellation-token

jacqueskang · web-flow · commit f24fe59ead55 · 2018-10-29T13:15:14.000+01:00
support cancellation token in IpcReader/IpcWriter
diff --git a/src/JKang.IpcServiceFramework.Client/IpcServiceClient.cs b/src/JKang.IpcServiceFramework.Client/IpcServiceClient.cs
@@ -4,6 +4,7 @@
 using System;
 using System.IO;
 using System.Linq.Expressions;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace JKang.IpcServiceFramework
@@ -23,10 +24,11 @@ protected IpcServiceClient(
             _converter = converter;
         }
 
-        public async Task InvokeAsync(Expression<Action<TInterface>> exp)
+        public async Task InvokeAsync(Expression<Action<TInterface>> exp,
+            CancellationToken cancellationToken = default(CancellationToken))
         {
             IpcRequest request = GetRequest(exp, new MyInterceptor());
-            IpcResponse response = await GetResponseAsync(request);
+            IpcResponse response = await GetResponseAsync(request, cancellationToken);
 
             if (response.Succeed)
             {
@@ -38,10 +40,11 @@ public async Task InvokeAsync(Expression<Action<TInterface>> exp)
             }
         }
 
-        public async Task<TResult> InvokeAsync<TResult>(Expression<Func<TInterface, TResult>> exp)
+        public async Task<TResult> InvokeAsync<TResult>(Expression<Func<TInterface, TResult>> exp,
+            CancellationToken cancellationToken = default(CancellationToken))
         {
             IpcRequest request = GetRequest(exp, new MyInterceptor<TResult>());
-            IpcResponse response = await GetResponseAsync(request);
+            IpcResponse response = await GetResponseAsync(request, cancellationToken);
 
             if (response.Succeed)
             {
@@ -83,19 +86,19 @@ private static IpcRequest GetRequest(Expression exp, MyInterceptor interceptor)
             };
         }
 
-        protected abstract Task<Stream> ConnectToServerAsync();
+        protected abstract Task<Stream> ConnectToServerAsync(CancellationToken cancellationToken);
 
-        private async Task<IpcResponse> GetResponseAsync(IpcRequest request)
+        private async Task<IpcResponse> GetResponseAsync(IpcRequest request, CancellationToken cancellationToken)
         {
-            using (Stream client = await ConnectToServerAsync())
+            using (Stream client = await ConnectToServerAsync(cancellationToken))
             using (var writer = new IpcWriter(client, _serializer, leaveOpen: true))
             using (var reader = new IpcReader(client, _serializer, leaveOpen: true))
             {
                 // send request
-                writer.Write(request);
+                await writer.WriteAsync(request, cancellationToken).ConfigureAwait(false);
 
                 // receive response
-                return reader.ReadIpcResponse();
+                return await reader.ReadIpcResponseAsync(cancellationToken).ConfigureAwait(false);
             }
         }
 
diff --git a/src/JKang.IpcServiceFramework.Client/NamedPipe/NamedPipeIpcServiceClient.cs b/src/JKang.IpcServiceFramework.Client/NamedPipe/NamedPipeIpcServiceClient.cs
@@ -1,6 +1,7 @@
 ﻿using JKang.IpcServiceFramework.Services;
 using System.IO;
 using System.IO.Pipes;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace JKang.IpcServiceFramework.NamedPipe
@@ -16,10 +17,10 @@ public NamedPipeIpcServiceClient(IIpcMessageSerializer serializer, IValueConvert
             _pipeName = pipeName;
         }
 
-        protected override async Task<Stream> ConnectToServerAsync()
+        protected override async Task<Stream> ConnectToServerAsync(CancellationToken cancellationToken)
         {
             var stream = new NamedPipeClientStream(".", _pipeName, PipeDirection.InOut, PipeOptions.None);
-            await stream.ConnectAsync();
+            await stream.ConnectAsync(cancellationToken);
             return stream;
         }
     }
diff --git a/src/JKang.IpcServiceFramework.Client/Tcp/TcpIpcServiceClient.cs b/src/JKang.IpcServiceFramework.Client/Tcp/TcpIpcServiceClient.cs
@@ -4,6 +4,7 @@
 using System.Net;
 using System.Net.Sockets;
 using System.Threading.Tasks;
+using System.Threading;
 
 namespace JKang.IpcServiceFramework.Tcp
 {
@@ -20,13 +21,28 @@ public TcpIpcServiceClient(IIpcMessageSerializer serializer, IValueConverter con
             _serverPort = serverPort;
         }
 
-        protected override async Task<Stream> ConnectToServerAsync()
+        protected override async Task<Stream> ConnectToServerAsync(CancellationToken cancellationToken)
         {
+            cancellationToken.ThrowIfCancellationRequested();
+
             var client = new TcpClient();
-            await client.ConnectAsync(_serverIp, _serverPort);
-            var stream = client.GetStream();
+            IAsyncResult result = client.BeginConnect(_serverIp, _serverPort, null, null);
+
+            await Task.Run(() =>
+            {
+                // poll every 1 second to check cancellation request
+                while (!result.AsyncWaitHandle.WaitOne(TimeSpan.FromMilliseconds(1000), false))
+                {
+                    if (cancellationToken.IsCancellationRequested)
+                    {
+                        client.Close();
+                        cancellationToken.ThrowIfCancellationRequested();
+                    }
+                }
+                client.EndConnect(result);
+            });
 
-            return stream;
+            return client.GetStream();
         }
     }
 }
diff --git a/src/JKang.IpcServiceFramework.Core/IO/IpcReader.cs b/src/JKang.IpcServiceFramework.Core/IO/IpcReader.cs
@@ -1,40 +1,49 @@
 ﻿using System;
 using System.IO;
-using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace JKang.IpcServiceFramework.IO
 {
     public class IpcReader : IDisposable
     {
-        private readonly BinaryReader _reader;
+        private readonly byte[] _lengthBuffer = new byte[4];
+        private readonly Stream _stream;
         private readonly IIpcMessageSerializer _serializer;
+        private readonly bool _leaveOpen;
 
         public IpcReader(Stream stream, IIpcMessageSerializer serializer)
             : this(stream, serializer, leaveOpen: false)
         { }
 
         public IpcReader(Stream stream, IIpcMessageSerializer serializer, bool leaveOpen)
         {
-            _reader = new BinaryReader(stream, Encoding.UTF8, leaveOpen);
+            _stream = stream;
             _serializer = serializer;
+            _leaveOpen = leaveOpen;
         }
 
-        public IpcRequest ReadIpcRequest()
+        public async Task<IpcRequest> ReadIpcRequestAsync(CancellationToken cancellationToken = default(CancellationToken))
         {
-            byte[] binary = ReadMessage();
+            byte[] binary = await ReadMessageAsync(cancellationToken);
             return _serializer.DeserializeRequest(binary);
         }
 
-        public IpcResponse ReadIpcResponse()
+        public async Task<IpcResponse> ReadIpcResponseAsync(CancellationToken cancellationToken = default(CancellationToken))
         {
-            byte[] binary = ReadMessage();
+            byte[] binary = await ReadMessageAsync(cancellationToken);
             return _serializer.DeserializeResponse(binary);
         }
 
-        private byte[] ReadMessage()
+        private async Task<byte[]> ReadMessageAsync(CancellationToken cancellationToken)
         {
-            int length = _reader.ReadInt32();
-            return _reader.ReadBytes(length);
+            await _stream.ReadAsync(_lengthBuffer, 0, _lengthBuffer.Length, cancellationToken);
+            int length = _lengthBuffer[0] | _lengthBuffer[1] << 8 | _lengthBuffer[2] << 16 | _lengthBuffer[3] << 24;
+
+            byte[] bytes = new byte[length];
+            await _stream.ReadAsync(bytes, 0, length, cancellationToken);
+
+            return bytes;
         }
 
         #region IDisposible
@@ -56,7 +65,10 @@ protected virtual void Dispose(bool disposing)
 
             if (disposing)
             {
-                _reader.Dispose();
+                if (!_leaveOpen)
+                {
+                    _stream.Dispose();
+                }
             }
 
             _disposed = true;
diff --git a/src/JKang.IpcServiceFramework.Core/IO/IpcWriter.cs b/src/JKang.IpcServiceFramework.Core/IO/IpcWriter.cs
@@ -1,40 +1,52 @@
 ﻿using System;
 using System.IO;
-using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace JKang.IpcServiceFramework.IO
 {
     public class IpcWriter : IDisposable
     {
-        private readonly BinaryWriter _writer;
+        private readonly byte[] _lengthBuffer = new byte[4];
+        private readonly Stream _stream;
         private readonly IIpcMessageSerializer _serializer;
+        private readonly bool _leaveOpen;
 
         public IpcWriter(Stream stream, IIpcMessageSerializer serializer)
             : this(stream, serializer, leaveOpen: false)
         { }
 
         public IpcWriter(Stream stream, IIpcMessageSerializer serializer, bool leaveOpen)
         {
-            _writer = new BinaryWriter(stream, Encoding.UTF8, leaveOpen);
+            _stream = stream;
             _serializer = serializer;
+            _leaveOpen = leaveOpen;
         }
 
-        public void Write(IpcRequest request)
+        public async Task WriteAsync(IpcRequest request,
+            CancellationToken cancellationToken = default(CancellationToken))
         {
             byte[] binary = _serializer.SerializeRequest(request);
-            WriteMessage(binary);
+            await WriteMessageAsync(binary, cancellationToken);
         }
 
-        public void Write(IpcResponse response)
+        public async Task WriteAsync(IpcResponse response,
+            CancellationToken cancellationToken = default(CancellationToken))
         {
             byte[] binary = _serializer.SerializeResponse(response);
-            WriteMessage(binary);
+            await WriteMessageAsync(binary, cancellationToken);
         }
 
-        private void WriteMessage(byte[] binary)
+        private async Task WriteMessageAsync(byte[] binary, CancellationToken cancellationToken)
         {
-            _writer.Write(binary.Length);
-            _writer.Write(binary);
+            int length = binary.Length;
+            _lengthBuffer[0] = (byte)length;
+            _lengthBuffer[1] = (byte)(length >> 8);
+            _lengthBuffer[2] = (byte)(length >> 16);
+            _lengthBuffer[3] = (byte)(length >> 24);
+
+            await _stream.WriteAsync(_lengthBuffer, 0, _lengthBuffer.Length, cancellationToken);
+            await _stream.WriteAsync(binary, 0, binary.Length, cancellationToken);
         }
 
         #region IDisposible
@@ -56,7 +68,10 @@ protected virtual void Dispose(bool disposing)
 
             if (disposing)
             {
-                _writer.Dispose();
+                if (!_leaveOpen)
+                {
+                    _stream.Dispose();
+                }
             }
 
             _disposed = true;
diff --git a/src/JKang.IpcServiceFramework.Server/IpcServiceEndpoint.cs b/src/JKang.IpcServiceFramework.Server/IpcServiceEndpoint.cs
@@ -24,8 +24,8 @@ protected IpcServiceEndpoint(string name, IServiceProvider serviceProvider)
         public abstract Task ListenAsync(CancellationToken cancellationToken = default(CancellationToken));
     }
 
-    public abstract class IpcServiceEndpoint<TContract>: IpcServiceEndpoint
-        where TContract: class
+    public abstract class IpcServiceEndpoint<TContract> : IpcServiceEndpoint
+        where TContract : class
     {
         private readonly IValueConverter _converter;
         private readonly IIpcMessageSerializer _serializer;
@@ -37,15 +37,22 @@ protected IpcServiceEndpoint(string name, IServiceProvider serviceProvider)
             _serializer = serviceProvider.GetRequiredService<IIpcMessageSerializer>();
         }
 
-        protected void Process(Stream server, ILogger logger)
+        protected async Task ProcessAsync(Stream server, ILogger logger, CancellationToken cancellationToken)
         {
             using (var writer = new IpcWriter(server, _serializer, leaveOpen: true))
             using (var reader = new IpcReader(server, _serializer, leaveOpen: true))
             {
                 try
                 {
+                    if (cancellationToken.IsCancellationRequested)
+                    {
+                        return;
+                    }
+
                     logger?.LogDebug($"[thread {Thread.CurrentThread.ManagedThreadId}] client connected, reading request...");
-                    IpcRequest request = reader.ReadIpcRequest();
+                    IpcRequest request = await reader.ReadIpcRequestAsync(cancellationToken).ConfigureAwait(false);
+
+                    cancellationToken.ThrowIfCancellationRequested();
 
                     logger?.LogDebug($"[thread {Thread.CurrentThread.ManagedThreadId}] request received, invoking corresponding method...");
                     IpcResponse response;
@@ -54,8 +61,10 @@ protected void Process(Stream server, ILogger logger)
                         response = GetReponse(request, scope);
                     }
 
+                    cancellationToken.ThrowIfCancellationRequested();
+
                     logger?.LogDebug($"[thread {Thread.CurrentThread.ManagedThreadId}] sending response...");
-                    writer.Write(response);
+                    await writer.WriteAsync(response, cancellationToken).ConfigureAwait(false);
 
                     logger?.LogDebug($"[thread {Thread.CurrentThread.ManagedThreadId}] done.");
                 }
diff --git a/src/JKang.IpcServiceFramework.Server/NamedPipe/NamedPipeIpcServiceEndpoint.cs b/src/JKang.IpcServiceFramework.Server/NamedPipe/NamedPipeIpcServiceEndpoint.cs
@@ -63,8 +63,8 @@ private void StartServerThread(object obj)
             {
                 using (var server = new NamedPipeServerStream(PipeName, PipeDirection.InOut, _options.ThreadCount))
                 {
-                    server.WaitForConnectionAsync().Wait(token);
-                    Task.Run(() => Process(server, _logger)).Wait(token);
+                    server.WaitForConnectionAsync(token).Wait();
+                    ProcessAsync(server, _logger, token).Wait();
                 }
             }
             catch when (token.IsCancellationRequested)
diff --git a/src/JKang.IpcServiceFramework.Server/Tcp/TcpIpcServiceEndpoint.cs b/src/JKang.IpcServiceFramework.Server/Tcp/TcpIpcServiceEndpoint.cs
@@ -40,7 +40,7 @@ public TcpIpcServiceEndpoint(String name, IServiceProvider serviceProvider, IPAd
                     {
                         TcpClient client = await _listener.AcceptTcpClientAsync();
                         NetworkStream server = client.GetStream();
-                        Process(server, _logger);
+                        await ProcessAsync(server, _logger, cancellationToken);
                     }
                 }
                 catch when (cancellationToken.IsCancellationRequested)

Original file line number	Diff line number	Diff line change
`@@ -4,6 +4,7 @@`
`4`	`4`	`using System;`
`5`	`5`	`using System.IO;`
`6`	`6`	`using System.Linq.Expressions;`
	`7`	`+using System.Threading;`
`7`	`8`	`using System.Threading.Tasks;`
`8`	`9`
`9`	`10`	`namespace JKang.IpcServiceFramework`
`@@ -23,10 +24,11 @@ protected IpcServiceClient(`
`23`	`24`	`_converter = converter;`
`24`	`25`	`}`
`25`	`26`
`26`		`- public async Task InvokeAsync(Expression<Action<TInterface>> exp)`
	`27`	`+ public async Task InvokeAsync(Expression<Action<TInterface>> exp,`
	`28`	`+ CancellationToken cancellationToken = default(CancellationToken))`
`27`	`29`	`{`
`28`	`30`	`IpcRequest request = GetRequest(exp, new MyInterceptor());`
`29`		`- IpcResponse response = await GetResponseAsync(request);`
	`31`	`+ IpcResponse response = await GetResponseAsync(request, cancellationToken);`
`30`	`32`
`31`	`33`	`if (response.Succeed)`
`32`	`34`	`{`
`@@ -38,10 +40,11 @@ public async Task InvokeAsync(Expression<Action<TInterface>> exp)`
`38`	`40`	`}`
`39`	`41`	`}`
`40`	`42`
`41`		`- public async Task<TResult> InvokeAsync<TResult>(Expression<Func<TInterface, TResult>> exp)`
	`43`	`+ public async Task<TResult> InvokeAsync<TResult>(Expression<Func<TInterface, TResult>> exp,`
	`44`	`+ CancellationToken cancellationToken = default(CancellationToken))`
`42`	`45`	`{`
`43`	`46`	`IpcRequest request = GetRequest(exp, new MyInterceptor<TResult>());`
`44`		`- IpcResponse response = await GetResponseAsync(request);`
	`47`	`+ IpcResponse response = await GetResponseAsync(request, cancellationToken);`
`45`	`48`
`46`	`49`	`if (response.Succeed)`
`47`	`50`	`{`
`@@ -83,19 +86,19 @@ private static IpcRequest GetRequest(Expression exp, MyInterceptor interceptor)`
`83`	`86`	`};`
`84`	`87`	`}`
`85`	`88`
`86`		`- protected abstract Task<Stream> ConnectToServerAsync();`
	`89`	`+ protected abstract Task<Stream> ConnectToServerAsync(CancellationToken cancellationToken);`
`87`	`90`
`88`		`- private async Task<IpcResponse> GetResponseAsync(IpcRequest request)`
	`91`	`+ private async Task<IpcResponse> GetResponseAsync(IpcRequest request, CancellationToken cancellationToken)`
`89`	`92`	`{`
`90`		`- using (Stream client = await ConnectToServerAsync())`
	`93`	`+ using (Stream client = await ConnectToServerAsync(cancellationToken))`
`91`	`94`	`using (var writer = new IpcWriter(client, _serializer, leaveOpen: true))`
`92`	`95`	`using (var reader = new IpcReader(client, _serializer, leaveOpen: true))`
`93`	`96`	`{`
`94`	`97`	`// send request`
`95`		`- writer.Write(request);`
	`98`	`+ await writer.WriteAsync(request, cancellationToken).ConfigureAwait(false);`
`96`	`99`
`97`	`100`	`// receive response`
`98`		`- return reader.ReadIpcResponse();`
	`101`	`+ return await reader.ReadIpcResponseAsync(cancellationToken).ConfigureAwait(false);`
`99`	`102`	`}`
`100`	`103`	`}`
`101`	`104`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`using JKang.IpcServiceFramework.Services;`
`2`	`2`	`using System.IO;`
`3`	`3`	`using System.IO.Pipes;`
	`4`	`+using System.Threading;`
`4`	`5`	`using System.Threading.Tasks;`
`5`	`6`
`6`	`7`	`namespace JKang.IpcServiceFramework.NamedPipe`
`@@ -16,10 +17,10 @@ public NamedPipeIpcServiceClient(IIpcMessageSerializer serializer, IValueConvert`
`16`	`17`	`_pipeName = pipeName;`
`17`	`18`	`}`
`18`	`19`
`19`		`- protected override async Task<Stream> ConnectToServerAsync()`
	`20`	`+ protected override async Task<Stream> ConnectToServerAsync(CancellationToken cancellationToken)`
`20`	`21`	`{`
`21`	`22`	`var stream = new NamedPipeClientStream(".", _pipeName, PipeDirection.InOut, PipeOptions.None);`
`22`		`- await stream.ConnectAsync();`
	`23`	`+ await stream.ConnectAsync(cancellationToken);`
`23`	`24`	`return stream;`
`24`	`25`	`}`
`25`	`26`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,40 +1,49 @@`
`1`	`1`	`using System;`
`2`	`2`	`using System.IO;`
`3`		`-using System.Text;`
	`3`	`+using System.Threading;`
	`4`	`+using System.Threading.Tasks;`
`4`	`5`
`5`	`6`	`namespace JKang.IpcServiceFramework.IO`
`6`	`7`	`{`
`7`	`8`	`public class IpcReader : IDisposable`
`8`	`9`	`{`
`9`		`- private readonly BinaryReader _reader;`
	`10`	`+ private readonly byte[] _lengthBuffer = new byte[4];`
	`11`	`+ private readonly Stream _stream;`
`10`	`12`	`private readonly IIpcMessageSerializer _serializer;`
	`13`	`+ private readonly bool _leaveOpen;`
`11`	`14`
`12`	`15`	`public IpcReader(Stream stream, IIpcMessageSerializer serializer)`
`13`	`16`	`: this(stream, serializer, leaveOpen: false)`
`14`	`17`	`{ }`
`15`	`18`
`16`	`19`	`public IpcReader(Stream stream, IIpcMessageSerializer serializer, bool leaveOpen)`
`17`	`20`	`{`
`18`		`- _reader = new BinaryReader(stream, Encoding.UTF8, leaveOpen);`
	`21`	`+ _stream = stream;`
`19`	`22`	`_serializer = serializer;`
	`23`	`+ _leaveOpen = leaveOpen;`
`20`	`24`	`}`
`21`	`25`
`22`		`- public IpcRequest ReadIpcRequest()`
	`26`	`+ public async Task<IpcRequest> ReadIpcRequestAsync(CancellationToken cancellationToken = default(CancellationToken))`
`23`	`27`	`{`
`24`		`- byte[] binary = ReadMessage();`
	`28`	`+ byte[] binary = await ReadMessageAsync(cancellationToken);`
`25`	`29`	`return _serializer.DeserializeRequest(binary);`
`26`	`30`	`}`
`27`	`31`
`28`		`- public IpcResponse ReadIpcResponse()`
	`32`	`+ public async Task<IpcResponse> ReadIpcResponseAsync(CancellationToken cancellationToken = default(CancellationToken))`
`29`	`33`	`{`
`30`		`- byte[] binary = ReadMessage();`
	`34`	`+ byte[] binary = await ReadMessageAsync(cancellationToken);`
`31`	`35`	`return _serializer.DeserializeResponse(binary);`
`32`	`36`	`}`
`33`	`37`
`34`		`- private byte[] ReadMessage()`
	`38`	`+ private async Task<byte[]> ReadMessageAsync(CancellationToken cancellationToken)`
`35`	`39`	`{`
`36`		`- int length = _reader.ReadInt32();`
`37`		`- return _reader.ReadBytes(length);`
	`40`	`+ await _stream.ReadAsync(_lengthBuffer, 0, _lengthBuffer.Length, cancellationToken);`
	`41`	`+ int length = _lengthBuffer[0] \| _lengthBuffer[1] << 8 \| _lengthBuffer[2] << 16 \| _lengthBuffer[3] << 24;`
	`42`	`+`
	`43`	`+ byte[] bytes = new byte[length];`
	`44`	`+ await _stream.ReadAsync(bytes, 0, length, cancellationToken);`
	`45`	`+`
	`46`	`+ return bytes;`
`38`	`47`	`}`
`39`	`48`
`40`	`49`	`#region IDisposible`
`@@ -56,7 +65,10 @@ protected virtual void Dispose(bool disposing)`
`56`	`65`
`57`	`66`	`if (disposing)`
`58`	`67`	`{`
`59`		`- _reader.Dispose();`
	`68`	`+ if (!_leaveOpen)`
	`69`	`+ {`
	`70`	`+ _stream.Dispose();`
	`71`	`+ }`
`60`	`72`	`}`
`61`	`73`
`62`	`74`	`_disposed = true;`
Original file line number	Diff line number	Diff line change
`@@ -24,8 +24,8 @@ protected IpcServiceEndpoint(string name, IServiceProvider serviceProvider)`
`24`	`24`	`public abstract Task ListenAsync(CancellationToken cancellationToken = default(CancellationToken));`
`25`	`25`	`}`
`26`	`26`
`27`		`- public abstract class IpcServiceEndpoint<TContract>: IpcServiceEndpoint`
`28`		`- where TContract: class`
	`27`	`+ public abstract class IpcServiceEndpoint<TContract> : IpcServiceEndpoint`
	`28`	`+ where TContract : class`
`29`	`29`	`{`
`30`	`30`	`private readonly IValueConverter _converter;`
`31`	`31`	`private readonly IIpcMessageSerializer _serializer;`
`@@ -37,15 +37,22 @@ protected IpcServiceEndpoint(string name, IServiceProvider serviceProvider)`
`37`	`37`	`_serializer = serviceProvider.GetRequiredService<IIpcMessageSerializer>();`
`38`	`38`	`}`
`39`	`39`
`40`		`- protected void Process(Stream server, ILogger logger)`
	`40`	`+ protected async Task ProcessAsync(Stream server, ILogger logger, CancellationToken cancellationToken)`
`41`	`41`	`{`
`42`	`42`	`using (var writer = new IpcWriter(server, _serializer, leaveOpen: true))`
`43`	`43`	`using (var reader = new IpcReader(server, _serializer, leaveOpen: true))`
`44`	`44`	`{`
`45`	`45`	`try`
`46`	`46`	`{`
	`47`	`+ if (cancellationToken.IsCancellationRequested)`
	`48`	`+ {`
	`49`	`+ return;`
	`50`	`+ }`
	`51`	`+`
`47`	`52`	`logger?.LogDebug($"[thread {Thread.CurrentThread.ManagedThreadId}] client connected, reading request...");`
`48`		`- IpcRequest request = reader.ReadIpcRequest();`
	`53`	`+ IpcRequest request = await reader.ReadIpcRequestAsync(cancellationToken).ConfigureAwait(false);`
	`54`	`+`
	`55`	`+ cancellationToken.ThrowIfCancellationRequested();`
`49`	`56`
`50`	`57`	`logger?.LogDebug($"[thread {Thread.CurrentThread.ManagedThreadId}] request received, invoking corresponding method...");`
`51`	`58`	`IpcResponse response;`
`@@ -54,8 +61,10 @@ protected void Process(Stream server, ILogger logger)`
`54`	`61`	`response = GetReponse(request, scope);`
`55`	`62`	`}`
`56`	`63`
	`64`	`+ cancellationToken.ThrowIfCancellationRequested();`
	`65`	`+`
`57`	`66`	`logger?.LogDebug($"[thread {Thread.CurrentThread.ManagedThreadId}] sending response...");`
`58`		`- writer.Write(response);`
	`67`	`+ await writer.WriteAsync(response, cancellationToken).ConfigureAwait(false);`
`59`	`68`
`60`	`69`	`logger?.LogDebug($"[thread {Thread.CurrentThread.ManagedThreadId}] done.");`
`61`	`70`	`}`
Original file line number	Diff line number	Diff line change
`@@ -63,8 +63,8 @@ private void StartServerThread(object obj)`
`63`	`63`	`{`
`64`	`64`	`using (var server = new NamedPipeServerStream(PipeName, PipeDirection.InOut, _options.ThreadCount))`
`65`	`65`	`{`
`66`		`- server.WaitForConnectionAsync().Wait(token);`
`67`		`- Task.Run(() => Process(server, _logger)).Wait(token);`
	`66`	`+ server.WaitForConnectionAsync(token).Wait();`
	`67`	`+ ProcessAsync(server, _logger, token).Wait();`
`68`	`68`	`}`
`69`	`69`	`}`
`70`	`70`	`catch when (token.IsCancellationRequested)`
Original file line number	Diff line number	Diff line change
`@@ -40,7 +40,7 @@ public TcpIpcServiceEndpoint(String name, IServiceProvider serviceProvider, IPAd`
`40`	`40`	`{`
`41`	`41`	`TcpClient client = await _listener.AcceptTcpClientAsync();`
`42`	`42`	`NetworkStream server = client.GetStream();`
`43`		`- Process(server, _logger);`
	`43`	`+ await ProcessAsync(server, _logger, cancellationToken);`
`44`	`44`	`}`
`45`	`45`	`}`
`46`	`46`	`catch when (cancellationToken.IsCancellationRequested)`