0Pricing
C# Academy · レッスン

UnaryとサーバーストリーミングRPC

Unary呼び出しとサーバー側ストリーミングを実装し、サーバーからクライアントへリアルタイムにデータを送信します。

「UnaryとサーバーストリーミングRPC」はCoddyKit上の無料C# Academyレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはC# Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 C# Academyコースには全4レッスンが含まれています。

gRPC の RPC パターン

gRPC は 4 つの RPC パターンをサポートしています。Unary（リクエスト／レスポンス）、Server Streaming（クライアントが 1 つ送信し、サーバーが複数送信）、Client Streaming、Bidirectional Streaming です。このレッスンでは最初の 2 つを扱います。

Unary RPC：基本パターン

Unary は最も単純なパターンで、1 つのリクエストに対して 1 つのレスポンスを返します。通常の関数呼び出しのように見えますが、Protobuf エンコーディングを使って HTTP/2 経由で通信します。

// .proto definition
service OrderService {
  rpc GetOrder (GetOrderRequest) returns (OrderResponse);
}

message GetOrderRequest { int32 id = 1; }
message OrderResponse   { int32 id = 1; string status = 2; double total = 3; }

Unary RPC の実装

生成された基底メソッドをオーバーライドします。ServerCallContext から、メタデータ、キャンセレーショントークン、デッドライン、接続相手の情報を取得できます。

public class OrderService : OrderService.OrderServiceBase
{
    private readonly IOrderRepository _repo;
    public OrderService(IOrderRepository repo) => _repo = repo;

    public override async Task<OrderResponse> GetOrder(
        GetOrderRequest request,
        ServerCallContext context)
    {
        var order = await _repo.GetByIdAsync(request.Id, context.CancellationToken);
        if (order is null)
            throw new RpcException(new Status(StatusCode.NotFound, "Order not found"));

        return new OrderResponse
        {
            Id     = order.Id,
            Status = order.Status.ToString(),
            Total  = (double)order.Total
        };
    }
}

クライアントから Unary RPC を呼び出す

生成されたクライアントには、同期的に見えるメソッドと完全な非同期メソッドの両方が用意されています。本番環境では必ず Async 版を使用してください。

var client = new OrderService.OrderServiceClient(channel);

// Async call with cancellation
var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
var reply = await client.GetOrderAsync(
    new GetOrderRequest { Id = 42 },
    cancellationToken: cts.Token);

Console.WriteLine($"Order {reply.Id}: {reply.Status} - ${reply.Total}");

サーバーストリーミング: Proto 定義

サーバーストリーミングでは、レスポンス型の前に stream キーワードを追加します。サーバーはストリームを閉じる前に、同じ接続上で複数のメッセージを送信します。

service StockService {
  // Client sends one symbol, server streams price updates
  rpc WatchStock (WatchRequest) returns (stream StockUpdate);
}

message WatchRequest  { string symbol = 1; }
message StockUpdate   { string symbol = 1; double price = 2; int64 timestamp = 3; }

サーバーストリーミングを実装する

メッセージの送信には IServerStreamWriter<T> を使用します。メソッドは Task を返すため、処理が完了するまで書き込み、その後で戻ります。

public override async Task WatchStock(
    WatchRequest request,
    IServerStreamWriter<StockUpdate> responseStream,
    ServerCallContext context)
{
    while (!context.CancellationToken.IsCancellationRequested)
    {
        var price = await _market.GetPriceAsync(request.Symbol);

        await responseStream.WriteAsync(new StockUpdate
        {
            Symbol    = request.Symbol,
            Price     = price,
            Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
        });

        await Task.Delay(1000, context.CancellationToken);
    }
}

クライアントでサーバーストリームを利用する

ResponseStream.ReadAllAsync() に対して await foreach を使用し、到着した各メッセージを処理します。

using var call = client.WatchStock(
    new WatchRequest { Symbol = "MSFT" });

await foreach (var update in call.ResponseStream.ReadAllAsync())
{
    Console.WriteLine($"{update.Symbol}: ${update.Price:F2}");
}
// Continues until server closes the stream or client cancels

サーバーストリームをキャンセルする

呼び出しを開始するときに CancellationToken を渡すと、クライアント側でストリームを停止できます。サーバーは独自のトークンを通じてキャンセルを受け取ります。

