Add exceed capacity test

src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs

@@ -16,6 +16,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Text;
 using System.Threading;
@@ -95,23 +96,27 @@ namespace Grpc.HealthCheck.Tests
             var impl = new HealthServiceImpl();
             var callTask = impl.Watch(new HealthCheckRequest { Service = "" }, writer, context);
 
-            var nextWriteTask = writer.WaitNextAsync();
+            // Calling Watch on a service that doesn't have a value set will initially return ServiceUnknown
+            var nextWriteTask = writer.WrittenMessagesReader.ReadAsync();
+            Assert.AreEqual(HealthCheckResponse.Types.ServingStatus.ServiceUnknown, (await nextWriteTask).Status);
+
+            nextWriteTask = writer.WrittenMessagesReader.ReadAsync();
             impl.SetStatus("", HealthCheckResponse.Types.ServingStatus.Serving);
             Assert.AreEqual(HealthCheckResponse.Types.ServingStatus.Serving, (await nextWriteTask).Status);
 
-            nextWriteTask = writer.WaitNextAsync();
+            nextWriteTask = writer.WrittenMessagesReader.ReadAsync();
             impl.SetStatus("", HealthCheckResponse.Types.ServingStatus.NotServing);
             Assert.AreEqual(HealthCheckResponse.Types.ServingStatus.NotServing, (await nextWriteTask).Status);
 
-            nextWriteTask = writer.WaitNextAsync();
+            nextWriteTask = writer.WrittenMessagesReader.ReadAsync();
             impl.SetStatus("", HealthCheckResponse.Types.ServingStatus.Unknown);
             Assert.AreEqual(HealthCheckResponse.Types.ServingStatus.Unknown, (await nextWriteTask).Status);
 
-            nextWriteTask = writer.WaitNextAsync();
+            // Setting status for a different service name will not update Watch results
+            nextWriteTask = writer.WrittenMessagesReader.ReadAsync();
             impl.SetStatus("grpc.test.TestService", HealthCheckResponse.Types.ServingStatus.Serving);
             Assert.IsFalse(nextWriteTask.IsCompleted);
 
-            nextWriteTask = writer.WaitNextAsync();
             impl.ClearStatus("");
             Assert.AreEqual(HealthCheckResponse.Types.ServingStatus.ServiceUnknown, (await nextWriteTask).Status);
 
@@ -119,6 +124,57 @@ namespace Grpc.HealthCheck.Tests
             cts.Cancel();
             await callTask;
         }
+
+        [Test]
+        public async Task Watch_ExceedMaximumCapacitySize_DiscardOldValues()
+        {
+            var cts = new CancellationTokenSource();
+            var context = new TestServerCallContext(cts.Token);
+            var writer = new TestResponseStreamWriter();
+
+            var impl = new HealthServiceImpl();
+            var callTask = impl.Watch(new HealthCheckRequest { Service = "" }, writer, context);
+
+            // Write new 10 statuses. Only last 5 statuses will be returned when we read them from watch writer
+            for (var i = 0; i < HealthServiceImpl.MaxStatusBufferSize * 2; i++)
+            {
+                // These statuses aren't "valid" but it is useful for testing to have an incrementing number
+                impl.SetStatus("", (HealthCheckResponse.Types.ServingStatus)i);
+            }
+
+            // Read messages in a background task
+            var statuses = new List<HealthCheckResponse.Types.ServingStatus>();
+            var readStatusesTask = Task.Run(async () => {
+                while (await writer.WrittenMessagesReader.WaitToReadAsync())
+                {
+                    if (writer.WrittenMessagesReader.TryRead(out var response))
+                    {
+                        statuses.Add(response.Status);
+                    }
+                }
+            });
+
+            // Tell server we're done watching and it can write what it has left and then exit
+            cts.Cancel();
+            await callTask;
+
+            // Ensure we've read all the queued statuses
+            writer.Complete();
+            await readStatusesTask;
+
+            // Collection will contain initial written message (ServiceUnknown) plus 5 queued messages
+            Assert.AreEqual(HealthServiceImpl.MaxStatusBufferSize + 1, statuses.Count);
+
+            // Initial written message
+            Assert.AreEqual(HealthCheckResponse.Types.ServingStatus.ServiceUnknown, statuses[0]);
+
+            // Last 5 queued messages
+            Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)5, statuses[1]);
+            Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)6, statuses[2]);
+            Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)7, statuses[3]);
+            Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)8, statuses[4]);
+            Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)9, statuses[5]);
+        }
 #endif
 
         private static HealthCheckResponse.Types.ServingStatus GetStatusHelper(HealthServiceImpl impl, string service)

src/csharp/Grpc.HealthCheck.Tests/TestResponseStreamWriter.cs

@@ -15,6 +15,7 @@
 #endregion
 
 #if GRPC_SUPPORT_WATCH
