RustのtonicのgRPCでバックエンドAPIを開発する方法まとめ

use std::{env, path::PathBuf};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // コンフィグ設定を定義
    let files = &["proto/sample/sample.proto"];
    let includes = &[
        "proto",
        "../root/go/pkg/mod/github.com/envoyproxy/protoc-gen-validate@v1.2.1",
    ];
    let file_descriptor_set_path = PathBuf::from(env::var("OUT_DIR").expect("OUT_DIR is not set"))
        .join("sample_descriptor.bin");
    let mut config = {
        let mut c = prost_build::Config::new();
        c.service_generator(
            tonic_build::configure()
            .message_attribute(".", "#[derive(::prost_validate::Validator)]")
            .service_generator(),
        );
        c
    };

    // コンフィグ設定にバリデーションを適用
    prost_validate_build::Builder::new()
        .file_descriptor_set_path(file_descriptor_set_path)
        .configure(&mut config, files, includes)?;

    // コンフィング設定からコード生成
    config.compile_protos(files, includes)?;

    Ok(())
}

use envy;
use serde::Deserialize;

// 環境変数のデフォルト値を返す関数
fn default_env() -> String {
    "local".to_string()
}

fn default_grpc_port() -> u16 {
    50051
}

fn default_rust_log() -> String {
    "info".to_string()
}

// 環境変数の構造体
#[derive(Deserialize, Debug)]
pub struct Config {
    #[serde(default = "default_env")]
    pub env: String,
    #[serde(default = "default_grpc_port")]
    pub grpc_port: u16,
    #[allow(dead_code)]
    #[serde(default = "default_rust_log")]
    pub rust_log: String,
}

// 環境変数を返す関数
pub fn get_config() -> Config {
    match envy::from_env::<Config>() {
        Ok(config) => config,
        Err(err) => {
            println!("環境変数の初期化エラー: {}", err);

            // 環境変数にデフォルト値を設定して返す
            Config {
                env: default_env(),
                grpc_port: default_grpc_port(),
                rust_log: default_rust_log(),
            }
        }
    }
}

use tonic::Request;

// 共通コンテキストの構造体
#[derive(Clone, Debug)]
pub struct Context {
    #[allow(dead_code)]
    pub request_id: String,
    #[allow(dead_code)]
    pub uri: String,
}

// コンテキスト作成関数
pub fn create_context<T>(req: &Request<T>) -> Context {
    let request_id = req
        .metadata()
        .get("x-request-id")
        .map(|value| value.to_str().unwrap_or("-"))
        .unwrap_or("-");

    let uri = req
        .metadata()
        .get("x-uri")
        .map(|value| value.to_str().unwrap_or("-"))
        .unwrap_or("-");

    Context {
        request_id: request_id.to_string(),
        uri: uri.to_string(),
    }
}

use thiserror::Error;
use tonic::Status;

#[derive(Error, Debug)]
pub enum CommonError {
    #[error("Internal Server Error")]
    InternalServerError,
    #[allow(dead_code)]
    #[error("{message}")]
    CustomError { status: Status, message: String },
}

use chrono::TimeZone;
use log;
use std::io::Write;

// 共通コンテキスト
use crate::contexts::context::Context;

// コンフィング設定
use crate::configs::config::get_config;

// ロガーの初期化用関数
pub fn init_logger() {
    // 日本時間を取得
    let jst = chrono::offset::FixedOffset::east_opt(9 * 3600)
        .unwrap()
        .from_utc_datetime(&chrono::Utc::now().naive_utc());

    // カスタムロガーの初期化
    env_logger::builder()
        .format(move |buf, record| {
            writeln!(
                buf,
                "{} {} {}",
                jst.format("%Y-%m-%d %H:%M:%S"),
                record.level(),
                record.args()
            )
        })
        .init();
}

// 共通コンテキストからログに追加する情報の文字列を取得する関数
fn get_info_from_request(ctx: &Context) -> String {
    format!("request_id={} uri={}", ctx.request_id, ctx.uri)
}

// ログ出力用関数
pub fn info(ctx: &Context, msg: &str) {
    let config = get_config();

    // ENV=testing以外の場合にログ出力
    if config.env != "testing" {
        let info = get_info_from_request(ctx);
        log::info!("[{}] {}", info, msg);
    }
}

// TODO: 使用する場合にコメントアウトを外す
// pub fn warn(ctx: &Context, msg: &str) {
//     let config = get_config();

//     // ENV=testing以外の場合にログ出力
//     if config.env != "testing" {
//         let info = get_info_from_request(ctx);
//         log::warn!("[{}] {}", info, msg);
//     }
// }

pub fn error(ctx: &Context, msg: &str) {
    let config = get_config();

    // ENV=testing以外の場合にログ出力
    if config.env != "testing" {
        let info = get_info_from_request(ctx);
        log::error!("[{}] {}", info, msg);
    }
}

