use reqwest::{Client, ClientBuilder, redirect::Policy};
use std::time::Duration;
 
/// 本番向けの Client 構築例
fn build_production_client() -> reqwest::Result<Client> {
    ClientBuilder::new()
        // タイムアウト設定
        .timeout(Duration::from_secs(30))           // リクエスト全体のタイムアウト
        .connect_timeout(Duration::from_secs(5))    // TCP接続確立のタイムアウト
        .read_timeout(Duration::from_secs(15))      // 読み取りタイムアウト
 
        // 接続プール
        .pool_max_idle_per_host(20)                  // ホストごとのアイドル接続数上限
        .pool_idle_timeout(Duration::from_secs(90))  // アイドル接続のTTL
 
        // TLS
        .min_tls_version(reqwest::tls::Version::TLS_1_2)
        .danger_accept_invalid_certs(false)          // 本番では必ず false
 
        // リダイレクト
        .redirect(Policy::limited(10))               // 最大10回のリダイレクト
 
        // ヘッダー
        .user_agent("my-rust-app/1.0")
        .default_headers({
            let mut headers = reqwest::header::HeaderMap::new();
            headers.insert("Accept", "application/json".parse().unwrap());
            headers
        })
 
        // 圧縮
        .gzip(true)
        .brotli(true)
        .deflate(true)
 
        .build()
}

use reqwest::Client;
use serde::{Deserialize, Serialize};
use std::time::Duration;
 
#[derive(Debug, Deserialize)]
struct User {
    id: u64,
    login: String,
    name: Option<String>,
    public_repos: Option<u32>,
}
 
#[derive(Serialize)]
struct CreateIssue {
    title: String,
    body: String,
    labels: Vec<String>,
}
 
#[derive(Debug, Deserialize)]
struct ApiResponse<T> {
    data: T,
    total_count: Option<u64>,
}
 
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // クライアントの構築 (接続プール付き — 再利用すべき)
    let client = Client::builder()
        .timeout(Duration::from_secs(30))
        .connect_timeout(Duration::from_secs(5))
        .pool_max_idle_per_host(10)
        .user_agent("my-rust-app/1.0")
        .build()?;
 
    // GET リクエスト + JSON デシリアライズ
    let user: User = client
        .get("https://api.github.com/users/rust-lang")
        .header("Accept", "application/json")
        .send()
        .await?
        .error_for_status()?  // 4xx/5xx をエラーに変換
        .json()
        .await?;
    println!("User: {} ({})", user.login, user.name.unwrap_or_default());
 
    // POST リクエスト
    let issue = CreateIssue {
        title: "Bug report".into(),
        body: "Description here".into(),
        labels: vec!["bug".into(), "triage".into()],
    };
    let response = client
        .post("https://api.example.com/issues")
        .bearer_auth("your-token")
        .json(&issue)
        .send()
        .await?;
    println!("Status: {}", response.status());
 
    // PUT リクエスト
    let response = client
        .put("https://api.example.com/users/1")
        .json(&serde_json::json!({
            "name": "Updated Name",
            "email": "new@example.com"
        }))
        .send()
        .await?;
    println!("PUT Status: {}", response.status());
 
    // DELETE リクエスト
    let response = client
        .delete("https://api.example.com/users/1")
        .bearer_auth("your-token")
        .send()
        .await?;
    println!("DELETE Status: {}", response.status());
 
    // PATCH リクエスト
    let response = client
        .patch("https://api.example.com/users/1")
        .json(&serde_json::json!({"name": "Patched"}))
        .send()
        .await?;
    println!("PATCH Status: {}", response.status());
 
    Ok(())
}

use reqwest::{Client, StatusCode};
use std::collections::HashMap;
 
async fn inspect_response(client: &Client, url: &str) -> anyhow::Result<()> {
    let response = client.get(url).send().await?;
 
    // ステータスコード
    let status = response.status();
    println!("Status: {} ({})", status.as_u16(), status.canonical_reason().unwrap_or(""));
 
    // レスポンスヘッダー
    for (name, value) in response.headers() {
        println!("  {}: {}", name, value.to_str().unwrap_or("<binary>"));
    }
 
    // Content-Type に基づく処理分岐
    let content_type = response.headers()
        .get("content-type")
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
 
    if content_type.contains("application/json") {
        let json: serde_json::Value = response.json().await?;
        println!("JSON: {}", serde_json::to_string_pretty(&json)?);
    } else if content_type.contains("text/") {
        let text = response.text().await?;
        println!("Text ({} bytes): {}", text.len(), &text[..200.min(text.len())]);
    } else {
        let bytes = response.bytes().await?;
        println!("Binary: {} bytes", bytes.len());
    }
 
    Ok(())
}
 
