//! End-to-end tests for the Knafeh RPC framework.
//!
//! These tests spin up a real QUIC/HTTP3 server and client, exercising the
//! full transport pipeline. TLS certificates are generated at runtime via
//! `rcgen` to avoid committing private keys.

mod common;

use std::net::SocketAddr;

use async_trait::async_trait;

use knafeh::codec::JsonCodec;
use knafeh::error::{KnafehError, RpcStatusCode};
use knafeh::rpc::message::Metadata;
use knafeh::rpc::message::{RpcRequest, RpcResponse};
use knafeh::rpc::service::{MethodDescriptor, MethodKind, Service};
use knafeh::rpc::stream::{rpc_stream_channel, RpcStreamRequest, RpcStreamResponse};
use knafeh::transport::tls::TlsConfig;
use knafeh::{Client, Server};

use common::{start_test_server, tls_server, EchoService, TEST_CERTS};

// EchoService imported from common module above.

struct GreeterService;

#[async_trait]
impl Service for GreeterService {
    fn name(&self) -> &str {
        "greeter"
    }

    fn methods(&self) -> Vec<MethodDescriptor> {
        vec![MethodDescriptor {
            name: "greet".to_string(),
            kind: MethodKind::Unary,
        }]
    }

    async fn call_unary(
        &self,
        _method: &str,
        request: RpcRequest,
    ) -> Result<RpcResponse, KnafehError> {
        let name: serde_json::Value =
            serde_json::from_slice(&request.body).map_err(|e| KnafehError::Codec(e.to_string()))?;
        let greeting = serde_json::json!({
            "message": format!("Hello, {}!", name["name"].as_str().unwrap_or("world"))
        });
        Ok(RpcResponse::ok(greeting.to_string().into_bytes()))
    }

    async fn call_server_stream(
        &self,
        _: &str,
        _: RpcRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }

    async fn call_client_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcResponse, KnafehError> {
        unimplemented!()
    }

    async fn call_bidi_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }
}

struct MetadataEchoService;

#[async_trait]
impl Service for MetadataEchoService {
    fn name(&self) -> &str {
        "metadata_echo"
    }

    fn methods(&self) -> Vec<MethodDescriptor> {
        vec![MethodDescriptor {
            name: "echo".to_string(),
            kind: MethodKind::Unary,
        }]
    }

    async fn call_unary(
        &self,
        _method: &str,
        request: RpcRequest,
    ) -> Result<RpcResponse, KnafehError> {
        // Copy request metadata into response metadata.
        let mut response = RpcResponse::ok(request.body);
        response.metadata = request.metadata;
        Ok(response)
    }

    async fn call_server_stream(
        &self,
        _: &str,
        _: RpcRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }

    async fn call_client_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcResponse, KnafehError> {
        unimplemented!()
    }

    async fn call_bidi_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }
}

// ---------------------------------------------------------------------------
// Test infrastructure
// ---------------------------------------------------------------------------

async fn test_client(addr: SocketAddr) -> Client {
    Client::builder()
        .endpoint(addr.to_string())
        .codec(JsonCodec::new())
        .build()
        .await
        .unwrap()
}