use hyper::header::HeaderValue;
use std::{
    pin::Pin,
    task::{Context, Poll},
};
use tonic::server::NamedService;
use tower::{Layer, Service};
use uuid::Uuid;

// 共通コンテキスト
use crate::contexts::context::Context as CommonContext;

// ロガー
use crate::loggers::logger;

// リクエスト用のミドルウェア
#[derive(Clone)]
pub struct RequestMiddleware<S> {
    inner: S,
}

impl<S> RequestMiddleware<S> {
    fn new(inner: S) -> Self {
        RequestMiddleware { inner }
    }
}

// リクエスト用のミドルウェアにServiceトレイトを実装
impl<B, S> Service<hyper::Request<B>> for RequestMiddleware<S>
where
    S: Service<hyper::Request<B>> + Clone + Send + 'static,
    S::Future: Send,
    B: Send + 'static,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, mut req: hyper::Request<B>) -> Self::Future {
        // リクエストからuriを取得
        let uri_path = req.uri().path();

        // uriをリクエストヘッダーに設定
        let uri_string = match HeaderValue::from_str(uri_path) {
            Ok(value) => {
                req.headers_mut().insert("x-uri", value.clone());
                value.to_str().unwrap().to_string()
            }
            Err(_) => {
                // エラーの場合は「-」を設定
                let header_value = HeaderValue::from_str("-").unwrap();
                req.headers_mut().insert("x-uri", header_value.clone());
                header_value.to_str().unwrap().to_string()
            }
        };

        // request-idをリクエストヘッダーに設定
        let uuid = Uuid::new_v4().to_string();
        req.headers_mut()
            .insert("x-request-id", uuid.clone().parse().unwrap());

        // リクエスト開始ログ
        let ctx = CommonContext {
            request_id: uuid,
            uri: uri_string,
        };
        logger::info(&ctx, "Start gRPC request !!");

        // 非同期処理のためself.innerをコピー
        let mut inner = self.inner.clone();

        // 非同期処理
        Box::pin(async move {
            let res = inner.call(req).await;

            // 処理完了後にリクエスト終了ログ
            logger::info(&ctx, "Finish gRPC request !!");

            res
        })
    }
}

// リクエスト用のミドルウェアにtonic::server::NamedServiceを実装
impl<S> NamedService for RequestMiddleware<S>
where
    S: NamedService,
{
    const NAME: &'static str = S::NAME;
}

// リクエスト用のミドルウェアのレイヤー
#[derive(Clone)]
pub struct RequestMiddlewareLayer;

// リクエスト用のミドルウェアのレイヤーにLayerトレイトの実装
impl<S> Layer<S> for RequestMiddlewareLayer {
    type Service = RequestMiddleware<S>;

    fn layer(&self, inner: S) -> Self::Service {
        RequestMiddleware::new(inner)
    }
}

use tonic::{Code, Request, Status};

#[allow(clippy::result_large_err)]
pub fn auth_interceptor(req: tonic::Request<()>) -> Result<Request<()>, Status> {
    // メタデータからx-uriを取得
    let uri = match req.metadata().get("x-uri") {
        Some(value) => value.to_str().unwrap_or_default(),
        None => "-",
    };

    // 対象のuriの場合はスキップ
    let skip_uri = vec![
        "/sample.SampleService/Hello",
    ];
    if skip_uri.contains(&uri) {
        return Ok(req);
    }

    // リクエストヘッダーからトークン取得
    let token = match req.metadata().get("authorization") {
        Some(value) => {
            let bearer_token = value.to_str().unwrap_or_default();
            bearer_token.trim_start_matches("Bearer ")
        }
        None => "",
    };

    // 認証トークンが設定されていない場合はエラー
    if token.is_empty() {
        let status = Status::new(
            Code::InvalidArgument,
            "認証用トークンが設定されていません。",
        );
        return Err(status);
    }

    // TODO: 認証チェック処理を追加

    // 戻り値
    Ok(req)
}

// 共通コンテキスト
use crate::contexts::context::Context;

// コンフィング設定
use crate::configs::config::get_config;

// ロガー
use crate::loggers::logger;

// 共通エラー用モジュール
use crate::errors::error::CommonError;

// サンプルリポジトリーの構造体
pub struct SampleRepository;

impl SampleRepository {
    // 初期化用メソッド
    pub fn new() -> Self {
        SampleRepository
    }
}

// サンプルリポジトリー用のトレイト（モック化もできるように定義）
#[mockall::automock]
#[async_trait::async_trait]
pub trait SampleRepositoryTrait {
    async fn sample_hello(&self, ctx: &Context) -> Result<String, CommonError>;
}