/// ステータスコード別のエラーハンドリング
async fn handle_api_response(client: &Client, url: &str) -> anyhow::Result<String> {
    let response = client.get(url).send().await?;
 
    match response.status() {
        StatusCode::OK => Ok(response.text().await?),
        StatusCode::NOT_FOUND => {
            anyhow::bail!("リソースが見つかりません: {}", url);
        }
        StatusCode::UNAUTHORIZED => {
            anyhow::bail!("認証が必要です");
        }
        StatusCode::TOO_MANY_REQUESTS => {
            // Rate limit — Retry-After ヘッダーを確認
            let retry_after = response.headers()
                .get("retry-after")
                .and_then(|v| v.to_str().ok())
                .and_then(|v| v.parse::<u64>().ok())
                .unwrap_or(60);
            anyhow::bail!("レート制限。{}秒後にリトライしてください", retry_after);
        }
        status if status.is_server_error() => {
            let body = response.text().await.unwrap_or_default();
            anyhow::bail!("サーバーエラー {}: {}", status, body);
        }
        status => {
            anyhow::bail!("予期しないステータス: {}", status);
        }
    }
}

use futures::StreamExt;
use tokio::io::AsyncWriteExt;
 
async fn download_file(url: &str, path: &str) -> anyhow::Result<()> {
    let client = reqwest::Client::new();
    let response = client.get(url).send().await?.error_for_status()?;
 
    let total_size = response.content_length().unwrap_or(0);
    let mut file = tokio::fs::File::create(path).await?;
    let mut downloaded: u64 = 0;
 
    let mut stream = response.bytes_stream();
    while let Some(chunk) = stream.next().await {
        let chunk = chunk?;
        file.write_all(&chunk).await?;
        downloaded += chunk.len() as u64;
        if total_size > 0 {
            let pct = (downloaded as f64 / total_size as f64) * 100.0;
            eprint!("\r進捗: {:.1}%", pct);
        }
    }
    eprintln!("\n完了: {} バイト", downloaded);
 
    Ok(())
}
 
/// レジューム対応ダウンロード
async fn resumable_download(url: &str, path: &str) -> anyhow::Result<()> {
    let client = reqwest::Client::new();
 
    // 既存ファイルのサイズを確認
    let existing_size = match tokio::fs::metadata(path).await {
        Ok(meta) => meta.len(),
        Err(_) => 0,
    };
 
    let mut request = client.get(url);
    if existing_size > 0 {
        // Range ヘッダーで途中から再開
        request = request.header("Range", format!("bytes={}-", existing_size));
        println!("{}バイトから再開", existing_size);
    }
 
    let response = request.send().await?.error_for_status()?;
 
    // 206 Partial Content の確認
    let is_partial = response.status() == reqwest::StatusCode::PARTIAL_CONTENT;
    let total_size = if is_partial {
        // Content-Range: bytes 1000-9999/10000 からトータルサイズを取得
        response.headers()
            .get("content-range")
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.split('/').last())
            .and_then(|v| v.parse::<u64>().ok())
            .unwrap_or(0)
    } else {
        response.content_length().unwrap_or(0)
    };
 
    let mut file = tokio::fs::OpenOptions::new()
        .create(true)
        .append(is_partial)
        .write(!is_partial)
        .truncate(!is_partial)
        .open(path)
        .await?;
 
    let mut downloaded = if is_partial { existing_size } else { 0 };
 
    let mut stream = response.bytes_stream();
    while let Some(chunk) = stream.next().await {
        let chunk = chunk?;
        file.write_all(&chunk).await?;
        downloaded += chunk.len() as u64;
        if total_size > 0 {
            let pct = (downloaded as f64 / total_size as f64) * 100.0;
            eprint!("\r進捗: {:.1}% ({}/{})", pct, downloaded, total_size);
        }
    }
    eprintln!("\nダウンロード完了: {} バイト", downloaded);
 
    Ok(())
}

use futures::stream::{self, StreamExt};
use reqwest::Client;
use std::sync::Arc;
use tokio::sync::Semaphore;
 
/// 並行数を制限した一括リクエスト (Semaphore方式)
async fn fetch_all_with_limit(
    client: &Client,
    urls: Vec<String>,
    max_concurrent: usize,
) -> Vec<Result<String, reqwest::Error>> {
    let semaphore = Arc::new(Semaphore::new(max_concurrent));
 
    let tasks: Vec<_> = urls.into_iter().map(|url| {
        let client = client.clone();
        let semaphore = semaphore.clone();
        tokio::spawn(async move {
            let _permit = semaphore.acquire().await.unwrap();
            client.get(&url).send().await?.text().await
        })
    }).collect();
 
    let mut results = Vec::new();
    for task in tasks {
        match task.await {
            Ok(result) => results.push(result),
            Err(e) => results.push(Err(reqwest::Error::from(
                std::io::Error::new(std::io::ErrorKind::Other, e)
            ).into())),
        }
    }
    results
}
 
