prism_mcp_rs/transport/

streaming_http.rs

//! Proprietary Prism HTTP transport extensions
//!
//! This module implements historical Prism chunked/compressed endpoints. It is
//! not the MCP 2026 Streamable HTTP conformance surface. Prefer
//! `HttpClientTransport` and `HttpServerTransport` for interoperable MCP; these
//! helpers require explicit `ProtocolMode::LegacyOnly` at the high-level client.
//!
//! ## Feature Requirements
//!
//! ```toml
//! # Cargo.toml - HTTP enhancements
//! [dependencies]
//! prism-mcp-rs = { version = "3", features = ["http", "chunked-encoding", "compression", "http2"] }
//! ```
//!
//! Individual features:
//!
//! ### Chunked Encoding Feature
//! ```toml
//! [dependencies]
//! prism-mcp-rs = { version = "3", features = ["chunked-encoding"] }
//! ```
//! Enables Prism-specific chunked request endpoints for matching legacy peers.
//!
//! ### Compression Feature  
//! ```toml
//! [dependencies]
//! prism-mcp-rs = { version = "3", features = ["compression"] }
//! ```
//! Enables response compression with Gzip, Brotli, and Zstd algorithms.
//!
//! ### HTTP/2 Feature
//! ```toml
//! [dependencies]
//! prism-mcp-rs = { version = "3", features = ["http2"] }
//! ```
//! Enables HTTP/2 protocol support with multiplexing and server push.
//!
//! ## Capabilities
//! - Chunked transfer encoding for large payloads
//! - Response compression: Gzip, Brotli, Zstd
//! - HTTP/2 protocol primitives
//! - Configurable compression strategies
//! - Backpressure and flow control
//!
//! Use only when both peers explicitly implement these endpoints and deployment
//! measurements justify the additional protocol surface.

use async_trait::async_trait;
use serde_json::Value;
use std::sync::Arc;
use std::time::Duration;
#[cfg(any(feature = "chunked-encoding", feature = "http2"))]
use std::time::Instant;
use tokio::sync::RwLock;

#[cfg(feature = "chunked-encoding")]
use tracing::{debug, warn};

use crate::core::error::{McpError, McpResult};
use crate::protocol::types::{JsonRpcNotification, JsonRpcRequest, JsonRpcResponse};
use crate::transport::traits::Transport;
#[cfg(feature = "chunked-encoding")]
use crate::transport::traits::{ConnectionState, TransportStats};

// Chunked encoding support
#[cfg(feature = "chunked-encoding")]
use bytes::{Bytes, BytesMut};
#[cfg(feature = "chunked-encoding")]
use tokio_stream::wrappers::ReceiverStream;

// HTTP/2 protocol support
#[cfg(feature = "http2")]
use h2::client::{Connection, SendRequest};
#[cfg(feature = "http2")]
use http::{Method, Request};
#[cfg(feature = "http2")]
use tokio::net::TcpStream;

// HTTP/2 specific types
#[cfg(feature = "http2")]
use http::HeaderMap;
#[cfg(feature = "http2")]
use std::collections::HashMap;
#[cfg(feature = "http2")]
use std::pin::Pin;

/// HTTP/2 Server Push handler for processing pushed streams
#[cfg(feature = "http2")]
pub type ServerPushHandler = Box<
    dyn Fn(PushPromise) -> Pin<Box<dyn std::future::Future<Output = McpResult<()>> + Send>>
        + Send
        + Sync,
>;

/// HTTP/2 Push Promise metadata
#[cfg(feature = "http2")]
#[derive(Debug, Clone)]
pub struct PushPromise {
    pub method: String,
    pub path: String,
    pub headers: HeaderMap,
    pub stream_id: u32,
    pub promised_stream_id: u32,
}

/// HTTP/2 stream management for bidirectional communication
#[cfg(feature = "http2")]
#[derive(Debug)]
pub struct Http2StreamManager {
    pub active_streams: HashMap<u32, StreamInfo>,
    pub push_promises: HashMap<u32, PushPromise>,
    pub max_concurrent_streams: usize,
    pub connection_window_size: u32,
}

#[cfg(feature = "http2")]
#[derive(Debug, Clone)]
pub struct StreamInfo {
    pub stream_id: u32,
    pub method: String,
    pub path: String,
    pub state: StreamState,
    pub created_at: Instant,
    pub bytes_sent: u64,
    pub bytes_received: u64,
}

#[cfg(feature = "http2")]
#[derive(Debug, Clone, PartialEq)]
pub enum StreamState {
    Open,
    HalfClosedLocal,
    HalfClosedRemote,
    Closed,
    Reserved(u32), // For push promised streams
}

#[cfg(feature = "http2")]
impl Http2StreamManager {
    pub fn new(max_concurrent_streams: usize) -> Self {
        Self {
            active_streams: HashMap::new(),
            push_promises: HashMap::new(),
            max_concurrent_streams,
            connection_window_size: 65535, // Default HTTP/2 window size
        }
    }

    pub fn can_create_stream(&self) -> bool {
        self.active_streams.len() < self.max_concurrent_streams
    }

    pub fn add_stream(&mut self, stream_id: u32, method: String, path: String) {
        let stream_info = StreamInfo {
            stream_id,
            method,
            path,
            state: StreamState::Open,
            created_at: Instant::now(),
            bytes_sent: 0,
            bytes_received: 0,
        };
        self.active_streams.insert(stream_id, stream_info);
    }

    pub fn add_push_promise(&mut self, stream_id: u32, promise: PushPromise) {
        let promised_stream_id = promise.promised_stream_id;
        self.push_promises.insert(stream_id, promise);
        // Mark stream as reserved for the push promise
        if let Some(stream_info) = self.active_streams.get_mut(&stream_id) {
            stream_info.state = StreamState::Reserved(promised_stream_id);
        }
    }

    pub fn close_stream(&mut self, stream_id: u32) {
        if let Some(mut stream_info) = self.active_streams.remove(&stream_id) {
            stream_info.state = StreamState::Closed;
            // Clean up any associated push promises
            self.push_promises.remove(&stream_id);
        }
    }

    pub fn update_stream_bytes(&mut self, stream_id: u32, sent: u64, received: u64) {
        if let Some(stream_info) = self.active_streams.get_mut(&stream_id) {
            stream_info.bytes_sent += sent;
            stream_info.bytes_received += received;
        }
    }

    pub fn get_stream_stats(&self) -> (usize, u64, u64) {
        let total_sent = self.active_streams.values().map(|s| s.bytes_sent).sum();
        let total_received = self.active_streams.values().map(|s| s.bytes_received).sum();
        (self.active_streams.len(), total_sent, total_received)
    }
}

/// HTTP/2 specific configuration
#[cfg(feature = "http2")]
#[derive(Debug, Clone)]
pub struct Http2Config {
    /// Maximum concurrent streams per connection
    pub max_concurrent_streams: usize,
    /// Initial connection window size
    pub initial_window_size: u32,
    /// Maximum frame size
    pub max_frame_size: u32,
    /// Enable server push
    pub enable_server_push: bool,
    /// Server push cache size
    pub push_cache_size: usize,
    /// Connection timeout
    pub connection_timeout: Duration,
    /// Keep-alive interval
    pub keep_alive_interval: Duration,
    /// Enable push promise validation
    pub validate_push_promises: bool,
}

#[cfg(feature = "http2")]
impl Default for Http2Config {
    fn default() -> Self {
        Self {
            max_concurrent_streams: 100,
            initial_window_size: 65535,
            max_frame_size: 16384,
            enable_server_push: true,
            push_cache_size: 1000,
            connection_timeout: Duration::from_secs(30),
            keep_alive_interval: Duration::from_secs(10),
            validate_push_promises: true,
        }
    }
}

// Content analysis and streaming decision types
#[derive(Debug, Clone, PartialEq)]
pub struct StreamingAnalysis {
    pub should_stream: bool,
    pub estimated_size: usize,
    pub content_type: ContentType,
    pub recommended_strategy: StreamingStrategy,
    pub estimated_chunks: usize,
}

