rust-mc-status 3.0.0

//! Tower integration example for the rust-mc-status library.
//!
//! Demonstrates McService, McRetryPolicy, and the full Tower middleware stack
//! including a real working custom Layer implementation.
//!
//! Run with:
//!   cargo run --example tower_usage --features tower
//!
//! Note: for tracing support add these dependencies to Cargo.toml (dev or normal):
//!   tracing = "0.1"
//!   tracing-subscriber = "0.3"

use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;

use tower::load_shed::LoadShedLayer;
use tower::{Layer, Service, ServiceBuilder, ServiceExt};
use rust_mc_status::{McClient, McError, McRetryPolicy, PingRequestTower, PingResponseTower};

// Optional tracing support – user must add dependencies
use tracing::{info, warn, span, Instrument, Level};
use tracing_subscriber::FmtSubscriber;

fn sep() { println!("\n{}\n", "─".repeat(50)); }

// ─── Custom Layer: LogLayer (plain printing) ─────────────────────────────────

#[derive(Clone)]
struct LogLayer {
    prefix: String,
}

impl LogLayer {
    fn new(prefix: impl Into<String>) -> Self {
        Self { prefix: prefix.into() }
    }
}

impl<S> Layer<S> for LogLayer {
    type Service = LogService<S>;
    fn layer(&self, inner: S) -> Self::Service {
        LogService { inner, prefix: self.prefix.clone() }
    }
}

#[derive(Clone)]
struct LogService<S> {
    inner:  S,
    prefix: String,
}

impl<S> Service<PingRequestTower> for LogService<S>
where
    S: Service<PingRequestTower, Response = PingResponseTower, Error = McError> + Send,
    S::Future: Send + 'static,
{
    type Response = PingResponseTower;
    type Error    = McError;
    type Future   = Pin<Box<dyn Future<Output = Result<PingResponseTower, McError>> + Send>>;

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

    fn call(&mut self, req: PingRequestTower) -> Self::Future {
        let prefix  = self.prefix.clone();
        let address = req.address.clone();
        let edition = req.edition;
        println!("{prefix} → {address} ({edition:?})");
        let fut = self.inner.call(req);
        Box::pin(async move {
            match fut.await {
                Ok(r)  => {
                    println!("{prefix} ✅ {address} {:.0} ms", r.status.latency);
                    Ok(r)
                }
                Err(e) => {
                    println!("{prefix} ❌ {address} {e}");
                    Err(e)
                }
            }
        })
    }
}

// ─── Custom Layer: MetricsLayer (atomic counters) ────────────────────────────

#[derive(Default)]
struct MetricsData {
    total:          AtomicU64,
    successes:      AtomicU64,
    failures:       AtomicU64,
    latency_us_sum: AtomicU64,
}

impl MetricsData {
    fn report(&self) -> String {
        let n   = self.successes.load(Ordering::Relaxed);
        let avg = if n == 0 {
            "n/a".into()
        } else {
            let us = self.latency_us_sum.load(Ordering::Relaxed);
            format!("{:.1} ms", us as f64 / n as f64 / 1_000.0)
        };
        format!(
            "total={}  ok={}  err={}  avg_latency={}",
            self.total.load(Ordering::Relaxed), n,
            self.failures.load(Ordering::Relaxed), avg,
        )
    }
}

#[derive(Clone)]
struct MetricsLayer {
    data: Arc<MetricsData>,
}

impl MetricsLayer {
    fn new() -> (Self, Arc<MetricsData>) {
        let data = Arc::new(MetricsData::default());
        (Self { data: Arc::clone(&data) }, data)
    }
}

impl<S> Layer<S> for MetricsLayer {
    type Service = MetricsService<S>;
    fn layer(&self, inner: S) -> Self::Service {
        MetricsService { inner, data: Arc::clone(&self.data) }
    }
}

#[derive(Clone)]
struct MetricsService<S> {
    inner: S,
    data:  Arc<MetricsData>,
}

