技術文章 » 用最少的程式碼打造一個Mini版的gRPC框架

用最少的程式碼打造一個Mini版的gRPC框架

2022-12-05 15:00:45

在《用最少的程式碼模擬gRPC四種訊息交換模式》中，我使用很簡單的程式碼模擬了gRPC四種訊息交換模式（Unary、Client Streaming、Server Streaming和Duplex Streaming），現在我們更近一步，試著使用極簡的方式打造一個gRPC框架（github地址）。這個gRPC是對ASP.NET Core gRPC實現原理的模擬，並不是想重新造一個輪子。

一、「標準」的gRPC定義、承載和呼叫
二、將gRPC方法抽象成委託
三、將委託轉換成RequestDelegate
   UnaryCallHandler
   ClientStreamingCallHandler
   ServerStreamingCallHandler
   DuplexStreamingCallHandler
四、路由註冊
五、為gRPC服務定義一個介面
六、重新定義和承載服務

一、「標準」的gRPC定義、承載和呼叫

可能有些讀者朋友們對ASP.NET Core gRPC還不是太熟悉，所以我們先來演示一下如何在一個ASP.NET Core應用中如何定義和承載一個簡單的gRPC服務，並使用自動生成的使用者端程式碼進行呼叫。我們新建一個空的解決方案，並在其中新增如下所示的三個專案。

image

我們在類庫專案Proto中定義瞭如下所示Greeter服務，並利用其中定義的四個操作分別模擬四種訊息交換模式。HelloRequest 和HelloReply 是它們涉及的兩個ProtoBuf訊息。

syntax = "proto3";
import "google/protobuf/empty.proto";

service Greeter {
  rpc SayHelloUnary (HelloRequest) returns ( HelloReply);
  rpc SayHelloServerStreaming (google.protobuf.Empty) returns (stream HelloReply);
  rpc SayHelloClientStreaming (stream HelloRequest) returns (HelloReply);
  rpc SayHelloDuplexStreaming (stream HelloRequest) returns (stream HelloReply);
}

message HelloRequest {
  string name = 1;
}

message HelloReply {
  string message = 1;
}

ASP.NET Core專案中定義瞭如下的GreeterServce服務實現了定義的四個操作，基礎類別GreeterBase是針對上面這個.proto檔案生成的型別。

public class GreeterService: GreeterBase
{
    public override Task<HelloReply> SayHelloUnary(HelloRequest request, ServerCallContext context)
    => Task.FromResult(new HelloReply { Message = $"Hello, {request.Name}" });

    public override async Task<HelloReply> SayHelloClientStreaming(IAsyncStreamReader<HelloRequest> reader, ServerCallContext context)
    {
        var list = new List<string>();
        while (await reader.MoveNext(CancellationToken.None))
        {
            list.Add(reader.Current.Name);
        }
        return new HelloReply { Message = $"Hello, {string.Join(",", list)}" };
    }

    public  override async Task SayHelloServerStreaming(Empty request, IServerStreamWriter<HelloReply> responseStream, ServerCallContext context)
    {
        await responseStream.WriteAsync(new HelloReply { Message = "Hello, Foo!" });
        await Task.Delay(1000);
        await responseStream.WriteAsync(new HelloReply { Message = "Hello, Bar!" });
        await Task.Delay(1000);
        await responseStream.WriteAsync(new HelloReply { Message = "Hello, Baz!" });
    }

    public override async Task SayHelloDuplexStreaming(IAsyncStreamReader<HelloRequest> reader, IServerStreamWriter<HelloReply> writer, ServerCallContext context)
    {
        while (await reader.MoveNext())
        {
            await writer.WriteAsync(new HelloReply { Message = $"Hello {reader.Current.Name}" });
        }
    }
}

具體的服務承載程式碼如下。我們採用Minimal API的形式，通過呼叫IServiceCollection介面的AddGrpc擴充套件方法註冊相關服務，並呼叫MapGrpcService<TService>將定義在GreeterServce中的四個方法對映我對應的路由終結點。

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddGrpc();
builder.WebHost.ConfigureKestrel(kestrel => kestrel.ConfigureEndpointDefaults(options => options.Protocols = HttpProtocols.Http2));
var app = builder.Build();
app.MapGrpcService<GreeterService>();
app.Run();

在控制檯專案Client中，我們利用生成出來的使用者端型別GreeterClient分別一對應的服務交換模式呼叫了四個gRPC方法。

var channel = GrpcChannel.ForAddress("http://localhost:5000");
var client = new GreeterClient(channel);

Console.WriteLine("Unary");
await UnaryCallAsync();

Console.WriteLine("\nServer Streaming");
await ServerStreamingCallAsync();

Console.WriteLine("\nClient Streaming");
await ClientStreamingCallAsync();

Console.WriteLine("\nDuplex Streaming");
await DuplexStreamingCallAsync();


Console.ReadLine();


async Task UnaryCallAsync()
{
    var request = new HelloRequest { Name = "foobar" };
    var reply = await client.SayHelloUnaryAsync(request);
    Console.WriteLine(reply.Message);
}