#[async_trait::async_trait]
impl SampleRepositoryTrait for SampleRepository {
    // 文字列「Sample Hello !!」を返す関数
    async fn sample_hello(&self, ctx: &Context) -> Result<String, CommonError> {
        let mut text = "Hello World !!".to_string();

        let config = get_config();
        if config.env == "testing" {
            text = "Hello World !! [ENV=testing]".to_string();
        }

        if text.is_empty() {
            logger::error(ctx, "textが空です");
            return Err(CommonError::InternalServerError);
        }

        Ok(text)
    }
}

// 共通コンテキスト
use crate::contexts::context::Context;

// 共通エラー用モジュール
use crate::errors::error::CommonError;

// ロガー
use crate::loggers::logger;

// リポジトリ用のモジュール
use crate::repositories::sample::sample_repository::SampleRepositoryTrait;

// 使用するリポジトリーをまとめる構造体
pub struct SampleCommonRepository {
    // Box<T>型で動的にメモリ領域確保
    // Send: オブジェクトが異なるスレッド間で安全に送信できることを保証
    // Sync: オブジェクトが複数のスレッドから同時にアクセスできることを保証
    // 'static: オブジェクトのライフタイムがプログラムが終了するまで破棄されない
    pub sample_repo: Box<dyn SampleRepositoryTrait + Send + Sync + 'static>,
}

// サンプルサービス
pub struct SampleService {
    repo: SampleCommonRepository,
}

impl SampleService {
    pub fn new(repo: SampleCommonRepository) -> Self {
        SampleService { repo }
    }
}

// サンプルサービス用のトレイト（モック化もできるように定義）
#[mockall::automock]
#[async_trait::async_trait]
pub trait SampleServiceTrait {
    async fn sample_get_text_hello(&self, ctx: &Context) -> Result<String, CommonError>;
}

#[async_trait::async_trait]
impl SampleServiceTrait for SampleService {
    async fn sample_get_text_hello(&self, ctx: &Context) -> Result<String, CommonError> {
        let text = match self.repo.sample_repo.sample_hello(ctx).await {
            Ok(text) => text,
            Err(err) => {
                logger::error(ctx, "sample_get_text_helloのsample_hello処理でエラー");
                return Err(err);
            }
        };

        Ok(text)
    }
}

use tonic::{Response, Status};

// 共通コンテキスト
use crate::contexts::context::Context;

// ロガー
use crate::loggers::logger;

// ビルドしたprotoファイルのモジュール
use crate::server::grpc::sample::sample_server::sample_proto;

// サービスのモジュール
use crate::services::sample::sample_service::{SampleService, SampleServiceTrait};

// 使用するサービスをまとめる構造体
pub struct SampleCommonService {
    pub sample_service: SampleService,
}

// 実行するユースケースの構造体
pub struct SampleHelloUsecase {
    pub service: SampleCommonService,
}

impl SampleHelloUsecase {
    pub async fn exec(
        &self,
        ctx: Context,
    ) -> Result<Response<sample_proto::HelloResponseBody>, Status> {
        // サンプルテキストを取得するサービスを実行
        let text = match self
            .service
            .sample_service
            .sample_get_text_hello(&ctx)
            .await
        {
            Ok(text) => text,
            Err(e) => {
                let msg = format!("Internal Server Error: {}", e);
                logger::error(&ctx, &msg);
                return Err(Status::unknown(msg));
            }
        };

        // レスポンスボディの設定
        let res_body = sample_proto::HelloResponseBody { message: text };

        // メタデータにrequest-idを追加
        let mut res = Response::new(res_body);
        res.metadata_mut()
            .insert("x-request-id", ctx.request_id.parse().unwrap());

        // 戻り値
        Ok(res)
    }
}

// tonic
use tonic::{Response, Status};

// 共通コンテキスト
use crate::contexts::context::Context;

// ビルドしたprotoファイルのモジュール
use crate::server::grpc::sample::sample_server::sample_proto;

// 実行するユースケースの構造体
pub struct SampleHelloAddTextUsecase {}

impl SampleHelloAddTextUsecase {
    pub async fn exec(
        &self,
        ctx: Context,
        req_body: sample_proto::HelloAddTextRequestBody,
    ) -> Result<Response<sample_proto::HelloAddTextResponseBody>, Status> {
        // レスポンスメッセージの設定
        let msg = format!("Hello {}", req_body.text);

        // レスポンスボディの設定
        let res_body = sample_proto::HelloAddTextResponseBody { message: msg };

        // メタデータにrequest-idを追加
        let mut res = Response::new(res_body);
        res.metadata_mut()
            .insert("x-request-id", ctx.request_id.parse().unwrap());

        // 戻り値
        Ok(res)
    }
}

use tonic::{service::InterceptorLayer, transport::Server};
use tonic_reflection::server::Builder as ReflectionBuilder;
use tower::ServiceBuilder;

// コンフィング設定
use crate::configs::config::get_config;