#[derive(Debug, Clone, PartialEq)]
pub enum ContentType {
    Standard,
    LargeText,
    Binary,
    Json,
}

#[derive(Debug, Clone, PartialEq)]
pub enum StreamingStrategy {
    Traditional,
    ChunkedStreaming,
    #[cfg(feature = "http2")]
    Http2ServerPush,
    #[cfg(feature = "http2")]
    Http2Multiplexed,
    #[cfg(feature = "compression")]
    CompressedStreaming,
}

/// Configuration for streaming HTTP transport
#[derive(Debug, Clone)]
pub struct StreamingConfig {
    /// Enable chunked transfer encoding for large requests
    pub enable_chunked_transfer: bool,
    /// Threshold in bytes to trigger chunked streaming
    pub chunk_threshold: usize,
    /// Chunk size for streaming operations
    pub chunk_size: usize,
    /// Enable compression support
    pub enable_compression: bool,
    /// Compression algorithm to use
    pub compression_type: CompressionType,
    /// Enable HTTP/2 Server Push if available
    pub enable_http2_server_push: bool,
    /// Timeout for streaming operations
    pub streaming_timeout_ms: u64,
    /// Maximum concurrent chunks
    pub max_concurrent_chunks: usize,
    /// Backpressure threshold
    pub backpressure_threshold: usize,
    /// Enable adaptive chunk sizing
    pub adaptive_chunk_sizing: bool,
    /// HTTP/2 specific configuration
    #[cfg(feature = "http2")]
    pub http2_config: Http2Config,
}

impl Default for StreamingConfig {
    fn default() -> Self {
        Self {
            enable_chunked_transfer: true,
            chunk_threshold: 8192, // 8KB threshold
            chunk_size: 16384,     // 16KB chunks
            enable_compression: true,
            compression_type: CompressionType::Gzip,
            enable_http2_server_push: false, // Opt-in
            streaming_timeout_ms: 60_000,
            max_concurrent_chunks: 10,
            backpressure_threshold: 1024 * 1024, // 1MB
            adaptive_chunk_sizing: true,
            #[cfg(feature = "http2")]
            http2_config: Http2Config::default(),
        }
    }
}

impl StreamingConfig {
    /// Configuration for memory-constrained environments
    pub fn memory_improved() -> Self {
        Self {
            chunk_threshold: 4096, // 4KB threshold
            chunk_size: 8192,      // 8KB chunks
            compression_type: CompressionType::Gzip,
            max_concurrent_chunks: 5,
            backpressure_threshold: 512 * 1024, // 512KB
            ..Default::default()
        }
    }

    /// Configuration for high-performance scenarios
    pub fn performance_improved() -> Self {
        Self {
            chunk_threshold: 32768, // 32KB threshold
            chunk_size: 65536,      // 64KB chunks
            enable_http2_server_push: true,
            max_concurrent_chunks: 20,
            backpressure_threshold: 4 * 1024 * 1024, // 4MB
            #[cfg(feature = "compression")]
            compression_type: CompressionType::Brotli,
            #[cfg(not(feature = "compression"))]
            compression_type: CompressionType::Gzip,
            #[cfg(feature = "http2")]
            http2_config: Http2Config {
                max_concurrent_streams: 200,
                initial_window_size: 131072, // 128KB
                max_frame_size: 32768,       // 32KB
                enable_server_push: true,
                push_cache_size: 2000,
                connection_timeout: Duration::from_secs(60),
                keep_alive_interval: Duration::from_secs(5),
                validate_push_promises: true,
            },
            ..Default::default()
        }
    }

    /// Builder method to set chunk size
    pub fn with_chunk_size(mut self, chunk_size: usize) -> Self {
        self.chunk_size = chunk_size;
        self
    }

    /// Builder method to set compression level
    pub fn with_compression_level(mut self, level: u8) -> Self {
        // Map compression level to compression type
        match level {
            0 => {
                self.compression_type = CompressionType::None;
                self.enable_compression = false;
            }
            1..=6 => {
                self.compression_type = CompressionType::Gzip;
                self.enable_compression = true;
            }
            #[cfg(feature = "compression")]
            7..=9 => {
                self.compression_type = CompressionType::Brotli;
                self.enable_compression = true;
            }
            #[cfg(not(feature = "compression"))]
            7..=9 => {
                self.compression_type = CompressionType::Gzip;
                self.enable_compression = true;
            }
            _ => {
                self.compression_type = CompressionType::Gzip;
                self.enable_compression = true;
            }
        }
        self
    }

    /// Builder method to set max concurrent streams
    pub fn with_max_concurrent_streams(mut self, max_streams: usize) -> Self {
        self.max_concurrent_chunks = max_streams;
        self
    }

    /// Get compression level as u8 for tests
    pub fn compression_level(&self) -> u8 {
        match self.compression_type {
            CompressionType::None => 0,
            CompressionType::Gzip => 6,
            #[cfg(feature = "compression")]
            CompressionType::Brotli => 8,
            #[cfg(feature = "compression")]
            CompressionType::Zstd => 7,
        }
    }

    /// Get max concurrent streams (alias for max_concurrent_chunks)
    pub fn max_concurrent_streams(&self) -> usize {
        self.max_concurrent_chunks
    }
}

#[derive(Debug, Clone, PartialEq)]
pub enum CompressionType {
    None,
    Gzip,
    #[cfg(feature = "compression")]
    Brotli,
    #[cfg(feature = "compression")]
    Zstd,
}

// For tests without streaming-compression feature
#[cfg(all(test, not(feature = "compression")))]
impl CompressionType {
    #[allow(dead_code)]
    pub fn test_brotli() -> Self {
        Self::Gzip
    }
    #[allow(dead_code)]
    pub fn test_zstd() -> Self {
        Self::Gzip
    }
}

/// smart content analyzer for streaming decisions
pub struct ContentAnalyzer {
    stats: Arc<RwLock<AnalysisStats>>,
}

#[derive(Debug, Default, Clone)]
pub struct AnalysisStats {
    pub total_requests: u64,
    pub large_requests: u64,
    pub binary_requests: u64,
    pub avg_size: f64,
}

impl ContentAnalyzer {
    pub fn new() -> Self {
        Self {
            stats: Arc::new(RwLock::new(AnalysisStats::default())),
        }
    }

    /// Analyze request to determine if streaming is beneficial
    pub async fn analyze_request(&self, request: &JsonRpcRequest) -> StreamingAnalysis {
        let content_size = serde_json::to_string(request).unwrap_or_default().len();
        let has_large_strings = self.has_large_string_content(request);
        let has_binary_data = self.has_binary_content(request);

        // Update statistics
        {
            let mut stats = self.stats.write().await;
            stats.total_requests += 1;
            if content_size > 8192 {
                stats.large_requests += 1;
            }
            if has_binary_data {
                stats.binary_requests += 1;
            }
            stats.avg_size = (stats.avg_size * (stats.total_requests - 1) as f64
                + content_size as f64)
                / stats.total_requests as f64;
        }

        let content_type = if has_binary_data {
            ContentType::Binary
        } else if has_large_strings {
            ContentType::LargeText
        } else if self.is_complex_json(request) {
            ContentType::Json
        } else {
            ContentType::Standard
        };

        let should_stream = content_size > 8192 || has_large_strings || has_binary_data;

        let recommended_strategy = if has_binary_data {
            #[cfg(feature = "compression")]
            {
                StreamingStrategy::CompressedStreaming
            }
            #[cfg(not(feature = "compression"))]
            {
                StreamingStrategy::ChunkedStreaming
            }
        } else if content_size > 100_000 {
            // Very large payloads benefit from multiplexing
            #[cfg(feature = "http2")]
            {
                StreamingStrategy::Http2Multiplexed
            }
            #[cfg(not(feature = "http2"))]
            {
                StreamingStrategy::ChunkedStreaming
            }
        } else if content_size > 32768 {
            #[cfg(feature = "http2")]
            {
                StreamingStrategy::Http2ServerPush
            }
            #[cfg(not(feature = "http2"))]
            {
                StreamingStrategy::ChunkedStreaming
            }
        } else if should_stream {
            StreamingStrategy::ChunkedStreaming
        } else {
            StreamingStrategy::Traditional
        };

        StreamingAnalysis {
            should_stream,
            estimated_size: content_size,
            content_type,
            recommended_strategy,
            estimated_chunks: (content_size / 16384).max(1),
        }
    }