fn echo_server() -> Server {
    Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(EchoService)
        .build()
        .unwrap()
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_e2e_unary_echo() {
    let (addr, _handle) = start_test_server(echo_server()).await;
    let client = test_client(addr).await;

    let response = client
        .call("echo/echo", br#"{"msg":"hello"}"#.to_vec())
        .await
        .unwrap();

    assert_eq!(response.status.code, RpcStatusCode::Ok);
    assert_eq!(response.body, br#"{"msg":"hello"}"#);
}

#[tokio::test]
async fn test_e2e_tls_verification_accepts_trusted_ca_and_rejects_untrusted_ca() {
    let (addr, _handle) = start_test_server(echo_server()).await;
    let endpoint = addr.to_string();

    let trusted_client = Client::builder()
        .endpoint(endpoint.clone())
        .tls(TlsConfig::client(&TEST_CERTS.ca_path))
        .codec(JsonCodec::new())
        .build()
        .await
        .unwrap();

    let response = trusted_client
        .call("echo/echo", br#"{"msg":"verified"}"#.to_vec())
        .await
        .unwrap();

    assert_eq!(response.status.code, RpcStatusCode::Ok);
    assert_eq!(response.body, br#"{"msg":"verified"}"#);

    let untrusted =
        rcgen::generate_simple_self_signed(vec!["localhost".to_string(), "127.0.0.1".to_string()])
            .unwrap();
    let untrusted_dir = tempfile::tempdir().unwrap();
    let untrusted_ca_path = untrusted_dir.path().join("untrusted-ca.pem");
    std::fs::write(&untrusted_ca_path, untrusted.cert.pem()).unwrap();

    let untrusted_client = Client::builder()
        .endpoint(endpoint)
        .tls(TlsConfig::client(&untrusted_ca_path))
        .codec(JsonCodec::new())
        .build()
        .await
        .unwrap();

    let err = untrusted_client
        .call("echo/echo", br#"{"msg":"untrusted"}"#.to_vec())
        .await
        .unwrap_err();

    assert!(matches!(
        err,
        KnafehError::Transport(_) | KnafehError::ConnectionClosed | KnafehError::Timeout
    ));
}

#[tokio::test]
async fn test_e2e_json_roundtrip() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(GreeterService)
        .build()
        .unwrap();

    let (addr, _handle) = start_test_server(server).await;
    let client = test_client(addr).await;

    let response = client
        .call("greeter/greet", br#"{"name":"Alice"}"#.to_vec())
        .await
        .unwrap();

    assert_eq!(response.status.code, RpcStatusCode::Ok);
    let body: serde_json::Value = serde_json::from_slice(&response.body).unwrap();
    assert_eq!(body["message"], "Hello, Alice!");
}

#[tokio::test]
async fn test_e2e_concurrent_calls() {
    let (addr, _handle) = start_test_server(echo_server()).await;
    let client = std::sync::Arc::new(test_client(addr).await);

    let mut handles = Vec::new();
    for i in 0..10 {
        let client = std::sync::Arc::clone(&client);
        handles.push(tokio::spawn(async move {
            let body = format!(r#"{{"i":{i}}}"#);
            let resp = client
                .call("echo/echo", body.as_bytes().to_vec())
                .await
                .unwrap();
            assert_eq!(resp.status.code, RpcStatusCode::Ok);
            assert_eq!(resp.body, body.as_bytes());
        }));
    }

    for h in handles {
        h.await.unwrap();
    }
}

#[tokio::test]
async fn test_e2e_metadata_roundtrip() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(MetadataEchoService)
        .build()
        .unwrap();

    let (addr, _handle) = start_test_server(server).await;
    let client = test_client(addr).await;

    let mut metadata = std::collections::HashMap::new();
    metadata.insert("trace-id".to_string(), "abc-123".to_string());

    let response = client
        .call_with_metadata("metadata_echo/echo", br#"{}"#.to_vec(), metadata)
        .await
        .unwrap();

    assert_eq!(response.status.code, RpcStatusCode::Ok);
    assert_eq!(
        response.metadata.get("trace-id"),
        Some(&"abc-123".to_string())
    );
}

#[tokio::test]
async fn test_e2e_method_not_found() {
    let (addr, _handle) = start_test_server(echo_server()).await;
    let client = test_client(addr).await;

    let response = client
        .call("nonexistent/method", br#"{}"#.to_vec())
        .await
        .unwrap();

    assert_eq!(response.status.code, RpcStatusCode::NotFound);
}

#[tokio::test]
async fn test_e2e_multiple_services() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(EchoService)
        .add_service(GreeterService)
        .build()
        .unwrap();

    let (addr, _handle) = start_test_server(server).await;
    let client = test_client(addr).await;

    let echo_resp = client
        .call("echo/echo", br#"{"x":1}"#.to_vec())
        .await
        .unwrap();
    assert_eq!(echo_resp.status.code, RpcStatusCode::Ok);
    assert_eq!(echo_resp.body, br#"{"x":1}"#);

    let greet_resp = client
        .call("greeter/greet", br#"{"name":"Bob"}"#.to_vec())
        .await
        .unwrap();
    assert_eq!(greet_resp.status.code, RpcStatusCode::Ok);
    let body: serde_json::Value = serde_json::from_slice(&greet_resp.body).unwrap();
    assert_eq!(body["message"], "Hello, Bob!");
}

#[tokio::test]
async fn test_e2e_connection_reuse() {
    let (addr, _handle) = start_test_server(echo_server()).await;
    let client = test_client(addr).await;

    // Make multiple sequential calls — they should reuse the same connection.
    for i in 0..5 {
        let body = format!(r#"{{"call":{i}}}"#);
        let resp = client
            .call("echo/echo", body.as_bytes().to_vec())
            .await
            .unwrap();
        assert_eq!(resp.status.code, RpcStatusCode::Ok);
    }

    // Pool should contain exactly 1 connection.
    assert_eq!(client.pool().size(), 1);
}

#[tokio::test]
async fn test_e2e_large_payload() {
    let (addr, _handle) = start_test_server(echo_server()).await;
    let client = test_client(addr).await;

    // Build a ~1MB JSON payload.
    let large_string = "x".repeat(1_000_000);
    let body = format!(r#"{{"data":"{large_string}"}}"#);

    let resp = client
        .call("echo/echo", body.as_bytes().to_vec())
        .await
        .unwrap();

    assert_eq!(resp.status.code, RpcStatusCode::Ok);
    assert_eq!(resp.body.len(), body.len());
}

#[tokio::test]
async fn test_e2e_empty_body() {
    let (addr, _handle) = start_test_server(echo_server()).await;
    let client = test_client(addr).await;

    // Empty JSON object.
    let resp = client.call("echo/echo", br#"{}"#.to_vec()).await.unwrap();
    assert_eq!(resp.status.code, RpcStatusCode::Ok);
    assert_eq!(resp.body, br#"{}"#);
}

#[tokio::test]
async fn test_e2e_middleware() {
    use knafeh::rpc::middleware::Interceptor;

    struct TagInterceptor;

    #[async_trait]
    impl Interceptor for TagInterceptor {
        async fn on_request(&self, request: &mut RpcRequest) -> Result<(), KnafehError> {
            request
                .metadata
                .insert("intercepted".to_string(), "true".to_string());
            Ok(())
        }
    }

    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(MetadataEchoService)
        .build()
        .unwrap();

    let (addr, _handle) = start_test_server(server).await;

    let client = Client::builder()
        .endpoint(addr.to_string())
        .codec(JsonCodec::new())
        .add_interceptor(TagInterceptor)
        .build()
        .await
        .unwrap();

    let resp = client
        .call("metadata_echo/echo", br#"{}"#.to_vec())
        .await
        .unwrap();
    assert_eq!(resp.status.code, RpcStatusCode::Ok);
    // The interceptor should have injected "intercepted" metadata into the
    // request. MetadataEchoService copies it to the response.
    assert_eq!(resp.metadata.get("intercepted"), Some(&"true".to_string()));
}

// ---------------------------------------------------------------------------
// B3: Interceptor chain (multiple interceptors)
// ---------------------------------------------------------------------------

/// Simulates: logging + auth + tracing interceptors stacked on client.
#[tokio::test]
async fn test_e2e_interceptor_chain() {
    use knafeh::rpc::middleware::Interceptor;

    struct PrefixInterceptor(&'static str);

    #[async_trait]
    impl Interceptor for PrefixInterceptor {
        async fn on_request(&self, req: &mut RpcRequest) -> Result<(), KnafehError> {
            req.metadata
                .insert(format!("x-{}", self.0), self.0.to_string());
            Ok(())
        }
    }

    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(MetadataEchoService)
        .build()
        .unwrap();
    let (addr, _handle) = start_test_server(server).await;

    let client = Client::builder()
        .endpoint(addr.to_string())
        .codec(JsonCodec::new())
        .add_interceptor(PrefixInterceptor("logging"))
        .add_interceptor(PrefixInterceptor("auth"))
        .add_interceptor(PrefixInterceptor("tracing"))
        .build()
        .await
        .unwrap();

    let resp = client
        .call("metadata_echo/echo", br#"{}"#.to_vec())
        .await
        .unwrap();
    assert_eq!(resp.status.code, RpcStatusCode::Ok);
    assert_eq!(resp.metadata.get("x-logging"), Some(&"logging".to_string()));
    assert_eq!(resp.metadata.get("x-auth"), Some(&"auth".to_string()));
    assert_eq!(resp.metadata.get("x-tracing"), Some(&"tracing".to_string()));
}

// ---------------------------------------------------------------------------
// B4: Interceptor rejection
// ---------------------------------------------------------------------------

/// Simulates: auth interceptor rejects unauthenticated request.
#[tokio::test]
async fn test_e2e_interceptor_rejection() {
    use knafeh::rpc::middleware::Interceptor;

    struct RejectInterceptor;

    #[async_trait]
    impl Interceptor for RejectInterceptor {
        async fn on_request(&self, _req: &mut RpcRequest) -> Result<(), KnafehError> {
            Err(KnafehError::Service {
                code: RpcStatusCode::Unauthenticated,
                message: "missing auth token".to_string(),
            })
        }
    }

    let (addr, _handle) = start_test_server(echo_server()).await;

    let client = Client::builder()
        .endpoint(addr.to_string())
        .codec(JsonCodec::new())
        .add_interceptor(RejectInterceptor)
        .build()
        .await
        .unwrap();

    let err = client
        .call("echo/echo", br#"{}"#.to_vec())
        .await
        .unwrap_err();

    match err {
        KnafehError::Service { code, message } => {
            assert_eq!(code, RpcStatusCode::Unauthenticated);
            assert_eq!(message, "missing auth token");
        }
        other => panic!("expected Service error, got: {other}"),
    }
}

// ---------------------------------------------------------------------------
// B5: Server-side interceptor
// ---------------------------------------------------------------------------

/// Simulates: server-side interceptor injects response metadata.
#[tokio::test]
async fn test_e2e_server_interceptor() {
    use knafeh::rpc::middleware::Interceptor;

    struct ServerTagInterceptor;

    #[async_trait]
    impl Interceptor for ServerTagInterceptor {
        async fn on_response(&self, resp: &mut RpcResponse) -> Result<(), KnafehError> {
            resp.metadata
                .insert("server-version".to_string(), "1.0".to_string());
            Ok(())
        }
    }

    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(MetadataEchoService)
        .add_interceptor(ServerTagInterceptor)
        .build()
        .unwrap();
    let (addr, _handle) = start_test_server(server).await;
    let client = test_client(addr).await;

    let resp = client
        .call("metadata_echo/echo", br#"{}"#.to_vec())
        .await
        .unwrap();
    assert_eq!(resp.status.code, RpcStatusCode::Ok);
    assert_eq!(
        resp.metadata.get("server-version"),
        Some(&"1.0".to_string())
    );
}

// ---------------------------------------------------------------------------
// D6: Large stream chunks
// ---------------------------------------------------------------------------

/// Simulates: streaming large records (100KB+ each).
#[tokio::test]
async fn test_e2e_server_stream_large_chunks() {
    // We'll reuse CountService but with a LargeChunkService instead.
    struct LargeChunkService;

    #[async_trait]
    impl Service for LargeChunkService {
        fn name(&self) -> &str {
            "large_chunk"
        }
        fn methods(&self) -> Vec<MethodDescriptor> {
            vec![MethodDescriptor {
                name: "stream".to_string(),
                kind: MethodKind::ServerStreaming,
            }]
        }
        async fn call_unary(&self, _: &str, _: RpcRequest) -> Result<RpcResponse, KnafehError> {
            unimplemented!()
        }
        async fn call_server_stream(
            &self,
            _: &str,
            _request: RpcRequest,
        ) -> Result<RpcStreamResponse, KnafehError> {
            let (sender, receiver) = rpc_stream_channel(8);
            tokio::spawn(async move {
                for i in 0..5u64 {
                    let big = "A".repeat(100_000);
                    let chunk = serde_json::json!({"i": i, "data": big})
                        .to_string()
                        .into_bytes();
                    if sender.send(chunk).await.is_err() {
                        break;
                    }
                }
            });
            Ok(RpcStreamResponse {
                metadata: Metadata::new(),
                receiver,
            })
        }
        async fn call_client_stream(
            &self,
            _: &str,
            _: RpcStreamRequest,
        ) -> Result<RpcResponse, KnafehError> {
            unimplemented!()
        }
        async fn call_bidi_stream(
            &self,
            _: &str,
            _: RpcStreamRequest,
        ) -> Result<RpcStreamResponse, KnafehError> {
            unimplemented!()
        }
    }

    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(LargeChunkService)
        .build()
        .unwrap();
    let (addr, _handle) = start_test_server(server).await;
    let client = test_client(addr).await;

    let mut stream = client
        .server_stream("large_chunk/stream", br#"{}"#.to_vec())
        .await
        .unwrap();

    let mut count = 0;
    while let Some(chunk) = stream.next().await {
        let chunk = chunk.unwrap();
        let val: serde_json::Value = serde_json::from_slice(&chunk).unwrap();
        assert_eq!(val["i"].as_u64().unwrap(), count);
        assert_eq!(val["data"].as_str().unwrap().len(), 100_000);
        count += 1;
    }
    assert_eq!(count, 5);
}

// ---------------------------------------------------------------------------
// Server-streaming test services
// ---------------------------------------------------------------------------

/// Streams N JSON chunks: {"i":0}, {"i":1}, ..., {"i":N-1}.
struct CountService;

#[async_trait]
impl Service for CountService {
    fn name(&self) -> &str {
        "count"
    }

    fn methods(&self) -> Vec<MethodDescriptor> {
        vec![MethodDescriptor {
            name: "count".to_string(),
            kind: MethodKind::ServerStreaming,
        }]
    }

    async fn call_unary(&self, _: &str, _: RpcRequest) -> Result<RpcResponse, KnafehError> {
        unimplemented!()
    }

    async fn call_server_stream(
        &self,
        _method: &str,
        request: RpcRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        let body: serde_json::Value =
            serde_json::from_slice(&request.body).map_err(|e| KnafehError::Codec(e.to_string()))?;
        let count = body["count"].as_u64().unwrap_or(0);

        let (sender, receiver) = rpc_stream_channel(32);

        tokio::spawn(async move {
            for i in 0..count {
                let chunk = serde_json::json!({"i": i}).to_string().into_bytes();
                if sender.send(chunk).await.is_err() {
                    break;
                }
            }
        });

        Ok(RpcStreamResponse {
            metadata: Metadata::new(),
            receiver,
        })
    }

    async fn call_client_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcResponse, KnafehError> {
        unimplemented!()
    }

    async fn call_bidi_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }
}

// ---------------------------------------------------------------------------
// Server-streaming tests
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_e2e_server_stream_basic() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(CountService)
        .build()
        .unwrap();

    let (addr, _handle) = start_test_server(server).await;
    let client = test_client(addr).await;

    let mut stream = client
        .server_stream("count/count", br#"{"count":5}"#.to_vec())
        .await
        .unwrap();

    let mut received = Vec::new();
    while let Some(chunk) = stream.next().await {
        let chunk = chunk.unwrap();
        let val: serde_json::Value = serde_json::from_slice(&chunk).unwrap();
        received.push(val["i"].as_u64().unwrap());
    }

    assert_eq!(received, vec![0, 1, 2, 3, 4]);
}

#[tokio::test]
async fn test_e2e_server_stream_empty() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(CountService)
        .build()
        .unwrap();

    let (addr, _handle) = start_test_server(server).await;
    let client = test_client(addr).await;

    let mut stream = client
        .server_stream("count/count", br#"{"count":0}"#.to_vec())
        .await
        .unwrap();

    // Zero chunks should yield nothing.
    assert!(stream.next().await.is_none());
}

#[tokio::test]
async fn test_e2e_server_stream_early_drop() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(CountService)
        .build()
        .unwrap();

    let (addr, _handle) = start_test_server(server).await;
    let client = test_client(addr).await;

    let mut stream = client
        .server_stream("count/count", br#"{"count":10000}"#.to_vec())
        .await
        .unwrap();

    // Read only 3 chunks then drop the stream.
    for _ in 0..3 {
        let chunk = stream.next().await.unwrap().unwrap();
        let _: serde_json::Value = serde_json::from_slice(&chunk).unwrap();
    }
    drop(stream);

    // No panic, no hang — server-side sender will error and stop.
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}

#[tokio::test]
async fn test_e2e_mixed_unary_and_streaming() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(EchoService)
        .add_service(CountService)
        .build()
        .unwrap();

    let (addr, _handle) = start_test_server(server).await;
    let client = test_client(addr).await;

    // Unary call should work.
    let resp = client
        .call("echo/echo", br#"{"x":1}"#.to_vec())
        .await
        .unwrap();
    assert_eq!(resp.status.code, RpcStatusCode::Ok);
    assert_eq!(resp.body, br#"{"x":1}"#);

    // Streaming call should also work on the same connection.
    let mut stream = client
        .server_stream("count/count", br#"{"count":3}"#.to_vec())
        .await
        .unwrap();

    let mut count = 0;
    while let Some(Ok(_)) = stream.next().await {
        count += 1;
    }
    assert_eq!(count, 3);
}

#[tokio::test]
async fn test_e2e_concurrent_server_streams() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(CountService)
        .build()
        .unwrap();

    let (addr, _handle) = start_test_server(server).await;
    let client = std::sync::Arc::new(test_client(addr).await);

    let mut handles = Vec::new();
    for _ in 0..5 {
        let client = std::sync::Arc::clone(&client);
        handles.push(tokio::spawn(async move {
            let mut stream = client
                .server_stream("count/count", br#"{"count":10}"#.to_vec())
                .await
                .unwrap();
            let mut count = 0;
            while let Some(Ok(_)) = stream.next().await {
                count += 1;
            }
            assert_eq!(count, 10);
        }));
    }

    for h in handles {
        h.await.unwrap();
    }
}