// ミドルウェア
use crate::middleware::request_middleware::RequestMiddlewareLayer;

// インターセプター
use crate::interceptors::interceptor::auth_interceptor;

// protoファイル
pub mod proto {
    pub const SAMPLE_DESCRIPTOR: &[u8] = tonic::include_file_descriptor_set!("sample_descriptor");
}

// gRPCサーバーのサービス
use crate::server::grpc::sample::sample_server::get_sample_server;

pub async fn router() -> Result<(), Box<dyn std::error::Error>> {
    // 環境変数取得
    let config = get_config();

    // アドレス設定
    let grpc_port = config.grpc_port;
    let addr = format!("[::]:{}", grpc_port).parse()?;

    // サーバー設定
    let sample_server = get_sample_server();

    // サーバーリフレクション設定（旧ツールではv1alphaを使う）
    let reflection_service_v1 = ReflectionBuilder::configure()
        .register_encoded_file_descriptor_set(proto::SAMPLE_DESCRIPTOR)
        .build_v1()?;

    let reflection_service_v1alpha = ReflectionBuilder::configure()
        .register_encoded_file_descriptor_set(proto::SAMPLE_DESCRIPTOR)
        .build_v1alpha()?;

    // サービス設定
    let service = ServiceBuilder::new()
        .layer(RequestMiddlewareLayer)
        .layer(InterceptorLayer::new(auth_interceptor))
        .service(sample_server);

    // サーバー起動
    Server::builder()
        .add_service(reflection_service_v1)
        .add_service(reflection_service_v1alpha)
        .add_service(service)
        .serve(addr)
        .await?;

    Ok(())
}

// モジュールのインポート
mod configs;
mod contexts;
mod errors;
mod interceptors;
mod loggers;
mod middleware;
mod repositories;
mod routers;
mod server;
mod services;
mod usecases;
use configs::config::get_config;
use loggers::logger::init_logger;
use routers::router::router;

#[tokio::main]
async fn main() {
    // 環境変数取得
    let config = get_config();

    // ロガーの初期化
    init_logger();

    // サーバー起動のログ出力
    log::info!("[ENV={}] Start rust_grpc !!", config.env);

    // サーバー起動
    router().await.unwrap();
}

// tonic
use tonic::{
    Response, Status,
    metadata::{Ascii, MetadataValue},
};

// tokio
use tokio::{
    spawn,
    sync::mpsc,
    time::{Duration, sleep},
};
use tokio_stream::wrappers::ReceiverStream;

// 変換用
use std::str::FromStr;

// 共通コンテキスト
use crate::contexts::context::Context;

// ロガー
use crate::loggers::logger;

// ビルドしたprotoファイルのモジュール
use crate::server::grpc::sample::sample_server::sample_proto;

// 実行するユースケースの構造体
pub struct SampleHelloServerStreamUsecase {}

// ストリーミング用の型定義
type HelloServerStreamStream =
    ReceiverStream<Result<sample_proto::HelloServerStreamResponseBody, Status>>;

impl SampleHelloServerStreamUsecase {
    pub async fn exec(
        &self,
        ctx: Context,
        req_body: sample_proto::HelloServerStreamRequestBody,
    ) -> Result<Response<HelloServerStreamStream>, Status> {
        // トレーラー用
        let x_request_id = MetadataValue::<Ascii>::from_str(ctx.request_id.as_str()).expect("-");

        // mpsc (multi-producer, single-consumer) チャンネルの作成
        // サーバーはこのチャンネルにデータ送信し、その後クライアントにストリーミングする
        // バッファサイズは適宜調整が必要だが、サーバーストリーミング機能なら1などでOK
        let (tx, rx) = mpsc::channel(1);

        spawn(async move {
            logger::info(&ctx, "Start Server Stream !!");
            // n件のデータをクライアントに返す（今回は3件）
            for i in 1..=3 {
                // レスポンスの設定
                let msg = format!("[{}] Hello {} !", i, req_body.text);
                let res_body = sample_proto::HelloServerStreamResponseBody { message: msg };

                // Okでラップしたレスポンスをtxで送信
                if let Err(e) = tx.send(Ok(res_body)).await {
                    // クライアントの接続切れなどでエラーの場合
                    let msg = format!("Failed to send data: {:?}", e);
                    logger::error(&ctx, &msg);
                    let mut status = Status::invalid_argument(msg);
                    status
                        .metadata_mut()
                        .insert("x-request-id", x_request_id.clone());
                    let _ = tx.send(Err(status)).await;
                    break;
                }

                // 1秒間待機処理
                sleep(Duration::from_secs(1)).await;
            }

            // トレーラーの設定
            let mut status = Status::ok("Stream finished successfully");
            status
                .metadata_mut()
                .insert("x-request-id", x_request_id.clone());

            // Errでラップしたステータスを送信
            if let Err(e) = tx.send(Err(status)).await {
                // クライアントの接続切れなどでエラーの場合
                let msg = format!("Failed to send data: {:?}", e);
                logger::error(&ctx, &msg);
                let mut status = Status::invalid_argument(msg);
                status.metadata_mut().insert("x-request-id", x_request_id);
                let _ = tx.send(Err(status)).await;
            }

            logger::info(&ctx, "Finish Server Stream !!");
        });

        // 戻り値
        Ok(Response::new(ReceiverStream::new(rx)))
    }
}