    fn has_large_string_content(&self, request: &JsonRpcRequest) -> bool {
        if let Some(params) = &request.params {
            self.find_large_strings(params, 4096)
        } else {
            false
        }
    }

    fn has_binary_content(&self, request: &JsonRpcRequest) -> bool {
        if let Some(params) = &request.params {
            self.find_binary_indicators(params)
        } else {
            false
        }
    }

    fn is_complex_json(&self, request: &JsonRpcRequest) -> bool {
        if let Some(params) = &request.params {
            self.count_json_depth(params) > 5
        } else {
            false
        }
    }

    fn find_large_strings(&self, value: &Value, threshold: usize) -> bool {
        Self::find_large_strings_recursive(value, threshold)
    }

    fn find_large_strings_recursive(value: &Value, threshold: usize) -> bool {
        match value {
            Value::String(s) => s.len() > threshold,
            Value::Array(arr) => arr
                .iter()
                .any(|v| Self::find_large_strings_recursive(v, threshold)),
            Value::Object(obj) => obj
                .values()
                .any(|v| Self::find_large_strings_recursive(v, threshold)),
            _ => false,
        }
    }

    fn find_binary_indicators(&self, value: &Value) -> bool {
        Self::find_binary_indicators_recursive(value)
    }

    fn find_binary_indicators_recursive(value: &Value) -> bool {
        match value {
            Value::String(s) => {
                // Look for base64 patterns or binary indicators
                s.len() > 1000
                    && (s
                        .chars()
                        .all(|c| c.is_ascii_alphanumeric() || c == '+' || c == '/' || c == '='))
            }
            Value::Array(arr) => arr.iter().any(Self::find_binary_indicators_recursive),
            Value::Object(obj) => obj.iter().any(|(k, v)| {
                k.contains("data")
                    || k.contains("blob")
                    || k.contains("binary")
                    || Self::find_binary_indicators_recursive(v)
            }),
            _ => false,
        }
    }

    fn count_json_depth(&self, value: &Value) -> usize {
        Self::count_json_depth_recursive(value)
    }

    fn count_json_depth_recursive(value: &Value) -> usize {
        match value {
            Value::Array(arr) => {
                1 + arr
                    .iter()
                    .map(Self::count_json_depth_recursive)
                    .max()
                    .unwrap_or(0)
            }
            Value::Object(obj) => {
                1 + obj
                    .values()
                    .map(Self::count_json_depth_recursive)
                    .max()
                    .unwrap_or(0)
            }
            _ => 0,
        }
    }

    pub async fn get_stats(&self) -> AnalysisStats {
        self.stats.read().await.clone()
    }

    /// Check if content should be streamed based on size and type
    pub fn should_stream(&self, content: &[u8]) -> bool {
        let size = content.len();
        let threshold = 8192; // 8KB default threshold

        if size > threshold {
            return true;
        }

        // Check for binary content that benefits from streaming
        self.is_binary_content(content)
    }

    /// Detect optimal compression type for content
    pub fn detect_optimal_compression_type(&self, content: &[u8]) -> CompressionType {
        if content.is_empty() {
            return CompressionType::None;
        }

        // Analyze content entropy to determine best compression
        let entropy = self.calculate_entropy(content);

        if entropy < 0.3 {
            // Low entropy - highly repetitive content, good for compression
            #[cfg(feature = "compression")]
            return CompressionType::Brotli;
            #[cfg(not(feature = "compression"))]
            return CompressionType::Gzip;
        } else if entropy < 0.7 {
            // Medium entropy - moderate compression benefit
            CompressionType::Gzip
        } else {
            // High entropy - minimal compression benefit
            CompressionType::None
        }
    }

    /// Get streaming threshold
    pub fn streaming_threshold(&self) -> usize {
        8192 // 8KB default
    }

    /// Calculate content entropy for compression analysis
    fn calculate_entropy(&self, content: &[u8]) -> f64 {
        if content.is_empty() {
            return 0.0;
        }

        // For very small content, return high entropy (low compression benefit)
        if content.len() < 50 {
            return 0.8;
        }

        let mut freq = [0u32; 256];
        for &byte in content {
            freq[byte as usize] += 1;
        }

        let len = content.len() as f64;
        let mut entropy = 0.0;

        for &count in &freq {
            if count > 0 {
                let p = count as f64 / len;
                entropy -= p * p.log2();
            }
        }

        entropy / 8.0 // Normalize to 0-1 range
    }

    /// Check if content is binary
    fn is_binary_content(&self, content: &[u8]) -> bool {
        if content.is_empty() {
            return false;
        }

        // Null bytes are a strong indicator of binary content
        if content.contains(&0) {
            // But if it's mostly nulls, it's probably test data
            let null_count = content.iter().filter(|&&b| b == 0).count();
            if (null_count as f64 / content.len() as f64) > 0.9 {
                return false; // Mostly null = test data, not binary
            }
            return true;
        }

        // Simple heuristic: if more than 30% of bytes are non-printable, consider binary
        let non_printable = content
            .iter()
            .filter(|&&b| b < 32 && b != b'\t' && b != b'\n' && b != b'\r')
            .count();

        (non_printable as f64 / content.len() as f64) > 0.3
    }
}

/// Streaming buffer with flow control
#[cfg(feature = "chunked-encoding")]
pub struct StreamingBuffer {
    buffer: BytesMut,
    chunk_size: usize,
    max_buffer_size: usize,
    flow_control: FlowControl,
    network_metrics: NetworkMetrics,
}

#[cfg(feature = "chunked-encoding")]
#[derive(Debug, Clone)]
pub struct FlowControl {
    pub max_concurrent_chunks: usize,
    pub backpressure_threshold: usize,
    pub adaptive_chunk_sizing: bool,
}

#[cfg(feature = "chunked-encoding")]
#[derive(Debug, Clone, Default)]
pub struct NetworkMetrics {
    pub high_bandwidth: bool,
    pub low_latency: bool,
    pub avg_latency_ms: f64,
    pub throughput_bps: f64,
}

#[cfg(feature = "chunked-encoding")]
impl StreamingBuffer {
    pub fn new(chunk_size: usize, max_buffer_size: usize) -> Self {
        Self {
            buffer: BytesMut::with_capacity(chunk_size),
            chunk_size,
            max_buffer_size,
            flow_control: FlowControl {
                max_concurrent_chunks: 10,
                backpressure_threshold: max_buffer_size / 2,
                adaptive_chunk_sizing: true,
            },
            network_metrics: NetworkMetrics::default(),
        }
    }

    /// Add data to buffer and yield chunks when ready
    pub async fn add_data(&mut self, data: &[u8]) -> McpResult<Vec<Bytes>> {
        // Check buffer size limit
        if self.buffer.len() + data.len() > self.max_buffer_size {
            return Err(McpError::transport(format!(
                "Buffer size limit exceeded: {} bytes",
                self.max_buffer_size
            )));
        }

        self.buffer.extend_from_slice(data);

        let mut chunks = Vec::new();
        while self.buffer.len() >= self.chunk_size {
            let chunk = self.buffer.split_to(self.chunk_size);
            chunks.push(chunk.freeze());

            // Apply flow control
            if chunks.len() >= self.flow_control.max_concurrent_chunks {
                break;
            }
        }

        // Adaptive chunk sizing based on network conditions
        if self.flow_control.adaptive_chunk_sizing {
            self.adjust_chunk_size().await;
        }

        Ok(chunks)
    }

