fixing #63 Exception on receiving larger message

Author: Jacques Kang

src/JKang.IpcServiceFramework.Core/IO/IpcReader.cs

@@ -37,11 +37,41 @@ public IpcReader(Stream stream, IIpcMessageSerializer serializer, bool leaveOpen
 
         private async Task<byte[]> ReadMessageAsync(CancellationToken cancellationToken)
         {
-            await _stream.ReadAsync(_lengthBuffer, 0, _lengthBuffer.Length, cancellationToken).ConfigureAwait(false);
-            int length = _lengthBuffer[0] | _lengthBuffer[1] << 8 | _lengthBuffer[2] << 16 | _lengthBuffer[3] << 24;
+            int headerLength = await _stream.ReadAsync(_lengthBuffer, 0, _lengthBuffer.Length, cancellationToken);
 
-            byte[] bytes = new byte[length];
-            await _stream.ReadAsync(bytes, 0, length, cancellationToken).ConfigureAwait(false);
+            if (headerLength != 4)
+            {
+                throw new ArgumentOutOfRangeException($"Header length must be 4 but was {headerLength}");
+            }
+
+            int expectedLength = _lengthBuffer[0] | _lengthBuffer[1] << 8 | _lengthBuffer[2] << 16 | _lengthBuffer[3] << 24;
+            byte[] bytes = new byte[expectedLength];
+            int totalBytesReceived = 0;
+            int remainingBytes = expectedLength;
+
+            using (var ms = new MemoryStream())
+            {
+                while (totalBytesReceived < expectedLength)
+                {
+                    int dataLength = await _stream.ReadAsync(bytes, 0, remainingBytes, cancellationToken);
+
+                    if (dataLength == 0)
+                    {
+                        break;             // end of stream or stream shut down.
+                    }
+
+                    ms.Write(bytes, 0, dataLength);
+                    totalBytesReceived += dataLength;
+                    remainingBytes -= dataLength;
+                }
+
+                bytes = ms.ToArray();
+            }
+
+            if (totalBytesReceived != expectedLength)
+            {
+                throw new System.ArgumentOutOfRangeException($"Data length must be {expectedLength} but was {totalBytesReceived}");
+            }
 
             return bytes;
         }

src/JKang.IpcServiceFramework.IntegrationTests/EdgeCaseTest.cs

@@ -0,0 +1,54 @@
+using AutoFixture.Xunit2;
+using Microsoft.Extensions.DependencyInjection;
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.IO;
+using System.Net;
+using System.Numerics;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace JKang.IpcServiceFramework.IntegrationTests
+{
+    public class EdgeCaseTest : IDisposable
+    {
+        private readonly CancellationTokenSource _cancellationToken;
+        private readonly int _port;
+        private readonly IpcServiceClient<ITestService> _client;
+
+        public EdgeCaseTest()
+        {
+            // configure DI
+            IServiceCollection services = new ServiceCollection()
+                .AddIpc(builder => builder.AddNamedPipe().AddService<ITestService, TestService>());
+            _port = new Random().Next(10000, 50000);
+            IIpcServiceHost host = new IpcServiceHostBuilder(services.BuildServiceProvider())
+                .AddTcpEndpoint<ITestService>(
+                    name: Guid.NewGuid().ToString(),
+                    ipEndpoint: IPAddress.Loopback,
+                    port: _port)
+                .Build();
+            _cancellationToken = new CancellationTokenSource();
+            host.RunAsync(_cancellationToken.Token);
+
+            _client = new IpcServiceClientBuilder<ITestService>()
+                .UseTcp(IPAddress.Loopback, _port)
+                .Build();
+        }
+
+        [Fact]
+        public async Task HugeMessage()
+        {
+            byte[] buffer = new byte[100000000]; // 100MB
+            new Random().NextBytes(buffer);
+            byte[] result = await _client.InvokeAsync(x => x.ReverseBytes(buffer));
+        }
+
+        public void Dispose()
+        {
+            _cancellationToken.Cancel();
+        }
+    }
+}