gRPC 大きなサイズのオブジェクトをストリームで受信する

gRPC の送受信データサイズには上限がある

gRPC ではリクエスト／レスポンスのデータサイズに上限があります。既定値は4MBです。
上限を変更することはできますが、データサイズが不定である場合はストリームを用いることが一般的です。リクエスト／レスポンスの一つ一つが上限を超えなければサイズ超過エラーは発生しません。

service Sample
{
    // 一つのリクエストに対して複数のレスポンスを受け取る
    rpc GetObjects (Request) returns (stream Response){}

    // 複数のリクエストに対して複数のレスポンスを受け取る
    rpc GetObjects (stream Request) returns (stream Response){}

    // 複数のリクエストに対して一つのレスポンスを受け取る
    rpc SetObjects (stream Request) returns (Response){}
}

リクエスト／レスポンスのデータサイズが大きかったら？

この方法を採用しても、リクエスト／レスポンスの一つ一つのデータサイズが上限を超える場合はサイズ超過エラーが発生します。

次のような場合の実装を考えてみました。

• 一回の API 呼び出しで複数の画像データを受信する
• 個々の画像サイズが非常に大きく不定である

gRPC APIの定義

今回は ProtocolBuffers で定義しました。bytes 型を使用しています。

LargeMessage.proto

syntax = "proto3";
option csharp_namespace = "Examples.GrpcModels.LargeMessage";

// 検索条件：今回のサンプルでは適当な内容
message Condition
{
    string Keyword = 1;
}

// 一定サイズのバイトデータを受信するためのレスポンス
message BufferResponse
{
    bool Eof = 1;
    bytes Data = 2;
}

// サービス定義
service LargeSearch
{
    rpc GetImages (Condition) returns (stream BufferResponse){}
}

サービスの実装

取得した画像のデータを1KBずつクライアントに送信しています。

サービスの実装

internal class LargeSearchServiceImpl : LargeSearch.LargeSearchBase
{
    public override async Task GetImages(Condition request, IServerStreamWriter<BufferResponse> responseStream, ServerCallContext context)
    {
        // レスポンスのインスタンスは使いまわす
        BufferResponse response = new BufferResponse();

        byte[] buffer = new byte[1024];

        foreach (Stream stream in GetImages(request))
        {
            using (stream)
            {
                while (true)
                {
                    int length = await stream.ReadAsync(buffer, 0, buffer.Length).ConfigureAwait(false);

                    response.Data = length <= 0 ? null : Google.Protobuf.ByteString.CopyFrom(buffer, 0, length);
                    response.Eof = length < buffer.Length;

                    await responseStream.WriteAsync(response).ConfigureAwait(false);

                    if (response.Eof) { break; }
                }
            }
        }

    }

    /// <summary>
    /// 画像データを取得します。
    /// </summary>
    /// <param name="condition">取得条件</param>
    /// <returns>画像データを格納しているストリーム</returns>
    private IEnumerable<Stream> GetImages(Condition condition)
    {
        // 実際には条件に該当するファイルなどから列挙するように実装します。
        yield return new FileStream(@"image1.jpg", FileMode.Open, FileAccess.Read);
        yield return new FileStream(@"image2.jpg", FileMode.Open, FileAccess.Read);
    }
}

クライアントの実装

レスポンスとして受け取ったバイトデータをストリームに格納し、Eof を受け取った時点で画像を生成しています。

private async Task GetImages()
{
    Grpc.Core.Channel channel = GetChannel();
    LargeSearch.LargeSearchClient client = new LargeSearch.LargeSearchClient(channel);

    Condition condition = new Condition
    {
        Keyword = "***"
    };

    using (MemoryStream stream = new MemoryStream())
    using (AsyncServerStreamingCall<BufferResponse> call = client.GetImages(condition))
    {
        while (await call.ResponseStream.MoveNext().ConfigureAwait(false))
        {
            BufferResponse response = call.ResponseStream.Current;

            if (response.Data != null && response.Data.Length > 0)
            {
                response.Data.WriteTo(stream);
            }

            if (response.Eof)
            {
                stream.Position = 0;
                ShowImage(stream);
                stream.SetLength(0);
            }
        }
    }

}

/// <summary>
/// 指定された画像を表示します。
/// </summary>
/// <param name="stream">画像データを格納しているストリーム</param>
private void ShowImage(Stream stream)
{
    if (this.InvokeRequired)
    {
        this.Invoke(new Action<Stream>(ShowImage), new object[] { stream });
        return;
    }
    // 単純にダイアログフォームで表示します。
    using (Image image = Image.FromStream(stream))
    using (Form frm = new Form())
    {
        PictureBox box = new PictureBox();
        box.Dock = DockStyle.Fill;
        box.Image = image;
        frm.Controls.Add(box);
        frm.ShowDialog(this);
    }
}