    /// Flush remaining data
    pub fn flush(&mut self) -> Option<Bytes> {
        if !self.buffer.is_empty() {
            Some(self.buffer.split().freeze())
        } else {
            None
        }
    }

    /// Dynamically adjust chunk size based on performance metrics
    async fn adjust_chunk_size(&mut self) {
        // Monitor network latency, throughput, etc.
        let metrics = &self.network_metrics;

        if metrics.high_bandwidth && metrics.low_latency {
            self.chunk_size = (self.chunk_size * 2).min(64 * 1024);
        } else if !metrics.low_latency {
            self.chunk_size = (self.chunk_size / 2).max(1024);
        }
    }

    pub fn update_network_metrics(&mut self, latency_ms: f64, throughput_bps: f64) {
        self.network_metrics.avg_latency_ms = latency_ms;
        self.network_metrics.throughput_bps = throughput_bps;
        self.network_metrics.low_latency = latency_ms < 50.0;
        self.network_metrics.high_bandwidth = throughput_bps > 10_000_000.0; // 10 Mbps
    }
}

/// Streaming compressor for multiple compression algorithms
#[cfg(feature = "compression")]
pub struct StreamingCompressor {
    compression_type: CompressionType,
    threshold: usize,
}

#[cfg(feature = "compression")]
impl StreamingCompressor {
    pub fn new(compression_type: CompressionType) -> Self {
        Self {
            compression_type,
            threshold: 1024, // Compress if > 1KB
        }
    }

    /// Compress payload if above threshold and beneficial
    pub async fn compress_if_beneficial(&self, data: &[u8]) -> McpResult<Vec<u8>> {
        if data.len() < self.threshold {
            return Ok(data.to_vec());
        }

        match self.compression_type {
            CompressionType::Gzip => self.compress_gzip(data).await,
            CompressionType::Brotli => self.compress_brotli(data).await,
            CompressionType::Zstd => self.compress_zstd(data).await,
            CompressionType::None => Ok(data.to_vec()),
        }
    }

    async fn compress_gzip(&self, data: &[u8]) -> McpResult<Vec<u8>> {
        use flate2::write::GzEncoder;
        use flate2::Compression;
        use std::io::Write;

        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
        encoder.write_all(data).map_err(McpError::io)?;
        encoder.finish().map_err(McpError::io)
    }

    async fn compress_brotli(&self, data: &[u8]) -> McpResult<Vec<u8>> {
        let mut output = Vec::new();
        let mut reader = std::io::Cursor::new(data);
        brotli::BrotliCompress(
            &mut reader,
            &mut output,
            &brotli::enc::BrotliEncoderParams::default(),
        )
        .map_err(|e| McpError::internal(format!("Brotli compression failed: {e}")))?;
        Ok(output)
    }

    async fn compress_zstd(&self, data: &[u8]) -> McpResult<Vec<u8>> {
        zstd::bulk::compress(data, 3)
            .map_err(|e| McpError::internal(format!("Zstd compression failed: {e}")))
    }

    /// Estimate compression ratio for decision making
    pub fn estimate_compression_ratio(&self, data: &[u8]) -> f64 {
        // Simple heuristic based on data characteristics
        let entropy = self.calculate_entropy(data);
        match self.compression_type {
            CompressionType::None => 1.0,
            CompressionType::Gzip => (1.0 - entropy * 0.6).max(0.3),
            CompressionType::Brotli => (1.0 - entropy * 0.7).max(0.25),
            CompressionType::Zstd => (1.0 - entropy * 0.65).max(0.28),
        }
    }

    fn calculate_entropy(&self, data: &[u8]) -> f64 {
        let mut counts = [0u32; 256];
        for &byte in data {
            counts[byte as usize] += 1;
        }

        let len = data.len() as f64;
        let mut entropy = 0.0;

        for &count in &counts {
            if count > 0 {
                let p = count as f64 / len;
                entropy -= p * p.log2();
            }
        }

        entropy / 8.0 // Normalize to 0-1
    }
}

/// Streaming HTTP client transport with advanced features
#[cfg(feature = "chunked-encoding")]
pub struct StreamingHttpClientTransport {
    client: reqwest::Client,
    base_url: String,
    config: StreamingConfig,
    content_analyzer: ContentAnalyzer,
    #[cfg(feature = "compression")]
    compressor: Option<StreamingCompressor>,
    #[cfg(feature = "http2")]
    h2_client: Option<SendRequest<bytes::Bytes>>,
    #[cfg(feature = "http2")]
    h2_connection: Option<Connection<TcpStream, bytes::Bytes>>,
    #[cfg(feature = "http2")]
    server_push_handlers: Arc<RwLock<HashMap<String, ServerPushHandler>>>,
    #[cfg(feature = "http2")]
    stream_manager: Arc<RwLock<Http2StreamManager>>,
    stats: Arc<RwLock<TransportStats>>,
    connection_state: Arc<RwLock<ConnectionState>>,
}

#[cfg(feature = "chunked-encoding")]
impl StreamingHttpClientTransport {
    /// Create streaming HTTP transport with default configuration
    pub async fn new<S: AsRef<str>>(base_url: S) -> McpResult<Self> {
        Self::with_config(base_url, StreamingConfig::default()).await
    }