/// buffer_unordered を使った並行リクエスト (Stream方式)
async fn fetch_all_stream(
    client: &Client,
    urls: Vec<String>,
    concurrency: usize,
) -> Vec<(String, Result<String, String>)> {
    stream::iter(urls)
        .map(|url| {
            let client = client.clone();
            async move {
                let result = client.get(&url)
                    .send()
                    .await
                    .and_then(|r| Ok(r))
                    .map_err(|e| e.to_string());
 
                let body = match result {
                    Ok(resp) => resp.text().await.map_err(|e| e.to_string()),
                    Err(e) => Err(e),
                };
                (url, body)
            }
        })
        .buffer_unordered(concurrency)
        .collect()
        .await
}
 
/// レート制限付き連続リクエスト
async fn rate_limited_requests(
    client: &Client,
    urls: Vec<String>,
    requests_per_second: u32,
) -> Vec<Result<String, reqwest::Error>> {
    let interval = std::time::Duration::from_millis(1000 / requests_per_second as u64);
    let mut results = Vec::new();
 
    for url in urls {
        let start = std::time::Instant::now();
        let result = client.get(&url).send().await?.text().await;
        results.push(result);
 
        // 次のリクエストまで待機
        let elapsed = start.elapsed();
        if elapsed < interval {
            tokio::time::sleep(interval - elapsed).await;
        }
    }
 
    results
}

use reqwest::{Client, multipart};
use tokio::fs::File;
use tokio::io::AsyncReadExt;
 
async fn upload_file(
    client: &Client,
    url: &str,
    file_path: &str,
    field_name: &str,
) -> anyhow::Result<reqwest::Response> {
    // ファイル読み込み
    let mut file = File::open(file_path).await?;
    let mut buffer = Vec::new();
    file.read_to_end(&mut buffer).await?;
 
    let file_name = std::path::Path::new(file_path)
        .file_name()
        .and_then(|n| n.to_str())
        .unwrap_or("upload")
        .to_string();
 
    // MIME タイプ推定
    let mime_type = match file_path.rsplit('.').next() {
        Some("png") => "image/png",
        Some("jpg") | Some("jpeg") => "image/jpeg",
        Some("pdf") => "application/pdf",
        Some("json") => "application/json",
        Some("csv") => "text/csv",
        _ => "application/octet-stream",
    };
 
    let part = multipart::Part::bytes(buffer)
        .file_name(file_name)
        .mime_str(mime_type)?;
 
    let form = multipart::Form::new()
        .part(field_name.to_string(), part)
        .text("description", "Uploaded from Rust client");
 
    let response = client
        .post(url)
        .multipart(form)
        .send()
        .await?
        .error_for_status()?;
 
    Ok(response)
}
 
/// 複数ファイルの同時アップロード
async fn upload_multiple_files(
    client: &Client,
    url: &str,
    file_paths: &[&str],
) -> anyhow::Result<reqwest::Response> {
    let mut form = multipart::Form::new();
 
    for (i, path) in file_paths.iter().enumerate() {
        let mut file = File::open(path).await?;
        let mut buffer = Vec::new();
        file.read_to_end(&mut buffer).await?;
 
        let file_name = std::path::Path::new(path)
            .file_name()
            .and_then(|n| n.to_str())
            .unwrap_or("file")
            .to_string();
 
        let part = multipart::Part::bytes(buffer)
            .file_name(file_name);
 
        form = form.part(format!("file_{}", i), part);
    }
 
    let response = client
        .post(url)
        .multipart(form)
        .send()
        .await?
        .error_for_status()?;
 
    Ok(response)
}

use reqwest::{Client, Response, StatusCode};
use std::time::Duration;
 
/// 指数バックオフ付きリトライ
async fn retry_request<F, Fut>(
    max_retries: u32,
    base_delay: Duration,
    mut request_fn: F,
) -> anyhow::Result<Response>
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = reqwest::Result<Response>>,
{
    let mut last_error = None;
 
    for attempt in 0..=max_retries {
        if attempt > 0 {
            let delay = base_delay * 2u32.pow(attempt - 1);
            // ジッタを追加 (0.5 ~ 1.5 倍)
            let jitter = rand::random::<f64>() + 0.5;
            let actual_delay = Duration::from_millis(
                (delay.as_millis() as f64 * jitter) as u64
            );
            tracing::warn!(
                "リトライ {}/{}: {}ms 後に再試行",
                attempt, max_retries, actual_delay.as_millis()
            );
            tokio::time::sleep(actual_delay).await;
        }
 
        match (request_fn)().await {
            Ok(response) => {
                let status = response.status();
                // リトライ可能なステータスコード
                if status == StatusCode::TOO_MANY_REQUESTS
                    || status == StatusCode::SERVICE_UNAVAILABLE
                    || status == StatusCode::GATEWAY_TIMEOUT
                    || status == StatusCode::BAD_GATEWAY
                {
                    last_error = Some(anyhow::anyhow!("リトライ可能なエラー: {}", status));
                    continue;
                }
                return Ok(response);
            }
            Err(e) => {
                if e.is_connect() || e.is_timeout() {
                    last_error = Some(anyhow::anyhow!("接続エラー: {}", e));
                    continue;
                }
                return Err(e.into());
            }
        }
    }
 
    Err(last_error.unwrap_or_else(|| anyhow::anyhow!("最大リトライ回数超過")))
}
 
