grpc-dotnet客户端流：如何优雅感知服务器提前完成调用？

客户端通过grpc-dotnet发起客户端流式调用上传文件，部分场景下服务器会忽略客户端流直接返回响应。客户端不知情的情况下继续调用RequestStream.WriteAsync()，最终抛出Grpc.Core.RpcException: 'Status(StatusCode="OK", Detail="")'异常。

核心疑问：有没有优雅的方式提前感知调用已完成，避免无效写入？

我翻查grpc-dotnet源码后，发现可通过反射访问AsyncClientStreamingCall.RequestStream.Call.ResponseFinished标记解决，但这种方式比捕获已知StatusCode的异常更不可取。

客户端伪代码 #

private static async Task CallUploadStream(GrpcService.GrpcServiceClient client)
{
    using var streamingCall = client.UploadStream();

    // First call
    await WriteAsync();

    // Delay
    await Task.Delay(TimeSpan.FromSeconds(1));

    // Second call
    await WriteAsync(); // Grpc.Core.RpcException: 'Status(StatusCode="OK", Detail="")'

    await streamingCall.RequestStream.CompleteAsync();
    await streamingCall;

    return;

    Task WriteAsync()
    {
        return streamingCall.RequestStream.WriteAsync(
            new UploadStreamRequest
            {
                Bytes = ByteString.CopyFrom(new byte[1]),
            }
        );
    }
}

Protobuf契约 #

syntax = "proto3";
package grpc.debug.contract.v1;
option csharp_namespace = "GrpcDebug.Contract";
import "google/protobuf/empty.proto";

service GrpcService {
  rpc UploadStream(stream UploadStreamRequest) returns (google.protobuf.Empty);
}

message UploadStreamRequest { bytes bytes = 1; }

立即返回响应的服务器代码 #

public class GrpcServiceV1 : GrpcService.GrpcServiceBase
{
    public override Task<Empty> UploadStream(IAsyncStreamReader<UploadStreamRequest> requestStream,
        ServerCallContext context)
    {
        return Task.FromResult(new Empty());
    }
}

1. 监听调用完成状态（推荐） #

利用AsyncClientStreamingCall的完成任务，提前启动监听逻辑标记调用状态，写入前检查状态避免无效操作。这种方式能主动感知调用完成，无需依赖异常捕获：

private static async Task CallUploadStream(GrpcService.GrpcServiceClient client)
{
    using var streamingCall = client.UploadStream();
    var callCompleted = new TaskCompletionSource<bool>();
    
    // 后台监听调用完成状态
    _ = Task.Run(async () =>
    {
        try
        {
            await streamingCall;
        }
        finally
        {
            callCompleted.TrySetResult(true);
        }
    });

    // 第一次写入
    await WriteAsync();

    await Task.Delay(TimeSpan.FromSeconds(1));

    // 写入前检查调用是否已完成
    if (!callCompleted.Task.IsCompleted)
    {
        await WriteAsync();
    }
    else
    {
        Console.WriteLine("调用已完成，跳过后续写入");
    }

    try
    {
        if (!callCompleted.Task.IsCompleted)
        {
            await streamingCall.RequestStream.CompleteAsync();
        }
        await streamingCall;
    }
    catch (RpcException ex) when (ex.Status.StatusCode == StatusCode.OK)
    {
        // 处理服务器提前返回的正常完成情况
    }

    return;

    Task WriteAsync()
    {
        return streamingCall.RequestStream.WriteAsync(
            new UploadStreamRequest
            {
                Bytes = ByteString.CopyFrom(new byte[1]),
            }
        );
    }
}

2. 捕获特定异常并终止写入 #

既然已知触发的是StatusCode.OK的RpcException，可以封装写入逻辑，捕获该异常后标记调用完成，后续写入直接跳过。这种方式实现简单，无需额外监听任务：

private static async Task CallUploadStream(GrpcService.GrpcServiceClient client)
{
    using var streamingCall = client.UploadStream();
    bool callIsCompleted = false;

    async Task SafeWriteAsync()
    {
        if (callIsCompleted) return;
        try
        {
            await streamingCall.RequestStream.WriteAsync(
                new UploadStreamRequest
                {
                    Bytes = ByteString.CopyFrom(new byte[1]),
                }
            );
        }
        catch (RpcException ex) when (ex.Status.StatusCode == StatusCode.OK)
        {
            callIsCompleted = true;
        }
    }

    // 第一次写入
    await SafeWriteAsync();

    await Task.Delay(TimeSpan.FromSeconds(1));

    // 第二次写入，若调用已完成则自动跳过
    await SafeWriteAsync();

    try
    {
        if (!callIsCompleted)
        {
            await streamingCall.RequestStream.CompleteAsync();
        }
        await streamingCall;
    }
    catch (RpcException ex) when (ex.Status.StatusCode == StatusCode.OK)
    {
        // 正常处理提前返回场景
    }
}

3. 服务器端优化（从根源规避） #

如果有权限修改服务器代码，建议不要直接返回响应，至少读取一个客户端请求后再返回，避免客户端收到过早的完成信号：

public override async Task<Empty> UploadStream(IAsyncStreamReader<UploadStreamRequest> requestStream,
    ServerCallContext context)
{
    // 至少读取一个请求，确保客户端有足够时间发送数据
    if (await requestStream.MoveNext(context.CancellationToken))
    {
        // 可在此处处理第一个请求的逻辑
    }
    return new Empty();
}

注：若服务器确实需要立即返回响应，客户端侧的处理仍是必要的，因为客户端流式调用中服务器先返回后，客户端无法收到额外的通知信号。

内容的提问来源于stack exchange，提问作者Oleksii Lotysh