    /// Create streaming HTTP transport with custom configuration
    pub async fn with_config<S: AsRef<str>>(
        base_url: S,
        config: StreamingConfig,
    ) -> McpResult<Self> {
        let client_builder =
            reqwest::Client::builder().timeout(Duration::from_millis(config.streaming_timeout_ms));

        // Note: For full HTTP/2 Server Push support, we would use h2 crate directly
        // reqwest 0.12.x doesn't expose all HTTP/2 features we need
        // The current implementation provides chunked streaming which works over HTTP/2
        #[cfg(feature = "http2")]
        let _http2_config = config.enable_http2_server_push; // Prepared for future h2 integration

        let client = client_builder
            .build()
            .map_err(|e| McpError::Http(format!("Failed to create streaming client: {e}")))?;

        #[cfg(feature = "compression")]
        let compressor =
            if config.enable_compression && config.compression_type != CompressionType::None {
                Some(StreamingCompressor::new(config.compression_type.clone()))
            } else {
                None
            };

        // Extract HTTP/2 config before moving config
        #[cfg(feature = "http2")]
        let max_streams = config.http2_config.max_concurrent_streams;

        Ok(Self {
            client,
            base_url: base_url.as_ref().to_string(),
            config,
            content_analyzer: ContentAnalyzer::new(),
            #[cfg(feature = "compression")]
            compressor,
            #[cfg(feature = "http2")]
            h2_client: None, // Will be initialized on first HTTP/2 request
            #[cfg(feature = "http2")]
            h2_connection: None,
            #[cfg(feature = "http2")]
            server_push_handlers: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(feature = "http2")]
            stream_manager: Arc::new(RwLock::new(Http2StreamManager::new(max_streams))),
            stats: Arc::new(RwLock::new(TransportStats::default())),
            connection_state: Arc::new(RwLock::new(ConnectionState::Disconnected)),
        })
    }

    /// Send request with smart streaming decision
    async fn send_request_smart(&mut self, request: JsonRpcRequest) -> McpResult<JsonRpcResponse> {
        let start_time = Instant::now();

        // Analyze request to determine streaming strategy
        let analysis = self.content_analyzer.analyze_request(&request).await;

        debug!(
            "Streaming analysis: should_stream={}, size={}, strategy={:?}",
            analysis.should_stream, analysis.estimated_size, analysis.recommended_strategy
        );

        let result = if analysis.should_stream && self.config.enable_chunked_transfer {
            match analysis.recommended_strategy {
                StreamingStrategy::ChunkedStreaming => {
                    match self.send_chunked_request(request.clone()).await {
                        Ok(response) => Ok(response),
                        Err(e) => {
                            warn!("Chunked request failed, falling back to traditional: {}", e);
                            self.send_traditional_request(request).await
                        }
                    }
                }
                #[cfg(feature = "compression")]
                StreamingStrategy::CompressedStreaming => {
                    match self.send_compressed_request(request.clone()).await {
                        Ok(response) => Ok(response),
                        Err(e) => {
                            warn!(
                                "Compressed request failed, falling back to traditional: {}",
                                e
                            );
                            self.send_traditional_request(request).await
                        }
                    }
                }
                #[cfg(feature = "http2")]
                StreamingStrategy::Http2ServerPush => {
                    match self.send_http2_request(request.clone()).await {
                        Ok(response) => Ok(response),
                        Err(e) => {
                            warn!("HTTP/2 request failed, falling back to chunked: {}", e);
                            match self.send_chunked_request(request.clone()).await {
                                Ok(response) => Ok(response),
                                Err(_) => self.send_traditional_request(request).await,
                            }
                        }
                    }
                }
                #[cfg(feature = "http2")]
                StreamingStrategy::Http2Multiplexed => {
                    // For multiplexed strategy, we use the same HTTP/2 implementation
                    // but could optimize for concurrent requests in the future
                    match self.send_http2_request(request.clone()).await {
                        Ok(response) => Ok(response),
                        Err(e) => {
                            warn!(
                                "HTTP/2 multiplexed request failed, falling back to chunked: {}",
                                e
                            );
                            match self.send_chunked_request(request.clone()).await {
                                Ok(response) => Ok(response),
                                Err(_) => self.send_traditional_request(request).await,
                            }
                        }
                    }
                }
                _ => self.send_traditional_request(request).await,
            }
        } else {
            self.send_traditional_request(request).await
        };

        // Update statistics
        let _duration = start_time.elapsed();
        {
            let mut stats = self.stats.write().await;
            stats.requests_sent += 1;
            stats.bytes_sent += analysis.estimated_size as u64;
            // Note: More detailed latency tracking could be added to TransportStats in future
        }

        result
    }

    async fn send_chunked_request(
        &mut self,
        request: JsonRpcRequest,
    ) -> McpResult<JsonRpcResponse> {
        let request_json = serde_json::to_string(&request)?.into_bytes();

        // Create streaming body
        let mut buffer =
            StreamingBuffer::new(self.config.chunk_size, self.config.backpressure_threshold);
        let chunks = buffer.add_data(&request_json).await?;

        // Add final chunk if there's remaining data
        let final_chunk = buffer.flush();

        // Create chunk stream
        let (tx, rx) = tokio::sync::mpsc::channel::<Result<bytes::Bytes, McpError>>(
            self.config.max_concurrent_chunks,
        );

        // Send chunks
        tokio::spawn(async move {
            for chunk in chunks {
                if tx.send(Ok(chunk)).await.is_err() {
                    break;
                }
            }
            if let Some(chunk) = final_chunk {
                let _ = tx.send(Ok(chunk)).await;
            }
        });

        let chunk_stream = ReceiverStream::new(rx);

        let url = format!("{}/mcp/stream", self.base_url);
        let response = self
            .client
            .post(&url)
            .header("Transfer-Encoding", "chunked")
            .header("Content-Type", "application/json")
            .body(reqwest::Body::wrap_stream(chunk_stream))
            .send()
            .await
            .map_err(|e| McpError::Http(format!("Chunked request failed: {e}")))?;

        self.handle_response(response).await
    }

    #[cfg(feature = "compression")]
    async fn send_compressed_request(
        &mut self,
        request: JsonRpcRequest,
    ) -> McpResult<JsonRpcResponse> {
        let request_json = serde_json::to_string(&request)?.into_bytes();

        // Compress if beneficial
        let request_bytes = if let Some(ref compressor) = self.compressor {
            compressor.compress_if_beneficial(&request_json).await?
        } else {
            request_json
        };

        let url = format!("{}/mcp/compressed", self.base_url);
        let mut request_builder = self
            .client
            .post(&url)
            .header("Content-Type", "application/json")
            .body(request_bytes);

        // Add compression headers
        if let Some(ref compressor) = self.compressor {
            let encoding = match compressor.compression_type {
                CompressionType::Gzip => "gzip",
                CompressionType::Brotli => "br",
                CompressionType::Zstd => "zstd",
                CompressionType::None => "identity",
            };
            request_builder = request_builder.header("Content-Encoding", encoding);
        }

        let response = request_builder
            .send()
            .await
            .map_err(|e| McpError::Http(format!("Compressed request failed: {e}")))?;

        self.handle_response(response).await
    }

    #[cfg(feature = "http2")]
    async fn send_http2_request(&mut self, request: JsonRpcRequest) -> McpResult<JsonRpcResponse> {
        // Initialize H2 client if not already done
        if self.h2_client.is_none() {
            self.init_h2_client().await?;
        }

        // Check if we can create a new stream
        let can_create_stream = {
            let stream_manager = self.stream_manager.read().await;
            stream_manager.can_create_stream()
        };

        if !can_create_stream {
            debug!("Max streams reached, falling back to chunked streaming");
            return self.send_chunked_request(request).await;
        }

        let request_json = serde_json::to_string(&request)?;
        let request_json_len = request_json.len();
        let request_bytes = request_json.into_bytes();

        // Use direct h2 client for full control
        if let Some(ref mut h2_client) = self.h2_client {
            // Create HTTP request
            let http_request = Request::builder()
                .method(Method::POST)
                .uri("/mcp/http2")
                .header("content-type", "application/json")
                .header("content-length", request_bytes.len())
                .body(())
                .map_err(|e| McpError::Http(format!("Failed to build HTTP/2 request: {e}")))?;

            // Send request and get stream
            let (response_future, mut send_stream) = h2_client
                .send_request(http_request, false)
                .map_err(|e| McpError::Http(format!("Failed to send HTTP/2 request: {e}")))?;

            // Send body data
            send_stream
                .send_data(Bytes::from(request_bytes), true)
                .map_err(|e| McpError::Http(format!("Failed to send HTTP/2 body: {e}")))?;

            // Wait for response
            let response = response_future
                .await
                .map_err(|e| McpError::Http(format!("HTTP/2 response error: {e}")))?;

            // Process response
            let (parts, mut body) = response.into_parts();

            if !parts.status.is_success() {
                return Err(McpError::Http(format!("HTTP/2 error: {}", parts.status)));
            }

            // Collect response body
            let mut response_bytes = Vec::new();
            while let Some(chunk) = body.data().await {
                let chunk = chunk.map_err(|e| McpError::Http(format!("HTTP/2 body error: {e}")))?;
                response_bytes.extend_from_slice(&chunk);
                // Send flow control update
                let _ = body.flow_control().release_capacity(chunk.len());
            }

            // Parse JSON response
            let response_text = String::from_utf8(response_bytes)
                .map_err(|e| McpError::Http(format!("Invalid UTF-8 in HTTP/2 response: {e}")))?;

            let json_response: JsonRpcResponse =
                serde_json::from_str(&response_text).map_err(|e| {
                    McpError::Http(format!("Failed to parse HTTP/2 JSON response: {e}"))
                })?;

            // Update stream manager
            {
                let mut stream_manager = self.stream_manager.write().await;
                stream_manager.update_stream_bytes(
                    0,
                    request_json_len as u64,
                    response_text.len() as u64,
                );
            }

            Ok(json_response)
        } else {
            // Fallback to chunked streaming if h2 client not available
            debug!("H2 client not available, falling back to chunked streaming");
            self.send_chunked_request(request).await
        }
    }

    async fn send_traditional_request(
        &mut self,
        request: JsonRpcRequest,
    ) -> McpResult<JsonRpcResponse> {
        let url = format!("{}/mcp", self.base_url);

        let response = self
            .client
            .post(&url)
            .json(&request)
            .send()
            .await
            .map_err(|e| McpError::Http(format!("Traditional request failed: {e}")))?;

        self.handle_response(response).await
    }

    async fn handle_response(&self, response: reqwest::Response) -> McpResult<JsonRpcResponse> {
        if !response.status().is_success() {
            return Err(McpError::Http(format!("HTTP error: {}", response.status())));
        }

        let json_response: JsonRpcResponse = response
            .json()
            .await
            .map_err(|e| McpError::Http(format!("Failed to parse response: {e}")))?;

        Ok(json_response)
    }

    pub async fn get_stats(&self) -> TransportStats {
        self.stats.read().await.clone()
    }

    pub async fn get_analysis_stats(&self) -> AnalysisStats {
        self.content_analyzer.get_stats().await
    }

    /// Register a server push handler for specific paths
    #[cfg(feature = "http2")]
    pub async fn register_push_handler<F>(&mut self, path: String, handler: F)
    where
        F: Fn(PushPromise) -> Pin<Box<dyn std::future::Future<Output = McpResult<()>> + Send>>
            + Send
            + Sync
            + 'static,
    {
        let boxed_handler: ServerPushHandler = Box::new(handler);
        let mut handlers = self.server_push_handlers.write().await;
        handlers.insert(path, boxed_handler);
    }

    /// Get HTTP/2 stream statistics
    #[cfg(feature = "http2")]
    pub async fn get_http2_stats(&self) -> (usize, u64, u64) {
        let stream_manager = self.stream_manager.read().await;
        stream_manager.get_stream_stats()
    }

    /// Initialize HTTP/2 client with direct h2 connection
    #[cfg(feature = "http2")]
    async fn init_h2_client(&mut self) -> McpResult<()> {
        use url::Url;

        // Parse the base URL to extract host and port
        let url = Url::parse(&self.base_url)
            .map_err(|e| McpError::Http(format!("Invalid base URL: {e}")))?;

        let host = url
            .host_str()
            .ok_or_else(|| McpError::Http("No host in base URL".to_string()))?;
        let port = url.port().unwrap_or(443); // Default to HTTPS port

        // Establish TCP connection
        let tcp_stream = TcpStream::connect(format!("{host}:{port}"))
            .await
            .map_err(|e| McpError::Http(format!("Failed to connect to {host}:{port}: {e}")))?;

        // Perform HTTP/2 handshake
        let (h2_client, connection) = h2::client::handshake(tcp_stream)
            .await
            .map_err(|e| McpError::Http(format!("HTTP/2 handshake failed: {e}")))?;

        // Store the client and connection
        self.h2_client = Some(h2_client);
        self.h2_connection = Some(connection);

        // Spawn connection driver task
        let connection = self.h2_connection.take().unwrap();
        tokio::spawn(async move {
            if let Err(e) = connection.await {
                tracing::error!("HTTP/2 connection error: {}", e);
            }
        });

        debug!("HTTP/2 client initialized successfully");
        Ok(())
    }

    /// Send multiplexed HTTP/2 requests for improved performance
    #[cfg(feature = "http2")]
    pub async fn send_multiplexed_requests(
        &mut self,
        requests: Vec<JsonRpcRequest>,
    ) -> McpResult<Vec<JsonRpcResponse>> {
        if self.h2_client.is_none() {
            self.init_h2_client().await?;
        }

        let mut response_futures = Vec::new();

        for request in requests {
            let request_json = serde_json::to_string(&request)?;
            let request_bytes = request_json.into_bytes();

            if let Some(ref mut h2_client) = self.h2_client {
                // Create HTTP request
                let http_request = Request::builder()
                    .method(Method::POST)
                    .uri("/mcp/http2")
                    .header("content-type", "application/json")
                    .header("content-length", request_bytes.len())
                    .body(())
                    .map_err(|e| McpError::Http(format!("Failed to build HTTP/2 request: {e}")))?;

                // Send request and get stream
                let (response_future, mut send_stream) = h2_client
                    .send_request(http_request, false)
                    .map_err(|e| McpError::Http(format!("Failed to send HTTP/2 request: {e}")))?;

                // Send body data asynchronously
                tokio::spawn(async move {
                    if let Err(e) = send_stream.send_data(Bytes::from(request_bytes), true) {
                        tracing::error!("Failed to send HTTP/2 body: {}", e);
                    }
                });

                response_futures.push(response_future);
            }
        }

        // Wait for all responses concurrently
        let mut responses = Vec::new();
        for response_future in response_futures {
            match response_future.await {
                Ok(response) => {
                    let (parts, mut body) = response.into_parts();

                    if !parts.status.is_success() {
                        return Err(McpError::Http(format!("HTTP/2 error: {}", parts.status)));
                    }

                    // Collect response body
                    let mut response_bytes = Vec::new();
                    while let Some(chunk) = body.data().await {
                        let chunk =
                            chunk.map_err(|e| McpError::Http(format!("HTTP/2 body error: {e}")))?;
                        response_bytes.extend_from_slice(&chunk);
                        let _ = body.flow_control().release_capacity(chunk.len());
                    }

                    // Parse JSON response
                    let response_text = String::from_utf8(response_bytes).map_err(|e| {
                        McpError::Http(format!("Invalid UTF-8 in HTTP/2 response: {e}"))
                    })?;

                    let json_response: JsonRpcResponse = serde_json::from_str(&response_text)
                        .map_err(|e| {
                            McpError::Http(format!("Failed to parse HTTP/2 JSON response: {e}"))
                        })?;

                    responses.push(json_response);
                }
                Err(e) => {
                    return Err(McpError::Http(format!("HTTP/2 response error: {e}")));
                }
            }
        }

        Ok(responses)
    }
}

