Merge pull request #42 from chequer-io/feature/QD-2289

QD-2289 Migrate gRPC C Binding Client to Managed

Author: tony-jang

JDBC.NET.Data/JDBC.NET.Data.csproj

@@ -10,14 +10,14 @@
     <RepositoryUrl>https://github.com/chequer-io/JDBC.NET</RepositoryUrl>
     <PackageProjectUrl>https://github.com/chequer-io/JDBC.NET</PackageProjectUrl>
     <PackageIcon>Logo.jpg</PackageIcon>
-    <Version>3.5.8</Version>
+    <Version>3.5.9</Version>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
   </PropertyGroup>
 
   <ItemGroup>
     <PackageReference Include="J2NET" Version="1.3.2" />
     <PackageReference Include="Google.Protobuf" Version="3.22.3" />
-    <PackageReference Include="Grpc" Version="2.46.6" />
+    <PackageReference Include="Grpc.Net.Client" Version="2.61.0" />
     <PackageReference Include="Grpc.Tools" Version="2.54.0">
         <PrivateAssets>all</PrivateAssets>
         <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>

JDBC.NET.Data/JdbcCallInterceptor.cs

@@ -0,0 +1,171 @@
+using System.Threading;
+using System.Threading.Tasks;
+using Grpc.Core;
+using Grpc.Core.Interceptors;
+using JDBC.NET.Data.Exceptions;
+
+namespace JDBC.NET.Data;
+
+internal sealed class JdbcCallInterceptor : Interceptor
+{
+    public override TResponse BlockingUnaryCall<TRequest, TResponse>(TRequest request, ClientInterceptorContext<TRequest, TResponse> context, BlockingUnaryCallContinuation<TRequest, TResponse> continuation)
+    {
+        try
+        {
+            return base.BlockingUnaryCall(request, context, continuation);
+        }
+        catch (RpcException e)
+        {
+            throw new JdbcException(e);
+        }
+    }
+
+    public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(TRequest request, ClientInterceptorContext<TRequest, TResponse> context, AsyncUnaryCallContinuation<TRequest, TResponse> continuation)
+    {
+        AsyncUnaryCall<TResponse> asyncCall = base.AsyncUnaryCall(request, context, continuation);
+
+        return new AsyncUnaryCall<TResponse>(
+            Wrap(asyncCall.ResponseAsync),
+            Wrap(asyncCall.ResponseHeadersAsync),
+            asyncCall.GetStatus,
+            asyncCall.GetTrailers,
+            asyncCall.Dispose
+        );
+    }
+
+    public override AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(ClientInterceptorContext<TRequest, TResponse> context, AsyncClientStreamingCallContinuation<TRequest, TResponse> continuation)
+    {
+        AsyncClientStreamingCall<TRequest, TResponse> asyncCall = base.AsyncClientStreamingCall(context, continuation);
+
+        return new AsyncClientStreamingCall<TRequest, TResponse>(
+            Wrap(asyncCall.RequestStream),
+            Wrap(asyncCall.ResponseAsync),
+            Wrap(asyncCall.ResponseHeadersAsync),
+            asyncCall.GetStatus,
+            asyncCall.GetTrailers,
+            asyncCall.Dispose
+        );
+    }
+
+    public override AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(ClientInterceptorContext<TRequest, TResponse> context, AsyncDuplexStreamingCallContinuation<TRequest, TResponse> continuation)
+    {
+        AsyncDuplexStreamingCall<TRequest, TResponse> asyncCall = base.AsyncDuplexStreamingCall(context, continuation);
+
+        return new AsyncDuplexStreamingCall<TRequest, TResponse>(
+            Wrap(asyncCall.RequestStream),
+            Wrap(asyncCall.ResponseStream),
+            Wrap(asyncCall.ResponseHeadersAsync),
+            asyncCall.GetStatus,
+            asyncCall.GetTrailers,
+            asyncCall.Dispose
+        );
+    }
+
+    public override AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(TRequest request, ClientInterceptorContext<TRequest, TResponse> context, AsyncServerStreamingCallContinuation<TRequest, TResponse> continuation)
+    {
+        AsyncServerStreamingCall<TResponse> asyncCall = base.AsyncServerStreamingCall(request, context, continuation);
+
+        return new AsyncServerStreamingCall<TResponse>(
+            Wrap(asyncCall.ResponseStream),
+            Wrap(asyncCall.ResponseHeadersAsync),
+            asyncCall.GetStatus,
+            asyncCall.GetTrailers,
+            asyncCall.Dispose
+        );
+    }
+
+    private static async Task<T> Wrap<T>(Task<T> task)
+    {
+        if (task.Status != TaskStatus.WaitingForActivation)
+            return await task;
+
+        try
+        {
+            return await task;
+        }
+        catch (RpcException e)
+        {
+            throw new JdbcException(e);
+        }
+    }
+
+    private static IClientStreamWriter<T> Wrap<T>(IClientStreamWriter<T> writer)
+    {
+        if (writer == null)
+            return null;
+
+        return new JdbcClientStreamWriter<T>(writer);
+    }
+
+    private static IAsyncStreamReader<T> Wrap<T>(IAsyncStreamReader<T> reader)
+    {
+        if (reader == null)
+            return null;
+
+        return new JdbcAsyncStreamReader<T>(reader);
+    }
+
+    private readonly struct JdbcClientStreamWriter<T> : IClientStreamWriter<T>
+    {
+        public WriteOptions WriteOptions
+        {
+            get => _writer.WriteOptions;
+            set => _writer.WriteOptions = value;
+        }
+
+        private readonly IClientStreamWriter<T> _writer;
+
+        public JdbcClientStreamWriter(IClientStreamWriter<T> writer)
+        {
+            _writer = writer;
+        }
+
+        public async Task WriteAsync(T message)
+        {
+            try
+            {
+                await _writer.WriteAsync(message);
+            }
+            catch (RpcException e)
+            {
+                throw new JdbcException(e);
+            }
+        }
+
+        public async Task CompleteAsync()
+        {
+            try
+            {
+                await _writer.CompleteAsync();
+            }
+            catch (RpcException e)
+            {
+                throw new JdbcException(e);
+            }
+        }
+    }
+
+    private readonly struct JdbcAsyncStreamReader<T> : IAsyncStreamReader<T>
+    {
+        public T Current => _reader.Current;
+
+        private readonly IAsyncStreamReader<T> _reader;
+
+        public JdbcAsyncStreamReader(IAsyncStreamReader<T> reader)
+        {
+            _reader = reader;
+        }
+
+        public async Task<bool> MoveNext(CancellationToken cancellationToken)
+        {
+            try
+            {
+                return await _reader.MoveNext(cancellationToken);
+            }
+            catch (RpcException e)
+            {
+                throw new JdbcException(e);
+            }
+        }
+    }
+}