async Task ServerStreamingCallAsync()
{
    var streamingCall = client.SayHelloServerStreaming(new Empty());
    var reader = streamingCall.ResponseStream;
    while (await reader.MoveNext(CancellationToken.None))
    {
        Console.WriteLine(reader.Current.Message);
    }
}


async Task ClientStreamingCallAsync()
{
    var streamingCall = client.SayHelloClientStreaming();
    var writer = streamingCall.RequestStream;

    await writer.WriteAsync(new HelloRequest { Name = "Foo" });
    await Task.Delay(1000);
    await writer.WriteAsync(new HelloRequest { Name = "Bar" });
    await Task.Delay(1000);
    await writer.WriteAsync(new HelloRequest { Name = "Baz" });
    await writer.CompleteAsync();

    var reply = await streamingCall.ResponseAsync;
    Console.WriteLine(reply.Message);
}

async Task DuplexStreamingCallAsync()
{
    var streamingCall = client.SayHelloDuplexStreaming();
    var writer = streamingCall.RequestStream;
    var reader = streamingCall.ResponseStream;
    _ = Task.Run(async () =>
    {
        await writer.WriteAsync(new HelloRequest { Name = "Foo" });
        await Task.Delay(1000);
        await writer.WriteAsync(new HelloRequest { Name = "Bar" });
        await Task.Delay(1000);
        await writer.WriteAsync(new HelloRequest { Name = "Baz" });
        await writer.CompleteAsync();
    });
    await foreach (var reply in reader.ReadAllAsync())
    {
        Console.WriteLine(reply.Message);
    }
}

如下所示的是使用者端控制檯上的輸出結果。

image

二、將gRPC方法抽象成委託

通過上面的演示我們也知道，承載的gRPC型別最終會將其實現的方法註冊成路由終結點，這一點其實和MVC是一樣的。但是gRPC的方法和定義在Controller型別中的Action方法不同之處在於，前者的簽名其實是固定的。如果我們將請求和響應訊息型別使用Request和Reply來表示，四種訊息交換模式的方法簽名就可以寫成如下的形式。

Task<Reply> Unary(Request request, ServerCallContext context);
Task<Reply> ClientStreaming(IAsyncStreamReader<Request> reader, ServerCallContext context);
Task ServerStreaming(Empty request, IServerStreamWriter<Reply> responseStream, ServerCallContext context);
Task DuplexStreaming(IAsyncStreamReader<Request> reader, IServerStreamWriter<Reply> writer, ServerCallContext context);

「流式」方法中用來讀取請求和寫入響應的IAsyncStreamReader<T>和IServerStreamWriter<T>定義如下。

public interface IAsyncStreamReader<out T>
{
    T Current { get; }
    Task<bool> MoveNext(CancellationToken cancellationToken = default);
}


public interface IAsyncStreamWriter<in T>
{
    Task WriteAsync(T message, CancellationToken cancellationToken = default);
}

public interface IServerStreamWriter<in T> : IAsyncStreamWriter<T>
{
}

public interface IClientStreamWriter<in T> : IAsyncStreamWriter<T>
{
    Task CompleteAsync();
}

表示伺服器端呼叫上下文的ServerCallContext 型別具有豐富的成員，但是它的本質就是對HttpContext上下文的封裝，所以我們對它進行了簡化。如下面的程式碼片段所示，我們給予這個上下文型別兩個屬性成員，一個是表示請求上下文的HttpContext，另一個則是用來設定響應狀態StatusCode，後者對應的列舉定義了完整的gRPC狀態碼。

public class ServerCallContext
{
    public StatusCode StatusCode { get; set; } = StatusCode.OK;
    public HttpContext HttpContext { get; }
    public ServerCallContext(HttpContext httpContext)=> HttpContext = httpContext;
}

public enum StatusCode
{
    OK = 0,
    Cancelled = 1,
    Unknown = 2,
    InvalidArgument = 3,
    DeadlineExceeded = 4,
    NotFound = 5,
    AlreadyExists = 6,
    PermissionDenied = 7,
    Unauthenticated = 0x10,
    ResourceExhausted = 8,
    FailedPrecondition = 9,
    Aborted = 10,
    OutOfRange = 11,
    Unimplemented = 12,
    Internal = 13,
    Unavailable = 14,
    DataLoss = 0xF
}

既然方法簽名固定，意味著我們可以將四種gRPC方法定義成如下四個對應的委託，泛型引數TService、TRequest和TResponse分別表示服務、請求和響應型別。

public delegate Task<TResponse> UnaryMethod<TService, TRequest, TResponse>(TService service, TRequest request, ServerCallContext context)
    where TService : class
    where TRequest : IMessage<TRequest>
    where TResponse : IMessage<TResponse>;

public delegate Task<TResponse> ClientStreamingMethod<TService, TRequest, TResponse>(TService service, IAsyncStreamReader<TRequest> reader, ServerCallContext context)
    where TService : class
    where TRequest : IMessage<TRequest>
    where TResponse : IMessage<TResponse>;