#[cfg(feature = "chunked-encoding")]
#[async_trait]
impl Transport for StreamingHttpClientTransport {
    async fn send_request(&mut self, request: JsonRpcRequest) -> McpResult<JsonRpcResponse> {
        {
            let mut state = self.connection_state.write().await;
            *state = ConnectionState::Connected;
        }

        let result = self.send_request_smart(request).await;

        // Update connection state based on result
        if result.is_err() {
            let mut state = self.connection_state.write().await;
            *state = ConnectionState::Error("Request failed".to_string());
        }

        result
    }

    async fn send_notification(&mut self, notification: JsonRpcNotification) -> McpResult<()> {
        let url = format!("{}/mcp/notification", self.base_url);

        let response = self
            .client
            .post(&url)
            .json(&notification)
            .send()
            .await
            .map_err(|e| McpError::Http(format!("Notification failed: {e}")))?;

        if !response.status().is_success() {
            return Err(McpError::Http(format!(
                "Notification error: {}",
                response.status()
            )));
        }

        Ok(())
    }

    async fn receive_notification(&mut self) -> McpResult<Option<JsonRpcNotification>> {
        // Streaming HTTP doesn't support server-initiated notifications in the same way
        // This would require Server-Sent Events or WebSocket upgrade
        Ok(None)
    }

    async fn close(&mut self) -> McpResult<()> {
        let mut state = self.connection_state.write().await;
        *state = ConnectionState::Disconnected;
        Ok(())
    }
}

// Stub implementations for when streaming features are not enabled
#[cfg(not(feature = "chunked-encoding"))]
pub struct StreamingHttpClientTransport;

#[cfg(not(feature = "chunked-encoding"))]
impl StreamingHttpClientTransport {
    pub async fn new<S: AsRef<str>>(_base_url: S) -> McpResult<Self> {
        Err(McpError::Transport(
            "Streaming HTTP feature not enabled".to_string(),
        ))
    }

    pub async fn with_config<S: AsRef<str>>(
        _base_url: S,
        _config: StreamingConfig,
    ) -> McpResult<Self> {
        Err(McpError::Transport(
            "Streaming HTTP feature not enabled".to_string(),
        ))
    }
}

#[cfg(not(feature = "chunked-encoding"))]
#[async_trait]
impl Transport for StreamingHttpClientTransport {
    async fn send_request(&mut self, _request: JsonRpcRequest) -> McpResult<JsonRpcResponse> {
        Err(McpError::Transport(
            "Streaming HTTP feature not enabled".to_string(),
        ))
    }