// tonic
use tonic::{
    Request, Response, Status, Streaming,
    metadata::{Ascii, MetadataValue},
};

// バリデーション用のトレイト
use prost_validate::Validator;

// 変換用
use std::str::FromStr;

// 共通コンテキスト
use crate::contexts::context::create_context;

// ロガー
use crate::loggers::logger;

// ビルドしたprotoファイルのモジュール
use crate::server::grpc::sample::sample_server::sample_proto;

// 実行するユースケースの構造体
pub struct SampleHelloClientStreamUsecase {}

impl SampleHelloClientStreamUsecase {
    pub async fn exec(
        &self,
        request: Request<Streaming<sample_proto::HelloClientStreamRequestBody>>,
    ) -> Result<Response<sample_proto::HelloClientStreamResponseBody>, Status> {
        // コンテキスト設定
        let ctx = create_context(&request);

        // トレーラー用
        let x_request_id = MetadataValue::<Ascii>::from_str(ctx.request_id.as_str()).expect("-");

        // リクエストからストリームを取り出す
        let mut stream = request.into_inner();

        logger::info(&ctx, "Start Client Stream !!");

        // ストリームからデータを1件ずつ受信し、ストリームが閉じるまでループ
        let mut lists = Vec::new();
        while let Some(req) = stream.message().await? {
            // バリデーションチェック
            match req.validate() {
                Ok(_) => {}
                Err(e) => {
                    let msg = format!("バリデーションエラー: {}", e);
                    logger::error(&ctx, &msg);
                    let mut status = Status::invalid_argument(msg);
                    status
                        .metadata_mut()
                        .insert("x-request-id", x_request_id.clone());
                    return Err(status);
                }
            }

            // データを配列に格納
            lists.push(req.text);
        }

        // 配列のデータから文字列を作成
        let msg = lists.join(",");

        // レスポンスを設定
        let mut res = Response::new(sample_proto::HelloClientStreamResponseBody { message: msg });

        // トレーラー設定
        res.metadata_mut().insert("x-request-id", x_request_id);

        logger::info(&ctx, "Finish Client Stream !!");

        // 戻り値
        Ok(res)
    }
}

// tonic
use tonic::{
    Request, Response, Status, Streaming,
    metadata::{Ascii, MetadataValue},
};

// tokio
use tokio::{
    spawn,
    sync::mpsc,
    time::{Duration, sleep},
};
use tokio_stream::{StreamExt, wrappers::ReceiverStream};

// 変換用
use std::str::FromStr;

// バリデーション用のトレイト
use prost_validate::Validator;

// 共通コンテキスト
use crate::contexts::context::create_context;

// ロガー
use crate::loggers::logger;

// ビルドしたprotoファイルのモジュール
use crate::server::grpc::sample::sample_server::sample_proto;

// 実行するユースケースの構造体
pub struct SampleHelloBidirectionalStreamUsecase {}

// ストリーミング用の型定義
type HelloBidirectionalStreamStream =
    ReceiverStream<Result<sample_proto::HelloBidirectionalStreamResponseBody, Status>>;