public delegate Task ServerStreamingMethod<TService, TRequest, TResponse>(TService service, TRequest request, IServerStreamWriter<TResponse> writer, ServerCallContext context)
    where TService : class
    where TRequest : IMessage<TRequest>
    where TResponse : IMessage<TResponse>;

public delegate Task DuplexStreamingMethod<TService, TRequest, TResponse>(TService service, IAsyncStreamReader<TRequest> reader, IServerStreamWriter<TResponse> writer, ServerCallContext context)
    where TService : class
    where TRequest : IMessage<TRequest>
    where TResponse : IMessage<TResponse>;

我們知道路由的本質就是建立一組路由模式（Pattern）和對應處理器之間的對映關係。路由模式很簡單，對應的路由模板為「{ServiceName}/{MethodName}」，並且採用Post請求方法。對應的處理器最終體現為一個RequestDelegate。那麼只要我們能夠將上述四種委託型別都轉換成RequestDelegate委託，一切都迎刃而解了。

三、將委託轉換成RequestDelegate

為了將四種委託型別轉化成RequestDelegate，我們將後者實現為一個ServiceCallHandler型別，併為其定義瞭如下兩個基礎類別。ServerCallHandlerBase的HandleCallAsync方法正好與RequestDelegate委託的簽名一致，所以這個方法最終會用來處理gRPC請求。不同的訊息交換模式採用不同的請求處理方式，只需實現抽象方法HandleCallAsyncCore就可以了。HandleCallAsync方法在呼叫此抽象方法之前將響應的ContentType設定成gRPC標準的響應型別「application/grpc」。在此之後將狀態碼設定為「grpc-status」首部，它將在HTTP2的DATA幀傳送完畢後，以HEADERS幀傳送到使用者端。這兩項操作都是gRPC協定的一部分。

public abstract class ServerCallHandlerBase
{
    public async Task HandleCallAsync(HttpContext httpContext)
    {
        try
        {
            var serverCallContext = new ServerCallContext(httpContext);
            var response = httpContext.Response;
            response.ContentType = "application/grpc";
            await HandleCallAsyncCore(serverCallContext);
            SetStatus(serverCallContext.StatusCode);
        }
        catch
        {
            SetStatus(StatusCode.Unknown);
        }
        void SetStatus(StatusCode statusCode)
        {
            httpContext.Response.AppendTrailer("grpc-status", ((int)statusCode).ToString());
        }
    }
    protected abstract Task HandleCallAsyncCore(ServerCallContext serverCallContext);
}

public abstract class ServerCallHandler<TService, TRequest, TResponse> : ServerCallHandlerBase
    where TService : class
    where TRequest : IMessage<TRequest>
    where TResponse : IMessage<TResponse>
{
    protected ServerCallHandler(MessageParser<TRequest> requestParser)=> RequestParser = requestParser;
    public MessageParser<TRequest> RequestParser { get; }
}

ServerCallHandler<TService, TRequest, TResponse>派生自ServerCallHandlerBase，並利用三個泛型引數TService、TRequest、TResponse來表示服務、請求和響應型別，RequestParser用來提供發序列化請求訊息的MessageParser<TRequest>物件。針對四種訊息交換模式的ServiceCallHandler型別均繼承這個泛型基礎類別。

UnaryCallHandler

基於Unary訊息交換模式的ServerCallHandler的具體型別為UnaryCallHandler<TService, TRequest, TResponse>，它由上述的UnaryMethod<TService, TRequest, TResponse>委託構建而成。在重寫的HandleCallAsyncCore方法中，我們利用HttpContext提供的IServiceProvider物件將服務範例建立出來後，從請求主體中將請求訊息讀取出來，然後交給指定的委託物件進行處理並得到響應訊息，該響應訊息最終用來對當前請求予以回覆。

internal class UnaryCallHandler<TService, TRequest, TResponse> : ServerCallHandler<TService, TRequest, TResponse>
    where TService : class
    where TRequest : IMessage<TRequest>
    where TResponse : IMessage<TResponse>
{
    private readonly UnaryMethod<TService, TRequest, TResponse> _handler;

    public UnaryCallHandler(UnaryMethod<TService, TRequest, TResponse> handler, MessageParser<TRequest> requestParser):base(requestParser)
    => _handler = handler;
        protected override async Task HandleCallAsyncCore(ServerCallContext serverCallContext)
    {
        using var scope = serverCallContext.HttpContext.RequestServices.CreateScope();
        var service = ActivatorUtilities.CreateInstance<TService>(scope.ServiceProvider);
        var httpContext = serverCallContext.HttpContext;
        var request = await httpContext.Request.BodyReader.ReadSingleMessageAsync<TRequest>(RequestParser);
        var reply = await _handler(service, request!, serverCallContext);
        await httpContext.Response.BodyWriter.WriteMessageAsync(reply);
    }
}