    async fn send_notification(&mut self, _notification: JsonRpcNotification) -> McpResult<()> {
        Err(McpError::Transport(
            "Streaming HTTP feature not enabled".to_string(),
        ))
    }

    async fn receive_notification(&mut self) -> McpResult<Option<JsonRpcNotification>> {
        Err(McpError::Transport(
            "Streaming HTTP feature not enabled".to_string(),
        ))
    }

    async fn close(&mut self) -> McpResult<()> {
        Ok(())
    }
}

// Default implementations and re-exports
impl Default for ContentAnalyzer {
    fn default() -> Self {
        Self::new()
    }
}

// Make ContentAnalyzer cloneable
impl Clone for ContentAnalyzer {
    fn clone(&self) -> Self {
        Self {
            stats: Arc::clone(&self.stats),
        }
    }
}

// Add missing helper types for tests
#[derive(Debug, Clone)]
pub struct AdaptiveBuffer {
    buffer: Vec<u8>,
    optimal_size: usize,
}

impl Default for AdaptiveBuffer {
    fn default() -> Self {
        Self::new()
    }
}

impl AdaptiveBuffer {
    pub fn new() -> Self {
        let buffer = Vec::with_capacity(16384); // Pre-allocate 16KB
        Self {
            buffer,
            optimal_size: 16384, // 16KB default
        }
    }

    pub fn write(&mut self, data: &[u8]) {
        self.buffer.extend_from_slice(data);
        // Adaptive sizing logic
        if self.buffer.len() > self.optimal_size * 2 {
            self.optimal_size = (self.optimal_size * 3) / 2;
        }
    }

    pub fn capacity(&self) -> usize {
        self.buffer.capacity()
    }

    pub fn optimal_size(&self) -> usize {
        self.optimal_size
    }

    pub fn len(&self) -> usize {
        self.buffer.len()
    }

    pub fn is_empty(&self) -> bool {
        self.buffer.is_empty()
    }

    pub fn read(&mut self, size: usize) -> Vec<u8> {
        let read_size = size.min(self.buffer.len());

        self.buffer.drain(..read_size).collect()
    }

    pub fn clear(&mut self) {
        self.buffer.clear();
    }
}

#[derive(Debug, Clone)]
pub struct FlowControlMetrics {
    window_size: usize,
    bytes_sent: usize,
    bytes_acked: usize,
    bytes_received: usize,
    requests_pending: usize,
    last_activity: std::time::Instant,
    rtt_estimate: Duration,
}

impl Default for FlowControlMetrics {
    fn default() -> Self {
        Self::new()
    }
}

impl FlowControlMetrics {
    pub fn new() -> Self {
        Self {
            window_size: 65536, // 64KB default
            bytes_sent: 0,
            bytes_acked: 0,
            bytes_received: 0,
            requests_pending: 0,
            last_activity: std::time::Instant::now(),
            rtt_estimate: Duration::from_millis(100),
        }
    }

    pub fn update_window_size(&mut self, new_size: usize) {
        self.window_size = new_size;
    }

    pub fn record_bytes_sent(&mut self, bytes: usize) {
        self.bytes_sent += bytes;
    }

    pub fn record_bytes_acked(&mut self, bytes: usize) {
        self.bytes_acked += bytes;
    }

    pub fn window_size(&self) -> usize {
        self.window_size
    }

    pub fn bytes_in_flight(&self) -> usize {
        self.bytes_sent.saturating_sub(self.bytes_acked)
    }

    pub fn available_window(&self) -> usize {
        self.window_size.saturating_sub(self.bytes_in_flight())
    }

    pub fn should_send(&self) -> bool {
        self.available_window() > 0
    }

    pub fn update_rtt(&mut self, rtt: Duration) {
        // Simple exponential moving average
        self.rtt_estimate = Duration::from_millis(
            (self.rtt_estimate.as_millis() as f64 * 0.875 + rtt.as_millis() as f64 * 0.125) as u64,
        );
    }

    pub fn estimated_rtt(&self) -> Duration {
        self.rtt_estimate
    }

    pub fn record_bytes_received(&mut self, bytes: usize) {
        self.bytes_received += bytes;
        self.last_activity = std::time::Instant::now();
    }

    pub fn record_request_start(&mut self) {
        self.requests_pending += 1;
        self.last_activity = std::time::Instant::now();
    }

    pub fn record_request_complete(&mut self) {
        if self.requests_pending > 0 {
            self.requests_pending -= 1;
        }
        self.last_activity = std::time::Instant::now();
    }

    pub fn calculate_throughput(&self) -> f64 {
        if self.rtt_estimate.as_millis() == 0 {
            return 0.0;
        }
        (self.bytes_received as f64 * 1000.0) / self.rtt_estimate.as_millis() as f64
    }

