クライアントと双方向ストリーミング

クライアントストリーミングと双方向ストリーミング

クライアントストリーミング: Proto 定義

service UploadService {
  // Client streams chunks, server returns a summary
  rpc UploadFile (stream FileChunk) returns (UploadSummary);
}

message FileChunk   { bytes data = 1; string filename = 2; }
message UploadSummary { int64 bytes_received = 1; string checksum = 2; }

サーバーでクライアントストリーミングを実装する

public override async Task<UploadSummary> UploadFile(
    IAsyncStreamReader<FileChunk> requestStream,
    ServerCallContext context)
{
    long totalBytes = 0;
    using var ms = new MemoryStream();

    await foreach (var chunk in requestStream.ReadAllAsync(context.CancellationToken))
    {
        await ms.WriteAsync(chunk.Data.Memory, context.CancellationToken);
        totalBytes += chunk.Data.Length;
    }

    var checksum = ComputeMd5(ms.ToArray());
    return new UploadSummary { BytesReceived = totalBytes, Checksum = checksum };
}

クライアントからクライアントストリーミングを呼び出す

using var call = client.UploadFile();

var fileBytes = await File.ReadAllBytesAsync("large-file.bin");
const int chunkSize = 64 * 1024; // 64 KB

for (int offset = 0; offset < fileBytes.Length; offset += chunkSize)
{
    var chunk = fileBytes.Skip(offset).Take(chunkSize).ToArray();
    await call.RequestStream.WriteAsync(new FileChunk
    {
        Filename = "large-file.bin",
        Data = Google.Protobuf.ByteString.CopyFrom(chunk)
    });
}

await call.RequestStream.CompleteAsync(); // signal end
var summary = await call;                // await server response
Console.WriteLine($"Uploaded {summary.BytesReceived} bytes");

双方向ストリーミング: Proto 定義

service ChatService {
  rpc Chat (stream ChatMessage) returns (stream ChatMessage);
}

message ChatMessage {
  string user    = 1;
  string content = 2;
  int64  sent_at = 3;
}

サーバーで双方向ストリーミングを実装する

public override async Task Chat(
    IAsyncStreamReader<ChatMessage> requestStream,
    IServerStreamWriter<ChatMessage> responseStream,
    ServerCallContext context)
{
    // Broadcast to all connected clients
    var readTask = Task.Run(async () =>
    {
        await foreach (var msg in requestStream.ReadAllAsync(context.CancellationToken))
        {
            _chatHub.Broadcast(msg);
        }
    });

    var writeTask = Task.Run(async () =>
    {
        await foreach (var msg in _chatHub.GetMessagesAsync(context.CancellationToken))
        {
            await responseStream.WriteAsync(msg);
        }
    });

    await Task.WhenAll(readTask, writeTask);
}

クライアントから双方向ストリーミングを行う

using var call = client.Chat();

// Read task
var readTask = Task.Run(async () =>
{
    await foreach (var msg in call.ResponseStream.ReadAllAsync())
        Console.WriteLine($"{msg.User}: {msg.Content}");
});

// Write task
while (Console.ReadLine() is string text && text != "exit")
{
    await call.RequestStream.WriteAsync(
        new ChatMessage { User = "Alice", Content = text,
                          SentAt = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() });
}
await call.RequestStream.CompleteAsync();
await readTask;

フロー制御とバックプレッシャー

// DO: Let gRPC flow control handle backpressure
await foreach (var chunk in requestStream.ReadAllAsync(ct))
{
    await ProcessChunkAsync(chunk); // naturally paced
}

// DON'T: Buffering all chunks defeats flow control
var allChunks = new List<FileChunk>();
await foreach (var chunk in requestStream.ReadAllAsync(ct))
    allChunks.Add(chunk); // may OOM on large uploads

ハーフクローズとストリームの完了

// Client signals it's done sending
await call.RequestStream.CompleteAsync(); // half-close

// Continue reading server responses after half-close
await foreach (var reply in call.ResponseStream.ReadAllAsync())
{
    Console.WriteLine(reply.Content);
}

スレッドセーフな双方向ストリーミングに Channels を使用する

private readonly Channel<ChatMessage> _broadcast =
    Channel.CreateUnbounded<ChatMessage>();

// Producer: receives from each client stream
public async Task ReceiveLoopAsync(
    IAsyncStreamReader<ChatMessage> reader, CancellationToken ct)
{
    await foreach (var msg in reader.ReadAllAsync(ct))
        await _broadcast.Writer.WriteAsync(msg, ct);
}

// Consumer: writes to each client's response stream
public async Task SendLoopAsync(
    IServerStreamWriter<ChatMessage> writer, CancellationToken ct)
{
    await foreach (var msg in _broadcast.Reader.ReadAllAsync(ct))
        await writer.WriteAsync(msg);
}

実例: リアルタイム分析パイプライン

service TelemetryService {
  rpc StreamTelemetry(stream TelemetryEvent) returns (stream AggregatedStats);
}

// Server implementation sends a rolling aggregate every 100 events:
public override async Task StreamTelemetry(
    IAsyncStreamReader<TelemetryEvent> requests,
    IServerStreamWriter<AggregatedStats> responses,
    ServerCallContext context)
{
    int count = 0;
    double total = 0;
    await foreach (var e in requests.ReadAllAsync(context.CancellationToken))
    {
        total += e.Value;
        if (++count % 100 == 0)
            await responses.WriteAsync(
                new AggregatedStats { Count = count, Average = total / count });
    }
}

確認問題

まとめ: クライアントストリーミングと双方向ストリーミング