/// 使用例
async fn fetch_with_retry(client: &Client, url: &str) -> anyhow::Result<String> {
    let url = url.to_string();
    let client = client.clone();
 
    let response = retry_request(3, Duration::from_millis(500), || {
        let client = client.clone();
        let url = url.clone();
        async move {
            client.get(&url).send().await
        }
    }).await?;
 
    Ok(response.text().await?)
}

use hyper::{body::Incoming, server::conn::http1, Request, Response};
use hyper::body::Bytes;
use http_body_util::Full;
use hyper::service::service_fn;
use hyper_util::rt::TokioIo;
use std::convert::Infallible;
use std::net::SocketAddr;
use tokio::net::TcpListener;
 
async fn handle_request(
    req: Request<Incoming>,
) -> Result<Response<Full<Bytes>>, Infallible> {
    let (method, path) = (req.method().clone(), req.uri().path().to_string());
 
    let response = match (method.as_str(), path.as_str()) {
        ("GET", "/") => {
            Response::builder()
                .status(200)
                .header("Content-Type", "text/plain; charset=utf-8")
                .body(Full::new(Bytes::from("Hello from hyper!")))
                .unwrap()
        }
        ("GET", "/health") => {
            Response::builder()
                .status(200)
                .header("Content-Type", "application/json")
                .body(Full::new(Bytes::from(r#"{"status":"ok"}"#)))
                .unwrap()
        }
        _ => {
            Response::builder()
                .status(404)
                .body(Full::new(Bytes::from("Not Found")))
                .unwrap()
        }
    };
 
    Ok(response)
}
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = SocketAddr::from(([0, 0, 0, 0], 3000));
    let listener = TcpListener::bind(addr).await?;
    println!("Listening on http://{}", addr);
 
    loop {
        let (stream, remote_addr) = listener.accept().await?;
        let io = TokioIo::new(stream);
 
        tokio::spawn(async move {
            if let Err(err) = http1::Builder::new()
                .serve_connection(io, service_fn(handle_request))
                .await
            {
                eprintln!("Error serving {}: {}", remote_addr, err);
            }
        });
    }
}

use hyper::{body::Incoming, server::conn::http2, Request, Response};
use hyper::body::Bytes;
use http_body_util::Full;
use hyper::service::service_fn;
use hyper_util::rt::{TokioIo, TokioExecutor};
use std::convert::Infallible;
use tokio::net::TcpListener;
 
async fn handle(
    _req: Request<Incoming>,
) -> Result<Response<Full<Bytes>>, Infallible> {
    Ok(Response::new(Full::new(Bytes::from("Hello HTTP/2!"))))
}
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("0.0.0.0:3000").await?;
 
    loop {
        let (stream, _) = listener.accept().await?;
        let io = TokioIo::new(stream);
 
        tokio::spawn(async move {
            // HTTP/2 は h2c (cleartext) モードで起動
            // 本番では TLS が必要
            if let Err(err) = http2::Builder::new(TokioExecutor::new())
                .serve_connection(io, service_fn(handle))
                .await
            {
                eprintln!("Error: {}", err);
            }
        });
    }
}

use hyper::{body::Incoming, server::conn::http1, Request, Response};
use hyper::body::Bytes;
use http_body_util::Full;
use hyper::service::service_fn;
use hyper_util::rt::TokioIo;
use std::convert::Infallible;
use tokio::net::TcpListener;
use tokio::signal;
 
async fn handle(
    _req: Request<Incoming>,
) -> Result<Response<Full<Bytes>>, Infallible> {
    Ok(Response::new(Full::new(Bytes::from("OK"))))
}
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("0.0.0.0:3000").await?;
    let (shutdown_tx, mut shutdown_rx) = tokio::sync::broadcast::channel::<()>(1);
 
    println!("サーバー起動。Ctrl+C で終了");
 
    // Ctrl+C ハンドラ
    let shutdown_tx_clone = shutdown_tx.clone();
    tokio::spawn(async move {
        signal::ctrl_c().await.unwrap();
        println!("\nシャットダウン開始...");
        let _ = shutdown_tx_clone.send(());
    });
 
    loop {
        tokio::select! {
            Ok((stream, addr)) = listener.accept() => {
                let io = TokioIo::new(stream);
                let mut shutdown_rx = shutdown_tx.subscribe();
 
                tokio::spawn(async move {
                    let conn = http1::Builder::new()
                        .serve_connection(io, service_fn(handle));
 
                    // graceful shutdown: 進行中のリクエストの完了を待つ
                    tokio::pin!(conn);
                    tokio::select! {
                        result = &mut conn => {
                            if let Err(e) = result {
                                eprintln!("Error serving {}: {}", addr, e);
                            }
                        }
                        _ = shutdown_rx.recv() => {
                            conn.as_mut().graceful_shutdown();
                            // 残りの処理を完了させる
                            if let Err(e) = conn.await {
                                eprintln!("Error during shutdown: {}", e);
                            }
                        }
                    }
                });
            }
            _ = shutdown_rx.recv() => {
                println!("新規接続の受付を停止");
                break;
            }
        }
    }
 
    println!("サーバー停止完了");
    Ok(())
}