var cts = new CancellationTokenSource();

// Cancel after 30 seconds
cts.CancelAfter(TimeSpan.FromSeconds(30));

using var call = client.WatchStock(
    new WatchRequest { Symbol = "AAPL" },
    cancellationToken: cts.Token);

try
{
    await foreach (var update in
        call.ResponseStream.ReadAllAsync(cts.Token))
    {
        Console.WriteLine($"Price: {update.Price}");
    }
}
catch (OperationCanceledException)
{
    Console.WriteLine("Stream cancelled.");
}

メタデータ: リクエストヘッダーとトレーラー

gRPC は、レスポンスの前にヘッダーとして送信するメタデータ（キーと値のペア）と、レスポンスの後に送信するトレーラーをサポートしています。認証トークン、トレース ID、ページネーションカーソルなどに使用します。

// Server: send initial metadata
await context.WriteResponseHeadersAsync(new Metadata
{
    { "x-correlation-id", Guid.NewGuid().ToString() }
});

// Server: set trailing metadata
context.ResponseTrailers.Add("x-total-count", "1500");

// Client: read headers
var headers  = await call.ResponseHeadersAsync;
var trailerId = headers.GetValue("x-correlation-id");

ブラウザクライアント向けの gRPC-Web

ブラウザでは生の gRPC を使用できません（HTTP/2 のトレーラーがサポートされていないためです）。Grpc.AspNetCore.Web と grpc-web JavaScript クライアント、または Blazor の Grpc.Net.Client.Web を使用してください。

// dotnet add package Grpc.AspNetCore.Web

app.UseGrpcWeb();
app.MapGrpcService<GreeterService>().EnableGrpcWeb();

// Blazor client:
var handler = new GrpcWebHandler(GrpcWebMode.GrpcWeb,
    new HttpClientHandler());
using var channel = GrpcChannel.ForAddress(
    "https://localhost:7042",
    new GrpcChannelOptions { HttpHandler = handler });

実例: ログストリーミングサービス

ログ集約サービスは、サーバーストリーミングを使用して、接続中の監視クライアントにアプリケーションのログエントリをリアルタイムでストリーミングします。

public override async Task StreamLogs(
    LogStreamRequest request,
    IServerStreamWriter<LogEntry> responseStream,
    ServerCallContext context)
{
    await foreach (var log in _logChannel.Reader.ReadAllAsync(
        context.CancellationToken))
    {
        if (log.Level >= request.MinLevel)
        {
            await responseStream.WriteAsync(new LogEntry
            {
                Level   = log.Level.ToString(),
                Message = log.Message,
                Timestamp = Timestamp.FromDateTime(log.Timestamp)
            });
        }
    }
}

確認問題

.proto ファイルでサーバーストリーミング RPC を示すキーワードは何ですか？

まとめ: Unary RPC とサーバーストリーミング RPC

重要なポイント:

  • Unary RPC: 1 つのリクエストに対して 1 つのレスポンスを返す、最も一般的なパターン
  • サーバーストリーミング: 1 つのリクエストに対して、時間をかけて複数のレスポンスメッセージを返す
  • IServerStreamWriter<T> に書き込んでサーバーストリーミングを実装する
  • ReadAllAsync() に対する await foreach でサーバーストリームを利用する
  • CancellationToken を使用して、どちら側からでもストリームを適切に停止する
  • gRPC-Web を使用すると、ブラウザクライアントから gRPC サービスを呼び出せる

よくある質問

「UnaryとサーバーストリーミングRPC」レッスンは無料ですか？

はい。「UnaryとサーバーストリーミングRPC」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、C# Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 C# Academyコースには全4レッスンが含まれています。

「UnaryとサーバーストリーミングRPC」で何を学びますか？

Unary呼び出しとサーバー側ストリーミングを実装し、サーバーからクライアントへリアルタイムにデータを送信します。 ブラウザで直接実行するハンズオンコードでC# Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

C# Academyを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのC# Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。

「UnaryとサーバーストリーミングRPC」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このC# Academyレッスンでコードを書いて実行できますか？

はい。すべてのC# Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. gRPCとProtobufの基礎
  2. UnaryとサーバーストリーミングRPC
  3. クライアントと双方向ストリーミング
  4. デッドライン、キャンセルとインターセプター
← C# Academyに戻る