[C#/COMMON/GRPC/.NET6] GRPC 스트리밍 사용하기

■ GRPC 스트리밍을 사용하는 방법을 보여준다.

[TestServer 프로젝트]

▶ weather.proto


syntax = "proto3";

import "google/protobuf/empty.proto";
import "google/protobuf/timestamp.proto";

option csharp_namespace = "TestServer";

package weather;

// 일기예보 프로토
service WeatherForecastProto
{
    rpc GetWeatherStream (google.protobuf.Empty) returns (stream WeatherForecastModel);
}

// 일기예보 모델
message WeatherForecastModel
{
    google.protobuf.Timestamp dateTimeStamp = 1;
    int32 temperatureC = 2;
    int32 temperatureF = 3;
    string summary = 4;
}

syntax = "proto3";

import "google/protobuf/empty.proto";

import "google/protobuf/timestamp.proto";

option csharp_namespace = "TestServer";

package weather;

// 일기예보 프로토

service WeatherForecastProto

{

rpc GetWeatherStream (google.protobuf.Empty) returns (stream WeatherForecastModel);

}

// 일기예보 모델

message WeatherForecastModel

{

google.protobuf.Timestamp dateTimeStamp = 1;

int32 temperatureC = 2;

int32 temperatureF = 3;

string summary = 4;

}

▶ WeatherForecastService.cs


using Google.Protobuf.WellKnownTypes;

using Grpc.Core;

namespace TestServer.Services
{
    /// <summary>
    /// 일기예보 서비스
    /// </summary>
    public class WeatherForecastService : WeatherForecastProto.WeatherForecastProtoBase
    {
        //////////////////////////////////////////////////////////////////////////////////////////////////// Field
        ////////////////////////////////////////////////////////////////////////////////////////// Private

        #region Field

        /// <summary>
        /// 로거
        /// </summary>
        private readonly ILogger<WeatherForecastService> logger;

        #endregion

        //////////////////////////////////////////////////////////////////////////////////////////////////// Constructor
        ////////////////////////////////////////////////////////////////////////////////////////// Public

        #region 생성자 - WeatherForecastService(logger)

        /// <summary>
        /// 생성자
        /// </summary>
        /// <param name="logger">로거</param>
        public WeatherForecastService(ILogger<WeatherForecastService> logger) => this.logger = logger;

        #endregion

        //////////////////////////////////////////////////////////////////////////////////////////////////// Method
        ////////////////////////////////////////////////////////////////////////////////////////// Public

        #region 기상 스트림 구하기 - GetWeatherStream(_, responseStream, context)

        /// <summary>
        /// 기상 스트림 구하기
        /// </summary>
        /// <param name="_">_</param>
        /// <param name="responseStream">응답 스트림</param>
        /// <param name="context">컨텍스트</param>
        /// <returns>태스크</returns>
        public override async Task GetWeatherStream(Empty _, IServerStreamWriter<WeatherForecastModel> responseStream, ServerCallContext context)
        {
            Random random = new Random();

            DateTime currentTime = DateTime.UtcNow;

            int i = 0;

            while(!context.CancellationToken.IsCancellationRequested && i < 20)
            {
                await Task.Delay(500);

                WeatherForecastModel weatherForecast = new WeatherForecastModel
                {
                    DateTimeStamp = Timestamp.FromDateTime(currentTime.AddDays(i++)),
                    TemperatureC  = random.Next(-20, 55),
                    Summary       = string.Empty
                };

                logger.LogInformation("Sending WeatherData response");

                await responseStream.WriteAsync(weatherForecast);
            }
        }

        #endregion
    }
}

using Google.Protobuf.WellKnownTypes;

using Grpc.Core;

namespace TestServer.Services

{

/// <summary>

/// 일기예보 서비스

/// </summary>

public class WeatherForecastService : WeatherForecastProto.WeatherForecastProtoBase

{

//////////////////////////////////////////////////////////////////////////////////////////////////// Field

////////////////////////////////////////////////////////////////////////////////////////// Private

#region Field

/// <summary>

/// 로거

/// </summary>

private readonly ILogger<WeatherForecastService> logger;

#endregion

//////////////////////////////////////////////////////////////////////////////////////////////////// Constructor

////////////////////////////////////////////////////////////////////////////////////////// Public

#region 생성자 - WeatherForecastService(logger)

/// <summary>

/// 생성자

/// </summary>

/// <param name="logger">로거</param>

public WeatherForecastService(ILogger<WeatherForecastService> logger) => this.logger = logger;

#endregion

//////////////////////////////////////////////////////////////////////////////////////////////////// Method

////////////////////////////////////////////////////////////////////////////////////////// Public

#region 기상 스트림 구하기 - GetWeatherStream(_, responseStream, context)

/// <summary>

/// 기상 스트림 구하기

/// </summary>

/// <param name="_">_</param>

/// <param name="responseStream">응답 스트림</param>

/// <param name="context">컨텍스트</param>

/// <returns>태스크</returns>

public override async Task GetWeatherStream(Empty _, IServerStreamWriter<WeatherForecastModel> responseStream, ServerCallContext context)

{

Random random = new Random();

DateTime currentTime = DateTime.UtcNow;

int i = 0;

while(!context.CancellationToken.IsCancellationRequested && i < 20)

{

await Task.Delay(500);

WeatherForecastModel weatherForecast = new WeatherForecastModel

{

DateTimeStamp = Timestamp.FromDateTime(currentTime.AddDays(i++)),

TemperatureC = random.Next(-20, 55),

Summary = string.Empty

};

logger.LogInformation("Sending WeatherData response");

await responseStream.WriteAsync(weatherForecast);

}

#endregion

}

▶ Program.cs


using TestServer.Services;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddGrpc();

var app = builder.Build();

app.UseRouting();

app.UseEndpoints
(
    endpoints =>
    {
        endpoints.MapGrpcService<WeatherForecastService>();
    }
);

app.Run();

using TestServer.Services;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddGrpc();

var app = builder.Build();

app.UseRouting();

app.UseEndpoints

(

endpoints =>

{

endpoints.MapGrpcService<WeatherForecastService>();

}

);

app.Run();

[TestClient 프로젝트]

▶ weather.proto


syntax = "proto3";

import "google/protobuf/empty.proto";
import "google/protobuf/timestamp.proto";

option csharp_namespace = "TestClient";

package weather;

// 일기예보 프로토
service WeatherForecastProto
{
    rpc GetWeatherStream (google.protobuf.Empty) returns (stream WeatherForecastModel);
}

// 일기예보 모델
message WeatherForecastModel
{
    google.protobuf.Timestamp dateTimeStamp = 1;
    int32 temperatureC = 2;
    int32 temperatureF = 3;
    string summary = 4;
}

syntax = "proto3";

import "google/protobuf/empty.proto";

import "google/protobuf/timestamp.proto";

option csharp_namespace = "TestClient";

package weather;

// 일기예보 프로토

service WeatherForecastProto

{

rpc GetWeatherStream (google.protobuf.Empty) returns (stream WeatherForecastModel);

}

// 일기예보 모델

message WeatherForecastModel

{

google.protobuf.Timestamp dateTimeStamp = 1;

int32 temperatureC = 2;

int32 temperatureF = 3;

string summary = 4;

}

▶ Program.cs


using Google.Protobuf.WellKnownTypes;

using Grpc.Core;
using Grpc.Net.Client;

using TestClient;

using var channel = GrpcChannel.ForAddress("http://localhost:5038");

var client = new WeatherForecastProto.WeatherForecastProtoClient(channel);

var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));

using var weatherCall = client.GetWeatherStream(new Empty(), cancellationToken : cts.Token);

try
{
    await foreach(WeatherForecastModel weatherForecast in weatherCall.ResponseStream.ReadAllAsync(cancellationToken : cts.Token))
    {
        Console.WriteLine($"{weatherForecast.DateTimeStamp.ToDateTime():s} | {weatherForecast.Summary} | {weatherForecast.TemperatureC} C");
    }
}
catch(RpcException exception) when (exception.StatusCode == StatusCode.Cancelled)
{
    Console.WriteLine("스트리밍 취소.");
}

using Google.Protobuf.WellKnownTypes;

using Grpc.Core;

using Grpc.Net.Client;

using TestClient;

using var channel = GrpcChannel.ForAddress("http://localhost:5038");

var client = new WeatherForecastProto.WeatherForecastProtoClient(channel);

var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));

using var weatherCall = client.GetWeatherStream(new Empty(), cancellationToken : cts.Token);

try

{

await foreach(WeatherForecastModel weatherForecast in weatherCall.ResponseStream.ReadAllAsync(cancellationToken : cts.Token))

{

Console.WriteLine($"{weatherForecast.DateTimeStamp.ToDateTime():s} | {weatherForecast.Summary} | {weatherForecast.TemperatureC} C");

}

catch(RpcException exception) when (exception.StatusCode == StatusCode.Cancelled)

{

Console.WriteLine("스트리밍 취소.");

}

TestSolution.zip