請求訊息是通過如下這個ReadSingleMessageAsync<TMessage>方法讀取出來的。按照gRPC協定，通過網路傳輸的請求和響應訊息都會在前面追加5個位元組，第一個位元組表示訊息是否經過加密，後面四個位元組是一個以大端序表示的整數，表示訊息的長度。對於其他訊息交換模式，也是呼叫Buffers的TryReadMessage<TRequest>方法從緩衝區中讀取請求訊息。

public static async Task<TMessage> ReadSingleMessageAsync<TMessage>(this PipeReader reader, MessageParser<TMessage> parser) where TMessage:IMessage<TMessage>
{
    while (true)
    {
        var result = await reader.ReadAsync();
        var buffer = result.Buffer;
        if (Buffers.TryReadMessage(parser, ref buffer, out var message))
        {
            return message!;
        }
        reader.AdvanceTo(buffer.Start, buffer.End);
        if (result.IsCompleted)
        {
            break;
        }
    }
    throw new IOException("Fails to read message.");
}

internal static class Buffers
{
    public static readonly int HeaderLength = 5;
    public static bool TryReadMessage<TRequest>(MessageParser<TRequest> parser, ref ReadOnlySequence<byte> buffer, out TRequest? message) where TRequest: IMessage<TRequest>
    {
        if (buffer.Length < HeaderLength)
        {
            message = default;
            return false;
        }

        Span<byte> lengthBytes = stackalloc byte[4];
        buffer.Slice(1, 4).CopyTo(lengthBytes);
        var length = BinaryPrimitives.ReadInt32BigEndian(lengthBytes);
        if (buffer.Length < length + HeaderLength)
        {
            message = default;
            return false;
        }

        message = parser.ParseFrom(buffer.Slice(HeaderLength, length));
        buffer = buffer.Slice(length + HeaderLength);
        return true;
    }
}

如下這個WriteMessageAsync擴充套件方法負責輸出響應訊息。

public static ValueTask<FlushResult> WriteMessageAsync(this PipeWriter writer, IMessage message)
{
    var length = message.CalculateSize();
    var span = writer.GetSpan(5 + length);
    span[0] = 0;
    BinaryPrimitives.WriteInt32BigEndian(span.Slice(1, 4), length);
    message.WriteTo(span.Slice(5, length));
    writer.Advance(5 + length);
    return writer.FlushAsync();
}

ClientStreamingCallHandler

ClientStreamingCallHandler<TService, TRequest, TResponse>代表Client Streaming模式下的ServerCallHandler，它由對應的ClientStreamingMethod<TService, TRequest, TResponse>委託建立而成。在重寫的HandleCallAsyncCore方法中，除了服務範例，它還需要一個用來以「流」的方式讀取請求的IAsyncStreamReader<TRequest>物件，它們都將作為引數傳遞給指定的委託，後者執行後會返回最終的響應訊息。此訊息同樣通過上面這個WriteMessageAsync擴充套件方法予以回覆。

internal class ClientStreamingCallHandler<TService, TRequest, TResponse> : ServerCallHandler<TService, TRequest, TResponse>
    where TService : class
    where TRequest : IMessage<TRequest>
    where TResponse : IMessage<TResponse>
{
    private readonly ClientStreamingMethod<TService, TRequest, TResponse> _handler;
    public ClientStreamingCallHandler(ClientStreamingMethod<TService, TRequest, TResponse> handler, MessageParser<TRequest> requestParser)
        :base(requestParser)
    {
        _handler = handler;
    }
    protected override async Task HandleCallAsyncCore(ServerCallContext serverCallContext)
    {
        using var scope = serverCallContext.HttpContext.RequestServices.CreateScope();
        var service = ActivatorUtilities.CreateInstance<TService>(scope.ServiceProvider);
        var reader = serverCallContext.HttpContext.Request.BodyReader;
        var writer = serverCallContext.HttpContext.Response.BodyWriter;
        var streamReader = new HttpContextStreamReader<TRequest>(serverCallContext.HttpContext, RequestParser);
        var response = await _handler(service, streamReader, serverCallContext);
        await writer.WriteMessageAsync(response);
    }
}

IAsyncStreamReader<T>介面的實現型別為如下這個HttpContextStreamReader<T>。在瞭解了請求訊息在網路中的結構之後，對於實現在該型別中針對請求的讀取操作，應該不難理解。

public class HttpContextStreamReader<T> : IAsyncStreamReader<T> where T : IMessage<T>
{
    private readonly PipeReader _reader;
    private readonly MessageParser<T> _parser;
    private ReadOnlySequence<byte> _buffer;
    public HttpContextStreamReader(HttpContext httpContext, MessageParser<T> parser)
    {
        _reader = httpContext.Request.BodyReader;
        _parser = parser;
    }
    public T Current { get; private set; } = default!;
    public async Task<bool> MoveNext(CancellationToken cancellationToken)
    {
        var completed = false;
        if (_buffer.IsEmpty)
        {
            var result = await _reader.ReadAsync(cancellationToken);
            _buffer = result.Buffer;
            completed = result.IsCompleted;
        }
        if (Buffers.TryReadMessage(_parser, ref _buffer, out var mssage))
        {
            Current = mssage!;
            _reader.AdvanceTo(_buffer.Start, _buffer.End);
            return true;
        }
        _reader.AdvanceTo(_buffer.Start, _buffer.End);
        _buffer = default;
        return !completed && await MoveNext(cancellationToken);
    }
}

