fix stream failure (#15)

* remove dead code

`_metricServer` isn't used, so there is no need for the ClientInterceptor
to be IDisposable. Updated the ServerInterceptor for symmetry.

* Also send failure to prometheus when a stream fails halfway through

When the stream is successfully started, but throws an exception
halfway through. Then the metric should be updated with a failure

Co-authored-by: Tom Moers <tom.moers@tomra.com>

NetGrpcPrometheus/ClientInterceptor.cs

@@ -12,9 +12,8 @@ namespace NetGrpcPrometheus
     /// <summary>
     /// Interceptor for intercepting calls on client side
     /// </summary>
-    public class ClientInterceptor : Interceptor, IDisposable
+    public class ClientInterceptor : Interceptor
     {
-        private readonly MetricServer _metricServer;
 
         private readonly MetricsBase _metrics;
 
@@ -178,10 +177,16 @@ public override AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRe
             {
                 AsyncServerStreamingCall<TResponse> streamingCall = continuation(request, context);
 
-                result = new AsyncServerStreamingCall<TResponse>(
-                    new WrapperStreamReader<TResponse>(streamingCall.ResponseStream,
-                        () => { _metrics.StreamReceivedCounterInc(method); }), streamingCall.ResponseHeadersAsync,
-                    streamingCall.GetStatus, streamingCall.GetTrailers, streamingCall.Dispose);
+                result =
+                    new AsyncServerStreamingCall<TResponse>(
+                        new WrapperStreamReader<TResponse>(
+                            streamingCall.ResponseStream,
+                            () => { _metrics.StreamReceivedCounterInc(method); },
+                            statusCode => { _metrics.ResponseCounterInc(method, statusCode); }),
+                        streamingCall.ResponseHeadersAsync,
+                        streamingCall.GetStatus,
+                        streamingCall.GetTrailers,
+                        streamingCall.Dispose);
 
                 _metrics.ResponseCounterInc(method, StatusCode.OK);
             }
@@ -216,8 +221,10 @@ public override AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreami
                 AsyncDuplexStreamingCall<TRequest, TResponse> streamingCall = continuation(context);
 
                 WrapperStreamReader<TResponse> responseStream =
-                    new WrapperStreamReader<TResponse>(streamingCall.ResponseStream,
-                        () => { _metrics.StreamReceivedCounterInc(method); });
+                    new WrapperStreamReader<TResponse>(
+                        streamingCall.ResponseStream,
+                        () => { _metrics.StreamReceivedCounterInc(method); },
+                        statusCode => { _metrics.ResponseCounterInc(method, statusCode); });
 
                 result = new AsyncDuplexStreamingCall<TRequest, TResponse>(
                     new WrapperClientStreamWriter<TRequest>(streamingCall.RequestStream,
@@ -240,10 +247,5 @@ public override AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreami
 
             return result;
         }
-
-        public void Dispose()
-        {
-            ((IDisposable) _metricServer)?.Dispose();
-        }
     }
 }

NetGrpcPrometheus/Helpers/WrapperStreamReader.cs

@@ -16,33 +16,38 @@ public class WrapperStreamReader<T> : IAsyncStreamReader<T>
 
         private readonly IAsyncStreamReader<T> _reader;
         private readonly Action _onMessage;
+        private readonly Action<StatusCode> _onError;
 
         /// <summary>
         /// Constructor for <see cref="IAsyncStreamReader{T}"/> wrapper
         /// </summary>
         /// <param name="reader">Stream reader that should be wrapped by this class</param>
         /// <param name="onMessage">Action that should be executed on each message received from the stream</param>
-        public WrapperStreamReader(IAsyncStreamReader<T> reader, Action onMessage)
+        public WrapperStreamReader(IAsyncStreamReader<T> reader, Action onMessage, Action<StatusCode> onError)
         {
             _reader = reader;
             _onMessage = onMessage;
+            _onError = onError;
         }
 
         public void Dispose()
         {
-            
+
         }
 
-        public Task<bool> MoveNext(CancellationToken cancellationToken)
+        public async Task<bool> MoveNext(CancellationToken cancellationToken)
         {
-            Task<bool> result = _reader.MoveNext(cancellationToken);
-
-            result.ContinueWith(task =>
+            try
             {
+                var result = await _reader.MoveNext(cancellationToken);
                 _onMessage.Invoke();
-            }, cancellationToken);
-
-            return result;
+                return result;
+            }
+            catch (RpcException e)
+            {
+                _onError(e.StatusCode);
+                throw;
+            }
         }
     }
 }

NetGrpcPrometheus/ServerInterceptor.cs

@@ -11,7 +11,7 @@ namespace NetGrpcPrometheus
     /// <summary>
     /// Interceptor for intercepting calls on server side 
     /// </summary>
-    public class ServerInterceptor : Interceptor, IDisposable
+    public class ServerInterceptor : Interceptor
     {
         private readonly MetricsBase _metrics;
 
@@ -115,8 +115,11 @@ public override Task<TResponse> ClientStreamingServerHandler<TRequest, TResponse
             try
             {
                 result = continuation(
-                    new WrapperStreamReader<TRequest>(requestStream,
-                        () => { _metrics.StreamReceivedCounterInc(method); }), context);
+                    new WrapperStreamReader<TRequest>(
+                        requestStream,
+                        () => { _metrics.StreamReceivedCounterInc(method); },
+                        statusCode => { _metrics.ResponseCounterInc(method, statusCode); }),
+                    context);
 
                 _metrics.ResponseCounterInc(method, StatusCode.OK);
             }
@@ -151,8 +154,10 @@ public override Task DuplexStreamingServerHandler<TRequest, TResponse>(
             try
             {
                 result = continuation(
-                    new WrapperStreamReader<TRequest>(requestStream,
-                        () => { _metrics.StreamReceivedCounterInc(method); }),
+                    new WrapperStreamReader<TRequest>(
+                        requestStream,
+                        () => { _metrics.StreamReceivedCounterInc(method); },
+                        statusCode => { _metrics.ResponseCounterInc(method, statusCode); }),
                     new WrapperServerStreamWriter<TResponse>(responseStream,
                         () => { _metrics.StreamSentCounterInc(method); }), context);
 
@@ -171,9 +176,5 @@ public override Task DuplexStreamingServerHandler<TRequest, TResponse>(
 
             return result;
         }
-
-        public void Dispose()
-        {
-        }
     }
 }

NetGrpcPrometheusTest/Helpers/TestClient.cs

@@ -134,7 +134,6 @@ public void Wait()
         public void Dispose()
         {
             _metricsServer.StopAsync().Wait();
-            _interceptor.Dispose();
         }
     }
 }

NetGrpcPrometheusTest/Helpers/TestServer.cs

@@ -50,7 +50,6 @@ public void Shutdown()
         public void Dispose()
         {
             Shutdown();
-            _interceptor?.Dispose();
         }
     }
 }