use futures::{SinkExt, StreamExt};
use tokio_tungstenite::{connect_async, tungstenite::Message};
 
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let url = "wss://echo.websocket.org";
    let (ws_stream, response) = connect_async(url).await?;
    println!("接続成功: {}", response.status());
 
    let (mut write, mut read) = ws_stream.split();
 
    // 送信タスク
    let send_task = tokio::spawn(async move {
        for i in 1..=5 {
            let msg = Message::Text(format!("メッセージ #{}", i));
            write.send(msg).await.unwrap();
            tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
        }
        write.send(Message::Close(None)).await.unwrap();
    });
 
    // 受信タスク
    let recv_task = tokio::spawn(async move {
        while let Some(msg) = read.next().await {
            match msg {
                Ok(Message::Text(text)) => println!("受信: {}", text),
                Ok(Message::Binary(data)) => println!("バイナリ: {} bytes", data.len()),
                Ok(Message::Ping(data)) => println!("Ping: {:?}", data),
                Ok(Message::Close(_)) => {
                    println!("切断");
                    break;
                }
                Err(e) => {
                    eprintln!("エラー: {}", e);
                    break;
                }
                _ => {}
            }
        }
    });
 
    let _ = tokio::join!(send_task, recv_task);
    Ok(())
}

use futures::{SinkExt, StreamExt};
use tokio_tungstenite::{connect_async, tungstenite::Message};
use std::time::Duration;
use tokio::sync::mpsc;
 
/// 再接続ロジック付きWebSocketクライアント
struct ReconnectingWsClient {
    url: String,
    max_retries: u32,
    base_delay: Duration,
}
 
impl ReconnectingWsClient {
    fn new(url: &str) -> Self {
        Self {
            url: url.to_string(),
            max_retries: 10,
            base_delay: Duration::from_secs(1),
        }
    }
 
    async fn run(
        &self,
        outgoing_rx: &mut mpsc::Receiver<String>,
        incoming_tx: mpsc::Sender<String>,
    ) {
        let mut retry_count = 0u32;
 
        loop {
            match self.connect_and_handle(outgoing_rx, &incoming_tx).await {
                Ok(()) => {
                    println!("WebSocket 正常切断");
                    break;
                }
                Err(e) => {
                    retry_count += 1;
                    if retry_count > self.max_retries {
                        eprintln!("最大リトライ回数超過: {}", e);
                        break;
                    }
 
                    let delay = self.base_delay * 2u32.pow((retry_count - 1).min(6));
                    eprintln!(
                        "接続エラー ({}回目): {}。{}秒後に再接続...",
                        retry_count, e, delay.as_secs()
                    );
                    tokio::time::sleep(delay).await;
                }
            }
        }
    }
 
    async fn connect_and_handle(
        &self,
        outgoing_rx: &mut mpsc::Receiver<String>,
        incoming_tx: &mpsc::Sender<String>,
    ) -> anyhow::Result<()> {
        let (ws_stream, _) = connect_async(&self.url).await?;
        println!("WebSocket 接続成功: {}", self.url);
 
        let (mut write, mut read) = ws_stream.split();
 
        loop {
            tokio::select! {
                // サーバーからのメッセージ受信
                msg = read.next() => {
                    match msg {
                        Some(Ok(Message::Text(text))) => {
                            if incoming_tx.send(text).await.is_err() {
                                break; // 受信側がドロップされた
                            }
                        }
                        Some(Ok(Message::Ping(data))) => {
                            write.send(Message::Pong(data)).await?;
                        }
                        Some(Ok(Message::Close(_))) => {
                            return Ok(()); // 正常切断
                        }
                        Some(Err(e)) => {
                            return Err(e.into());
                        }
                        None => {
                            return Err(anyhow::anyhow!("ストリーム終了"));
                        }
                        _ => {}
                    }
                }
                // アプリからの送信メッセージ
                msg = outgoing_rx.recv() => {
                    match msg {
                        Some(text) => {
                            write.send(Message::Text(text)).await?;
                        }
                        None => {
                            // 送信チャネルがクローズ
                            write.send(Message::Close(None)).await?;
                            return Ok(());
                        }
                    }
                }
            }
        }
 
        Ok(())
    }
}