ServerStreamingCallHandler

ServerStreamingCallHandler<TService, TRequest, TResponse>代表Server Streaming模式下的ServerCallHandler，它由對應的ServerStreamingMethod<TService, TRequest, TResponse>委託建立而成。在重寫的HandleCallAsyncCore方法中，除了服務範例，它還需要一個用來以「流」的方式寫入響應的IAsyncStreamWriter<TResponse>物件，它們都將作為引數傳遞給指定的委託。

internal class ServerStreamingCallHandler<TService, TRequest, TResponse> : ServerCallHandler<TService, TRequest, TResponse>
    where TService : class
    where TRequest : IMessage<TRequest>
    where TResponse : IMessage<TResponse>
{
    private readonly ServerStreamingMethod<TService, TRequest, TResponse> _handler;
    public ServerStreamingCallHandler(ServerStreamingMethod<TService, TRequest, TResponse> handler, MessageParser<TRequest> requestParser):base(requestParser)
        => _handler = handler;
    protected override async Task HandleCallAsyncCore(ServerCallContext serverCallContext)
    {
        using var scope = serverCallContext.HttpContext.RequestServices.CreateScope();
        var service = ActivatorUtilities.CreateInstance<TService>(scope.ServiceProvider);
        var httpContext = serverCallContext.HttpContext;
        var streamWriter = new HttpContextStreamWriter<TResponse>(httpContext);
        var request = await httpContext.Request.BodyReader.ReadSingleMessageAsync(RequestParser);
        await _handler(service, request, streamWriter, serverCallContext);
    }
}

IAsyncStreamWriter<T>介面的實現型別為如下這個HttpContextStreamWriter<T>，它直接呼叫上面定義的WriteMessageAsync擴充套件方法將指定的訊息寫入響應主體的輸出流。

public class HttpContextStreamWriter<T> : IServerStreamWriter<T> where T : IMessage<T>
{
    private readonly PipeWriter _writer;
    public HttpContextStreamWriter(HttpContext httpContext) => _writer = httpContext.Response.BodyWriter;
    public Task WriteAsync(T message, CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        return _writer.WriteMessageAsync(message).AsTask();
    }
}

DuplexStreamingCallHandler

DuplexStreamingCallHandler<TService, TRequest, TResponse>代表Duplex Streaming模式下的ServerCallHandler，它由對應的DuplexStreamingMethod<TService, TRequest, TResponse>委託建立而成。在重寫的HandleCallAsyncCore方法中，除了服務範例，它還需要分別建立以「流」的方式讀/寫請求/響應的IAsyncStreamReader<TRequest>和IAsyncStreamWriter<TResponse>物件，對應的型別分別為上面定義的HttpContextStreamReader<TRequest>和HttpContextStreamWriter<TResponse>。

internal class DuplexStreamingCallHandler<TService, TRequest, TResponse> : ServerCallHandler<TService, TRequest, TResponse>
    where TService : class
    where TRequest : IMessage<TRequest>
    where TResponse : IMessage<TResponse>
{
    private readonly DuplexStreamingMethod<TService, TRequest, TResponse> _handler;
    public DuplexStreamingCallHandler(DuplexStreamingMethod<TService, TRequest, TResponse> handler, MessageParser<TRequest> requestParser) :base(requestParser)
        => _handler = handler;
    protected override async Task HandleCallAsyncCore(ServerCallContext serverCallContext)
    {
        using var scope = serverCallContext.HttpContext.RequestServices.CreateScope();
        var service = ActivatorUtilities.CreateInstance<TService>(scope.ServiceProvider);
        var reader = serverCallContext.HttpContext.Request.BodyReader;
        var writer = serverCallContext.HttpContext.Response.BodyWriter;
        var streamReader = new HttpContextStreamReader<TRequest>(serverCallContext.HttpContext, RequestParser);
        var streamWriter = new HttpContextStreamWriter<TResponse>(serverCallContext.HttpContext);
        await _handler(service, streamReader, streamWriter, serverCallContext);
    }
}

四、路由註冊

目前我們將針對四種訊息交換模式的gRPC方法抽象成對應的泛型委託，並且可以利用它們建立ServerCallHandler，後者可以提供作為路由終結點處理器的RequestDelegate委託。列舉和對應ServerCallHandler之間的對映關係如下所示：

  • UnaryMethod<TService, TRequest, TResponse>：UnaryCallHandler<TService, TRequest, TResponse>
  • ClientStreamingMethod<TService, TRequest, TResponse>：ClientStreamingCallHandler<TService, TRequest, TResponse>
  • ServerStreamingMethod<TService, TRequest, TResponse>：ServerStreamingCallHandler<TService, TRequest, TResponse>
  • DuplexStreamingMethod<TService, TRequest, TResponse>：DuplexStreamingCallHandler<TService, TRequest, TResponse>