    pub fn should_throttle(&self) -> bool {
        self.requests_pending > 10 || self.available_window() < 1024
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;
    use std::time::Duration;
    use tokio::time::timeout;

    #[test]
    fn test_streaming_config_default() {
        let config = StreamingConfig::default();
        assert_eq!(config.chunk_size, 16384);
        assert!(config.enable_compression);
        assert_eq!(config.compression_level(), 6);
        assert_eq!(config.max_concurrent_streams(), 10);
    }

    #[test]
    fn test_streaming_config_builder() {
        let config = StreamingConfig::default()
            .with_chunk_size(4096)
            .with_compression_level(0)
            .with_compression_level(3)
            .with_max_concurrent_streams(50);

        assert_eq!(config.chunk_size, 4096);
        assert!(config.enable_compression); // Level 3 enables compression
        assert_eq!(config.compression_level(), 6); // Level 3 maps to Gzip which returns 6
        assert_eq!(config.max_concurrent_streams(), 50);
    }

    #[test]
    fn test_content_analyzer_creation() {
        let _analyzer = ContentAnalyzer::new();
        // Basic creation test - if this compiles, the constructor works
        let _default_analyzer = ContentAnalyzer::default();
    }

    #[test]
    fn test_content_analyzer_should_stream() {
        let analyzer = ContentAnalyzer::new();

        // Test small content (should not stream)
        let small_data = "small data".as_bytes();
        assert!(!analyzer.should_stream(small_data));

        // Test large content (should stream)
        let large_data = vec![0u8; 10_000];
        assert!(analyzer.should_stream(&large_data));

        // Test boundary case
        let boundary_data = vec![0u8; analyzer.streaming_threshold()];
        assert!(!analyzer.should_stream(&boundary_data)); // Equal to threshold

        let over_boundary = vec![0u8; analyzer.streaming_threshold() + 1];
        assert!(analyzer.should_stream(&over_boundary)); // Over threshold
    }

    #[test]
    fn test_content_analyzer_detect_compression_type() {
        let analyzer = ContentAnalyzer::new();

        // Test JSON content
        let json_data = r#"{"key": "value", "number": 123}"#.as_bytes();
        let compression = analyzer.detect_optimal_compression_type(json_data);
        match compression {
            CompressionType::Gzip => {
                // JSON should benefit from compression
            }
            #[cfg(feature = "compression")]
            CompressionType::Brotli => {
                // JSON should benefit from compression
            }
            #[cfg(feature = "compression")]
            CompressionType::Zstd => {
                // JSON should benefit from compression
            }
            CompressionType::None => {
                // Also acceptable for small JSON
            }
        }

        // Test already compressed data (should not compress further)
        let compressed_data = vec![0x1f, 0x8b, 0x08]; // Gzip header
        let compression = analyzer.detect_optimal_compression_type(&compressed_data);
        assert_eq!(compression, CompressionType::None);

        // Test random data (should use light compression)
        let random_data = vec![42u8; 1000]; // Repetitive data
        let compression = analyzer.detect_optimal_compression_type(&random_data);
        match compression {
            CompressionType::Gzip => {
                // Repetitive data should compress well
            }
            #[cfg(feature = "compression")]
            CompressionType::Brotli => {
                // Repetitive data should compress well
            }
            #[cfg(feature = "compression")]
            CompressionType::Zstd => {
                // Repetitive data should compress well
            }
            CompressionType::None => {
                // Also acceptable
            }
        }
    }

    #[test]
    fn test_adaptive_buffer_creation() {
        let buffer = AdaptiveBuffer::new();
        assert_eq!(buffer.len(), 0);
        assert!(buffer.capacity() >= 8192); // Should have initial capacity
    }

    #[test]
    fn test_adaptive_buffer_write_read() {
        let mut buffer = AdaptiveBuffer::new();

        let test_data = b"Hello, streaming world!";
        buffer.write(test_data);

        assert_eq!(buffer.len(), test_data.len());

        let read_data = buffer.read(10);
        assert_eq!(read_data.len(), 10);
        assert_eq!(&read_data[..], &test_data[..10]);

        // Buffer should have remaining data
        assert_eq!(buffer.len(), test_data.len() - 10);
    }

    #[test]
    fn test_adaptive_buffer_resize() {
        let mut buffer = AdaptiveBuffer::new();
        let initial_capacity = buffer.capacity();

        // Write data larger than initial capacity
        let large_data = vec![42u8; initial_capacity + 1000];
        buffer.write(&large_data);

        // Buffer should have resized
        assert!(buffer.capacity() > initial_capacity);
        assert_eq!(buffer.len(), large_data.len());
    }

    #[test]
    fn test_adaptive_buffer_clear() {
        let mut buffer = AdaptiveBuffer::new();
        buffer.write(b"test data");

        assert!(!buffer.is_empty());
        buffer.clear();
        assert_eq!(buffer.len(), 0);
    }

    #[test]
    fn test_flow_control_metrics_creation() {
        let metrics = FlowControlMetrics::new();
        assert_eq!(metrics.bytes_sent, 0);
        assert_eq!(metrics.bytes_received, 0);
        assert_eq!(metrics.requests_pending, 0);
        assert!(metrics.last_activity.elapsed().as_secs() < 1); // Should be recent
    }

    #[test]
    fn test_flow_control_metrics_update() {
        let mut metrics = FlowControlMetrics::new();

        metrics.record_bytes_sent(1024);
        assert_eq!(metrics.bytes_sent, 1024);

        metrics.record_bytes_received(512);
        assert_eq!(metrics.bytes_received, 512);

        metrics.record_request_start();
        assert_eq!(metrics.requests_pending, 1);

        metrics.record_request_complete();
        assert_eq!(metrics.requests_pending, 0);
    }

    #[test]
    fn test_flow_control_metrics_throughput() {
        let mut metrics = FlowControlMetrics::new();

        // Simulate some activity
        metrics.record_bytes_sent(1000);
        metrics.record_bytes_received(500);

        let throughput = metrics.calculate_throughput();
        assert!(throughput > 0.0);
    }

    #[test]
    fn test_flow_control_metrics_should_throttle() {
        let mut metrics = FlowControlMetrics::new();

        // Normal load - should not throttle
        metrics.record_request_start();
        assert!(!metrics.should_throttle());

        // High load - add many pending requests
        for _ in 0..150 {
            metrics.record_request_start();
        }
        assert!(metrics.should_throttle());
    }

    #[tokio::test]
    #[cfg(not(feature = "chunked-encoding"))]
    async fn test_streaming_client_creation_without_feature() {
        // When streaming-http feature is not enabled, should return error
        let result = StreamingHttpClientTransport::new("http://localhost:3000").await;
        assert!(result.is_err());

        if let Err(McpError::Transport(msg)) = result {
            assert!(msg.contains("Streaming HTTP feature not enabled"));
        } else {
            panic!("Expected Transport error");
        }
    }

    #[tokio::test]
    #[cfg(not(feature = "chunked-encoding"))]
    async fn test_streaming_client_with_config_without_feature() {
        // When streaming-http feature is not enabled, should return error
        let config = StreamingConfig::default();
        let result =
            StreamingHttpClientTransport::with_config("http://localhost:3000", config).await;
        assert!(result.is_err());

        if let Err(McpError::Transport(msg)) = result {
            assert!(msg.contains("Streaming HTTP feature not enabled"));
        } else {
            panic!("Expected Transport error");
        }
    }

    #[tokio::test]
    async fn test_transport_interface_without_feature() {
        // Test Transport trait implementation when feature is disabled
        let result = StreamingHttpClientTransport::new("http://localhost:3000").await;
        if let Ok(mut transport) = result {
            // This branch won't be reached when feature is disabled
            let request = JsonRpcRequest {
                jsonrpc: "2.0".to_string(),
                id: json!(1),
                method: "test".to_string(),
                params: None,
            };

            let response = transport.send_request(request).await;
            assert!(response.is_err());
        }
    }

    #[test]
    fn test_compression_type_enum() {
        // Test that CompressionType enum values are as expected
        let none = CompressionType::None;
        let gzip = CompressionType::Gzip;
        #[cfg(feature = "compression")]
        let brotli = CompressionType::Brotli;
        #[cfg(not(feature = "compression"))]
        let brotli = CompressionType::test_brotli();

        #[cfg(feature = "compression")]
        let zstd = CompressionType::Zstd;
        #[cfg(not(feature = "compression"))]
        let zstd = CompressionType::test_zstd();

        // Test that they can be compared
        assert_eq!(none, CompressionType::None);
        assert_ne!(gzip, CompressionType::None);

        // Only test inequality when compression features are enabled
        #[cfg(feature = "compression")]
        {
            assert_ne!(brotli, gzip);
            assert_ne!(zstd, brotli);
        }

        #[cfg(not(feature = "compression"))]
        {
            // When feature is disabled, test methods return Gzip
            assert_eq!(brotli, gzip);
            assert_eq!(zstd, gzip);
        }
    }

    #[test]
    fn test_performance_metrics() {
        // Test performance-related functionality
        let analyzer = ContentAnalyzer::new();

        // Test that analysis is fast for small data
        let small_data = b"small";
        let start = std::time::Instant::now();
        let _should_stream = analyzer.should_stream(small_data);
        let _compression = analyzer.detect_optimal_compression_type(small_data);
        let duration = start.elapsed();

        // Should be fast (less than 1ms for small data)
        assert!(duration.as_millis() < 10);
    }

    #[test]
    fn test_edge_cases() {
        let analyzer = ContentAnalyzer::new();

        // Test empty data
        let empty_data = b"";
        assert!(!analyzer.should_stream(empty_data));
        assert_eq!(
            analyzer.detect_optimal_compression_type(empty_data),
            CompressionType::None
        );

        // Test single byte
        let single_byte = b"a";
        assert!(!analyzer.should_stream(single_byte));

        // Test adaptive buffer with empty data
        let mut buffer = AdaptiveBuffer::new();
        buffer.write(b"");
        assert_eq!(buffer.len(), 0);

        let empty_read = buffer.read(10);
        assert_eq!(empty_read.len(), 0);
    }

    #[test]
    fn test_streaming_config_validation() {
        let config = StreamingConfig::default();

        // Test that default values are reasonable
        assert!(config.chunk_size > 0);
        assert!(config.chunk_size <= 64 * 1024); // Not too large
        assert!(config.compression_level() <= 9); // Valid compression level
        assert!(config.max_concurrent_streams() > 0);
        assert!(config.max_concurrent_streams() <= 1000); // Reasonable limit
    }

    #[tokio::test]
    async fn test_concurrent_operations() {
        // Test that multiple operations can be performed concurrently
        let analyzer = ContentAnalyzer::new();
        let mut handles = vec![];

        for i in 0..10 {
            let analyzer_clone = analyzer.clone();
            let handle = tokio::spawn(async move {
                let data = format!("test data {i}");
                analyzer_clone.should_stream(data.as_bytes())
            });
            handles.push(handle);
        }

        // All operations should complete successfully
        for handle in handles {
            let result = timeout(Duration::from_millis(100), handle).await;
            assert!(result.is_ok());
        }
    }
}