use axum::{
    extract::ws::{Message, WebSocket, WebSocketUpgrade},
    extract::State,
    response::IntoResponse,
    routing::get,
    Router,
};
use std::sync::Arc;
use tokio::sync::broadcast;
 
/// 共有状態: ブロードキャストチャネルでチャットルームを実現
#[derive(Clone)]
struct ChatState {
    tx: broadcast::Sender<String>,
}
 
async fn ws_handler(
    ws: WebSocketUpgrade,
    State(state): State<ChatState>,
) -> impl IntoResponse {
    ws.on_upgrade(move |socket| handle_socket(socket, state))
}
 
async fn handle_socket(socket: WebSocket, state: ChatState) {
    let (mut sender, mut receiver) = socket.split();
    let mut rx = state.tx.subscribe();
 
    // サーバー → クライアント (ブロードキャスト受信 → WebSocket送信)
    let mut send_task = tokio::spawn(async move {
        while let Ok(msg) = rx.recv().await {
            if sender.send(Message::Text(msg)).await.is_err() {
                break;
            }
        }
    });
 
    // クライアント → サーバー (WebSocket受信 → ブロードキャスト送信)
    let tx = state.tx.clone();
    let mut recv_task = tokio::spawn(async move {
        while let Some(Ok(msg)) = receiver.next().await {
            if let Message::Text(text) = msg {
                let _ = tx.send(text);
            }
        }
    });
 
    // どちらかのタスクが終了したら両方を停止
    tokio::select! {
        _ = &mut send_task => recv_task.abort(),
        _ = &mut recv_task => send_task.abort(),
    }
}
 
// WebSocket サーバーの起動
// #[tokio::main]
// async fn main() {
//     let (tx, _) = broadcast::channel(100);
//     let state = ChatState { tx };
//     let app = Router::new()
//         .route("/ws", get(ws_handler))
//         .with_state(state);
//     let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
//     axum::serve(listener, app).await.unwrap();
// }

use futures::{SinkExt, StreamExt};
use tokio_tungstenite::tungstenite::Message;
use std::time::Duration;
 
/// Heartbeat 付き WebSocket 接続管理
async fn ws_with_heartbeat(
    ws_stream: tokio_tungstenite::WebSocketStream<
        tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>
    >,
    heartbeat_interval: Duration,
    pong_timeout: Duration,
) -> anyhow::Result<()> {
    let (mut write, mut read) = ws_stream.split();
    let mut heartbeat = tokio::time::interval(heartbeat_interval);
    let mut last_pong = std::time::Instant::now();
 
    loop {
        tokio::select! {
            // 受信メッセージの処理
            msg = read.next() => {
                match msg {
                    Some(Ok(Message::Text(text))) => {
                        println!("受信: {}", text);
                    }
                    Some(Ok(Message::Pong(_))) => {
                        last_pong = std::time::Instant::now();
                    }
                    Some(Ok(Message::Ping(data))) => {
                        write.send(Message::Pong(data)).await?;
                    }
                    Some(Ok(Message::Close(_))) | None => {
                        println!("接続終了");
                        break;
                    }
                    Some(Err(e)) => {
                        eprintln!("エラー: {}", e);
                        break;
                    }
                    _ => {}
                }
            }
 
            // 定期 Heartbeat
            _ = heartbeat.tick() => {
                // Pong タイムアウトチェック
                if last_pong.elapsed() > pong_timeout {
                    eprintln!("Pong タイムアウト。接続切断");
                    break;
                }
                write.send(Message::Ping(vec![].into())).await?;
            }
        }
    }
 
    // クリーンな切断
    let _ = write.send(Message::Close(None)).await;
    Ok(())
}

// proto/greeter.proto
syntax = "proto3";
package greeter;
 
service Greeter {
    // Unary RPC: 単一リクエスト → 単一レスポンス
    rpc SayHello (HelloRequest) returns (HelloReply);
 
    // Server Streaming RPC: 単一リクエスト → ストリームレスポンス
    rpc SayHelloStream (HelloRequest) returns (stream HelloReply);
 
    // Client Streaming RPC: ストリームリクエスト → 単一レスポンス
    rpc RecordGreetings (stream HelloRequest) returns (GreetingSummary);
 
    // Bidirectional Streaming RPC: ストリーム ↔ ストリーム
    rpc Chat (stream ChatMessage) returns (stream ChatMessage);
}
 