impl<S> Service<PingRequestTower> for MetricsService<S>
where
    S: Service<PingRequestTower, Response = PingResponseTower, Error = McError> + Send,
    S::Future: Send + 'static,
{
    type Response = PingResponseTower;
    type Error    = McError;
    type Future   = Pin<Box<dyn Future<Output = Result<PingResponseTower, McError>> + Send>>;

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

    fn call(&mut self, req: PingRequestTower) -> Self::Future {
        let data = Arc::clone(&self.data);
        data.total.fetch_add(1, Ordering::Relaxed);
        let fut = self.inner.call(req);
        Box::pin(async move {
            match fut.await {
                Ok(r) => {
                    data.successes.fetch_add(1, Ordering::Relaxed);
                    data.latency_us_sum.fetch_add(
                        (r.status.latency * 1_000.0) as u64,
                        Ordering::Relaxed,
                    );
                    Ok(r)
                }
                Err(e) => {
                    data.failures.fetch_add(1, Ordering::Relaxed);
                    Err(e)
                }
            }
        })
    }
}

// ─── Layer: TracingLayer (structured logs with fixed target) ─────────────────

#[derive(Clone)]
struct TracingLayer {
    prefix: String,  // used only in log message, not as target
}

impl TracingLayer {
    fn new(prefix: impl Into<String>) -> Self {
        Self { prefix: prefix.into() }
    }
}

impl<S> Layer<S> for TracingLayer {
    type Service = TracingService<S>;
    fn layer(&self, inner: S) -> Self::Service {
        TracingService { inner, prefix: self.prefix.clone() }
    }
}

#[derive(Clone)]
struct TracingService<S> {
    inner:  S,
    prefix: String,
}

impl<S> Service<PingRequestTower> for TracingService<S>
where
    S: Service<PingRequestTower, Response = PingResponseTower, Error = McError> + Send,
    S::Future: Send + 'static,
{
    type Response = PingResponseTower;
    type Error    = McError;
    type Future   = Pin<Box<dyn Future<Output = Result<PingResponseTower, McError>> + Send>>;

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

    fn call(&mut self, req: PingRequestTower) -> Self::Future {
        let prefix = self.prefix.clone();
        let address = req.address.clone();
        let edition = req.edition;
        let span = span!(Level::INFO, "ping", %address, ?edition);
        let fut = self.inner.call(req);
        Box::pin(async move {
            let result = fut.await;
            match &result {
                Ok(r) => {
                    info!(
                        prefix = %prefix,
                        latency_ms = r.status.latency,
                        players = ?r.status.players(),
                        "Ping succeeded"
                    );
                }
                Err(e) => {
                    warn!(
                        prefix = %prefix,
                        error = %e,
                        "Ping failed"
                    );
                }
            }
            result
        }.instrument(span))
    }
}

// ─── Layer: PollTimingLayer (measure poll_ready latency) ─────────────────────

#[derive(Clone)]
struct PollTimingLayer;

impl<S> Layer<S> for PollTimingLayer {
    type Service = PollTimingService<S>;
    fn layer(&self, inner: S) -> Self::Service {
        PollTimingService { inner }
    }
}

#[derive(Clone)]
struct PollTimingService<S> {
    inner: S,
}

impl<S> Service<PingRequestTower> for PollTimingService<S>
where
    S: Service<PingRequestTower, Response = PingResponseTower, Error = McError> + Send,
{
    type Response = PingResponseTower;
    type Error    = McError;
    type Future   = S::Future;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), McError>> {
        let start = std::time::Instant::now();
        let res = self.inner.poll_ready(cx);
        let elapsed = start.elapsed().as_micros();
        if elapsed > 1000 {
            eprintln!("[poll_ready] waited {} µs", elapsed);
        }
        res
    }

    fn call(&mut self, req: PingRequestTower) -> Self::Future {
        self.inner.call(req)
    }
}