impl SampleHelloBidirectionalStreamUsecase {
    pub async fn exec(
        &self,
        request: Request<Streaming<sample_proto::HelloBidirectionalStreamRequestBody>>,
    ) -> Result<Response<HelloBidirectionalStreamStream>, Status> {
        // コンテキスト設定
        let ctx = create_context(&request);

        // トレーラー用
        let x_request_id = MetadataValue::<Ascii>::from_str(ctx.request_id.as_str()).expect("-");

        // リクエストからストリームを取り出す
        let mut stream = request.into_inner();

        // mpsc (multi-producer, single-consumer) チャンネルの作成
        // サーバーはこのチャンネルにデータ送信し、その後クライアントにストリーミングする
        // バッファサイズは適宜調整が必要
        let (tx, rx) = mpsc::channel(1);

        spawn(async move {
            logger::info(&ctx, "Start Bidirectional Stream !!");
            while let Some(result) = stream.next().await {
                match result {
                    Ok(req) => {
                        // バリデーションチェック
                        match req.validate() {
                            Ok(_) => {}
                            Err(e) => {
                                let msg = format!("バリデーションエラー: {}", e);
                                logger::error(&ctx, &msg);
                                let mut status = Status::invalid_argument(msg);
                                status
                                    .metadata_mut()
                                    .insert("x-request-id", x_request_id.clone());
                                let _ = tx.send(Err(status)).await;
                                break;
                            }
                        }

                        // レスポンスを設定
                        let res = sample_proto::HelloBidirectionalStreamResponseBody {
                            message: req.text,
                        };

                        // Okでラップしたレスポンスをtxで送信
                        if let Err(e) = tx.send(Ok(res)).await {
                            // クライアントの接続切れなどでエラーの場合
                            let msg = format!("Failed to send data: {:?}", e);
                            logger::error(&ctx, &msg);
                            let mut status = Status::invalid_argument(msg);
                            status
                                .metadata_mut()
                                .insert("x-request-id", x_request_id.clone());
                            let _ = tx.send(Err(status)).await;
                            break;
                        }

                        // 1秒間待機処理
                        sleep(Duration::from_secs(1)).await;
                    }
                    Err(e) => {
                        // クライアントの接続切れなどでエラーの場合
                        let msg = format!("Failed to send data: {:?}", e);
                        logger::error(&ctx, &msg);
                        let mut status = Status::invalid_argument(msg);
                        status
                            .metadata_mut()
                            .insert("x-request-id", x_request_id.clone());
                        let _ = tx.send(Err(status)).await;
                        break;
                    }
                }
            }

            // トレーラーの設定
            let mut status = Status::ok("Stream finished successfully");
            status
                .metadata_mut()
                .insert("x-request-id", x_request_id.clone());

            // Errでラップしたステータスを送信
            if let Err(e) = tx.send(Err(status)).await {
                // クライアントの接続切れなどでエラーの場合
                let msg = format!("Failed to send data: {:?}", e);
                logger::error(&ctx, &msg);
                let mut status = Status::invalid_argument(msg);
                status.metadata_mut().insert("x-request-id", x_request_id);
                let _ = tx.send(Err(status)).await;
            }

            logger::info(&ctx, "Finish Bidirectional Stream !!");
        });

        Ok(Response::new(ReceiverStream::new(rx)))
    }
}

#[cfg(test)]

// sample_serverのストリーミングのテスト
mod sample_server_streaming_test {
    use crate::configs::config::get_config;
    use crate::contexts::context::Context;
    use crate::server::grpc::sample::sample_server::sample_proto::{
        HelloBidirectionalStreamRequestBody, HelloClientStreamRequestBody,
        HelloServerStreamRequestBody, sample_service_client::SampleServiceClient,
    };
    use crate::usecases::sample::hello_server_stream_usecase::SampleHelloServerStreamUsecase;
    use tokio_stream::{StreamExt, iter};
    use tonic::Request;

    #[tokio::test]
    async fn hello_server_stream_should_return_succeed_by_usecase() {
        /* ユースケースを実行して検証する場合 */

        // インスタンス化
        let usecase = SampleHelloServerStreamUsecase {};

        // コンテキストの設定
        let ctx = Context {
            request_id: "5ccba39d-fc9e-482a-aa6b-b94a450a53d0".to_string(),
            uri: "/sample.SampleService/HelloServerStream".to_string(),
        };

        // リクエストの設定
        let req = HelloServerStreamRequestBody {
            text: "Server Stream".to_string(),
        };

        // ユースケースの実行
        let res = usecase.exec(ctx, req).await;

        // 検証
        assert!(res.is_ok());

        // ストリームの検証
        let mut stream = res.unwrap().into_inner();
        let mut i = 1;
        while let Some(result) = stream.next().await {
            match result {
                Ok(res) => {
                    let msg = format!("[{}] Hello Server Stream !", i);
                    assert_eq!(res.message, msg);
                }
                Err(_) => {}
            }
            i += 1;
        }
    }

    #[tokio::test]
    async fn hello_server_stream_should_return_succeed_by_client() {
        let config = get_config();

        // サーバーのインスタンス作成
        let mut client =
            SampleServiceClient::connect(format!("http://localhost:{}", config.grpc_port))
                .await
                .unwrap();

        // リクエストの作成
        let mut request = Request::new(HelloServerStreamRequestBody {
            text: "Server Stream".to_string(),
        });

        // メタデータにBearerトークンを追加
        request
            .metadata_mut()
            .insert("authorization", "Bearer token".parse().unwrap());

        // テストの実行
        let res = client.hello_server_stream(request).await;

        // 検証
        assert!(res.is_ok());

        // ストリームの検証
        let mut stream = res.unwrap().into_inner();
        let mut i = 1;
        while let Some(result) = stream.next().await {
            match result {
                Ok(res) => {
                    let msg = format!("[{}] Hello Server Stream !", i);
                    assert_eq!(res.message, msg);
                }
                Err(_) => {}
            }
            i += 1;
        }
    }