message HelloRequest {
    string name = 1;
}
 
message HelloReply {
    string message = 1;
    int64 timestamp = 2;
}
 
message GreetingSummary {
    int32 count = 1;
    string summary = 2;
}
 
message ChatMessage {
    string user = 1;
    string content = 2;
    int64 timestamp = 3;
}

// build.rs
fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic_build::configure()
        .build_server(true)
        .build_client(true)
        .out_dir("src/generated")   // 生成先 (任意)
        .compile_protos(
            &["proto/greeter.proto"],
            &["proto"],
        )?;
    Ok(())
}

// src/server.rs
use tonic::{transport::Server, Request, Response, Status, Streaming};
use greeter::greeter_server::{Greeter, GreeterServer};
use greeter::{HelloReply, HelloRequest, GreetingSummary, ChatMessage};
use tokio_stream::wrappers::ReceiverStream;
use tokio_stream::StreamExt;
use std::time::{SystemTime, UNIX_EPOCH};
 
pub mod greeter {
    tonic::include_proto!("greeter");
}
 
fn now_timestamp() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_secs() as i64
}
 
#[derive(Debug, Default)]
pub struct MyGreeter;
 
#[tonic::async_trait]
impl Greeter for MyGreeter {
    // 1. Unary RPC
    async fn say_hello(
        &self,
        request: Request<HelloRequest>,
    ) -> Result<Response<HelloReply>, Status> {
        let metadata = request.metadata();
        // メタデータからの情報取得
        let request_id = metadata
            .get("x-request-id")
            .and_then(|v| v.to_str().ok())
            .unwrap_or("unknown");
        println!("Unary RPC: request_id={}", request_id);
 
        let name = request.into_inner().name;
        if name.is_empty() {
            return Err(Status::invalid_argument("name は必須です"));
        }
 
        let reply = HelloReply {
            message: format!("こんにちは、{}!", name),
            timestamp: now_timestamp(),
        };
        Ok(Response::new(reply))
    }
 
    // 2. Server Streaming RPC
    type SayHelloStreamStream = ReceiverStream<Result<HelloReply, Status>>;
 
    async fn say_hello_stream(
        &self,
        request: Request<HelloRequest>,
    ) -> Result<Response<Self::SayHelloStreamStream>, Status> {
        let name = request.into_inner().name;
        let (tx, rx) = tokio::sync::mpsc::channel(4);
 
        tokio::spawn(async move {
            for i in 1..=5 {
                let reply = HelloReply {
                    message: format!("{}回目: こんにちは、{}!", i, name),
                    timestamp: now_timestamp(),
                };
                if tx.send(Ok(reply)).await.is_err() { break; }
                tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
            }
        });
 
        Ok(Response::new(ReceiverStream::new(rx)))
    }
 
    // 3. Client Streaming RPC
    async fn record_greetings(
        &self,
        request: Request<Streaming<HelloRequest>>,
    ) -> Result<Response<GreetingSummary>, Status> {
        let mut stream = request.into_inner();
        let mut names = Vec::new();
 
        while let Some(req) = stream.next().await {
            let req = req?;
            println!("Client Stream 受信: {}", req.name);
            names.push(req.name);
        }
 
        let summary = GreetingSummary {
            count: names.len() as i32,
            summary: format!("受信した名前: {}", names.join(", ")),
        };
 
        Ok(Response::new(summary))
    }
 
    // 4. Bidirectional Streaming RPC
    type ChatStream = ReceiverStream<Result<ChatMessage, Status>>;
 
    async fn chat(
        &self,
        request: Request<Streaming<ChatMessage>>,
    ) -> Result<Response<Self::ChatStream>, Status> {
        let mut inbound = request.into_inner();
        let (tx, rx) = tokio::sync::mpsc::channel(32);
 
        tokio::spawn(async move {
            while let Some(result) = inbound.next().await {
                match result {
                    Ok(msg) => {
                        println!("[{}] {}", msg.user, msg.content);
                        // エコーバック (実際にはブロードキャストなど)
                        let reply = ChatMessage {
                            user: "Server".to_string(),
                            content: format!("Echo: {}", msg.content),
                            timestamp: now_timestamp(),
                        };
                        if tx.send(Ok(reply)).await.is_err() { break; }
                    }
                    Err(e) => {
                        eprintln!("ストリームエラー: {}", e);
                        break;
                    }
                }
            }
        });
 
        Ok(Response::new(ReceiverStream::new(rx)))
    }
}
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = "[::1]:50051".parse()?;
    println!("gRPC サーバー起動: {}", addr);
 
    Server::builder()
        .add_service(GreeterServer::new(MyGreeter::default()))
        .serve(addr)
        .await?;
    Ok(())
}

use greeter::greeter_client::GreeterClient;
use greeter::{HelloRequest, ChatMessage};
use tokio_stream::StreamExt;
 