+using System.Threading.Channels;
 using System.Threading.Tasks;
 
 using Grpc.Core;
@@ -24,24 +25,29 @@ namespace Grpc.HealthCheck.Tests
 {
     internal class TestResponseStreamWriter : IServerStreamWriter<HealthCheckResponse>
     {
-        private TaskCompletionSource<HealthCheckResponse> _tcs;
+        private Channel<HealthCheckResponse> _channel;
 
-        public WriteOptions WriteOptions { get; set; }
-
-        public Task<HealthCheckResponse> WaitNextAsync()
+        public TestResponseStreamWriter(int maxCapacity = 1)
         {
-            _tcs = new TaskCompletionSource<HealthCheckResponse>();
-            return _tcs.Task;
+            _channel = System.Threading.Channels.Channel.CreateBounded<HealthCheckResponse>(new BoundedChannelOptions(maxCapacity) {
+                SingleReader = false,
+                SingleWriter = true,
+                FullMode = BoundedChannelFullMode.Wait
+            });
         }
 
+        public ChannelReader<HealthCheckResponse> WrittenMessagesReader => _channel.Reader;
+
+        public WriteOptions WriteOptions { get; set; }
+
         public Task WriteAsync(HealthCheckResponse message)
         {
-            if (_tcs != null)
-            {
-                _tcs.TrySetResult(message);
-            }
+            return _channel.Writer.WriteAsync(message).AsTask();
+        }
 
-            return Task.FromResult<object>(null);
+        public void Complete()
+        {
+            _channel.Writer.Complete();
         }
     }
 }

src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs

@@ -39,6 +39,9 @@ namespace Grpc.HealthCheck
     /// </summary>
     public class HealthServiceImpl : Grpc.Health.V1.Health.HealthBase
     {
+        // The maximum number of statuses to buffer on the server.
+        internal const int MaxStatusBufferSize = 5;
+
         private readonly object statusLock = new object();
         private readonly Dictionary<string, HealthCheckResponse.Types.ServingStatus> statusMap =
             new Dictionary<string, HealthCheckResponse.Types.ServingStatus>();
@@ -159,8 +162,11 @@ namespace Grpc.HealthCheck
 
             // Channel is used to to marshall multiple callers updating status into a single queue.
             // This is required because IServerStreamWriter is not thread safe.
-            // The channel will buffer up to XXX messages, after which it will drop the oldest messages.
-            Channel<HealthCheckResponse> channel = Channel.CreateBounded<HealthCheckResponse>(new BoundedChannelOptions(capacity: 5) {
+            //
+            // A queue of unwritten statuses could build up if flow control causes responseStream.WriteAsync to await.
+            // When this number is exceeded the server will discard older statuses. The discarded intermediate statues
+            // will never be sent to the client.
+            Channel<HealthCheckResponse> channel = Channel.CreateBounded<HealthCheckResponse>(new BoundedChannelOptions(capacity: MaxStatusBufferSize) {
                 SingleReader = true,
                 SingleWriter = false,
                 FullMode = BoundedChannelFullMode.DropOldest
@@ -199,7 +205,7 @@ namespace Grpc.HealthCheck
                 channel.Writer.Complete();
             });
 
-            // Read messages. WaitToReadyAsync will wait until new messages are available.
+            // Read messages. WaitToReadAsync will wait until new messages are available.
             // Loop will exit when the call is canceled and the writer is marked as complete.
             while (await channel.Reader.WaitToReadAsync())
             {

src/csharp/Grpc.HealthCheck/Properties/AssemblyInfo.cs

@@ -1,4 +1,4 @@
-#region Copyright notice and license
+#region Copyright notice and license
 
 // Copyright 2015 gRPC authors.
 //
@@ -27,3 +27,13 @@ using System.Runtime.CompilerServices;
 [assembly: AssemblyCopyright("Google Inc.  All rights reserved.")]
 [assembly: AssemblyTrademark("")]
 [assembly: AssemblyCulture("")]
+
+#if SIGNED
+[assembly: InternalsVisibleTo("Grpc.HealthCheck.Tests,PublicKey=" +
+    "00240000048000009400000006020000002400005253413100040000010001002f5797a92c6fcde81bd4098f43" +
+    "0442bb8e12768722de0b0cb1b15e955b32a11352740ee59f2c94c48edc8e177d1052536b8ac651bce11ce5da3a" +
+    "27fc95aff3dc604a6971417453f9483c7b5e836756d5b271bf8f2403fe186e31956148c03d804487cf642f8cc0" +
+    "71394ee9672dfe5b55ea0f95dfd5a7f77d22c962ccf51320d3")]
+#else
+[assembly: InternalsVisibleTo("Grpc.HealthCheck.Tests")]
+#endif