// ─── Main ─────────────────────────────────────────────────────────────────────

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize tracing subscriber (compact format)
    let subscriber = FmtSubscriber::builder()
        .with_target(false)
        .with_level(true)
        .with_thread_ids(false)
        .compact()
        .finish();
    tracing::subscriber::set_global_default(subscriber).unwrap();

    println!("=== Tower Integration ===\n");

    // ─── 1. McService — direct call (no middleware) ──────────────────────────

    println!("1. McService — direct call (no middleware)");
    sep();

    let mut svc = McClient::builder()
        .timeout(Duration::from_secs(15))
        .build()
        .into_service();

    svc.ready().await?;
    match svc.call(PingRequestTower::java("mc.hypixel.net")).await {
        Ok(r)  => println!("   ✅ {:.0} ms  players: {}/{}",
            r.status.latency,
            r.status.players().map(|(o, _)| o).unwrap_or(0),
            r.status.players().map(|(_, m)| m).unwrap_or(0)),
        Err(e) => println!("   ❌ {e}"),
    }

    sep();

    // ─── 2. Rate limiting ────────────────────────────────────────────────────

    println!("2. RateLimit — 2 requests / second");
    sep();

    let mut svc = ServiceBuilder::new()
        .rate_limit(2, Duration::from_secs(1))
        .service(McClient::builder().timeout(Duration::from_secs(15)).build().into_service());

    for i in 1..=3u8 {
        svc.ready().await?;
        let wall = std::time::Instant::now();
        match svc.call(PingRequestTower::java("mc.hypixel.net")).await {
            Ok(r)  => println!("   [{i}] ✅ {:.0} ms  wall: {:.0} ms",
                r.status.latency, wall.elapsed().as_secs_f64() * 1000.0),
            Err(e) => println!("   [{i}] ❌ {e}"),
        }
    }
    println!("   (request 3 waited for the rate-limit window to refill)");

    sep();

    // ─── 3. Retry ────────────────────────────────────────────────────────────

    println!("3. McRetryPolicy — up to 3 attempts total");
    sep();

    // Force timeouts — all attempts exhaust
    let mut svc_short = ServiceBuilder::new()
        .retry(McRetryPolicy::new(3))
        .service(McClient::builder().timeout(Duration::from_millis(50)).build().into_service());

    svc_short.ready().await?;
    match svc_short.call(PingRequestTower::java("mc.hypixel.net")).await {
        Ok(r)  => println!("   ✅ Unexpectedly succeeded: {:.0} ms", r.status.latency),
        Err(e) => println!("   ❌ All attempts exhausted: {e}"),
    }

    // Real timeout — succeeds on first try
    let mut svc_real = ServiceBuilder::new()
        .retry(McRetryPolicy::new(3))
        .service(McClient::builder().timeout(Duration::from_secs(15)).build().into_service());

    svc_real.ready().await?;
    match svc_real.call(PingRequestTower::java("mc.hypixel.net")).await {
        Ok(r)  => println!("   ✅ Succeeded: {:.0} ms", r.status.latency),
        Err(e) => println!("   ❌ {e}"),
    }

    sep();

    // ─── 4. Buffer — Clone + concurrent ──────────────────────────────────────

    println!("4. Buffer — concurrent calls from multiple tasks");
    sep();

    let svc = ServiceBuilder::new()
        .buffer(16)
        .service(McClient::builder().timeout(Duration::from_secs(15)).build().into_service());

    let tasks: Vec<_> = [
        PingRequestTower::java("mc.hypixel.net"),
        PingRequestTower::java("mc.hypixel.net"),
        PingRequestTower::java("does-not-exist.invalid"),
    ]
    .into_iter()
    .map(|req| {
        let mut s = svc.clone();
        tokio::spawn(async move {
            match ServiceExt::<PingRequestTower>::ready(&mut s).await {
                Ok(s) => match s.call(req).await {
                    Ok(r)  => println!("   ✅ {:.0} ms", r.status.latency),
                    Err(e) => println!("   ❌ {e}"),
                },
                Err(e) => println!("   ❌ ready() failed: {e}"),
            }
        })
    })
    .collect();

    for t in tasks { let _ = t.await; }

    sep();

    // ─── 5. LoadShedLayer — защита от перегрузки ─────────────────────────────

    println!("5. LoadShedLayer — сбрасывает запросы при перегрузке");
    sep();

    let svc = ServiceBuilder::new()
        .buffer(2)
        .layer(LoadShedLayer::new())
        .service(McClient::builder().timeout(Duration::from_secs(15)).build().into_service());

    let mut tasks = vec![];
    for i in 1..=5u8 {
        let mut s = svc.clone();
        tasks.push(tokio::spawn(async move {
            match s.ready().await {
                Ok(s) => match s.call(PingRequestTower::java("mc.hypixel.net")).await {
                    Ok(r)  => println!("   [{i}] ✅ {:.0} ms", r.status.latency),
                    Err(e) => println!("   [{i}] ❌ {}", e),
                },
                Err(e) => println!("   [{i}] ❌ ready error: {}", e),
            }
        }));
    }
    for t in tasks { let _ = t.await; }
    println!("   (ожидаемо: часть запросов будет сброшена из-за переполнения очереди)");

    sep();

    // ─── 6. Custom LogLayer ──────────────────────────────────────────────────

    println!("6. Custom LogLayer");
    sep();

    let mut svc = ServiceBuilder::new()
        .layer(LogLayer::new("[log]"))
        .service(McClient::builder().timeout(Duration::from_secs(15)).build().into_service());

    svc.ready().await?;
    let _ = svc.call(PingRequestTower::java("mc.hypixel.net")).await;
    svc.ready().await?;
    let _ = svc.call(PingRequestTower::java("does-not-exist.invalid")).await;

    sep();

    // ─── 7. Custom MetricsLayer ──────────────────────────────────────────────

    println!("7. Custom MetricsLayer");
    sep();

    let (metrics_layer, metrics_data) = MetricsLayer::new();
    let mut svc = ServiceBuilder::new()
        .layer(metrics_layer)
        .service(McClient::builder().timeout(Duration::from_secs(15)).build().into_service());

    for req in [
        PingRequestTower::java("mc.hypixel.net"),
        PingRequestTower::java("does-not-exist.invalid"),
        PingRequestTower::bedrock("geo.hivebedrock.network:19132"),
    ] {
        svc.ready().await?;
        let _ = svc.call(req).await;
    }
    println!("   Metrics: {}", metrics_data.report());

    sep();

    // ─── 8. Structured logs with tracing (fixed target) ──────────────────────

    println!("8. TracingLayer — structured logs (compact format)");
    sep();

    let mut svc = ServiceBuilder::new()
        .layer(TracingLayer::new("mcstatus"))
        .service(McClient::builder().timeout(Duration::from_secs(15)).build().into_service());

    svc.ready().await?;
    let _ = svc.call(PingRequestTower::java("mc.hypixel.net")).await;
    svc.ready().await?;
    let _ = svc.call(PingRequestTower::java("does-not-exist.invalid")).await;

    sep();

    // ─── 9. PollTimingLayer — измерение времени poll_ready ───────────────────

    println!("9. PollTimingLayer — логирует долгие ожидания в poll_ready");
    sep();

    let mut svc = ServiceBuilder::new()
        .layer(PollTimingLayer)
        .rate_limit(1, Duration::from_secs(2))
        .service(McClient::builder().timeout(Duration::from_secs(15)).build().into_service());

    for i in 1..=3 {
        svc.ready().await?;
        let _ = svc.call(PingRequestTower::java("mc.hypixel.net")).await;
        println!("   request {i} done");
    }

    sep();

    // ─── 10. Full stack — everything together ────────────────────────────────

    println!("10. Full stack — Buffer + Log + Metrics + RateLimit + Retry");
    sep();

    let (metrics_layer, metrics_data) = MetricsLayer::new();
    let svc = ServiceBuilder::new()
        .buffer(32)
        .layer(LogLayer::new("[full]"))
        .layer(metrics_layer)
        .rate_limit(10, Duration::from_secs(1))
        .retry(McRetryPolicy::new(2))
        .service(McClient::builder().timeout(Duration::from_secs(15)).build().into_service());

    let reqs = [
        PingRequestTower::java("mc.hypixel.net"),
        PingRequestTower::bedrock("geo.hivebedrock.network:19132"),
        PingRequestTower::java("does-not-exist.invalid"),
    ];

    let tasks: Vec<_> = reqs.into_iter()
        .map(|req| {
            let mut s = svc.clone();
            tokio::spawn(async move {
                match ServiceExt::<PingRequestTower>::ready(&mut s).await {
                    Ok(s) => { let _ = s.call(req).await; }
                    Err(e) => println!("   ready() error: {e}"),
                }
            })
        })
        .collect();

    for t in tasks { let _ = t.await; }
    println!("\n   Final metrics: {}", metrics_data.report());

    Ok(())
}