    #[tokio::test]
    async fn hello_client_stream_should_return_succeed() {
        let config = get_config();

        // サーバーのインスタンス作成
        let mut client =
            SampleServiceClient::connect(format!("http://localhost:{}", config.grpc_port))
                .await
                .unwrap();

        // 送信するメッセージリスト作成
        let message_lists = vec![
            HelloClientStreamRequestBody {
                text: "A".to_string(),
            },
            HelloClientStreamRequestBody {
                text: "B".to_string(),
            },
            HelloClientStreamRequestBody {
                text: "C".to_string(),
            },
        ];

        // ストリームの作成
        let mut request = Request::new(iter(message_lists));

        // メタデータにBearerトークンを追加
        request
            .metadata_mut()
            .insert("authorization", "Bearer token".parse().unwrap());

        // テストの実行
        let res = client.hello_client_stream(request).await;

        // 検証
        assert!(res.is_ok());
        let res_body = res.unwrap().into_inner();
        assert_eq!(res_body.message, "A,B,C");
    }

    #[tokio::test]
    async fn hello_bidirectional_stream_should_return_succeed() {
        let config = get_config();

        // サーバーのインスタンス作成
        let mut client =
            SampleServiceClient::connect(format!("http://localhost:{}", config.grpc_port))
                .await
                .unwrap();

        // 送信するメッセージリスト作成
        let message_lists = vec![
            HelloBidirectionalStreamRequestBody {
                text: "A".to_string(),
            },
            HelloBidirectionalStreamRequestBody {
                text: "B".to_string(),
            },
        ];

        // ストリームの作成
        let mut request = Request::new(iter(message_lists.clone()));

        // メタデータにBearerトークンを追加
        request
            .metadata_mut()
            .insert("authorization", "Bearer token".parse().unwrap());

        // テストの実行
        let res = client.hello_bidirectional_stream(request).await;

        // 検証
        assert!(res.is_ok());

        // ストリームの検証
        let mut stream = res.unwrap().into_inner();
        let mut i = 0;
        while let Some(result) = stream.next().await {
            match result {
                Ok(res) => {
                    let msg = message_lists[i].text.clone();
                    assert_eq!(res.message, msg);
                }
                Err(_) => {}
            }
            i += 1;
        }
    }
}

// tonic
use tonic::{Request, Response, Status, Streaming};

// tokio
use tokio_stream::wrappers::ReceiverStream;

// バリデーション用のトレイト
use prost_validate::Validator;

// 共通コンテキスト
use crate::contexts::context::create_context;

// ロガー
use crate::loggers::logger;

// ビルドしたprotoファイルのインポート
pub mod sample_proto {
    // protoファイルに定義したpakage名を指定
    tonic::include_proto!("sample");
}

// リポジトリ
use crate::repositories::sample::sample_repository::SampleRepository;

// サービス
use crate::services::sample::sample_service::{SampleCommonRepository, SampleService};

// ユースケース
use crate::usecases::sample::hello_add_text_usecase::SampleHelloAddTextUsecase;
use crate::usecases::sample::hello_bidirectional_stream_usecase::SampleHelloBidirectionalStreamUsecase;
use crate::usecases::sample::hello_client_stream_usecase::SampleHelloClientStreamUsecase;
use crate::usecases::sample::hello_server_stream_usecase::SampleHelloServerStreamUsecase;
use crate::usecases::sample::hello_usecase::{SampleCommonService, SampleHelloUsecase};

// 構造体定義
#[derive(Debug, Default)]
pub struct SampleServer {}

// protoファイルの関数の実装をメソッド定義
#[tonic::async_trait]
impl sample_proto::sample_service_server::SampleService for SampleServer {
    async fn hello(
        &self,
        request: Request<sample_proto::Empty>,
    ) -> Result<Response<sample_proto::HelloResponseBody>, Status> {
        // コンテキスト設定
        let ctx = create_context(&request);

        // インスタンス化
        let sample_repo = Box::new(SampleRepository::new());
        let sample_common_repo = SampleCommonRepository { sample_repo };
        let sample_service = SampleService::new(sample_common_repo);
        let sample_common_service = SampleCommonService { sample_service };
        let usecase = SampleHelloUsecase {
            service: sample_common_service,
        };

        // ユースケースの実行
        usecase.exec(ctx).await
    }

    async fn hello_add_text(
        &self,
        request: Request<sample_proto::HelloAddTextRequestBody>,
    ) -> Result<Response<sample_proto::HelloAddTextResponseBody>, Status> {
        // コンテキスト設定
        let ctx = create_context(&request);

        // リクエストボディを取得
        let req_body = request.into_inner();

        // バリデーションチェック
        match req_body.validate() {
            Ok(_) => {}
            Err(e) => {
                let msg = format!("バリデーションエラー: {}", e);
                logger::error(&ctx, &msg);
                return Err(Status::invalid_argument(msg));
            }
        }

        // インスタンス化
        let usecase = SampleHelloAddTextUsecase {};

        // ユースケースの実行
        usecase.exec(ctx, req_body).await
    }

