将服务器流式读取到客户端 c#

Question

我正在使用服务器流式处理应用程序。

问题是客户端没有从服务器流中读取数据。

这是我的proto 服务：

service UserService {
    rpc GetData(Id) returns (stream DataResponse) {}
}
message Id {
    int32 id = 1;
}

message DataResponse {
    bytes data = 1;
}

c#服务器是这样的：

public override async Task GetData(Id request, IServerStreamWriter<DataResponse> response, ServerCallContext context)
{
    var user = {} // get user
    foreach (var d in user.Data)
    {
        await response.WriteAsync(new DataResponse { Data = d });
    }
}

它之所以有效，是因为我有一个 NodeJS 客户端，我可以在其中调用服务器并完美运行。

Node 中的客户端是

let call = client.getData({id:1})
call.on('data', function (response) {
    // do things
})
call.on('end', function () {
    // do things
})

而c#客户端是：

AsyncServerStreamingCall<DataResponse> response = client.GetData(new Id{Id_ = 1});
while(await response.ResponseStream.MoveNext())
{
    Console.WriteLine("Into while loop"); // <-- This is never executed
    DataResponse current = response.ResponseStream.Current;
    Console.WriteLine($"{current.Data}");
}

我还添加了一个try/catch，它不输出任何内容，所以看起来MoveNext() 总是false。

这里有什么问题？为什么 NodeJS 客户端可以工作而 c# 客户端无法读取流？我错过了什么吗？

这是完整的 client.cs 类：

class Program
{
    const int Port = 50051;
    static void Main(string[] args)
    {
        try
        {
            Channel channel = new Channel("127.0.0.1:50051", ChannelCredentials.Insecure);
            var client = new UserService.UserServiceClient(channel);
            GetDataStreaming(client);
        }
        catch (RpcException ex)
        {
            Console.WriteLine($"Error: {{Code: {ex.StatusCode}, Status: {ex.Status.Detail}}}");
        }
    }

    private static async void GetDataStreaming(UserService.UserServiceClient client)
    {
        
            AsyncServerStreamingCall<DataResponse> response = client.GetData(new Id { Id_ = 1 });
            while (await response.ResponseStream.MoveNext())
            {
                Console.WriteLine("Into while loop");
                DataResponse current = response.ResponseStream.Current;
                Console.WriteLine($"{current.Data.ToStringUtf8()}");
            }

    }

}

Answer 1

问题是您的客户端在客户端收到响应之前已经结束。当您在 Main 中调用 GetDataStreaming(client) 时，它不会等待并完成。

要解决此问题，请将 async void GetDataStreaming 更改为 async Task GetDataStreaming。

private static async Task GetDataStreaming(UserService.UserServiceClient client)
{        
        AsyncServerStreamingCall<DataResponse> response = client.GetData(new Id { Id_ = 1 });
        while (await response.ResponseStream.MoveNext())
        {
            Console.WriteLine("Into while loop");
            DataResponse current = response.ResponseStream.Current;
            Console.WriteLine($"{current.Data.ToStringUtf8()}");
        }
}

把static void Main改成static async Task Main，最后你也应该调用channel.ShutdownAsync方法。

static async Task Main(string[] args)
{
    try
    {
        Channel channel = new Channel("127.0.0.1:50051", ChannelCredentials.Insecure);
        var client = new UserService.UserServiceClient(channel);
        await GetDataStreaming(client);
        await channel.ShutdownAsync();
    }
    catch (RpcException ex)
    {
        Console.WriteLine($"Error: {{Code: {ex.StatusCode}, Status: {ex.Status.Detail}}}");
    }
}

另一种选择是将async void GetDataStreaming 更改为async Task GetDataStreaming 并在Main 方法中等到Task 完成。

static void Main(string[] args)
{
    try
    {
        Channel channel = new Channel("127.0.0.1:50051", ChannelCredentials.Insecure);
        var client = new UserService.UserServiceClient(channel);
        var task = GetDataStreaming(client);
        task.Wait();
    }
    catch (RpcException ex)
    {
        Console.WriteLine($"Error: {{Code: {ex.StatusCode}, Status: {ex.Status.Detail}}}");
    }
}