現在我們將整個路由註冊的流程串起來，為此我們定義瞭如下這個IServiceBinder<TService>介面，它提供了兩種方式將定義在服務型別TService中的gRPC方法註冊成對應的路由終結點。

public interface IServiceBinder<TService> where TService : class
{
    IServiceBinder<TService> AddUnaryMethod<TRequest, TResponse>(string methodName, Func<TService, Func<TRequest, ServerCallContext, Task<TResponse>>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>;

    IServiceBinder<TService> AddClientStreamingMethod<TRequest, TResponse>(string methodName, Func<TService, Func<IAsyncStreamReader<TRequest>, ServerCallContext, Task<TResponse>>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>;

    IServiceBinder<TService> AddServerStreamingMethod<TRequest, TResponse>(string methodName, Func<TService, Func<TRequest, IServerStreamWriter<TResponse>, ServerCallContext, Task>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>;

    IServiceBinder<TService> AddDuplexStreamingMethod<TRequest, TResponse>(string methodName, Func<TService, Func<IAsyncStreamReader<TRequest>, IServerStreamWriter<TResponse>, ServerCallContext, Task>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>;


    IServiceBinder<TService> AddUnaryMethod<TRequest, TResponse>(Expression<Func<TService, Task<TResponse>>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>;
    IServiceBinder<TService> AddClientStreamingMethod<TRequest, TResponse>( Expression<Func<TService, Task<TResponse>>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>;

    IServiceBinder<TService> AddServerStreamingMethod<TRequest, TResponse>( Expression<Func<TService, Task>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>;

    IServiceBinder<TService> AddDuplexStreamingMethod<TRequest, TResponse>( Expression<Func<TService, Task>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>;
}

路由終結點由路由模式和處理器兩個元素組成，路由模式主要體現在由gRPC服務和操作名稱組成的路由模板，我們預設使用服務型別的名稱和方法名稱（提出Async字尾）。為了能夠對這兩個名稱進行客製化，我們定義瞭如下兩個特性GrpcServiceAttribute和GrpcMethodAttribute，它們可以分別標註在服務型別和操作方法上來指定一個任意的名稱。

[AttributeUsage(AttributeTargets.Class)]
public class GrpcServiceAttribute: Attribute
{
    public string? ServiceName { get; set; }

}

[AttributeUsage(AttributeTargets.Method)]
public class GrpcMethodAttribute : Attribute
{
    public string? MethodName { get; set; }
}

如下所示的ServiceBinder<TService> 是對IServiceBinder<TService> 介面的實現，它是對一個IEndpointRouteBuilder 物件的封裝。對於實現的第一組方法，我們利用提供的方法名稱與解析TService型別得到的服務名稱合併，進而得到路由終結點的URL模板。這些方法還提供了一個針對gRPC方法簽名的Func<TService,Func<…>>委託，我們利用它來將提供用於構建對應ServiceCallHandler的委託。我們最終利用IEndpointRouteBuilder 物件完成針對路由終結點的註冊。

public class ServiceBinder<TService> : IServiceBinder<TService> where TService : class
{
    private readonly IEndpointRouteBuilder _routeBuilder;
    public ServiceBinder(IEndpointRouteBuilder routeBuilder) => _routeBuilder = routeBuilder;

    public IServiceBinder<TService> AddUnaryMethod<TRequest, TResponse>(string methodName, Func<TService, Func<TRequest, ServerCallContext, Task<TResponse>>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>
    {
        Task<TResponse> GetMethod(TService service, TRequest request, ServerCallContext context) => methodAccessor(service)(request, context);
        var callHandler = new UnaryCallHandler<TService, TRequest, TResponse>(GetMethod, parser);
        _routeBuilder.MapPost(ServiceBinder<TService>.GetPath(methodName), callHandler.HandleCallAsync);
        return this;
    }

    public IServiceBinder<TService> AddClientStreamingMethod<TRequest, TResponse>(string methodName, Func<TService, Func<IAsyncStreamReader<TRequest>, ServerCallContext, Task<TResponse>>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>
    {
        Task<TResponse> GetMethod(TService service, IAsyncStreamReader<TRequest> reader, ServerCallContext context) => methodAccessor(service)(reader, context);
        var callHandler = new ClientStreamingCallHandler<TService, TRequest, TResponse>(GetMethod, parser);
        _routeBuilder.MapPost(ServiceBinder<TService>.GetPath(methodName), callHandler.HandleCallAsync);
        return this;
    }

    public IServiceBinder<TService> AddServerStreamingMethod<TRequest, TResponse>(string methodName, Func<TService, Func<TRequest, IServerStreamWriter<TResponse>, ServerCallContext, Task>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>
    {
        ServerStreamingMethod<TService, TRequest, TResponse> handler = (service, request, writer, context) => methodAccessor(service)(request, writer, context);
        var callHandler = new ServerStreamingCallHandler<TService, TRequest, TResponse>(handler, parser);
        _routeBuilder.MapPost(ServiceBinder<TService>.GetPath(methodName), callHandler.HandleCallAsync);
        return this;
    }

    public IServiceBinder<TService> AddDuplexStreamingMethod<TRequest, TResponse>(string methodName, Func<TService, Func<IAsyncStreamReader<TRequest>, IServerStreamWriter<TResponse>, ServerCallContext, Task>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>
    {
        DuplexStreamingMethod<TService, TRequest, TResponse> handler = (service, reader, writer, context) => methodAccessor(service)(reader, writer, context);
        var callHandler = new DuplexStreamingCallHandler<TService, TRequest, TResponse>(handler, parser);
        _routeBuilder.MapPost(ServiceBinder<TService>.GetPath(methodName), callHandler.HandleCallAsync);
        return this;
    }

    private static string GetPath(string methodName)
    {
        var serviceName = typeof(TService).GetCustomAttribute<GrpcServiceAttribute>()?.ServiceName ?? typeof(TService).Name;
        if (methodName.EndsWith("Async"))
        {
            methodName = methodName.Substring(0, methodName.Length - 5);
        }
        return $"{serviceName}/{methodName}";
    }

    public IServiceBinder<TService> AddUnaryMethod<TRequest, TResponse>(Expression<Func<TService, Task<TResponse>>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>
    {
        var method = CreateDelegate<UnaryMethod<TService, TRequest,TResponse>>(methodAccessor, out var methodName);
        var serviceName = typeof(TService).GetCustomAttribute<GrpcServiceAttribute>()?.ServiceName ?? typeof(TService).Name;
        var callHandler = new UnaryCallHandler<TService, TRequest, TResponse>(method, parser);
        _routeBuilder.MapPost(ServiceBinder<TService>.GetPath(methodName), callHandler.HandleCallAsync);
        return this;
    }

    public IServiceBinder<TService> AddClientStreamingMethod<TRequest, TResponse>( Expression<Func<TService, Task<TResponse>>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>
    {
        var method = CreateDelegate<ClientStreamingMethod<TService, TRequest, TResponse>>(methodAccessor, out var methodName);
        var serviceName = typeof(TService).GetCustomAttribute<GrpcServiceAttribute>()?.ServiceName ?? typeof(TService).Name;
        var callHandler = new ClientStreamingCallHandler<TService, TRequest, TResponse>(method, parser);
        _routeBuilder.MapPost(ServiceBinder<TService>.GetPath(methodName), callHandler.HandleCallAsync);
        return this;
    }

    public IServiceBinder<TService> AddServerStreamingMethod<TRequest, TResponse>(Expression<Func<TService, Task>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>
    {
        var method = CreateDelegate<ServerStreamingMethod<TService, TRequest, TResponse>>(methodAccessor, out var methodName);
        var serviceName = typeof(TService).GetCustomAttribute<GrpcServiceAttribute>()?.ServiceName ?? typeof(TService).Name;
        var callHandler = new ServerStreamingCallHandler<TService, TRequest, TResponse>(method, parser);
        _routeBuilder.MapPost(ServiceBinder<TService>.GetPath(methodName), callHandler.HandleCallAsync);
        return this;
    }

    public IServiceBinder<TService> AddDuplexStreamingMethod<TRequest, TResponse>(Expression<Func<TService, Task>> methodAccessor, MessageParser<TRequest> parser)
        where TRequest : IMessage<TRequest>
        where TResponse : IMessage<TResponse>
    {
        var method = CreateDelegate<DuplexStreamingMethod<TService, TRequest, TResponse>>(methodAccessor, out var methodName);
        var serviceName = typeof(TService).GetCustomAttribute<GrpcServiceAttribute>()?.ServiceName ?? typeof(TService).Name;
        var callHandler = new DuplexStreamingCallHandler<TService, TRequest, TResponse>(method, parser);
        _routeBuilder.MapPost(ServiceBinder<TService>.GetPath(methodName), callHandler.HandleCallAsync);
        return this;
    }

    private TDelegate CreateDelegate<TDelegate>(LambdaExpression expression, out string methodName) where TDelegate : Delegate
    {
        var method = ((MethodCallExpression)expression.Body).Method;
        methodName = method.GetCustomAttribute<GrpcMethodAttribute>()?.MethodName ?? method.Name;
        return (TDelegate)Delegate.CreateDelegate(typeof(TDelegate), method);
    }
}

由於第二組方法提供的針對gRPC方法呼叫的表示式，所以我們可以得到描述方法的MethodInfo物件，該物件不但解決了委託物件的建立問題，還可以提供方法的名稱，所以這組方法無需提供gRPC方法的名稱。但是提供的表示式並不能嚴格匹配方法的簽名，所以無法提供編譯時的錯誤檢驗，所以各有優缺點。

五、為gRPC服務定義一個介面

由於路由終結點的註冊是針對服務型別進行的，所以我們決定讓服務型別自身來完成所有的路由註冊工作。在這裡我們使用C# 11中一個叫做「靜態介面方法」的特性，為服務型別定義如下這個IGrpcService<TService>介面，服務型別TService定義的所有gRPC方法的路由註冊全部在靜態方法Bind中完成，該方法將上述的IServiceBinder<TService>作為引數。

public interface  IGrpcService<TService> where TService:class
{
     static abstract void Bind(IServiceBinder<TService> binder);
}

我們定義瞭如下這個針對IEndpointRouteBuilder 介面的擴充套件方法完成針對指定服務型別的路由註冊。為了與現有的方法區別開來，我特意將其命名為MapGrpcService2。該方法根據指定的IEndpointRouteBuilder 物件將ServiceBinder<TService>物件建立出來，並作為引數呼叫服務型別的靜態Bind方法。到此為止，整個Mini版的gRPC伺服器端框架就構建完成了，接下來我們看看它能否工作。

public static class EndpointRouteBuilderExtensions
{
    public static IEndpointRouteBuilder MapGrpcService2<TService>(this IEndpointRouteBuilder routeBuilder) where TService : class, IGrpcService<TService>
    {

        var binder = new ServiceBinder<TService>(routeBuilder);
        TService.Bind(binder);
        return routeBuilder;
    }
}

六、重新定義和承載服務

我們開篇演示了ASP.NET Core gRPC的服務定義、承載和呼叫。如果我們上面構建的Mini版gRPC框架能夠正常工作，意味著使用者端程式碼可以保持不變，我們現在就來試試看。我們在Server專案中將GreeterService服務型別改成如下的形式，它不再繼承任何基礎類別，只實現IGrpcService<GreeterService>介面。針對四種訊息交換模式的四個方法的實現方法保持不變，在實現的靜態Bind方法中，我們採用兩種形式完成了針對這四個方法的路由註冊。

[GrpcService(ServiceName = "Greeter")]
public class GreeterService: IGrpcService<GreeterService>
{
    public Task<HelloReply> SayHelloUnaryAsync(HelloRequest request, ServerCallContext context)
    => Task.FromResult(new HelloReply { Message = $"Hello, {request.Name}" });

    public async Task<HelloReply> SayHelloClientStreamingAsync(IAsyncStreamReader<HelloRequest> reader, ServerCallContext context)
    {
        var list = new List<string>();
        while (await reader.MoveNext(CancellationToken.None))
        {
            list.Add(reader.Current.Name);
        }
        return new HelloReply { Message = $"Hello, {string.Join(",", list)}" };
    }

    public  async Task SayHelloServerStreamingAsync(Empty request, IServerStreamWriter<HelloReply> responseStream, ServerCallContext context)
    {
        await responseStream.WriteAsync(new HelloReply { Message = "Hello, Foo!" });
        await Task.Delay(1000);
        await responseStream.WriteAsync(new HelloReply { Message = "Hello, Bar!" });
        await Task.Delay(1000);
        await responseStream.WriteAsync(new HelloReply { Message = "Hello, Baz!" });
    }

    public async Task SayHelloDuplexStreamingAsync(IAsyncStreamReader<HelloRequest> reader, IServerStreamWriter<HelloReply> writer, ServerCallContext context)
    {
        while (await reader.MoveNext())
        {
            await writer.WriteAsync(new HelloReply { Message = $"Hello {reader.Current.Name}" });
        }
    }

        public static void Bind(IServiceBinder<GreeterService> binder)
        {
            binder
               .AddUnaryMethod<HelloRequest, HelloReply>(it =>it.SayHelloUnaryAsync(default!,default!), HelloRequest.Parser)
               .AddClientStreamingMethod<HelloRequest, HelloReply>(it => it.SayHelloClientStreamingAsync(default!, default!), HelloRequest.Parser)

               .AddServerStreamingMethod<Empty, HelloReply>(nameof(SayHelloServerStreamingAsync), it => it.SayHelloServerStreamingAsync, Empty.Parser)
               .AddDuplexStreamingMethod<HelloRequest, HelloReply>(nameof(SayHelloDuplexStreamingAsync), it => it.SayHelloDuplexStreamingAsync, HelloRequest.Parser);
        }
    }

服務承載程式直接將針對MapGrpcService<GreeterService>方法的呼叫換成MapGrpcService2<GreeterService>。由於整個框架根本不需要預先註冊任何的服務，所以針對AddGrpc擴充套件方法的呼叫也可以刪除。

using GrpcMini;
using Microsoft.AspNetCore.Server.Kestrel.Core;

var builder = WebApplication.CreateBuilder(args);
builder.WebHost.ConfigureKestrel(kestrel => kestrel.ConfigureEndpointDefaults(options => options.Protocols = HttpProtocols.Http2));
var app = builder.Build();
app.MapGrpcService2<Server.Greeter>();
app.Run();

再次執行我們的程式，使用者端依然可以得到相同的輸出。

image