    // サーバーストリーミングの追加（typeの定義必須）
    // ※protoファイルで定義した名称+Streamという型の定義が必要になる
    type HelloServerStreamStream =
        ReceiverStream<Result<sample_proto::HelloServerStreamResponseBody, Status>>;

    async fn hello_server_stream(
        &self,
        request: Request<sample_proto::HelloServerStreamRequestBody>,
    ) -> Result<Response<Self::HelloServerStreamStream>, Status> {
        // コンテキスト設定
        let ctx = create_context(&request);

        // リクエストボディを取得
        let req_body = request.into_inner();

        // バリデーションチェック
        match req_body.validate() {
            Ok(_) => {}
            Err(e) => {
                let msg = format!("バリデーションエラー: {}", e);
                logger::error(&ctx, &msg);
                return Err(Status::invalid_argument(msg));
            }
        }

        // インスタンス化
        let usecase = SampleHelloServerStreamUsecase {};

        // ユースケースの実行
        usecase.exec(ctx, req_body).await
    }

    // クライアントストリーミングの追加
    async fn hello_client_stream(
        &self,
        request: Request<Streaming<sample_proto::HelloClientStreamRequestBody>>,
    ) -> Result<Response<sample_proto::HelloClientStreamResponseBody>, Status> {
        // インスタンス化
        let usecase = SampleHelloClientStreamUsecase {};

        // ユースケースの実行
        usecase.exec(request).await
    }

    // 双方向ストリーミングの追加（typeの定義必須）
    // ※protoファイルで定義した名称+Streamという型の定義が必要になる
    type HelloBidirectionalStreamStream =
        ReceiverStream<Result<sample_proto::HelloBidirectionalStreamResponseBody, Status>>;

    async fn hello_bidirectional_stream(
        &self,
        request: Request<Streaming<sample_proto::HelloBidirectionalStreamRequestBody>>,
    ) -> Result<Response<Self::HelloBidirectionalStreamStream>, Status> {
        // インスタンス化
        let usecase = SampleHelloBidirectionalStreamUsecase {};

        // ユースケースの実行
        usecase.exec(request).await
    }
}

// ルーターに設定するサーバー定義を返す関数
pub fn get_sample_server() -> sample_proto::sample_service_server::SampleServiceServer<SampleServer>
{
    let sample_server = SampleServer::default();
    sample_proto::sample_service_server::SampleServiceServer::new(sample_server)
}

# ####################
# # ビルドステージ
# ####################
FROM rust:1.87.0-alpine3.21 AS builder

WORKDIR /build

# ビルドに必要なパッケージをインストール
RUN apk update && \
    apk add --no-cache \
        openssl-dev \
        alpine-sdk \
        protobuf-dev \
        curl

# gRPCのバリデーション用にgoをインストール
# goのバージョンを変更したい場合は以下リンク先のページで必要情報をご確認下さい。
# https://go.dev/dl/
ENV GO_VERSION=1.24.4
ENV GO_SHA256=d5501ee5aca0f258d5fe9bfaed401958445014495dc115f202d43d5210b45241
ENV GO_OS=linux
ENV GO_ARCH=arm64
ENV PATH="/usr/local/go/bin:/root/go/bin:${PATH}"

RUN curl -fsSLO "https://go.dev/dl/go${GO_VERSION}.${GO_OS}-${GO_ARCH}.tar.gz" && \
    echo "${GO_SHA256} go${GO_VERSION}.${GO_OS}-${GO_ARCH}.tar.gz" | sha256sum -c - && \
    tar -C /usr/local -xzf "go${GO_VERSION}.${GO_OS}-${GO_ARCH}.tar.gz" && \
    rm "go${GO_VERSION}.${GO_OS}-${GO_ARCH}.tar.gz" && \
    rm -rf /var/lib/apt/lists/*

# バリデーション用にgoのライブラリをインストール
RUN go install github.com/envoyproxy/protoc-gen-validate@v1.2.1

COPY . .

# ビルド
RUN cargo build --release --locked

# ####################
# # 実行ステージ
# ####################
FROM alpine:3.21 AS runner

WORKDIR /app

# コンテナ用ユーザー作成
RUN addgroup --system --gid 1001 appuser && \
adduser --system --uid 1001 appuser

# ビルドステージで作成したバイナリをコピー
COPY --from=builder --chown=appuser:appuser /build/target/release/rust_grpc .

# ポートを設定
EXPOSE 50051

# コンテナ起動ユーザー設定
USER appuser

# APIサーバー起動コマンド
CMD ["./rust_grpc"]