pub mod greeter {
    tonic::include_proto!("greeter");
}
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut client = GreeterClient::connect("http://[::1]:50051").await?;
 
    // 1. Unary RPC
    println!("=== Unary RPC ===");
    let mut request = tonic::Request::new(HelloRequest {
        name: "Rust".into(),
    });
    // メタデータの付与
    request.metadata_mut().insert(
        "x-request-id",
        "req-12345".parse().unwrap(),
    );
    let response = client.say_hello(request).await?;
    println!("応答: {}", response.into_inner().message);
 
    // 2. Server Streaming RPC
    println!("\n=== Server Streaming RPC ===");
    let request = tonic::Request::new(HelloRequest {
        name: "World".into(),
    });
    let mut stream = client.say_hello_stream(request).await?.into_inner();
    while let Some(reply) = stream.next().await {
        let reply = reply?;
        println!("ストリーム: {} (ts={})", reply.message, reply.timestamp);
    }
 
    // 3. Client Streaming RPC
    println!("\n=== Client Streaming RPC ===");
    let names = vec!["Alice", "Bob", "Charlie"];
    let request_stream = tokio_stream::iter(
        names.into_iter().map(|name| HelloRequest { name: name.into() })
    );
    let response = client.record_greetings(request_stream).await?;
    let summary = response.into_inner();
    println!("サマリー: {} 件 — {}", summary.count, summary.summary);
 
    // 4. Bidirectional Streaming RPC
    println!("\n=== Bidirectional Streaming RPC ===");
    let (tx, rx) = tokio::sync::mpsc::channel(32);
 
    // 送信ストリーム
    tokio::spawn(async move {
        for i in 1..=3 {
            let msg = ChatMessage {
                user: "Client".to_string(),
                content: format!("メッセージ #{}", i),
                timestamp: 0,
            };
            tx.send(msg).await.unwrap();
            tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        }
    });
 
    let outbound = tokio_stream::wrappers::ReceiverStream::new(rx);
    let mut inbound = client.chat(outbound).await?.into_inner();
 
    while let Some(reply) = inbound.next().await {
        let reply = reply?;
        println!("[{}] {}", reply.user, reply.content);
    }
 
    Ok(())
}

use tonic::{transport::Channel, Request, Status};
use tonic::service::Interceptor;
 
/// 認証インターセプター
#[derive(Clone)]
struct AuthInterceptor {
    token: String,
}
 
impl Interceptor for AuthInterceptor {
    fn call(&mut self, mut request: Request<()>) -> Result<Request<()>, Status> {
        let token = format!("Bearer {}", self.token);
        request.metadata_mut().insert(
            "authorization",
            token.parse().map_err(|_| Status::internal("invalid token"))?,
        );
        // リクエストIDも付与
        let request_id = uuid::Uuid::new_v4().to_string();
        request.metadata_mut().insert(
            "x-request-id",
            request_id.parse().unwrap(),
        );
        Ok(request)
    }
}
 
/// サーバー側の認証チェック
fn check_auth(request: &Request<impl std::fmt::Debug>) -> Result<String, Status> {
    let token = request.metadata()
        .get("authorization")
        .and_then(|v| v.to_str().ok())
        .ok_or_else(|| Status::unauthenticated("認証トークンなし"))?;
 
    let token = token
        .strip_prefix("Bearer ")
        .ok_or_else(|| Status::unauthenticated("Bearer トークン形式不正"))?;
 
    // トークン検証 (実際にはJWT検証など)
    if token == "valid-token" {
        Ok("user-123".to_string())
    } else {
        Err(Status::unauthenticated("無効なトークン"))
    }
}
 
// 使用例
// let channel = Channel::from_static("http://[::1]:50051").connect().await?;
// let interceptor = AuthInterceptor { token: "valid-token".into() };
// let client = GreeterClient::with_interceptor(channel, interceptor);

use tonic::transport::{Server, ServerTlsConfig, Identity};
use tonic_health::server::HealthReporter;
 
async fn start_grpc_server() -> Result<(), Box<dyn std::error::Error>> {
    // TLS 設定
    let cert = tokio::fs::read("server.pem").await?;
    let key = tokio::fs::read("server.key").await?;
    let identity = Identity::from_pem(cert, key);
 
    let tls_config = ServerTlsConfig::new().identity(identity);
 
    // Health check サービス
    let (mut health_reporter, health_service) = tonic_health::server::health_reporter();
    health_reporter
        .set_serving::<GreeterServer<MyGreeter>>()
        .await;
 
    let addr = "[::1]:50051".parse()?;
 
    Server::builder()
        .tls_config(tls_config)?
        .add_service(health_service)
        .add_service(GreeterServer::new(MyGreeter::default()))
        .serve(addr)
        .await?;
 
    Ok(())
}