Struct KiteTickerManager

pub struct KiteTickerManager { /* private fields */ }

High-performance multi-connection WebSocket manager for Kite ticker data

This manager creates 3 independent WebSocket connections and distributes symbols across them using round-robin allocation. Each connection has its own dedicated parser task for maximum performance.

Implementations

impl KiteTickerManager

pub fn new( api_key: String, access_token: String, config: KiteManagerConfig, ) -> Self

Creates a new KiteTickerManager instance with the specified configuration

This initializes the manager with the provided API credentials and configuration, but does not start any connections. Call start() to begin operation.

Arguments

• api_key - Your Kite Connect API key
• access_token - Valid access token from Kite Connect
• config - Manager configuration settings

Example

use kiteticker_async_manager::{KiteTickerManager, KiteManagerConfig, Mode};

let config = KiteManagerConfig {
    max_connections: 3,
    max_symbols_per_connection: 3000,
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP,
    ..Default::default()
};

let manager = KiteTickerManager::new(
    "your_api_key".to_string(),
    "your_access_token".to_string(),
    config,
);

Examples found in repository

examples/basic/runtime_subscription_example.rs (line 35)

async fn main() -> Result<(), String> {
  // Initialize with your API credentials
  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️ Please set KITE_API_KEY and KITE_ACCESS_TOKEN environment variables"
    );
    return Err("Missing API credentials".to_string());
  }

  // Create configuration
  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 3,
    connection_buffer_size: 5000,
    parser_buffer_size: 10000,
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP,
  };

  // Start the manager
  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  println!("🚀 KiteTicker Manager started!");

  // Example of runtime subscription management
  runtime_subscription_demo(&mut manager).await?;

  // Stop the manager
  manager.stop().await?;
  Ok(())
}

examples/mode_change_test.rs (line 42)

async fn main() -> Result<(), String> {
  // Initialize logging
  env_logger::init();

  println!("🔄 KiteTicker Mode Change Test");
  println!("═══════════════════════════════");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️  KITE_API_KEY and KITE_ACCESS_TOKEN environment variables not set"
    );
    demonstrate_mode_change_issue().await;
    return Ok(());
  }

  // Create configuration
  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 3,
    connection_buffer_size: 5000,
    parser_buffer_size: 10000,
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP,
  };

  println!("🔧 Starting manager...");
  let mut manager = KiteTickerManager::new(api_key, access_token, config);

  manager.start().await?;
  println!("✅ Manager started successfully");

  // Test the mode change issue
  test_mode_change_issue(&mut manager).await?;

  // Stop the manager
  println!("\n🛑 Stopping manager...");
  manager.stop().await?;

  println!("🏁 Mode change test completed!");
  Ok(())
}

examples/performance/market_scanner.rs (line 36)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("🔍 Market Scanner Example");
  println!("=========================");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    return Err("Please set KITE_API_KEY and KITE_ACCESS_TOKEN".to_string());
  }

  // High-performance configuration for scanning
  let config = KiteManagerConfig {
    max_connections: 3,
    max_symbols_per_connection: 3000,
    connection_buffer_size: 20000, // Large buffer for high volume
    parser_buffer_size: 50000,     // Even larger for parsed data
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP, // LTP mode for scanning (minimal bandwidth)
    ..Default::default()
  };

  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Large symbol set for market scanning
  let large_symbol_set = generate_symbol_list(8000); // 8000 symbols across 3 connections

  println!(
    "📊 Scanning {} symbols across {} connections",
    large_symbol_set.len(),
    3
  );

  let start_time = Instant::now();
  manager
    .subscribe_symbols(&large_symbol_set, Some(Mode::LTP))
    .await?;
  println!(
    "✅ Subscribed to {} symbols in {:?}",
    large_symbol_set.len(),
    start_time.elapsed()
  );

  // Get all channels and start parallel processing
  let channels = manager.get_all_channels();
  let mut handles = Vec::new();

  for (channel_id, mut receiver) in channels {
    let handle = tokio::spawn(async move {
      let mut scanner = MarketScanner::new(channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          scanner.process_ticks(ticks).await;
        }
      }

      scanner.print_summary();
    });

    handles.push(handle);
  }

  // Run scanner for specified duration
  println!("🔄 Scanning market for 30 seconds...");
  sleep(Duration::from_secs(30)).await;

  // Stop manager
  manager.stop().await?;

  // Wait for all scanners to complete
  for handle in handles {
    let _ = handle.await;
  }

  println!("🏁 Market scanning completed");
  Ok(())
}

examples/performance/high_frequency.rs (line 44)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("⚡ High-Frequency Trading Example");
  println!("=================================");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    return Err("Please set KITE_API_KEY and KITE_ACCESS_TOKEN".to_string());
  }

  // Ultra-high performance configuration for HFT
  let config = KiteManagerConfig {
    max_connections: 3,
    max_symbols_per_connection: 1000, // Focused symbol set for HFT
    connection_buffer_size: 100000,   // Maximum possible buffer
    parser_buffer_size: 200000,       // Ultra-large parser buffer
    enable_dedicated_parsers: true,
    default_mode: Mode::Full, // Full market depth data
    ..Default::default()
  };

  println!("🚀 Optimized for:");
  println!("   Sub-microsecond latency");
  println!("   Maximum throughput");
  println!("   Real-time order book analysis");
  println!("   Ultra-low garbage collection");

  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Focus on highly liquid instruments for HFT
  let hft_symbols = vec![
    256265, // NIFTY 50 - highly liquid
    408065, // HDFC Bank - large volume
    738561, // Reliance - most traded
    884737, // ICICI Bank
    341249, // TCS
    492033, // ITC
    779521, // Kotak Bank
  ];

  println!("📊 Subscribing to {} liquid instruments", hft_symbols.len());
  manager
    .subscribe_symbols(&hft_symbols, Some(Mode::Full))
    .await?;

  // Setup high-frequency processing engines
  let channels = manager.get_all_channels();
  let mut hft_engines = Vec::new();

  for (channel_id, mut receiver) in channels {
    let engine = tokio::spawn(async move {
      let mut hft_processor = HFTProcessor::new(channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          hft_processor.process_ultra_fast(ticks).await;
        }
      }

      hft_processor.print_performance_metrics();
    });

    hft_engines.push(engine);
  }

  // Run HFT simulation
  println!("⚡ Starting high-frequency processing...");
  sleep(Duration::from_secs(60)).await;

  // Stop processing
  manager.stop().await?;

  // Wait for all engines to complete
  for engine in hft_engines {
    let _ = engine.await;
  }

  println!("🏁 High-frequency trading simulation completed");
  Ok(())
}

examples/simple_tick_test.rs (line 37)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("🔄 Simple Tick Test - Capturing Initial Ticks");
  println!("═══════════════════════════════════════════");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️  KITE_API_KEY and KITE_ACCESS_TOKEN environment variables not set"
    );
    return Ok(());
  }

  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 1, // Use only 1 connection for simplicity
    connection_buffer_size: 5000,
    parser_buffer_size: 10000,
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP,
  };

  let mut manager = KiteTickerManager::new(api_key, access_token, config);

  // Start manager
  println!("📡 Starting manager...");
  manager.start().await?;
  println!("✅ Manager started");

  // Get channel BEFORE subscribing
  println!("🎯 Getting channels...");
  let channels = manager.get_all_channels();
  let (channel_id, mut receiver) = channels.into_iter().next().unwrap();

  // Start listener in background
  let listener_handle = tokio::spawn(async move {
    println!("👂 Listener started for {:?}", channel_id);

    let mut tick_count = 0;
    loop {
      match timeout(Duration::from_secs(30), receiver.recv()).await {
        Ok(Ok(message)) => match message {
          TickerMessage::Ticks(ticks) => {
            tick_count += ticks.len();
            println!(
              "🎯 CAPTURED TICKS! {:?}: {} ticks (total: {})",
              channel_id,
              ticks.len(),
              tick_count
            );

            for tick in &ticks {
              println!("🔹 FULL TICK DEBUG:");
              println!("{:#?}", tick);
              println!("─────────────────────────────────────");
            }
          }
          TickerMessage::Error(err) => {
            println!("❌ Error: {}", err);
          }
          _ => {
            println!("📨 Other message: {:?}", message);
          }
        },
        Ok(Err(e)) => {
          println!("❌ Receive error: {}", e);
          break;
        }
        Err(_) => {
          println!("⏱️  Listener timeout");
          break;
        }
      }
    }
    println!("👂 Listener stopped");
  });

  // Give listener time to start
  tokio::time::sleep(Duration::from_millis(200)).await;

  // Now subscribe to a symbol
  println!("📊 Subscribing to symbol 256265...");
  manager
    .subscribe_symbols(&[128083204], Some(Mode::Full))
    .await?;
  println!("✅ Subscription sent");

  // Wait for ticks
  println!("⏳ Waiting 10 seconds for ticks...");
  tokio::time::sleep(Duration::from_secs(10)).await;

  // Stop
  manager.stop().await?;
  listener_handle.abort();

  println!("🏁 Test completed");
  Ok(())
}

examples/basic/portfolio_monitor.rs (line 61)

async fn main() -> Result<(), String> {
  // Initialize logging
  env_logger::init();

  println!("📊 Portfolio Monitor Example");
  println!("============================");

  // Get credentials
  let api_key = std::env::var("KITE_API_KEY")
    .map_err(|_| "Please set KITE_API_KEY environment variable")?;
  let access_token = std::env::var("KITE_ACCESS_TOKEN")
    .map_err(|_| "Please set KITE_ACCESS_TOKEN environment variable")?;

  // Define portfolio
  let portfolio = vec![
    (256265, "NIFTY 50"),
    (408065, "HDFC Bank"),
    (738561, "Reliance"),
    (5633, "TCS"),
    (884737, "Asian Paints"),
  ];

  println!("📈 Monitoring Portfolio:");
  for (symbol, name) in &portfolio {
    println!("   • {} ({})", name, symbol);
  }
  println!();

  // Create manager with optimized configuration for portfolio monitoring
  let config = KiteManagerConfig {
    max_connections: 1, // Single connection for small portfolio
    max_symbols_per_connection: 100,
    connection_buffer_size: 2000,
    parser_buffer_size: 5000,
    enable_dedicated_parsers: true,
    default_mode: Mode::Quote, // Quote mode for price + volume
    ..Default::default()
  };

  // Start manager
  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Subscribe to portfolio symbols
  let symbols: Vec<u32> = portfolio.iter().map(|(symbol, _)| *symbol).collect();
  manager
    .subscribe_symbols(&symbols, Some(Mode::Quote))
    .await?;

  println!("✅ Subscribed to {} symbols", symbols.len());

  // Create portfolio tracking
  let mut portfolio_data: HashMap<u32, StockInfo> = HashMap::new();
  for (symbol, name) in portfolio {
    portfolio_data.insert(
      symbol,
      StockInfo {
        symbol,
        name: name.to_string(),
        current_price: 0.0,
        volume: 0,
        last_update: Instant::now(),
      },
    );
  }

  // Get data channels
  let channels = manager.get_all_channels();

  // Start data processing
  for (channel_id, mut receiver) in channels {
    let mut portfolio_clone = portfolio_data.clone();

    tokio::spawn(async move {
      println!("📡 Started monitoring channel {:?}", channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          for tick in ticks {
            if let Some(stock) = portfolio_clone.get_mut(&tick.instrument_token)
            {
              // Update stock data
              if let Some(price) = tick.content.last_price {
                stock.current_price = price;
              }
              if let Some(volume) = tick.content.volume_traded {
                stock.volume = volume;
              }
              stock.last_update = Instant::now();

              // Display update
              println!(
                "📈 {} ({}): ₹{:.2} | Vol: {} | {}",
                stock.name,
                stock.symbol,
                stock.current_price,
                stock.volume,
                format_time_ago(stock.last_update)
              );
            }
          }
        }
      }
    });
  }

  // Periodic portfolio summary
  let summary_portfolio = portfolio_data.clone();
  tokio::spawn(async move {
    let mut interval = tokio::time::interval(Duration::from_secs(30));

    loop {
      interval.tick().await;
      print_portfolio_summary(&summary_portfolio);
    }
  });

  // Monitor for demo duration
  println!("📊 Monitoring portfolio for 2 minutes...\n");
  sleep(Duration::from_secs(120)).await;

  // Final summary
  println!("\n🏁 Demo completed!");
  print_portfolio_summary(&portfolio_data);

  // Stop manager
  manager.stop().await?;
  println!("✅ Portfolio monitor stopped");

  Ok(())
}

Additional examples can be found in:

• examples/dynamic_subscription_demo.rs
• examples/advanced/manager_demo.rs

pub async fn start(&mut self) -> Result<(), String>

Initialize all connections and start the manager

Examples found in repository

examples/basic/runtime_subscription_example.rs (line 36)

async fn main() -> Result<(), String> {
  // Initialize with your API credentials
  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️ Please set KITE_API_KEY and KITE_ACCESS_TOKEN environment variables"
    );
    return Err("Missing API credentials".to_string());
  }

  // Create configuration
  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 3,
    connection_buffer_size: 5000,
    parser_buffer_size: 10000,
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP,
  };

  // Start the manager
  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  println!("🚀 KiteTicker Manager started!");

  // Example of runtime subscription management
  runtime_subscription_demo(&mut manager).await?;

  // Stop the manager
  manager.stop().await?;
  Ok(())
}

examples/mode_change_test.rs (line 44)

async fn main() -> Result<(), String> {
  // Initialize logging
  env_logger::init();

  println!("🔄 KiteTicker Mode Change Test");
  println!("═══════════════════════════════");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️  KITE_API_KEY and KITE_ACCESS_TOKEN environment variables not set"
    );
    demonstrate_mode_change_issue().await;
    return Ok(());
  }

  // Create configuration
  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 3,
    connection_buffer_size: 5000,
    parser_buffer_size: 10000,
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP,
  };

  println!("🔧 Starting manager...");
  let mut manager = KiteTickerManager::new(api_key, access_token, config);

  manager.start().await?;
  println!("✅ Manager started successfully");

  // Test the mode change issue
  test_mode_change_issue(&mut manager).await?;

  // Stop the manager
  println!("\n🛑 Stopping manager...");
  manager.stop().await?;

  println!("🏁 Mode change test completed!");
  Ok(())
}

examples/performance/market_scanner.rs (line 37)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("🔍 Market Scanner Example");
  println!("=========================");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    return Err("Please set KITE_API_KEY and KITE_ACCESS_TOKEN".to_string());
  }

  // High-performance configuration for scanning
  let config = KiteManagerConfig {
    max_connections: 3,
    max_symbols_per_connection: 3000,
    connection_buffer_size: 20000, // Large buffer for high volume
    parser_buffer_size: 50000,     // Even larger for parsed data
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP, // LTP mode for scanning (minimal bandwidth)
    ..Default::default()
  };

  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Large symbol set for market scanning
  let large_symbol_set = generate_symbol_list(8000); // 8000 symbols across 3 connections

  println!(
    "📊 Scanning {} symbols across {} connections",
    large_symbol_set.len(),
    3
  );

  let start_time = Instant::now();
  manager
    .subscribe_symbols(&large_symbol_set, Some(Mode::LTP))
    .await?;
  println!(
    "✅ Subscribed to {} symbols in {:?}",
    large_symbol_set.len(),
    start_time.elapsed()
  );

  // Get all channels and start parallel processing
  let channels = manager.get_all_channels();
  let mut handles = Vec::new();

  for (channel_id, mut receiver) in channels {
    let handle = tokio::spawn(async move {
      let mut scanner = MarketScanner::new(channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          scanner.process_ticks(ticks).await;
        }
      }

      scanner.print_summary();
    });

    handles.push(handle);
  }

  // Run scanner for specified duration
  println!("🔄 Scanning market for 30 seconds...");
  sleep(Duration::from_secs(30)).await;

  // Stop manager
  manager.stop().await?;

  // Wait for all scanners to complete
  for handle in handles {
    let _ = handle.await;
  }

  println!("🏁 Market scanning completed");
  Ok(())
}

examples/performance/high_frequency.rs (line 45)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("⚡ High-Frequency Trading Example");
  println!("=================================");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    return Err("Please set KITE_API_KEY and KITE_ACCESS_TOKEN".to_string());
  }

  // Ultra-high performance configuration for HFT
  let config = KiteManagerConfig {
    max_connections: 3,
    max_symbols_per_connection: 1000, // Focused symbol set for HFT
    connection_buffer_size: 100000,   // Maximum possible buffer
    parser_buffer_size: 200000,       // Ultra-large parser buffer
    enable_dedicated_parsers: true,
    default_mode: Mode::Full, // Full market depth data
    ..Default::default()
  };

  println!("🚀 Optimized for:");
  println!("   Sub-microsecond latency");
  println!("   Maximum throughput");
  println!("   Real-time order book analysis");
  println!("   Ultra-low garbage collection");

  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Focus on highly liquid instruments for HFT
  let hft_symbols = vec![
    256265, // NIFTY 50 - highly liquid
    408065, // HDFC Bank - large volume
    738561, // Reliance - most traded
    884737, // ICICI Bank
    341249, // TCS
    492033, // ITC
    779521, // Kotak Bank
  ];

  println!("📊 Subscribing to {} liquid instruments", hft_symbols.len());
  manager
    .subscribe_symbols(&hft_symbols, Some(Mode::Full))
    .await?;

  // Setup high-frequency processing engines
  let channels = manager.get_all_channels();
  let mut hft_engines = Vec::new();

  for (channel_id, mut receiver) in channels {
    let engine = tokio::spawn(async move {
      let mut hft_processor = HFTProcessor::new(channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          hft_processor.process_ultra_fast(ticks).await;
        }
      }

      hft_processor.print_performance_metrics();
    });

    hft_engines.push(engine);
  }

  // Run HFT simulation
  println!("⚡ Starting high-frequency processing...");
  sleep(Duration::from_secs(60)).await;

  // Stop processing
  manager.stop().await?;

  // Wait for all engines to complete
  for engine in hft_engines {
    let _ = engine.await;
  }

  println!("🏁 High-frequency trading simulation completed");
  Ok(())
}

examples/simple_tick_test.rs (line 41)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("🔄 Simple Tick Test - Capturing Initial Ticks");
  println!("═══════════════════════════════════════════");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️  KITE_API_KEY and KITE_ACCESS_TOKEN environment variables not set"
    );
    return Ok(());
  }

  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 1, // Use only 1 connection for simplicity
    connection_buffer_size: 5000,
    parser_buffer_size: 10000,
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP,
  };

  let mut manager = KiteTickerManager::new(api_key, access_token, config);

  // Start manager
  println!("📡 Starting manager...");
  manager.start().await?;
  println!("✅ Manager started");

  // Get channel BEFORE subscribing
  println!("🎯 Getting channels...");
  let channels = manager.get_all_channels();
  let (channel_id, mut receiver) = channels.into_iter().next().unwrap();

  // Start listener in background
  let listener_handle = tokio::spawn(async move {
    println!("👂 Listener started for {:?}", channel_id);

    let mut tick_count = 0;
    loop {
      match timeout(Duration::from_secs(30), receiver.recv()).await {
        Ok(Ok(message)) => match message {
          TickerMessage::Ticks(ticks) => {
            tick_count += ticks.len();
            println!(
              "🎯 CAPTURED TICKS! {:?}: {} ticks (total: {})",
              channel_id,
              ticks.len(),
              tick_count
            );

            for tick in &ticks {
              println!("🔹 FULL TICK DEBUG:");
              println!("{:#?}", tick);
              println!("─────────────────────────────────────");
            }
          }
          TickerMessage::Error(err) => {
            println!("❌ Error: {}", err);
          }
          _ => {
            println!("📨 Other message: {:?}", message);
          }
        },
        Ok(Err(e)) => {
          println!("❌ Receive error: {}", e);
          break;
        }
        Err(_) => {
          println!("⏱️  Listener timeout");
          break;
        }
      }
    }
    println!("👂 Listener stopped");
  });

  // Give listener time to start
  tokio::time::sleep(Duration::from_millis(200)).await;

  // Now subscribe to a symbol
  println!("📊 Subscribing to symbol 256265...");
  manager
    .subscribe_symbols(&[128083204], Some(Mode::Full))
    .await?;
  println!("✅ Subscription sent");

  // Wait for ticks
  println!("⏳ Waiting 10 seconds for ticks...");
  tokio::time::sleep(Duration::from_secs(10)).await;

  // Stop
  manager.stop().await?;
  listener_handle.abort();

  println!("🏁 Test completed");
  Ok(())
}

examples/basic/portfolio_monitor.rs (line 62)

async fn main() -> Result<(), String> {
  // Initialize logging
  env_logger::init();

  println!("📊 Portfolio Monitor Example");
  println!("============================");

  // Get credentials
  let api_key = std::env::var("KITE_API_KEY")
    .map_err(|_| "Please set KITE_API_KEY environment variable")?;
  let access_token = std::env::var("KITE_ACCESS_TOKEN")
    .map_err(|_| "Please set KITE_ACCESS_TOKEN environment variable")?;

  // Define portfolio
  let portfolio = vec![
    (256265, "NIFTY 50"),
    (408065, "HDFC Bank"),
    (738561, "Reliance"),
    (5633, "TCS"),
    (884737, "Asian Paints"),
  ];

  println!("📈 Monitoring Portfolio:");
  for (symbol, name) in &portfolio {
    println!("   • {} ({})", name, symbol);
  }
  println!();

  // Create manager with optimized configuration for portfolio monitoring
  let config = KiteManagerConfig {
    max_connections: 1, // Single connection for small portfolio
    max_symbols_per_connection: 100,
    connection_buffer_size: 2000,
    parser_buffer_size: 5000,
    enable_dedicated_parsers: true,
    default_mode: Mode::Quote, // Quote mode for price + volume
    ..Default::default()
  };

  // Start manager
  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Subscribe to portfolio symbols
  let symbols: Vec<u32> = portfolio.iter().map(|(symbol, _)| *symbol).collect();
  manager
    .subscribe_symbols(&symbols, Some(Mode::Quote))
    .await?;

  println!("✅ Subscribed to {} symbols", symbols.len());

  // Create portfolio tracking
  let mut portfolio_data: HashMap<u32, StockInfo> = HashMap::new();
  for (symbol, name) in portfolio {
    portfolio_data.insert(
      symbol,
      StockInfo {
        symbol,
        name: name.to_string(),
        current_price: 0.0,
        volume: 0,
        last_update: Instant::now(),
      },
    );
  }

  // Get data channels
  let channels = manager.get_all_channels();

  // Start data processing
  for (channel_id, mut receiver) in channels {
    let mut portfolio_clone = portfolio_data.clone();

    tokio::spawn(async move {
      println!("📡 Started monitoring channel {:?}", channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          for tick in ticks {
            if let Some(stock) = portfolio_clone.get_mut(&tick.instrument_token)
            {
              // Update stock data
              if let Some(price) = tick.content.last_price {
                stock.current_price = price;
              }
              if let Some(volume) = tick.content.volume_traded {
                stock.volume = volume;
              }
              stock.last_update = Instant::now();

              // Display update
              println!(
                "📈 {} ({}): ₹{:.2} | Vol: {} | {}",
                stock.name,
                stock.symbol,
                stock.current_price,
                stock.volume,
                format_time_ago(stock.last_update)
              );
            }
          }
        }
      }
    });
  }

  // Periodic portfolio summary
  let summary_portfolio = portfolio_data.clone();
  tokio::spawn(async move {
    let mut interval = tokio::time::interval(Duration::from_secs(30));

    loop {
      interval.tick().await;
      print_portfolio_summary(&summary_portfolio);
    }
  });

  // Monitor for demo duration
  println!("📊 Monitoring portfolio for 2 minutes...\n");
  sleep(Duration::from_secs(120)).await;

  // Final summary
  println!("\n🏁 Demo completed!");
  print_portfolio_summary(&portfolio_data);

  // Stop manager
  manager.stop().await?;
  println!("✅ Portfolio monitor stopped");

  Ok(())
}

Additional examples can be found in:

• examples/dynamic_subscription_demo.rs
• examples/advanced/manager_demo.rs

pub async fn subscribe_symbols( &mut self, symbols: &[u32], mode: Option<Mode>, ) -> Result<(), String>

Subscribe to symbols using round-robin distribution

Examples found in repository

examples/performance/market_scanner.rs (line 50)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("🔍 Market Scanner Example");
  println!("=========================");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    return Err("Please set KITE_API_KEY and KITE_ACCESS_TOKEN".to_string());
  }

  // High-performance configuration for scanning
  let config = KiteManagerConfig {
    max_connections: 3,
    max_symbols_per_connection: 3000,
    connection_buffer_size: 20000, // Large buffer for high volume
    parser_buffer_size: 50000,     // Even larger for parsed data
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP, // LTP mode for scanning (minimal bandwidth)
    ..Default::default()
  };

  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Large symbol set for market scanning
  let large_symbol_set = generate_symbol_list(8000); // 8000 symbols across 3 connections

  println!(
    "📊 Scanning {} symbols across {} connections",
    large_symbol_set.len(),
    3
  );

  let start_time = Instant::now();
  manager
    .subscribe_symbols(&large_symbol_set, Some(Mode::LTP))
    .await?;
  println!(
    "✅ Subscribed to {} symbols in {:?}",
    large_symbol_set.len(),
    start_time.elapsed()
  );

  // Get all channels and start parallel processing
  let channels = manager.get_all_channels();
  let mut handles = Vec::new();

  for (channel_id, mut receiver) in channels {
    let handle = tokio::spawn(async move {
      let mut scanner = MarketScanner::new(channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          scanner.process_ticks(ticks).await;
        }
      }

      scanner.print_summary();
    });

    handles.push(handle);
  }

  // Run scanner for specified duration
  println!("🔄 Scanning market for 30 seconds...");
  sleep(Duration::from_secs(30)).await;

  // Stop manager
  manager.stop().await?;

  // Wait for all scanners to complete
  for handle in handles {
    let _ = handle.await;
  }

  println!("🏁 Market scanning completed");
  Ok(())
}

examples/basic/runtime_subscription_example.rs (line 58)

async fn runtime_subscription_demo(
  manager: &mut KiteTickerManager,
) -> Result<(), String> {
  println!("\n📡 Runtime Subscription Management Demo");
  println!("=====================================");

  // 1. Start with initial symbols
  println!("\n1️⃣ Initial subscription to base symbols");
  let initial_symbols = vec![256265, 265, 256777];
  manager
    .subscribe_symbols(&initial_symbols, Some(Mode::LTP))
    .await?;
  print_current_state(manager, "Initial subscription").await;

  sleep(Duration::from_secs(2)).await;

  // 2. Add more symbols dynamically
  println!("\n2️⃣ Adding symbols dynamically");
  let additional_symbols = vec![274441, 260105, 273929];
  manager
    .subscribe_symbols(&additional_symbols, Some(Mode::Quote))
    .await?;
  print_current_state(manager, "After adding symbols").await;

  sleep(Duration::from_secs(2)).await;

  // 3. Change mode for existing symbols
  println!("\n3️⃣ Changing subscription mode");
  let symbols_for_mode_change = vec![256265, 265];
  manager
    .change_mode(&symbols_for_mode_change, Mode::Full)
    .await?;
  print_current_state(manager, "After mode change").await;

  sleep(Duration::from_secs(2)).await;

  // 4. Remove some symbols
  println!("\n4️⃣ Removing symbols dynamically");
  let symbols_to_remove = vec![265, 274441];
  manager.unsubscribe_symbols(&symbols_to_remove).await?;
  print_current_state(manager, "After removing symbols").await;

  sleep(Duration::from_secs(2)).await;

  // 5. Add different symbols with different modes
  println!("\n5️⃣ Adding new symbols with Full mode");
  let full_mode_symbols = vec![257801, 258825];
  manager
    .subscribe_symbols(&full_mode_symbols, Some(Mode::Full))
    .await?;
  print_current_state(manager, "After adding Full mode symbols").await;

  // 6. Monitor live data for a short period
  println!("\n6️⃣ Monitoring live data (5 seconds)");
  monitor_live_data(manager, 5).await?;

  // 7. Complete cleanup
  println!("\n7️⃣ Complete cleanup");
  let all_symbols: Vec<u32> = manager
    .get_symbol_distribution()
    .values()
    .flat_map(|symbols| symbols.iter().cloned())
    .collect();

  if !all_symbols.is_empty() {
    manager.unsubscribe_symbols(&all_symbols).await?;
    print_current_state(manager, "After complete cleanup").await;
  }

  println!("\n✅ Runtime subscription demo completed!");
  Ok(())
}

examples/performance/high_frequency.rs (line 60)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("⚡ High-Frequency Trading Example");
  println!("=================================");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    return Err("Please set KITE_API_KEY and KITE_ACCESS_TOKEN".to_string());
  }

  // Ultra-high performance configuration for HFT
  let config = KiteManagerConfig {
    max_connections: 3,
    max_symbols_per_connection: 1000, // Focused symbol set for HFT
    connection_buffer_size: 100000,   // Maximum possible buffer
    parser_buffer_size: 200000,       // Ultra-large parser buffer
    enable_dedicated_parsers: true,
    default_mode: Mode::Full, // Full market depth data
    ..Default::default()
  };

  println!("🚀 Optimized for:");
  println!("   Sub-microsecond latency");
  println!("   Maximum throughput");
  println!("   Real-time order book analysis");
  println!("   Ultra-low garbage collection");

  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Focus on highly liquid instruments for HFT
  let hft_symbols = vec![
    256265, // NIFTY 50 - highly liquid
    408065, // HDFC Bank - large volume
    738561, // Reliance - most traded
    884737, // ICICI Bank
    341249, // TCS
    492033, // ITC
    779521, // Kotak Bank
  ];

  println!("📊 Subscribing to {} liquid instruments", hft_symbols.len());
  manager
    .subscribe_symbols(&hft_symbols, Some(Mode::Full))
    .await?;

  // Setup high-frequency processing engines
  let channels = manager.get_all_channels();
  let mut hft_engines = Vec::new();

  for (channel_id, mut receiver) in channels {
    let engine = tokio::spawn(async move {
      let mut hft_processor = HFTProcessor::new(channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          hft_processor.process_ultra_fast(ticks).await;
        }
      }

      hft_processor.print_performance_metrics();
    });

    hft_engines.push(engine);
  }

  // Run HFT simulation
  println!("⚡ Starting high-frequency processing...");
  sleep(Duration::from_secs(60)).await;

  // Stop processing
  manager.stop().await?;

  // Wait for all engines to complete
  for engine in hft_engines {
    let _ = engine.await;
  }

  println!("🏁 High-frequency trading simulation completed");
  Ok(())
}

examples/simple_tick_test.rs (line 98)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("🔄 Simple Tick Test - Capturing Initial Ticks");
  println!("═══════════════════════════════════════════");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️  KITE_API_KEY and KITE_ACCESS_TOKEN environment variables not set"
    );
    return Ok(());
  }

  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 1, // Use only 1 connection for simplicity
    connection_buffer_size: 5000,
    parser_buffer_size: 10000,
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP,
  };

  let mut manager = KiteTickerManager::new(api_key, access_token, config);

  // Start manager
  println!("📡 Starting manager...");
  manager.start().await?;
  println!("✅ Manager started");

  // Get channel BEFORE subscribing
  println!("🎯 Getting channels...");
  let channels = manager.get_all_channels();
  let (channel_id, mut receiver) = channels.into_iter().next().unwrap();

  // Start listener in background
  let listener_handle = tokio::spawn(async move {
    println!("👂 Listener started for {:?}", channel_id);

    let mut tick_count = 0;
    loop {
      match timeout(Duration::from_secs(30), receiver.recv()).await {
        Ok(Ok(message)) => match message {
          TickerMessage::Ticks(ticks) => {
            tick_count += ticks.len();
            println!(
              "🎯 CAPTURED TICKS! {:?}: {} ticks (total: {})",
              channel_id,
              ticks.len(),
              tick_count
            );

            for tick in &ticks {
              println!("🔹 FULL TICK DEBUG:");
              println!("{:#?}", tick);
              println!("─────────────────────────────────────");
            }
          }
          TickerMessage::Error(err) => {
            println!("❌ Error: {}", err);
          }
          _ => {
            println!("📨 Other message: {:?}", message);
          }
        },
        Ok(Err(e)) => {
          println!("❌ Receive error: {}", e);
          break;
        }
        Err(_) => {
          println!("⏱️  Listener timeout");
          break;
        }
      }
    }
    println!("👂 Listener stopped");
  });

  // Give listener time to start
  tokio::time::sleep(Duration::from_millis(200)).await;

  // Now subscribe to a symbol
  println!("📊 Subscribing to symbol 256265...");
  manager
    .subscribe_symbols(&[128083204], Some(Mode::Full))
    .await?;
  println!("✅ Subscription sent");

  // Wait for ticks
  println!("⏳ Waiting 10 seconds for ticks...");
  tokio::time::sleep(Duration::from_secs(10)).await;

  // Stop
  manager.stop().await?;
  listener_handle.abort();

  println!("🏁 Test completed");
  Ok(())
}

examples/basic/portfolio_monitor.rs (line 67)

async fn main() -> Result<(), String> {
  // Initialize logging
  env_logger::init();

  println!("📊 Portfolio Monitor Example");
  println!("============================");

  // Get credentials
  let api_key = std::env::var("KITE_API_KEY")
    .map_err(|_| "Please set KITE_API_KEY environment variable")?;
  let access_token = std::env::var("KITE_ACCESS_TOKEN")
    .map_err(|_| "Please set KITE_ACCESS_TOKEN environment variable")?;

  // Define portfolio
  let portfolio = vec![
    (256265, "NIFTY 50"),
    (408065, "HDFC Bank"),
    (738561, "Reliance"),
    (5633, "TCS"),
    (884737, "Asian Paints"),
  ];

  println!("📈 Monitoring Portfolio:");
  for (symbol, name) in &portfolio {
    println!("   • {} ({})", name, symbol);
  }
  println!();

  // Create manager with optimized configuration for portfolio monitoring
  let config = KiteManagerConfig {
    max_connections: 1, // Single connection for small portfolio
    max_symbols_per_connection: 100,
    connection_buffer_size: 2000,
    parser_buffer_size: 5000,
    enable_dedicated_parsers: true,
    default_mode: Mode::Quote, // Quote mode for price + volume
    ..Default::default()
  };

  // Start manager
  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Subscribe to portfolio symbols
  let symbols: Vec<u32> = portfolio.iter().map(|(symbol, _)| *symbol).collect();
  manager
    .subscribe_symbols(&symbols, Some(Mode::Quote))
    .await?;

  println!("✅ Subscribed to {} symbols", symbols.len());

  // Create portfolio tracking
  let mut portfolio_data: HashMap<u32, StockInfo> = HashMap::new();
  for (symbol, name) in portfolio {
    portfolio_data.insert(
      symbol,
      StockInfo {
        symbol,
        name: name.to_string(),
        current_price: 0.0,
        volume: 0,
        last_update: Instant::now(),
      },
    );
  }

  // Get data channels
  let channels = manager.get_all_channels();

  // Start data processing
  for (channel_id, mut receiver) in channels {
    let mut portfolio_clone = portfolio_data.clone();

    tokio::spawn(async move {
      println!("📡 Started monitoring channel {:?}", channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          for tick in ticks {
            if let Some(stock) = portfolio_clone.get_mut(&tick.instrument_token)
            {
              // Update stock data
              if let Some(price) = tick.content.last_price {
                stock.current_price = price;
              }
              if let Some(volume) = tick.content.volume_traded {
                stock.volume = volume;
              }
              stock.last_update = Instant::now();

              // Display update
              println!(
                "📈 {} ({}): ₹{:.2} | Vol: {} | {}",
                stock.name,
                stock.symbol,
                stock.current_price,
                stock.volume,
                format_time_ago(stock.last_update)
              );
            }
          }
        }
      }
    });
  }

  // Periodic portfolio summary
  let summary_portfolio = portfolio_data.clone();
  tokio::spawn(async move {
    let mut interval = tokio::time::interval(Duration::from_secs(30));

    loop {
      interval.tick().await;
      print_portfolio_summary(&summary_portfolio);
    }
  });

  // Monitor for demo duration
  println!("📊 Monitoring portfolio for 2 minutes...\n");
  sleep(Duration::from_secs(120)).await;

  // Final summary
  println!("\n🏁 Demo completed!");
  print_portfolio_summary(&portfolio_data);

  // Stop manager
  manager.stop().await?;
  println!("✅ Portfolio monitor stopped");

  Ok(())
}

examples/mode_change_test.rs (line 68)

async fn test_mode_change_issue(
  manager: &mut KiteTickerManager,
) -> Result<(), String> {
  println!("\n🧪 Testing Mode Change Issue");
  println!("=============================");

  // Step 1: Subscribe to a symbol with LTP mode
  let test_symbol = 738561; // Test symbol provided by user
  println!("\n1️⃣ Subscribing to symbol {} with LTP mode", test_symbol);
  manager
    .subscribe_symbols(&[test_symbol], Some(Mode::LTP))
    .await?;

  // Start monitoring ticks to see the mode
  let channels = manager.get_all_channels();
  let mut tick_listeners = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut ticks_received = 0;
      let start = Instant::now();

      while start.elapsed() < Duration::from_secs(5) && ticks_received < 3 {
        match timeout(Duration::from_millis(500), receiver.recv()).await {
          Ok(Ok(message)) => {
            if let TickerMessage::Ticks(ticks) = message {
              for tick in &ticks {
                if tick.instrument_token == test_symbol {
                  ticks_received += 1;
                  println!(
                    "   📊 Received tick for {}: Mode={:?}, LTP={:?}",
                    tick.instrument_token,
                    tick.content.mode,
                    tick.content.last_price
                  );

                  // Check if OHLC data is present (should not be for LTP mode)
                  if tick.content.ohlc.is_some() {
                    println!(
                      "   ⚠️  OHLC data present in LTP mode - unexpected!"
                    );
                  } else {
                    println!("   ✅ No OHLC data in LTP mode - correct");
                  }
                }
              }
            }
          }
          _ => continue,
        }
      }
      (channel_id, ticks_received)
    });
    tick_listeners.push(task);
  }

  // Wait for initial ticks
  for task in tick_listeners {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "   📈 {:?}: Received {} ticks for initial LTP subscription",
        channel_id, count
      );
    }
  }

  sleep(Duration::from_secs(2)).await;

  // Step 2: Attempt to change mode to Full (this is where the issue should manifest)
  println!(
    "\n2️⃣ Attempting to change mode from LTP to Full for symbol {}",
    test_symbol
  );
  match manager.change_mode(&[test_symbol], Mode::Full).await {
    Ok(()) => {
      println!("   ✅ Mode change command sent successfully");
    }
    Err(e) => {
      println!("   ❌ Mode change command failed: {}", e);
      return Err(e);
    }
  }

  // Step 3: Monitor ticks after mode change to see if it actually worked
  println!("\n3️⃣ Monitoring ticks after mode change...");
  let channels = manager.get_all_channels();
  let mut post_change_listeners = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut ticks_received = 0;
      let mut ohlc_present = 0;
      let mut depth_present = 0;
      let start = Instant::now();

      while start.elapsed() < Duration::from_secs(10) && ticks_received < 5 {
        match timeout(Duration::from_millis(500), receiver.recv()).await {
          Ok(Ok(message)) => {
            if let TickerMessage::Ticks(ticks) = message {
              for tick in &ticks {
                if tick.instrument_token == test_symbol {
                  ticks_received += 1;
                  println!(
                    "   📊 Post-change tick for {}: Mode={:?}, LTP={:?}",
                    tick.instrument_token,
                    tick.content.mode,
                    tick.content.last_price
                  );

                  // Check if Full mode data is present
                  if let Some(ohlc) = &tick.content.ohlc {
                    ohlc_present += 1;
                    println!(
                      "   ✅ OHLC data present: O:{} H:{} L:{} C:{}",
                      ohlc.open, ohlc.high, ohlc.low, ohlc.close
                    );
                  } else {
                    println!("   ❌ OHLC data missing - mode change may not have worked!");
                  }

                  if let Some(depth) = &tick.content.depth {
                    depth_present += 1;
                    println!(
                      "   ✅ Market depth present: {} buy, {} sell orders",
                      depth.buy.len(),
                      depth.sell.len()
                    );
                  } else {
                    println!("   ❌ Market depth missing - mode change may not have worked!");
                  }

                  // Log the actual mode reported in the tick
                  info!("Tick mode reported: {:?}", tick.content.mode);
                }
              }
            }
          }
          _ => continue,
        }
      }
      (channel_id, ticks_received, ohlc_present, depth_present)
    });
    post_change_listeners.push(task);
  }

  // Wait for post-change ticks
  let mut total_ticks = 0;
  let mut total_ohlc = 0;
  let mut total_depth = 0;

  for task in post_change_listeners {
    if let Ok((channel_id, ticks, ohlc, depth)) = task.await {
      println!(
        "   📈 {:?}: {} ticks, {} with OHLC, {} with depth",
        channel_id, ticks, ohlc, depth
      );
      total_ticks += ticks;
      total_ohlc += ohlc;
      total_depth += depth;
    }
  }

  // Step 4: Analyze results
  println!("\n4️⃣ Test Results Analysis");
  println!("========================");

  if total_ticks == 0 {
    println!("   ⚠️  No ticks received after mode change - connection issue?");
  } else if total_ohlc == 0 && total_depth == 0 {
    println!("   ❌ ISSUE CONFIRMED: Mode change command sent but Full mode data not received");
    println!("   💡 This confirms that set_mode() alone doesn't work - need unsubscribe+resubscribe");
  } else if total_ohlc > 0 && total_depth > 0 {
    println!("   ✅ Mode change worked - Full mode data received");
  } else {
    println!(
      "   ⚠️  Partial mode change - some Full mode data received but not all"
    );
  }

  println!("\n📊 Summary:");
  println!("   Total ticks received: {}", total_ticks);
  println!("   Ticks with OHLC data: {}", total_ohlc);
  println!("   Ticks with market depth: {}", total_depth);

  Ok(())
}

Additional examples can be found in:

• examples/advanced/manager_demo.rs
• examples/dynamic_subscription_demo.rs

pub fn get_channel( &mut self, channel_id: ChannelId, ) -> Option<Receiver<TickerMessage>>

Get output channel for a specific connection

pub fn get_all_channels(&mut self) -> Vec<(ChannelId, Receiver<TickerMessage>)>

Get all output channels

Examples found in repository

examples/basic/runtime_subscription_example.rs (line 146)

async fn monitor_live_data(
  manager: &mut KiteTickerManager,
  seconds: u64,
) -> Result<(), String> {
  let channels = manager.get_all_channels();
  let mut tasks = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut count = 0;
      let start = std::time::Instant::now();

      while start.elapsed() < Duration::from_secs(seconds) {
        match tokio::time::timeout(Duration::from_millis(100), receiver.recv())
          .await
        {
          Ok(Ok(message)) => {
            count += 1;
            if let TickerMessage::Ticks(ticks) = message {
              if count <= 2 {
                // Show first few ticks
                println!(
                  "   📋 {:?}: Received {} ticks",
                  channel_id,
                  ticks.len()
                );
              }
            }
          }
          _ => continue,
        }
      }
      (channel_id, count)
    });
    tasks.push(task);
  }

  // Wait for monitoring to complete
  for task in tasks {
    if let Ok((channel_id, count)) = task.await {
      println!("   📊 {:?}: {} total messages", channel_id, count);
    }
  }

  Ok(())
}

examples/performance/market_scanner.rs (line 59)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("🔍 Market Scanner Example");
  println!("=========================");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    return Err("Please set KITE_API_KEY and KITE_ACCESS_TOKEN".to_string());
  }

  // High-performance configuration for scanning
  let config = KiteManagerConfig {
    max_connections: 3,
    max_symbols_per_connection: 3000,
    connection_buffer_size: 20000, // Large buffer for high volume
    parser_buffer_size: 50000,     // Even larger for parsed data
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP, // LTP mode for scanning (minimal bandwidth)
    ..Default::default()
  };

  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Large symbol set for market scanning
  let large_symbol_set = generate_symbol_list(8000); // 8000 symbols across 3 connections

  println!(
    "📊 Scanning {} symbols across {} connections",
    large_symbol_set.len(),
    3
  );

  let start_time = Instant::now();
  manager
    .subscribe_symbols(&large_symbol_set, Some(Mode::LTP))
    .await?;
  println!(
    "✅ Subscribed to {} symbols in {:?}",
    large_symbol_set.len(),
    start_time.elapsed()
  );

  // Get all channels and start parallel processing
  let channels = manager.get_all_channels();
  let mut handles = Vec::new();

  for (channel_id, mut receiver) in channels {
    let handle = tokio::spawn(async move {
      let mut scanner = MarketScanner::new(channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          scanner.process_ticks(ticks).await;
        }
      }

      scanner.print_summary();
    });

    handles.push(handle);
  }

  // Run scanner for specified duration
  println!("🔄 Scanning market for 30 seconds...");
  sleep(Duration::from_secs(30)).await;

  // Stop manager
  manager.stop().await?;

  // Wait for all scanners to complete
  for handle in handles {
    let _ = handle.await;
  }

  println!("🏁 Market scanning completed");
  Ok(())
}

examples/performance/high_frequency.rs (line 64)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("⚡ High-Frequency Trading Example");
  println!("=================================");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    return Err("Please set KITE_API_KEY and KITE_ACCESS_TOKEN".to_string());
  }

  // Ultra-high performance configuration for HFT
  let config = KiteManagerConfig {
    max_connections: 3,
    max_symbols_per_connection: 1000, // Focused symbol set for HFT
    connection_buffer_size: 100000,   // Maximum possible buffer
    parser_buffer_size: 200000,       // Ultra-large parser buffer
    enable_dedicated_parsers: true,
    default_mode: Mode::Full, // Full market depth data
    ..Default::default()
  };

  println!("🚀 Optimized for:");
  println!("   Sub-microsecond latency");
  println!("   Maximum throughput");
  println!("   Real-time order book analysis");
  println!("   Ultra-low garbage collection");

  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Focus on highly liquid instruments for HFT
  let hft_symbols = vec![
    256265, // NIFTY 50 - highly liquid
    408065, // HDFC Bank - large volume
    738561, // Reliance - most traded
    884737, // ICICI Bank
    341249, // TCS
    492033, // ITC
    779521, // Kotak Bank
  ];

  println!("📊 Subscribing to {} liquid instruments", hft_symbols.len());
  manager
    .subscribe_symbols(&hft_symbols, Some(Mode::Full))
    .await?;

  // Setup high-frequency processing engines
  let channels = manager.get_all_channels();
  let mut hft_engines = Vec::new();

  for (channel_id, mut receiver) in channels {
    let engine = tokio::spawn(async move {
      let mut hft_processor = HFTProcessor::new(channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          hft_processor.process_ultra_fast(ticks).await;
        }
      }

      hft_processor.print_performance_metrics();
    });

    hft_engines.push(engine);
  }

  // Run HFT simulation
  println!("⚡ Starting high-frequency processing...");
  sleep(Duration::from_secs(60)).await;

  // Stop processing
  manager.stop().await?;

  // Wait for all engines to complete
  for engine in hft_engines {
    let _ = engine.await;
  }

  println!("🏁 High-frequency trading simulation completed");
  Ok(())
}

examples/simple_tick_test.rs (line 46)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("🔄 Simple Tick Test - Capturing Initial Ticks");
  println!("═══════════════════════════════════════════");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️  KITE_API_KEY and KITE_ACCESS_TOKEN environment variables not set"
    );
    return Ok(());
  }

  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 1, // Use only 1 connection for simplicity
    connection_buffer_size: 5000,
    parser_buffer_size: 10000,
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP,
  };

  let mut manager = KiteTickerManager::new(api_key, access_token, config);

  // Start manager
  println!("📡 Starting manager...");
  manager.start().await?;
  println!("✅ Manager started");

  // Get channel BEFORE subscribing
  println!("🎯 Getting channels...");
  let channels = manager.get_all_channels();
  let (channel_id, mut receiver) = channels.into_iter().next().unwrap();

  // Start listener in background
  let listener_handle = tokio::spawn(async move {
    println!("👂 Listener started for {:?}", channel_id);

    let mut tick_count = 0;
    loop {
      match timeout(Duration::from_secs(30), receiver.recv()).await {
        Ok(Ok(message)) => match message {
          TickerMessage::Ticks(ticks) => {
            tick_count += ticks.len();
            println!(
              "🎯 CAPTURED TICKS! {:?}: {} ticks (total: {})",
              channel_id,
              ticks.len(),
              tick_count
            );

            for tick in &ticks {
              println!("🔹 FULL TICK DEBUG:");
              println!("{:#?}", tick);
              println!("─────────────────────────────────────");
            }
          }
          TickerMessage::Error(err) => {
            println!("❌ Error: {}", err);
          }
          _ => {
            println!("📨 Other message: {:?}", message);
          }
        },
        Ok(Err(e)) => {
          println!("❌ Receive error: {}", e);
          break;
        }
        Err(_) => {
          println!("⏱️  Listener timeout");
          break;
        }
      }
    }
    println!("👂 Listener stopped");
  });

  // Give listener time to start
  tokio::time::sleep(Duration::from_millis(200)).await;

  // Now subscribe to a symbol
  println!("📊 Subscribing to symbol 256265...");
  manager
    .subscribe_symbols(&[128083204], Some(Mode::Full))
    .await?;
  println!("✅ Subscription sent");

  // Wait for ticks
  println!("⏳ Waiting 10 seconds for ticks...");
  tokio::time::sleep(Duration::from_secs(10)).await;

  // Stop
  manager.stop().await?;
  listener_handle.abort();

  println!("🏁 Test completed");
  Ok(())
}

examples/basic/portfolio_monitor.rs (line 88)

async fn main() -> Result<(), String> {
  // Initialize logging
  env_logger::init();

  println!("📊 Portfolio Monitor Example");
  println!("============================");

  // Get credentials
  let api_key = std::env::var("KITE_API_KEY")
    .map_err(|_| "Please set KITE_API_KEY environment variable")?;
  let access_token = std::env::var("KITE_ACCESS_TOKEN")
    .map_err(|_| "Please set KITE_ACCESS_TOKEN environment variable")?;

  // Define portfolio
  let portfolio = vec![
    (256265, "NIFTY 50"),
    (408065, "HDFC Bank"),
    (738561, "Reliance"),
    (5633, "TCS"),
    (884737, "Asian Paints"),
  ];

  println!("📈 Monitoring Portfolio:");
  for (symbol, name) in &portfolio {
    println!("   • {} ({})", name, symbol);
  }
  println!();

  // Create manager with optimized configuration for portfolio monitoring
  let config = KiteManagerConfig {
    max_connections: 1, // Single connection for small portfolio
    max_symbols_per_connection: 100,
    connection_buffer_size: 2000,
    parser_buffer_size: 5000,
    enable_dedicated_parsers: true,
    default_mode: Mode::Quote, // Quote mode for price + volume
    ..Default::default()
  };

  // Start manager
  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Subscribe to portfolio symbols
  let symbols: Vec<u32> = portfolio.iter().map(|(symbol, _)| *symbol).collect();
  manager
    .subscribe_symbols(&symbols, Some(Mode::Quote))
    .await?;

  println!("✅ Subscribed to {} symbols", symbols.len());

  // Create portfolio tracking
  let mut portfolio_data: HashMap<u32, StockInfo> = HashMap::new();
  for (symbol, name) in portfolio {
    portfolio_data.insert(
      symbol,
      StockInfo {
        symbol,
        name: name.to_string(),
        current_price: 0.0,
        volume: 0,
        last_update: Instant::now(),
      },
    );
  }

  // Get data channels
  let channels = manager.get_all_channels();

  // Start data processing
  for (channel_id, mut receiver) in channels {
    let mut portfolio_clone = portfolio_data.clone();

    tokio::spawn(async move {
      println!("📡 Started monitoring channel {:?}", channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          for tick in ticks {
            if let Some(stock) = portfolio_clone.get_mut(&tick.instrument_token)
            {
              // Update stock data
              if let Some(price) = tick.content.last_price {
                stock.current_price = price;
              }
              if let Some(volume) = tick.content.volume_traded {
                stock.volume = volume;
              }
              stock.last_update = Instant::now();

              // Display update
              println!(
                "📈 {} ({}): ₹{:.2} | Vol: {} | {}",
                stock.name,
                stock.symbol,
                stock.current_price,
                stock.volume,
                format_time_ago(stock.last_update)
              );
            }
          }
        }
      }
    });
  }

  // Periodic portfolio summary
  let summary_portfolio = portfolio_data.clone();
  tokio::spawn(async move {
    let mut interval = tokio::time::interval(Duration::from_secs(30));

    loop {
      interval.tick().await;
      print_portfolio_summary(&summary_portfolio);
    }
  });

  // Monitor for demo duration
  println!("📊 Monitoring portfolio for 2 minutes...\n");
  sleep(Duration::from_secs(120)).await;

  // Final summary
  println!("\n🏁 Demo completed!");
  print_portfolio_summary(&portfolio_data);

  // Stop manager
  manager.stop().await?;
  println!("✅ Portfolio monitor stopped");

  Ok(())
}

examples/mode_change_test.rs (line 72)

async fn test_mode_change_issue(
  manager: &mut KiteTickerManager,
) -> Result<(), String> {
  println!("\n🧪 Testing Mode Change Issue");
  println!("=============================");

  // Step 1: Subscribe to a symbol with LTP mode
  let test_symbol = 738561; // Test symbol provided by user
  println!("\n1️⃣ Subscribing to symbol {} with LTP mode", test_symbol);
  manager
    .subscribe_symbols(&[test_symbol], Some(Mode::LTP))
    .await?;

  // Start monitoring ticks to see the mode
  let channels = manager.get_all_channels();
  let mut tick_listeners = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut ticks_received = 0;
      let start = Instant::now();

      while start.elapsed() < Duration::from_secs(5) && ticks_received < 3 {
        match timeout(Duration::from_millis(500), receiver.recv()).await {
          Ok(Ok(message)) => {
            if let TickerMessage::Ticks(ticks) = message {
              for tick in &ticks {
                if tick.instrument_token == test_symbol {
                  ticks_received += 1;
                  println!(
                    "   📊 Received tick for {}: Mode={:?}, LTP={:?}",
                    tick.instrument_token,
                    tick.content.mode,
                    tick.content.last_price
                  );

                  // Check if OHLC data is present (should not be for LTP mode)
                  if tick.content.ohlc.is_some() {
                    println!(
                      "   ⚠️  OHLC data present in LTP mode - unexpected!"
                    );
                  } else {
                    println!("   ✅ No OHLC data in LTP mode - correct");
                  }
                }
              }
            }
          }
          _ => continue,
        }
      }
      (channel_id, ticks_received)
    });
    tick_listeners.push(task);
  }

  // Wait for initial ticks
  for task in tick_listeners {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "   📈 {:?}: Received {} ticks for initial LTP subscription",
        channel_id, count
      );
    }
  }

  sleep(Duration::from_secs(2)).await;

  // Step 2: Attempt to change mode to Full (this is where the issue should manifest)
  println!(
    "\n2️⃣ Attempting to change mode from LTP to Full for symbol {}",
    test_symbol
  );
  match manager.change_mode(&[test_symbol], Mode::Full).await {
    Ok(()) => {
      println!("   ✅ Mode change command sent successfully");
    }
    Err(e) => {
      println!("   ❌ Mode change command failed: {}", e);
      return Err(e);
    }
  }

  // Step 3: Monitor ticks after mode change to see if it actually worked
  println!("\n3️⃣ Monitoring ticks after mode change...");
  let channels = manager.get_all_channels();
  let mut post_change_listeners = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut ticks_received = 0;
      let mut ohlc_present = 0;
      let mut depth_present = 0;
      let start = Instant::now();

      while start.elapsed() < Duration::from_secs(10) && ticks_received < 5 {
        match timeout(Duration::from_millis(500), receiver.recv()).await {
          Ok(Ok(message)) => {
            if let TickerMessage::Ticks(ticks) = message {
              for tick in &ticks {
                if tick.instrument_token == test_symbol {
                  ticks_received += 1;
                  println!(
                    "   📊 Post-change tick for {}: Mode={:?}, LTP={:?}",
                    tick.instrument_token,
                    tick.content.mode,
                    tick.content.last_price
                  );

                  // Check if Full mode data is present
                  if let Some(ohlc) = &tick.content.ohlc {
                    ohlc_present += 1;
                    println!(
                      "   ✅ OHLC data present: O:{} H:{} L:{} C:{}",
                      ohlc.open, ohlc.high, ohlc.low, ohlc.close
                    );
                  } else {
                    println!("   ❌ OHLC data missing - mode change may not have worked!");
                  }

                  if let Some(depth) = &tick.content.depth {
                    depth_present += 1;
                    println!(
                      "   ✅ Market depth present: {} buy, {} sell orders",
                      depth.buy.len(),
                      depth.sell.len()
                    );
                  } else {
                    println!("   ❌ Market depth missing - mode change may not have worked!");
                  }

                  // Log the actual mode reported in the tick
                  info!("Tick mode reported: {:?}", tick.content.mode);
                }
              }
            }
          }
          _ => continue,
        }
      }
      (channel_id, ticks_received, ohlc_present, depth_present)
    });
    post_change_listeners.push(task);
  }

  // Wait for post-change ticks
  let mut total_ticks = 0;
  let mut total_ohlc = 0;
  let mut total_depth = 0;

  for task in post_change_listeners {
    if let Ok((channel_id, ticks, ohlc, depth)) = task.await {
      println!(
        "   📈 {:?}: {} ticks, {} with OHLC, {} with depth",
        channel_id, ticks, ohlc, depth
      );
      total_ticks += ticks;
      total_ohlc += ohlc;
      total_depth += depth;
    }
  }

  // Step 4: Analyze results
  println!("\n4️⃣ Test Results Analysis");
  println!("========================");

  if total_ticks == 0 {
    println!("   ⚠️  No ticks received after mode change - connection issue?");
  } else if total_ohlc == 0 && total_depth == 0 {
    println!("   ❌ ISSUE CONFIRMED: Mode change command sent but Full mode data not received");
    println!("   💡 This confirms that set_mode() alone doesn't work - need unsubscribe+resubscribe");
  } else if total_ohlc > 0 && total_depth > 0 {
    println!("   ✅ Mode change worked - Full mode data received");
  } else {
    println!(
      "   ⚠️  Partial mode change - some Full mode data received but not all"
    );
  }

  println!("\n📊 Summary:");
  println!("   Total ticks received: {}", total_ticks);
  println!("   Ticks with OHLC data: {}", total_ohlc);
  println!("   Ticks with market depth: {}", total_depth);

  Ok(())
}

Additional examples can be found in:

• examples/dynamic_subscription_demo.rs
• examples/advanced/manager_demo.rs

pub async fn get_stats(&self) -> Result<ManagerStats, String>

Get manager statistics

Examples found in repository

examples/basic/runtime_subscription_example.rs (line 134)

async fn print_current_state(manager: &KiteTickerManager, context: &str) {
  println!("\n📊 Current State ({})", context);

  // Show distribution
  let distribution = manager.get_symbol_distribution();
  let mut total = 0;
  for (channel_id, symbols) in &distribution {
    println!("   {:?}: {} symbols", channel_id, symbols.len());
    total += symbols.len();
  }
  println!("   📈 Total: {} symbols", total);

  // Show stats
  if let Ok(stats) = manager.get_stats().await {
    println!(
      "   📊 Stats: {} connections, {} messages",
      stats.active_connections, stats.total_messages_received
    );
  }
}

examples/advanced/manager_demo.rs (line 240)

pub async fn main() -> Result<(), String> {
  // Initialize logging
  env_logger::init();

  println!("🚀 KiteTicker Multi-Connection Manager Demo");
  println!("═══════════════════════════════════════════");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️  KITE_API_KEY and KITE_ACCESS_TOKEN environment variables not set"
    );
    println!("   This demo will show the manager architecture without live connections");
    demonstrate_offline_architecture().await;
    return Ok(());
  }

  // Create high-performance configuration - RESTORED TO 3 CONNECTIONS
  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 3,            // BACK TO 3 CONNECTIONS!
    connection_buffer_size: 10000, // High buffer for performance
    parser_buffer_size: 20000,     // Even higher for parsed messages
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true, // Use dedicated parser tasks
    default_mode: Mode::Full,       // Full mode for maximum data
  };

  println!("🔧 Configuration:");
  println!("   Max connections: {}", config.max_connections);
  println!(
    "   Max symbols per connection: {}",
    config.max_symbols_per_connection
  );
  println!(
    "   Connection buffer size: {}",
    config.connection_buffer_size
  );
  println!("   Parser buffer size: {}", config.parser_buffer_size);
  println!("   Dedicated parsers: {}", config.enable_dedicated_parsers);
  println!();

  // Create and start the manager
  println!("📡 Starting multi-connection manager...");
  let start_time = Instant::now();

  let mut manager = KiteTickerManager::new(api_key, access_token, config);

  match timeout(Duration::from_secs(30), manager.start()).await {
    Ok(Ok(())) => {
      println!("✅ Manager started in {:?}", start_time.elapsed());
    }
    Ok(Err(e)) => {
      println!("❌ Manager failed to start: {}", e);
      return Err(e);
    }
    Err(_) => {
      println!("⏱️  Manager startup timeout");
      return Err("Manager startup timeout".to_string());
    }
  }

  // Test with market symbols for proper distribution
  let symbols = vec![
    256265, 265, 256777, 274441, 260105, 273929, 260617, 257033, 257289,
    257545, 257801, 258825, 259081, 259337, 259593, 259849, 260873, 261129,
    261385, 261641, 261897, 262153, 262409, 262665, 262921, 263177, 263433,
    263689, 263945, 264457, 264713, 264969, 265225, 265737, 265993, 266249,
    266505, 266761, 267017, 267273, 267529, 267785, 268041, 268297, 268553,
    268809, 269065, 269321, 269577, 269833, 270089, 270345, 270601, 270857,
    271113, 271625, 271881, 272137, 272393, 273417, 273673, 274185, 274697,
    274953, 275209, 275465, 275721, 275977, 276233, 276489, 276745, 277001,
    277257, 277513, 277769, 278025, 278281, 278537, 278793, 279049, 279305,
    279561, 279817, 280073, 280329, 280585, 280841, 281097, 281353, 281865,
    282121, 282377, 282633, 282889, 283145, 283401, 283657, 283913, 284169,
    284425, 284681, 284937, 285193, 285449, 285961, 286217, 286473, 286729,
    286985, 287241, 287497, 287753, 288009, 288265, 288521, 288777, 289033,
    289289, 289545, 289801, 290057, 290313, 290569, 290825, 291081, 291337,
    291593, 291849, 292105, 292361, 292617, 292873, 293129, 293385, 293641,
    293897, 294153, 294409, 294665, 294921, 295177, 295433, 295689, 398345,
    398601, 398857, 399113, 399369, 399625, 399881, 400137, 400393, 400905,
    401161, 401673, 401929, 402185, 402441, 402697, 402953, 403209, 403465,
    403721, 403977, 404233, 404489, 404745, 405001, 405257, 405513, 405769,
    406025, 406281, 406537, 406793, 407049, 407305, 407561, 407817, 408073,
    408329, 408585, 408841, 409097, 409353, 409609, 409865, 410121, 410377,
    410633, 410889, 411145, 411401, 411657, 411913, 412169, 412425, 412681,
    412937, 413193, 413449, 413705, 413961, 414217, 414473, 414729, 414985,
  ];

  println!("📊 Subscribing to symbols across connections...");

  // Subscribe to different symbol sets
  manager
    .subscribe_symbols(&symbols, Some(Mode::Full))
    .await?;
  // manager.subscribe_symbols(&bank_nifty, Some(Mode::Quote)).await?;
  // manager.subscribe_symbols(&it_stocks, Some(Mode::LTP)).await?;

  println!(
    "✅ Subscribed to {} total symbols",
    symbols.len() //  + bank_nifty.len() + it_stocks.len()
  );

  // Get symbol distribution
  let distribution = manager.get_symbol_distribution();
  println!("\n📈 Symbol distribution across connections:");
  for (channel_id, symbols) in &distribution {
    println!("   {:?}: {} symbols", channel_id, symbols.len());
  }

  // Get all output channels
  let channels = manager.get_all_channels();
  println!("\n🔀 Created {} output channels", channels.len());

  // Start monitoring each channel
  let mut channel_tasks = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut message_count = 0;
      let mut tick_count = 0;
      let start_time = Instant::now();
      let mut last_report = Instant::now();

      println!("🎯 Starting monitoring for {:?}", channel_id);

      loop {
        match timeout(Duration::from_secs(30), receiver.recv()).await {
          Ok(Ok(message)) => {
            message_count += 1;

            match message {
              TickerMessage::Ticks(ticks) => {
                tick_count += ticks.len();

                // Show first few ticks for demonstration
                if message_count <= 3 {
                  for tick in &ticks {
                    println!(
                      "📋 {:?}: Tick {} @ {:?}",
                      channel_id,
                      tick.instrument_token,
                      tick.content.last_price.unwrap_or(0.0)
                    );
                  }
                }
              }
              TickerMessage::Error(e) => {
                println!("⚠️  {:?}: Error: {}", channel_id, e);
              }
              _ => {
                println!("📨 {:?}: Other message", channel_id);
              }
            }

            // Report performance every 10 seconds
            if last_report.elapsed() >= Duration::from_secs(10) {
              let elapsed = start_time.elapsed();
              let messages_per_sec =
                message_count as f64 / elapsed.as_secs_f64();
              let ticks_per_sec = tick_count as f64 / elapsed.as_secs_f64();

              println!("📊 {:?} Performance:", channel_id);
              println!(
                "   Messages: {} ({:.1}/sec)",
                message_count, messages_per_sec
              );
              println!("   Ticks: {} ({:.1}/sec)", tick_count, ticks_per_sec);

              last_report = Instant::now();
            }
          }
          Ok(Err(e)) => {
            println!("❌ {:?}: Channel error: {}", channel_id, e);
            break;
          }
          Err(_) => {
            println!("⏱️  {:?}: No messages for 30s", channel_id);
          }
        }
      }

      (channel_id, message_count, tick_count)
    });

    channel_tasks.push(task);
  }

  // Monitor overall system health
  let health_task = tokio::spawn(async move {
    loop {
      sleep(Duration::from_secs(15)).await;

      println!("\n🏥 System Health Check:");
      println!("   All connections active ✅");
      println!("   Parsers running ✅");
      println!("   Memory usage optimized ✅");
    }
  });

  // Run for demonstration period
  println!(
    "\n📈 Monitoring performance for 60 seconds (Ctrl+C to stop early)..."
  );

  let demo_duration = Duration::from_secs(60);
  let demo_start = Instant::now();

  // Wait for demo duration or Ctrl+C
  tokio::select! {
      _ = sleep(demo_duration) => {
          println!("\n⏰ Demo duration completed");
      }
      _ = tokio::signal::ctrl_c() => {
          println!("\n🛑 Received Ctrl+C, stopping...");
      }
  }

  // Abort monitoring tasks
  health_task.abort();
  for task in channel_tasks {
    task.abort();
  }

  // Get final statistics
  println!("\n📊 Final Statistics:");

  if let Ok(stats) = manager.get_stats().await {
    println!("   Total runtime: {:?}", demo_start.elapsed());
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
    println!("   Total errors: {}", stats.total_errors);

    for (i, conn_stats) in stats.connection_stats.iter().enumerate() {
      println!(
        "   Connection {}: {} symbols, {} messages, {} errors",
        i,
        conn_stats.symbol_count,
        conn_stats.messages_received,
        conn_stats.errors_count
      );
    }
  }

  let processor_stats = manager.get_processor_stats().await;
  println!("\n🔧 Parser Performance:");
  for (channel_id, stats) in processor_stats {
    println!(
      "   {:?}: {:.1} msg/sec, {:?} avg latency",
      channel_id, stats.messages_per_second, stats.processing_latency_avg
    );
  }

  // Stop the manager
  println!("\n🛑 Stopping manager...");
  manager.stop().await?;

  println!("🏁 Demo completed successfully!");
  Ok(())
}

examples/dynamic_subscription_demo.rs (line 380)

async fn demo_dynamic_subscription(
  manager: &mut KiteTickerManager,
) -> Result<(), String> {
  println!("\n🎯 True Dynamic Subscription Demo");
  println!("==================================");

  // Start with a small initial set
  let initial_symbols = vec![256265, 265, 256777]; // 3 symbols
  let additional_batch_1 = vec![274441, 260105, 273929]; // 3 more symbols
  let additional_batch_2 = vec![260617, 257033, 257289, 257545]; // 4 more symbols
  let symbols_to_remove = vec![265, 274441]; // Remove some symbols
  let final_batch = vec![257801, 258825]; // Add final symbols

  // Step 1: Initial subscription with small set
  println!(
    "\n📊 Step 1: Initial subscription to {} symbols",
    initial_symbols.len()
  );
  println!("Starting with: {:?}", initial_symbols);

  // Start listening for ticks BEFORE subscribing
  let channels_before_sub = manager.get_all_channels();
  let mut tick_listeners = Vec::new();

  for (channel_id, mut receiver) in channels_before_sub {
    let task = tokio::spawn(async move {
      let start = std::time::Instant::now();
      let mut tick_count = 0;

      // Listen for initial ticks for 5 seconds
      while start.elapsed() < Duration::from_secs(5) {
        match timeout(Duration::from_millis(500), receiver.recv()).await {
          Ok(Ok(message)) => {
            if let TickerMessage::Ticks(ticks) = message {
              tick_count += ticks.len();
              println!(
                "🎯 {:?}: Received {} ticks immediately after subscription!",
                channel_id,
                ticks.len()
              );

              // Debug: Show detailed tick information
              debug!("Processing {} ticks on {:?}", ticks.len(), channel_id);

              for (idx, tick) in ticks.iter().enumerate() {
                println!("  🔹 Tick {}: Symbol: {}, LTP: {:?}, Volume: {:?}, Change: {:?}",
                                    idx + 1,
                                    tick.instrument_token,
                                    tick.content.last_price,
                                    tick.content.volume_traded,
                                    tick.content.net_change
                                );

                // Debug: Show tick structure details
                debug!(
                  "Tick {} Details for symbol {}:",
                  idx + 1,
                  tick.instrument_token
                );
                debug!("  - Raw instrument_token: {}", tick.instrument_token);
                debug!("  - Raw last_price: {:?}", tick.content.last_price);
                debug!(
                  "  - Raw volume_traded: {:?}",
                  tick.content.volume_traded
                );
                debug!("  - Raw change: {:?}", tick.content.net_change);
                debug!("  - Raw last_quantity: <not available>");
                debug!(
                  "  - Raw average_price: {:?}",
                  tick.content.avg_traded_price
                );
                debug!(
                  "  - Raw buy_quantity: {:?}",
                  tick.content.total_buy_qty
                );
                debug!(
                  "  - Raw sell_quantity: {:?}",
                  tick.content.total_sell_qty
                );
                debug!("  - Raw oi: {:?}", tick.content.oi);
                debug!("  - Raw mode: {:?}", tick.content.mode);

                if let Some(ohlc) = &tick.content.ohlc {
                  debug!(
                    "  - OHLC available: O:{} H:{} L:{} C:{}",
                    ohlc.open, ohlc.high, ohlc.low, ohlc.close
                  );
                }

                if let Some(depth) = &tick.content.depth {
                  debug!(
                    "  - Market depth available: {} buy, {} sell",
                    depth.buy.len(),
                    depth.sell.len()
                  );
                }
              }
            }
          }
          _ => continue,
        }
      }
      (channel_id, tick_count)
    });
    tick_listeners.push(task);
  }

  // Now subscribe to symbols
  manager
    .subscribe_symbols(&initial_symbols, Some(Mode::LTP))
    .await?;

  println!("✅ Subscription sent, waiting for initial ticks...");

  // Wait for the listeners to finish
  for task in tick_listeners {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "📊 {:?}: Received {} total ticks during initial subscription",
        channel_id, count
      );
    }
  }

  print_distribution(manager, "After initial subscription").await;

  // Step 2: Wait and monitor initial data
  println!("\n⏳ Step 2: Monitoring initial data flow");
  monitor_ticks_briefly(manager, 5, "Initial Subscription Data").await;

  if let Ok(stats) = manager.get_stats().await {
    println!("✅ Current Statistics:");
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
  }

  // Step 3: Dynamic addition - Batch 1
  println!(
    "\n➕ Step 3: DYNAMIC ADDITION - Adding {} more symbols",
    additional_batch_1.len()
  );
  println!("Adding: {:?}", additional_batch_1);
  manager
    .subscribe_symbols(&additional_batch_1, Some(Mode::Quote))
    .await?;

  // Give time for new tick data to arrive
  sleep(Duration::from_secs(2)).await;

  print_distribution(manager, "After first dynamic addition").await;
  monitor_ticks_briefly(manager, 3, "After First Addition").await;

  // Step 4: Dynamic addition - Batch 2
  println!(
    "\n➕ Step 4: DYNAMIC ADDITION - Adding {} more symbols",
    additional_batch_2.len()
  );
  println!("Adding: {:?}", additional_batch_2);
  manager
    .subscribe_symbols(&additional_batch_2, Some(Mode::Full))
    .await?;

  // Give time for new tick data to arrive
  sleep(Duration::from_secs(2)).await;

  print_distribution(manager, "After second dynamic addition").await;
  monitor_ticks_briefly(manager, 3, "After Second Addition").await;

  // Step 5: Dynamic removal
  println!(
    "\n➖ Step 5: DYNAMIC REMOVAL - Removing {} symbols",
    symbols_to_remove.len()
  );
  println!("Removing: {:?}", symbols_to_remove);
  match manager.unsubscribe_symbols(&symbols_to_remove).await {
    Ok(()) => {
      print_distribution(manager, "After dynamic removal").await;
      println!("✅ Dynamic removal successful!");
    }
    Err(e) => {
      println!("⚠️  Dynamic removal failed: {}", e);
    }
  }
  monitor_ticks_briefly(manager, 3, "After Symbol Removal").await;

  // Step 6: Final addition and mode change demo
  println!(
    "\n➕ Step 6: FINAL ADDITION - Adding {} symbols",
    final_batch.len()
  );
  println!("Adding: {:?}", final_batch);
  manager
    .subscribe_symbols(&final_batch, Some(Mode::LTP))
    .await?;

  print_distribution(manager, "After final addition").await;

  // Step 7: Mode change demonstration
  println!("\n🔄 Step 7: MODE CHANGE - Changing subscription mode");
  let symbols_for_mode_change = vec![256265, 260105]; // Change some existing symbols to Full mode
  println!("Changing {:?} to Full mode", symbols_for_mode_change);
  match manager
    .change_mode(&symbols_for_mode_change, Mode::Full)
    .await
  {
    Ok(()) => println!("✅ Mode change successful!"),
    Err(e) => println!("⚠️  Mode change failed: {}", e),
  }

  // Step 8: Final statistics and monitoring
  println!("\n📈 Step 8: Final Statistics and Performance");
  sleep(Duration::from_secs(3)).await; // Let some data flow

  if let Ok(stats) = manager.get_stats().await {
    println!("✅ Final Manager Statistics:");
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
    println!("   Total errors: {}", stats.total_errors);

    for (i, conn_stats) in stats.connection_stats.iter().enumerate() {
      println!(
        "   Connection {}: {} symbols, {} messages",
        i + 1,
        conn_stats.symbol_count,
        conn_stats.messages_received
      );
    }
  }

  // Step 9: Show processor performance
  println!("\n⚡ Step 9: Final Parser Performance");
  let processor_stats = manager.get_processor_stats().await;
  for (channel_id, stats) in processor_stats {
    println!(
      "   {:?}: {:.1} msg/sec, {:?} avg latency",
      channel_id, stats.messages_per_second, stats.processing_latency_avg
    );
  }

  // Step 10: Monitor live data for a short period
  println!("\n📺 Step 10: Live Data Monitoring (10 seconds)");
  println!("Monitoring real-time tick data from all dynamically managed connections...");

  let channels = manager.get_all_channels();
  let mut tasks = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut count = 0;
      let start = std::time::Instant::now();

      while start.elapsed() < Duration::from_secs(10) {
        match timeout(Duration::from_secs(2), receiver.recv()).await {
          Ok(Ok(message)) => {
            count += 1;
            if let TickerMessage::Ticks(ticks) = message {
              println!("📋 {:?}: {} ticks received", channel_id, ticks.len());

              // Debug: Final monitoring with comprehensive details
              debug!(
                "Final monitoring - Message #{}, {} ticks on {:?}",
                count,
                ticks.len(),
                channel_id
              );

              for tick in &ticks {
                println!(
                  "  🔹 Symbol: {}, LTP: {:?}, Volume: {:?}, Change: {:?}",
                  tick.instrument_token,
                  tick.content.last_price,
                  tick.content.volume_traded,
                  tick.content.net_change
                );

                // Debug: Show complete tick analysis
                debug!(
                  "Final Debug Analysis for Symbol {}:",
                  tick.instrument_token
                );
                debug!(
                  "  - Timestamp: Now = {:?}",
                  std::time::SystemTime::now()
                );
                debug!("  - Data completeness check:");
                debug!("    * Last Price: {:?} (✅)", tick.content.last_price);
                debug!(
                  "    * Volume: {:?} ({})",
                  tick.content.volume_traded,
                  if tick.content.volume_traded.is_some() {
                    "✅"
                  } else {
                    "❌"
                  }
                );
                debug!(
                  "    * Change: {:?} ({})",
                  tick.content.net_change,
                  if tick.content.net_change.is_some() {
                    "✅"
                  } else {
                    "❌"
                  }
                );
                debug!("    * Mode: {:?} (✅)", tick.content.mode);

                // Validate mode-specific data
                match tick.content.mode {
                  Mode::Full => {
                    debug!("  - Full mode validation:");
                    debug!(
                      "    * OHLC: {}",
                      if tick.content.ohlc.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    debug!(
                      "    * Depth: {}",
                      if tick.content.depth.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    if let Some(ohlc) = &tick.content.ohlc {
                      debug!(
                        "    * OHLC Values: O:{} H:{} L:{} C:{}",
                        ohlc.open, ohlc.high, ohlc.low, ohlc.close
                      );
                    }
                    if let Some(depth) = &tick.content.depth {
                      debug!(
                        "    * Depth Levels: {} buy, {} sell",
                        depth.buy.len(),
                        depth.sell.len()
                      );
                    }
                  }
                  Mode::Quote => {
                    debug!("  - Quote mode validation:");
                    debug!(
                      "    * OHLC: {}",
                      if tick.content.ohlc.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    if let Some(ohlc) = &tick.content.ohlc {
                      debug!(
                        "    * OHLC Values: O:{} H:{} L:{} C:{}",
                        ohlc.open, ohlc.high, ohlc.low, ohlc.close
                      );
                    }
                  }
                  Mode::LTP => {
                    debug!("  - LTP mode validation: ✅ Basic data only");
                  }
                }
              }
            } else {
              println!("📋 {:?}: Non-tick message received", channel_id);
              debug!("Message type: {:?}", message);
            }
          }
          _ => continue,
        }
      }
      (channel_id, count)
    });
    tasks.push(task);
  }

  // Wait for monitoring to complete
  for task in tasks {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "📊 {:?}: {} total messages in 10 seconds",
        channel_id, count
      );
    }
  }

  // Final cleanup demonstration
  println!("\n🧹 Step 11: Final Cleanup Demonstration");
  let current_symbols: Vec<u32> = manager
    .get_symbol_distribution()
    .values()
    .flat_map(|symbols| symbols.iter().cloned())
    .collect();

  println!(
    "Attempting to unsubscribe from {} remaining symbols...",
    current_symbols.len()
  );

  match manager.unsubscribe_symbols(&current_symbols).await {
    Ok(()) => {
      print_distribution(manager, "After complete cleanup").await;
      println!("✅ Complete cleanup successful!");
    }
    Err(e) => {
      println!("⚠️  Cleanup encountered issues: {}", e);
      println!("💡 This demonstrates the current architecture capabilities");
    }
  }

  println!("\n✅ Dynamic Subscription Demo Completed!");
  println!("🎯 Key Dynamic Features Demonstrated:");
  println!("   ✅ Runtime symbol addition across multiple batches");
  println!("   ✅ Runtime symbol removal with proper tracking");
  println!("   ✅ Dynamic mode changes for existing symbols");
  println!("   ✅ Real-time capacity distribution and monitoring");
  println!("   ✅ Independent message processing per connection");
  println!("   ✅ High-performance parser tasks with statistics");
  println!("   ✅ Complete subscription lifecycle management");

  Ok(())
}

pub async fn get_health(&self) -> Result<HealthSummary, String>

Get health summary

pub async fn get_processor_stats(&self) -> Vec<(ChannelId, ProcessorStats)>

Get processor statistics for all channels

Examples found in repository

examples/advanced/manager_demo.rs (line 258)

pub async fn main() -> Result<(), String> {
  // Initialize logging
  env_logger::init();

  println!("🚀 KiteTicker Multi-Connection Manager Demo");
  println!("═══════════════════════════════════════════");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️  KITE_API_KEY and KITE_ACCESS_TOKEN environment variables not set"
    );
    println!("   This demo will show the manager architecture without live connections");
    demonstrate_offline_architecture().await;
    return Ok(());
  }

  // Create high-performance configuration - RESTORED TO 3 CONNECTIONS
  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 3,            // BACK TO 3 CONNECTIONS!
    connection_buffer_size: 10000, // High buffer for performance
    parser_buffer_size: 20000,     // Even higher for parsed messages
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true, // Use dedicated parser tasks
    default_mode: Mode::Full,       // Full mode for maximum data
  };

  println!("🔧 Configuration:");
  println!("   Max connections: {}", config.max_connections);
  println!(
    "   Max symbols per connection: {}",
    config.max_symbols_per_connection
  );
  println!(
    "   Connection buffer size: {}",
    config.connection_buffer_size
  );
  println!("   Parser buffer size: {}", config.parser_buffer_size);
  println!("   Dedicated parsers: {}", config.enable_dedicated_parsers);
  println!();

  // Create and start the manager
  println!("📡 Starting multi-connection manager...");
  let start_time = Instant::now();

  let mut manager = KiteTickerManager::new(api_key, access_token, config);

  match timeout(Duration::from_secs(30), manager.start()).await {
    Ok(Ok(())) => {
      println!("✅ Manager started in {:?}", start_time.elapsed());
    }
    Ok(Err(e)) => {
      println!("❌ Manager failed to start: {}", e);
      return Err(e);
    }
    Err(_) => {
      println!("⏱️  Manager startup timeout");
      return Err("Manager startup timeout".to_string());
    }
  }

  // Test with market symbols for proper distribution
  let symbols = vec![
    256265, 265, 256777, 274441, 260105, 273929, 260617, 257033, 257289,
    257545, 257801, 258825, 259081, 259337, 259593, 259849, 260873, 261129,
    261385, 261641, 261897, 262153, 262409, 262665, 262921, 263177, 263433,
    263689, 263945, 264457, 264713, 264969, 265225, 265737, 265993, 266249,
    266505, 266761, 267017, 267273, 267529, 267785, 268041, 268297, 268553,
    268809, 269065, 269321, 269577, 269833, 270089, 270345, 270601, 270857,
    271113, 271625, 271881, 272137, 272393, 273417, 273673, 274185, 274697,
    274953, 275209, 275465, 275721, 275977, 276233, 276489, 276745, 277001,
    277257, 277513, 277769, 278025, 278281, 278537, 278793, 279049, 279305,
    279561, 279817, 280073, 280329, 280585, 280841, 281097, 281353, 281865,
    282121, 282377, 282633, 282889, 283145, 283401, 283657, 283913, 284169,
    284425, 284681, 284937, 285193, 285449, 285961, 286217, 286473, 286729,
    286985, 287241, 287497, 287753, 288009, 288265, 288521, 288777, 289033,
    289289, 289545, 289801, 290057, 290313, 290569, 290825, 291081, 291337,
    291593, 291849, 292105, 292361, 292617, 292873, 293129, 293385, 293641,
    293897, 294153, 294409, 294665, 294921, 295177, 295433, 295689, 398345,
    398601, 398857, 399113, 399369, 399625, 399881, 400137, 400393, 400905,
    401161, 401673, 401929, 402185, 402441, 402697, 402953, 403209, 403465,
    403721, 403977, 404233, 404489, 404745, 405001, 405257, 405513, 405769,
    406025, 406281, 406537, 406793, 407049, 407305, 407561, 407817, 408073,
    408329, 408585, 408841, 409097, 409353, 409609, 409865, 410121, 410377,
    410633, 410889, 411145, 411401, 411657, 411913, 412169, 412425, 412681,
    412937, 413193, 413449, 413705, 413961, 414217, 414473, 414729, 414985,
  ];

  println!("📊 Subscribing to symbols across connections...");

  // Subscribe to different symbol sets
  manager
    .subscribe_symbols(&symbols, Some(Mode::Full))
    .await?;
  // manager.subscribe_symbols(&bank_nifty, Some(Mode::Quote)).await?;
  // manager.subscribe_symbols(&it_stocks, Some(Mode::LTP)).await?;

  println!(
    "✅ Subscribed to {} total symbols",
    symbols.len() //  + bank_nifty.len() + it_stocks.len()
  );

  // Get symbol distribution
  let distribution = manager.get_symbol_distribution();
  println!("\n📈 Symbol distribution across connections:");
  for (channel_id, symbols) in &distribution {
    println!("   {:?}: {} symbols", channel_id, symbols.len());
  }

  // Get all output channels
  let channels = manager.get_all_channels();
  println!("\n🔀 Created {} output channels", channels.len());

  // Start monitoring each channel
  let mut channel_tasks = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut message_count = 0;
      let mut tick_count = 0;
      let start_time = Instant::now();
      let mut last_report = Instant::now();

      println!("🎯 Starting monitoring for {:?}", channel_id);

      loop {
        match timeout(Duration::from_secs(30), receiver.recv()).await {
          Ok(Ok(message)) => {
            message_count += 1;

            match message {
              TickerMessage::Ticks(ticks) => {
                tick_count += ticks.len();

                // Show first few ticks for demonstration
                if message_count <= 3 {
                  for tick in &ticks {
                    println!(
                      "📋 {:?}: Tick {} @ {:?}",
                      channel_id,
                      tick.instrument_token,
                      tick.content.last_price.unwrap_or(0.0)
                    );
                  }
                }
              }
              TickerMessage::Error(e) => {
                println!("⚠️  {:?}: Error: {}", channel_id, e);
              }
              _ => {
                println!("📨 {:?}: Other message", channel_id);
              }
            }

            // Report performance every 10 seconds
            if last_report.elapsed() >= Duration::from_secs(10) {
              let elapsed = start_time.elapsed();
              let messages_per_sec =
                message_count as f64 / elapsed.as_secs_f64();
              let ticks_per_sec = tick_count as f64 / elapsed.as_secs_f64();

              println!("📊 {:?} Performance:", channel_id);
              println!(
                "   Messages: {} ({:.1}/sec)",
                message_count, messages_per_sec
              );
              println!("   Ticks: {} ({:.1}/sec)", tick_count, ticks_per_sec);

              last_report = Instant::now();
            }
          }
          Ok(Err(e)) => {
            println!("❌ {:?}: Channel error: {}", channel_id, e);
            break;
          }
          Err(_) => {
            println!("⏱️  {:?}: No messages for 30s", channel_id);
          }
        }
      }

      (channel_id, message_count, tick_count)
    });

    channel_tasks.push(task);
  }

  // Monitor overall system health
  let health_task = tokio::spawn(async move {
    loop {
      sleep(Duration::from_secs(15)).await;

      println!("\n🏥 System Health Check:");
      println!("   All connections active ✅");
      println!("   Parsers running ✅");
      println!("   Memory usage optimized ✅");
    }
  });

  // Run for demonstration period
  println!(
    "\n📈 Monitoring performance for 60 seconds (Ctrl+C to stop early)..."
  );

  let demo_duration = Duration::from_secs(60);
  let demo_start = Instant::now();

  // Wait for demo duration or Ctrl+C
  tokio::select! {
      _ = sleep(demo_duration) => {
          println!("\n⏰ Demo duration completed");
      }
      _ = tokio::signal::ctrl_c() => {
          println!("\n🛑 Received Ctrl+C, stopping...");
      }
  }

  // Abort monitoring tasks
  health_task.abort();
  for task in channel_tasks {
    task.abort();
  }

  // Get final statistics
  println!("\n📊 Final Statistics:");

  if let Ok(stats) = manager.get_stats().await {
    println!("   Total runtime: {:?}", demo_start.elapsed());
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
    println!("   Total errors: {}", stats.total_errors);

    for (i, conn_stats) in stats.connection_stats.iter().enumerate() {
      println!(
        "   Connection {}: {} symbols, {} messages, {} errors",
        i,
        conn_stats.symbol_count,
        conn_stats.messages_received,
        conn_stats.errors_count
      );
    }
  }

  let processor_stats = manager.get_processor_stats().await;
  println!("\n🔧 Parser Performance:");
  for (channel_id, stats) in processor_stats {
    println!(
      "   {:?}: {:.1} msg/sec, {:?} avg latency",
      channel_id, stats.messages_per_second, stats.processing_latency_avg
    );
  }

  // Stop the manager
  println!("\n🛑 Stopping manager...");
  manager.stop().await?;

  println!("🏁 Demo completed successfully!");
  Ok(())
}

examples/dynamic_subscription_demo.rs (line 483)

async fn demo_dynamic_subscription(
  manager: &mut KiteTickerManager,
) -> Result<(), String> {
  println!("\n🎯 True Dynamic Subscription Demo");
  println!("==================================");

  // Start with a small initial set
  let initial_symbols = vec![256265, 265, 256777]; // 3 symbols
  let additional_batch_1 = vec![274441, 260105, 273929]; // 3 more symbols
  let additional_batch_2 = vec![260617, 257033, 257289, 257545]; // 4 more symbols
  let symbols_to_remove = vec![265, 274441]; // Remove some symbols
  let final_batch = vec![257801, 258825]; // Add final symbols

  // Step 1: Initial subscription with small set
  println!(
    "\n📊 Step 1: Initial subscription to {} symbols",
    initial_symbols.len()
  );
  println!("Starting with: {:?}", initial_symbols);

  // Start listening for ticks BEFORE subscribing
  let channels_before_sub = manager.get_all_channels();
  let mut tick_listeners = Vec::new();

  for (channel_id, mut receiver) in channels_before_sub {
    let task = tokio::spawn(async move {
      let start = std::time::Instant::now();
      let mut tick_count = 0;

      // Listen for initial ticks for 5 seconds
      while start.elapsed() < Duration::from_secs(5) {
        match timeout(Duration::from_millis(500), receiver.recv()).await {
          Ok(Ok(message)) => {
            if let TickerMessage::Ticks(ticks) = message {
              tick_count += ticks.len();
              println!(
                "🎯 {:?}: Received {} ticks immediately after subscription!",
                channel_id,
                ticks.len()
              );

              // Debug: Show detailed tick information
              debug!("Processing {} ticks on {:?}", ticks.len(), channel_id);

              for (idx, tick) in ticks.iter().enumerate() {
                println!("  🔹 Tick {}: Symbol: {}, LTP: {:?}, Volume: {:?}, Change: {:?}",
                                    idx + 1,
                                    tick.instrument_token,
                                    tick.content.last_price,
                                    tick.content.volume_traded,
                                    tick.content.net_change
                                );

                // Debug: Show tick structure details
                debug!(
                  "Tick {} Details for symbol {}:",
                  idx + 1,
                  tick.instrument_token
                );
                debug!("  - Raw instrument_token: {}", tick.instrument_token);
                debug!("  - Raw last_price: {:?}", tick.content.last_price);
                debug!(
                  "  - Raw volume_traded: {:?}",
                  tick.content.volume_traded
                );
                debug!("  - Raw change: {:?}", tick.content.net_change);
                debug!("  - Raw last_quantity: <not available>");
                debug!(
                  "  - Raw average_price: {:?}",
                  tick.content.avg_traded_price
                );
                debug!(
                  "  - Raw buy_quantity: {:?}",
                  tick.content.total_buy_qty
                );
                debug!(
                  "  - Raw sell_quantity: {:?}",
                  tick.content.total_sell_qty
                );
                debug!("  - Raw oi: {:?}", tick.content.oi);
                debug!("  - Raw mode: {:?}", tick.content.mode);

                if let Some(ohlc) = &tick.content.ohlc {
                  debug!(
                    "  - OHLC available: O:{} H:{} L:{} C:{}",
                    ohlc.open, ohlc.high, ohlc.low, ohlc.close
                  );
                }

                if let Some(depth) = &tick.content.depth {
                  debug!(
                    "  - Market depth available: {} buy, {} sell",
                    depth.buy.len(),
                    depth.sell.len()
                  );
                }
              }
            }
          }
          _ => continue,
        }
      }
      (channel_id, tick_count)
    });
    tick_listeners.push(task);
  }

  // Now subscribe to symbols
  manager
    .subscribe_symbols(&initial_symbols, Some(Mode::LTP))
    .await?;

  println!("✅ Subscription sent, waiting for initial ticks...");

  // Wait for the listeners to finish
  for task in tick_listeners {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "📊 {:?}: Received {} total ticks during initial subscription",
        channel_id, count
      );
    }
  }

  print_distribution(manager, "After initial subscription").await;

  // Step 2: Wait and monitor initial data
  println!("\n⏳ Step 2: Monitoring initial data flow");
  monitor_ticks_briefly(manager, 5, "Initial Subscription Data").await;

  if let Ok(stats) = manager.get_stats().await {
    println!("✅ Current Statistics:");
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
  }

  // Step 3: Dynamic addition - Batch 1
  println!(
    "\n➕ Step 3: DYNAMIC ADDITION - Adding {} more symbols",
    additional_batch_1.len()
  );
  println!("Adding: {:?}", additional_batch_1);
  manager
    .subscribe_symbols(&additional_batch_1, Some(Mode::Quote))
    .await?;

  // Give time for new tick data to arrive
  sleep(Duration::from_secs(2)).await;

  print_distribution(manager, "After first dynamic addition").await;
  monitor_ticks_briefly(manager, 3, "After First Addition").await;

  // Step 4: Dynamic addition - Batch 2
  println!(
    "\n➕ Step 4: DYNAMIC ADDITION - Adding {} more symbols",
    additional_batch_2.len()
  );
  println!("Adding: {:?}", additional_batch_2);
  manager
    .subscribe_symbols(&additional_batch_2, Some(Mode::Full))
    .await?;

  // Give time for new tick data to arrive
  sleep(Duration::from_secs(2)).await;

  print_distribution(manager, "After second dynamic addition").await;
  monitor_ticks_briefly(manager, 3, "After Second Addition").await;

  // Step 5: Dynamic removal
  println!(
    "\n➖ Step 5: DYNAMIC REMOVAL - Removing {} symbols",
    symbols_to_remove.len()
  );
  println!("Removing: {:?}", symbols_to_remove);
  match manager.unsubscribe_symbols(&symbols_to_remove).await {
    Ok(()) => {
      print_distribution(manager, "After dynamic removal").await;
      println!("✅ Dynamic removal successful!");
    }
    Err(e) => {
      println!("⚠️  Dynamic removal failed: {}", e);
    }
  }
  monitor_ticks_briefly(manager, 3, "After Symbol Removal").await;

  // Step 6: Final addition and mode change demo
  println!(
    "\n➕ Step 6: FINAL ADDITION - Adding {} symbols",
    final_batch.len()
  );
  println!("Adding: {:?}", final_batch);
  manager
    .subscribe_symbols(&final_batch, Some(Mode::LTP))
    .await?;

  print_distribution(manager, "After final addition").await;

  // Step 7: Mode change demonstration
  println!("\n🔄 Step 7: MODE CHANGE - Changing subscription mode");
  let symbols_for_mode_change = vec![256265, 260105]; // Change some existing symbols to Full mode
  println!("Changing {:?} to Full mode", symbols_for_mode_change);
  match manager
    .change_mode(&symbols_for_mode_change, Mode::Full)
    .await
  {
    Ok(()) => println!("✅ Mode change successful!"),
    Err(e) => println!("⚠️  Mode change failed: {}", e),
  }

  // Step 8: Final statistics and monitoring
  println!("\n📈 Step 8: Final Statistics and Performance");
  sleep(Duration::from_secs(3)).await; // Let some data flow

  if let Ok(stats) = manager.get_stats().await {
    println!("✅ Final Manager Statistics:");
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
    println!("   Total errors: {}", stats.total_errors);

    for (i, conn_stats) in stats.connection_stats.iter().enumerate() {
      println!(
        "   Connection {}: {} symbols, {} messages",
        i + 1,
        conn_stats.symbol_count,
        conn_stats.messages_received
      );
    }
  }

  // Step 9: Show processor performance
  println!("\n⚡ Step 9: Final Parser Performance");
  let processor_stats = manager.get_processor_stats().await;
  for (channel_id, stats) in processor_stats {
    println!(
      "   {:?}: {:.1} msg/sec, {:?} avg latency",
      channel_id, stats.messages_per_second, stats.processing_latency_avg
    );
  }

  // Step 10: Monitor live data for a short period
  println!("\n📺 Step 10: Live Data Monitoring (10 seconds)");
  println!("Monitoring real-time tick data from all dynamically managed connections...");

  let channels = manager.get_all_channels();
  let mut tasks = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut count = 0;
      let start = std::time::Instant::now();

      while start.elapsed() < Duration::from_secs(10) {
        match timeout(Duration::from_secs(2), receiver.recv()).await {
          Ok(Ok(message)) => {
            count += 1;
            if let TickerMessage::Ticks(ticks) = message {
              println!("📋 {:?}: {} ticks received", channel_id, ticks.len());

              // Debug: Final monitoring with comprehensive details
              debug!(
                "Final monitoring - Message #{}, {} ticks on {:?}",
                count,
                ticks.len(),
                channel_id
              );

              for tick in &ticks {
                println!(
                  "  🔹 Symbol: {}, LTP: {:?}, Volume: {:?}, Change: {:?}",
                  tick.instrument_token,
                  tick.content.last_price,
                  tick.content.volume_traded,
                  tick.content.net_change
                );

                // Debug: Show complete tick analysis
                debug!(
                  "Final Debug Analysis for Symbol {}:",
                  tick.instrument_token
                );
                debug!(
                  "  - Timestamp: Now = {:?}",
                  std::time::SystemTime::now()
                );
                debug!("  - Data completeness check:");
                debug!("    * Last Price: {:?} (✅)", tick.content.last_price);
                debug!(
                  "    * Volume: {:?} ({})",
                  tick.content.volume_traded,
                  if tick.content.volume_traded.is_some() {
                    "✅"
                  } else {
                    "❌"
                  }
                );
                debug!(
                  "    * Change: {:?} ({})",
                  tick.content.net_change,
                  if tick.content.net_change.is_some() {
                    "✅"
                  } else {
                    "❌"
                  }
                );
                debug!("    * Mode: {:?} (✅)", tick.content.mode);

                // Validate mode-specific data
                match tick.content.mode {
                  Mode::Full => {
                    debug!("  - Full mode validation:");
                    debug!(
                      "    * OHLC: {}",
                      if tick.content.ohlc.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    debug!(
                      "    * Depth: {}",
                      if tick.content.depth.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    if let Some(ohlc) = &tick.content.ohlc {
                      debug!(
                        "    * OHLC Values: O:{} H:{} L:{} C:{}",
                        ohlc.open, ohlc.high, ohlc.low, ohlc.close
                      );
                    }
                    if let Some(depth) = &tick.content.depth {
                      debug!(
                        "    * Depth Levels: {} buy, {} sell",
                        depth.buy.len(),
                        depth.sell.len()
                      );
                    }
                  }
                  Mode::Quote => {
                    debug!("  - Quote mode validation:");
                    debug!(
                      "    * OHLC: {}",
                      if tick.content.ohlc.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    if let Some(ohlc) = &tick.content.ohlc {
                      debug!(
                        "    * OHLC Values: O:{} H:{} L:{} C:{}",
                        ohlc.open, ohlc.high, ohlc.low, ohlc.close
                      );
                    }
                  }
                  Mode::LTP => {
                    debug!("  - LTP mode validation: ✅ Basic data only");
                  }
                }
              }
            } else {
              println!("📋 {:?}: Non-tick message received", channel_id);
              debug!("Message type: {:?}", message);
            }
          }
          _ => continue,
        }
      }
      (channel_id, count)
    });
    tasks.push(task);
  }

  // Wait for monitoring to complete
  for task in tasks {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "📊 {:?}: {} total messages in 10 seconds",
        channel_id, count
      );
    }
  }

  // Final cleanup demonstration
  println!("\n🧹 Step 11: Final Cleanup Demonstration");
  let current_symbols: Vec<u32> = manager
    .get_symbol_distribution()
    .values()
    .flat_map(|symbols| symbols.iter().cloned())
    .collect();

  println!(
    "Attempting to unsubscribe from {} remaining symbols...",
    current_symbols.len()
  );

  match manager.unsubscribe_symbols(&current_symbols).await {
    Ok(()) => {
      print_distribution(manager, "After complete cleanup").await;
      println!("✅ Complete cleanup successful!");
    }
    Err(e) => {
      println!("⚠️  Cleanup encountered issues: {}", e);
      println!("💡 This demonstrates the current architecture capabilities");
    }
  }

  println!("\n✅ Dynamic Subscription Demo Completed!");
  println!("🎯 Key Dynamic Features Demonstrated:");
  println!("   ✅ Runtime symbol addition across multiple batches");
  println!("   ✅ Runtime symbol removal with proper tracking");
  println!("   ✅ Dynamic mode changes for existing symbols");
  println!("   ✅ Real-time capacity distribution and monitoring");
  println!("   ✅ Independent message processing per connection");
  println!("   ✅ High-performance parser tasks with statistics");
  println!("   ✅ Complete subscription lifecycle management");

  Ok(())
}

pub fn get_symbol_distribution(&self) -> HashMap<ChannelId, Vec<u32>>

Get symbol distribution across connections

Examples found in repository

examples/basic/runtime_subscription_example.rs (line 107)

async fn runtime_subscription_demo(
  manager: &mut KiteTickerManager,
) -> Result<(), String> {
  println!("\n📡 Runtime Subscription Management Demo");
  println!("=====================================");

  // 1. Start with initial symbols
  println!("\n1️⃣ Initial subscription to base symbols");
  let initial_symbols = vec![256265, 265, 256777];
  manager
    .subscribe_symbols(&initial_symbols, Some(Mode::LTP))
    .await?;
  print_current_state(manager, "Initial subscription").await;

  sleep(Duration::from_secs(2)).await;

  // 2. Add more symbols dynamically
  println!("\n2️⃣ Adding symbols dynamically");
  let additional_symbols = vec![274441, 260105, 273929];
  manager
    .subscribe_symbols(&additional_symbols, Some(Mode::Quote))
    .await?;
  print_current_state(manager, "After adding symbols").await;

  sleep(Duration::from_secs(2)).await;

  // 3. Change mode for existing symbols
  println!("\n3️⃣ Changing subscription mode");
  let symbols_for_mode_change = vec![256265, 265];
  manager
    .change_mode(&symbols_for_mode_change, Mode::Full)
    .await?;
  print_current_state(manager, "After mode change").await;

  sleep(Duration::from_secs(2)).await;

  // 4. Remove some symbols
  println!("\n4️⃣ Removing symbols dynamically");
  let symbols_to_remove = vec![265, 274441];
  manager.unsubscribe_symbols(&symbols_to_remove).await?;
  print_current_state(manager, "After removing symbols").await;

  sleep(Duration::from_secs(2)).await;

  // 5. Add different symbols with different modes
  println!("\n5️⃣ Adding new symbols with Full mode");
  let full_mode_symbols = vec![257801, 258825];
  manager
    .subscribe_symbols(&full_mode_symbols, Some(Mode::Full))
    .await?;
  print_current_state(manager, "After adding Full mode symbols").await;

  // 6. Monitor live data for a short period
  println!("\n6️⃣ Monitoring live data (5 seconds)");
  monitor_live_data(manager, 5).await?;

  // 7. Complete cleanup
  println!("\n7️⃣ Complete cleanup");
  let all_symbols: Vec<u32> = manager
    .get_symbol_distribution()
    .values()
    .flat_map(|symbols| symbols.iter().cloned())
    .collect();

  if !all_symbols.is_empty() {
    manager.unsubscribe_symbols(&all_symbols).await?;
    print_current_state(manager, "After complete cleanup").await;
  }

  println!("\n✅ Runtime subscription demo completed!");
  Ok(())
}

async fn print_current_state(manager: &KiteTickerManager, context: &str) {
  println!("\n📊 Current State ({})", context);

  // Show distribution
  let distribution = manager.get_symbol_distribution();
  let mut total = 0;
  for (channel_id, symbols) in &distribution {
    println!("   {:?}: {} symbols", channel_id, symbols.len());
    total += symbols.len();
  }
  println!("   📈 Total: {} symbols", total);

  // Show stats
  if let Ok(stats) = manager.get_stats().await {
    println!(
      "   📊 Stats: {} connections, {} messages",
      stats.active_connections, stats.total_messages_received
    );
  }
}

examples/advanced/manager_demo.rs (line 117)

pub async fn main() -> Result<(), String> {
  // Initialize logging
  env_logger::init();

  println!("🚀 KiteTicker Multi-Connection Manager Demo");
  println!("═══════════════════════════════════════════");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️  KITE_API_KEY and KITE_ACCESS_TOKEN environment variables not set"
    );
    println!("   This demo will show the manager architecture without live connections");
    demonstrate_offline_architecture().await;
    return Ok(());
  }

  // Create high-performance configuration - RESTORED TO 3 CONNECTIONS
  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 3,            // BACK TO 3 CONNECTIONS!
    connection_buffer_size: 10000, // High buffer for performance
    parser_buffer_size: 20000,     // Even higher for parsed messages
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true, // Use dedicated parser tasks
    default_mode: Mode::Full,       // Full mode for maximum data
  };

  println!("🔧 Configuration:");
  println!("   Max connections: {}", config.max_connections);
  println!(
    "   Max symbols per connection: {}",
    config.max_symbols_per_connection
  );
  println!(
    "   Connection buffer size: {}",
    config.connection_buffer_size
  );
  println!("   Parser buffer size: {}", config.parser_buffer_size);
  println!("   Dedicated parsers: {}", config.enable_dedicated_parsers);
  println!();

  // Create and start the manager
  println!("📡 Starting multi-connection manager...");
  let start_time = Instant::now();

  let mut manager = KiteTickerManager::new(api_key, access_token, config);

  match timeout(Duration::from_secs(30), manager.start()).await {
    Ok(Ok(())) => {
      println!("✅ Manager started in {:?}", start_time.elapsed());
    }
    Ok(Err(e)) => {
      println!("❌ Manager failed to start: {}", e);
      return Err(e);
    }
    Err(_) => {
      println!("⏱️  Manager startup timeout");
      return Err("Manager startup timeout".to_string());
    }
  }

  // Test with market symbols for proper distribution
  let symbols = vec![
    256265, 265, 256777, 274441, 260105, 273929, 260617, 257033, 257289,
    257545, 257801, 258825, 259081, 259337, 259593, 259849, 260873, 261129,
    261385, 261641, 261897, 262153, 262409, 262665, 262921, 263177, 263433,
    263689, 263945, 264457, 264713, 264969, 265225, 265737, 265993, 266249,
    266505, 266761, 267017, 267273, 267529, 267785, 268041, 268297, 268553,
    268809, 269065, 269321, 269577, 269833, 270089, 270345, 270601, 270857,
    271113, 271625, 271881, 272137, 272393, 273417, 273673, 274185, 274697,
    274953, 275209, 275465, 275721, 275977, 276233, 276489, 276745, 277001,
    277257, 277513, 277769, 278025, 278281, 278537, 278793, 279049, 279305,
    279561, 279817, 280073, 280329, 280585, 280841, 281097, 281353, 281865,
    282121, 282377, 282633, 282889, 283145, 283401, 283657, 283913, 284169,
    284425, 284681, 284937, 285193, 285449, 285961, 286217, 286473, 286729,
    286985, 287241, 287497, 287753, 288009, 288265, 288521, 288777, 289033,
    289289, 289545, 289801, 290057, 290313, 290569, 290825, 291081, 291337,
    291593, 291849, 292105, 292361, 292617, 292873, 293129, 293385, 293641,
    293897, 294153, 294409, 294665, 294921, 295177, 295433, 295689, 398345,
    398601, 398857, 399113, 399369, 399625, 399881, 400137, 400393, 400905,
    401161, 401673, 401929, 402185, 402441, 402697, 402953, 403209, 403465,
    403721, 403977, 404233, 404489, 404745, 405001, 405257, 405513, 405769,
    406025, 406281, 406537, 406793, 407049, 407305, 407561, 407817, 408073,
    408329, 408585, 408841, 409097, 409353, 409609, 409865, 410121, 410377,
    410633, 410889, 411145, 411401, 411657, 411913, 412169, 412425, 412681,
    412937, 413193, 413449, 413705, 413961, 414217, 414473, 414729, 414985,
  ];

  println!("📊 Subscribing to symbols across connections...");

  // Subscribe to different symbol sets
  manager
    .subscribe_symbols(&symbols, Some(Mode::Full))
    .await?;
  // manager.subscribe_symbols(&bank_nifty, Some(Mode::Quote)).await?;
  // manager.subscribe_symbols(&it_stocks, Some(Mode::LTP)).await?;

  println!(
    "✅ Subscribed to {} total symbols",
    symbols.len() //  + bank_nifty.len() + it_stocks.len()
  );

  // Get symbol distribution
  let distribution = manager.get_symbol_distribution();
  println!("\n📈 Symbol distribution across connections:");
  for (channel_id, symbols) in &distribution {
    println!("   {:?}: {} symbols", channel_id, symbols.len());
  }

  // Get all output channels
  let channels = manager.get_all_channels();
  println!("\n🔀 Created {} output channels", channels.len());

  // Start monitoring each channel
  let mut channel_tasks = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut message_count = 0;
      let mut tick_count = 0;
      let start_time = Instant::now();
      let mut last_report = Instant::now();

      println!("🎯 Starting monitoring for {:?}", channel_id);

      loop {
        match timeout(Duration::from_secs(30), receiver.recv()).await {
          Ok(Ok(message)) => {
            message_count += 1;

            match message {
              TickerMessage::Ticks(ticks) => {
                tick_count += ticks.len();

                // Show first few ticks for demonstration
                if message_count <= 3 {
                  for tick in &ticks {
                    println!(
                      "📋 {:?}: Tick {} @ {:?}",
                      channel_id,
                      tick.instrument_token,
                      tick.content.last_price.unwrap_or(0.0)
                    );
                  }
                }
              }
              TickerMessage::Error(e) => {
                println!("⚠️  {:?}: Error: {}", channel_id, e);
              }
              _ => {
                println!("📨 {:?}: Other message", channel_id);
              }
            }

            // Report performance every 10 seconds
            if last_report.elapsed() >= Duration::from_secs(10) {
              let elapsed = start_time.elapsed();
              let messages_per_sec =
                message_count as f64 / elapsed.as_secs_f64();
              let ticks_per_sec = tick_count as f64 / elapsed.as_secs_f64();

              println!("📊 {:?} Performance:", channel_id);
              println!(
                "   Messages: {} ({:.1}/sec)",
                message_count, messages_per_sec
              );
              println!("   Ticks: {} ({:.1}/sec)", tick_count, ticks_per_sec);

              last_report = Instant::now();
            }
          }
          Ok(Err(e)) => {
            println!("❌ {:?}: Channel error: {}", channel_id, e);
            break;
          }
          Err(_) => {
            println!("⏱️  {:?}: No messages for 30s", channel_id);
          }
        }
      }

      (channel_id, message_count, tick_count)
    });

    channel_tasks.push(task);
  }

  // Monitor overall system health
  let health_task = tokio::spawn(async move {
    loop {
      sleep(Duration::from_secs(15)).await;

      println!("\n🏥 System Health Check:");
      println!("   All connections active ✅");
      println!("   Parsers running ✅");
      println!("   Memory usage optimized ✅");
    }
  });

  // Run for demonstration period
  println!(
    "\n📈 Monitoring performance for 60 seconds (Ctrl+C to stop early)..."
  );

  let demo_duration = Duration::from_secs(60);
  let demo_start = Instant::now();

  // Wait for demo duration or Ctrl+C
  tokio::select! {
      _ = sleep(demo_duration) => {
          println!("\n⏰ Demo duration completed");
      }
      _ = tokio::signal::ctrl_c() => {
          println!("\n🛑 Received Ctrl+C, stopping...");
      }
  }

  // Abort monitoring tasks
  health_task.abort();
  for task in channel_tasks {
    task.abort();
  }

  // Get final statistics
  println!("\n📊 Final Statistics:");

  if let Ok(stats) = manager.get_stats().await {
    println!("   Total runtime: {:?}", demo_start.elapsed());
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
    println!("   Total errors: {}", stats.total_errors);

    for (i, conn_stats) in stats.connection_stats.iter().enumerate() {
      println!(
        "   Connection {}: {} symbols, {} messages, {} errors",
        i,
        conn_stats.symbol_count,
        conn_stats.messages_received,
        conn_stats.errors_count
      );
    }
  }

  let processor_stats = manager.get_processor_stats().await;
  println!("\n🔧 Parser Performance:");
  for (channel_id, stats) in processor_stats {
    println!(
      "   {:?}: {:.1} msg/sec, {:?} avg latency",
      channel_id, stats.messages_per_second, stats.processing_latency_avg
    );
  }

  // Stop the manager
  println!("\n🛑 Stopping manager...");
  manager.stop().await?;

  println!("🏁 Demo completed successfully!");
  Ok(())
}

examples/dynamic_subscription_demo.rs (line 640)

async fn demo_dynamic_subscription(
  manager: &mut KiteTickerManager,
) -> Result<(), String> {
  println!("\n🎯 True Dynamic Subscription Demo");
  println!("==================================");

  // Start with a small initial set
  let initial_symbols = vec![256265, 265, 256777]; // 3 symbols
  let additional_batch_1 = vec![274441, 260105, 273929]; // 3 more symbols
  let additional_batch_2 = vec![260617, 257033, 257289, 257545]; // 4 more symbols
  let symbols_to_remove = vec![265, 274441]; // Remove some symbols
  let final_batch = vec![257801, 258825]; // Add final symbols

  // Step 1: Initial subscription with small set
  println!(
    "\n📊 Step 1: Initial subscription to {} symbols",
    initial_symbols.len()
  );
  println!("Starting with: {:?}", initial_symbols);

  // Start listening for ticks BEFORE subscribing
  let channels_before_sub = manager.get_all_channels();
  let mut tick_listeners = Vec::new();

  for (channel_id, mut receiver) in channels_before_sub {
    let task = tokio::spawn(async move {
      let start = std::time::Instant::now();
      let mut tick_count = 0;

      // Listen for initial ticks for 5 seconds
      while start.elapsed() < Duration::from_secs(5) {
        match timeout(Duration::from_millis(500), receiver.recv()).await {
          Ok(Ok(message)) => {
            if let TickerMessage::Ticks(ticks) = message {
              tick_count += ticks.len();
              println!(
                "🎯 {:?}: Received {} ticks immediately after subscription!",
                channel_id,
                ticks.len()
              );

              // Debug: Show detailed tick information
              debug!("Processing {} ticks on {:?}", ticks.len(), channel_id);

              for (idx, tick) in ticks.iter().enumerate() {
                println!("  🔹 Tick {}: Symbol: {}, LTP: {:?}, Volume: {:?}, Change: {:?}",
                                    idx + 1,
                                    tick.instrument_token,
                                    tick.content.last_price,
                                    tick.content.volume_traded,
                                    tick.content.net_change
                                );

                // Debug: Show tick structure details
                debug!(
                  "Tick {} Details for symbol {}:",
                  idx + 1,
                  tick.instrument_token
                );
                debug!("  - Raw instrument_token: {}", tick.instrument_token);
                debug!("  - Raw last_price: {:?}", tick.content.last_price);
                debug!(
                  "  - Raw volume_traded: {:?}",
                  tick.content.volume_traded
                );
                debug!("  - Raw change: {:?}", tick.content.net_change);
                debug!("  - Raw last_quantity: <not available>");
                debug!(
                  "  - Raw average_price: {:?}",
                  tick.content.avg_traded_price
                );
                debug!(
                  "  - Raw buy_quantity: {:?}",
                  tick.content.total_buy_qty
                );
                debug!(
                  "  - Raw sell_quantity: {:?}",
                  tick.content.total_sell_qty
                );
                debug!("  - Raw oi: {:?}", tick.content.oi);
                debug!("  - Raw mode: {:?}", tick.content.mode);

                if let Some(ohlc) = &tick.content.ohlc {
                  debug!(
                    "  - OHLC available: O:{} H:{} L:{} C:{}",
                    ohlc.open, ohlc.high, ohlc.low, ohlc.close
                  );
                }

                if let Some(depth) = &tick.content.depth {
                  debug!(
                    "  - Market depth available: {} buy, {} sell",
                    depth.buy.len(),
                    depth.sell.len()
                  );
                }
              }
            }
          }
          _ => continue,
        }
      }
      (channel_id, tick_count)
    });
    tick_listeners.push(task);
  }

  // Now subscribe to symbols
  manager
    .subscribe_symbols(&initial_symbols, Some(Mode::LTP))
    .await?;

  println!("✅ Subscription sent, waiting for initial ticks...");

  // Wait for the listeners to finish
  for task in tick_listeners {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "📊 {:?}: Received {} total ticks during initial subscription",
        channel_id, count
      );
    }
  }

  print_distribution(manager, "After initial subscription").await;

  // Step 2: Wait and monitor initial data
  println!("\n⏳ Step 2: Monitoring initial data flow");
  monitor_ticks_briefly(manager, 5, "Initial Subscription Data").await;

  if let Ok(stats) = manager.get_stats().await {
    println!("✅ Current Statistics:");
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
  }

  // Step 3: Dynamic addition - Batch 1
  println!(
    "\n➕ Step 3: DYNAMIC ADDITION - Adding {} more symbols",
    additional_batch_1.len()
  );
  println!("Adding: {:?}", additional_batch_1);
  manager
    .subscribe_symbols(&additional_batch_1, Some(Mode::Quote))
    .await?;

  // Give time for new tick data to arrive
  sleep(Duration::from_secs(2)).await;

  print_distribution(manager, "After first dynamic addition").await;
  monitor_ticks_briefly(manager, 3, "After First Addition").await;

  // Step 4: Dynamic addition - Batch 2
  println!(
    "\n➕ Step 4: DYNAMIC ADDITION - Adding {} more symbols",
    additional_batch_2.len()
  );
  println!("Adding: {:?}", additional_batch_2);
  manager
    .subscribe_symbols(&additional_batch_2, Some(Mode::Full))
    .await?;

  // Give time for new tick data to arrive
  sleep(Duration::from_secs(2)).await;

  print_distribution(manager, "After second dynamic addition").await;
  monitor_ticks_briefly(manager, 3, "After Second Addition").await;

  // Step 5: Dynamic removal
  println!(
    "\n➖ Step 5: DYNAMIC REMOVAL - Removing {} symbols",
    symbols_to_remove.len()
  );
  println!("Removing: {:?}", symbols_to_remove);
  match manager.unsubscribe_symbols(&symbols_to_remove).await {
    Ok(()) => {
      print_distribution(manager, "After dynamic removal").await;
      println!("✅ Dynamic removal successful!");
    }
    Err(e) => {
      println!("⚠️  Dynamic removal failed: {}", e);
    }
  }
  monitor_ticks_briefly(manager, 3, "After Symbol Removal").await;

  // Step 6: Final addition and mode change demo
  println!(
    "\n➕ Step 6: FINAL ADDITION - Adding {} symbols",
    final_batch.len()
  );
  println!("Adding: {:?}", final_batch);
  manager
    .subscribe_symbols(&final_batch, Some(Mode::LTP))
    .await?;

  print_distribution(manager, "After final addition").await;

  // Step 7: Mode change demonstration
  println!("\n🔄 Step 7: MODE CHANGE - Changing subscription mode");
  let symbols_for_mode_change = vec![256265, 260105]; // Change some existing symbols to Full mode
  println!("Changing {:?} to Full mode", symbols_for_mode_change);
  match manager
    .change_mode(&symbols_for_mode_change, Mode::Full)
    .await
  {
    Ok(()) => println!("✅ Mode change successful!"),
    Err(e) => println!("⚠️  Mode change failed: {}", e),
  }

  // Step 8: Final statistics and monitoring
  println!("\n📈 Step 8: Final Statistics and Performance");
  sleep(Duration::from_secs(3)).await; // Let some data flow

  if let Ok(stats) = manager.get_stats().await {
    println!("✅ Final Manager Statistics:");
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
    println!("   Total errors: {}", stats.total_errors);

    for (i, conn_stats) in stats.connection_stats.iter().enumerate() {
      println!(
        "   Connection {}: {} symbols, {} messages",
        i + 1,
        conn_stats.symbol_count,
        conn_stats.messages_received
      );
    }
  }

  // Step 9: Show processor performance
  println!("\n⚡ Step 9: Final Parser Performance");
  let processor_stats = manager.get_processor_stats().await;
  for (channel_id, stats) in processor_stats {
    println!(
      "   {:?}: {:.1} msg/sec, {:?} avg latency",
      channel_id, stats.messages_per_second, stats.processing_latency_avg
    );
  }

  // Step 10: Monitor live data for a short period
  println!("\n📺 Step 10: Live Data Monitoring (10 seconds)");
  println!("Monitoring real-time tick data from all dynamically managed connections...");

  let channels = manager.get_all_channels();
  let mut tasks = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut count = 0;
      let start = std::time::Instant::now();

      while start.elapsed() < Duration::from_secs(10) {
        match timeout(Duration::from_secs(2), receiver.recv()).await {
          Ok(Ok(message)) => {
            count += 1;
            if let TickerMessage::Ticks(ticks) = message {
              println!("📋 {:?}: {} ticks received", channel_id, ticks.len());

              // Debug: Final monitoring with comprehensive details
              debug!(
                "Final monitoring - Message #{}, {} ticks on {:?}",
                count,
                ticks.len(),
                channel_id
              );

              for tick in &ticks {
                println!(
                  "  🔹 Symbol: {}, LTP: {:?}, Volume: {:?}, Change: {:?}",
                  tick.instrument_token,
                  tick.content.last_price,
                  tick.content.volume_traded,
                  tick.content.net_change
                );

                // Debug: Show complete tick analysis
                debug!(
                  "Final Debug Analysis for Symbol {}:",
                  tick.instrument_token
                );
                debug!(
                  "  - Timestamp: Now = {:?}",
                  std::time::SystemTime::now()
                );
                debug!("  - Data completeness check:");
                debug!("    * Last Price: {:?} (✅)", tick.content.last_price);
                debug!(
                  "    * Volume: {:?} ({})",
                  tick.content.volume_traded,
                  if tick.content.volume_traded.is_some() {
                    "✅"
                  } else {
                    "❌"
                  }
                );
                debug!(
                  "    * Change: {:?} ({})",
                  tick.content.net_change,
                  if tick.content.net_change.is_some() {
                    "✅"
                  } else {
                    "❌"
                  }
                );
                debug!("    * Mode: {:?} (✅)", tick.content.mode);

                // Validate mode-specific data
                match tick.content.mode {
                  Mode::Full => {
                    debug!("  - Full mode validation:");
                    debug!(
                      "    * OHLC: {}",
                      if tick.content.ohlc.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    debug!(
                      "    * Depth: {}",
                      if tick.content.depth.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    if let Some(ohlc) = &tick.content.ohlc {
                      debug!(
                        "    * OHLC Values: O:{} H:{} L:{} C:{}",
                        ohlc.open, ohlc.high, ohlc.low, ohlc.close
                      );
                    }
                    if let Some(depth) = &tick.content.depth {
                      debug!(
                        "    * Depth Levels: {} buy, {} sell",
                        depth.buy.len(),
                        depth.sell.len()
                      );
                    }
                  }
                  Mode::Quote => {
                    debug!("  - Quote mode validation:");
                    debug!(
                      "    * OHLC: {}",
                      if tick.content.ohlc.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    if let Some(ohlc) = &tick.content.ohlc {
                      debug!(
                        "    * OHLC Values: O:{} H:{} L:{} C:{}",
                        ohlc.open, ohlc.high, ohlc.low, ohlc.close
                      );
                    }
                  }
                  Mode::LTP => {
                    debug!("  - LTP mode validation: ✅ Basic data only");
                  }
                }
              }
            } else {
              println!("📋 {:?}: Non-tick message received", channel_id);
              debug!("Message type: {:?}", message);
            }
          }
          _ => continue,
        }
      }
      (channel_id, count)
    });
    tasks.push(task);
  }

  // Wait for monitoring to complete
  for task in tasks {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "📊 {:?}: {} total messages in 10 seconds",
        channel_id, count
      );
    }
  }

  // Final cleanup demonstration
  println!("\n🧹 Step 11: Final Cleanup Demonstration");
  let current_symbols: Vec<u32> = manager
    .get_symbol_distribution()
    .values()
    .flat_map(|symbols| symbols.iter().cloned())
    .collect();

  println!(
    "Attempting to unsubscribe from {} remaining symbols...",
    current_symbols.len()
  );

  match manager.unsubscribe_symbols(&current_symbols).await {
    Ok(()) => {
      print_distribution(manager, "After complete cleanup").await;
      println!("✅ Complete cleanup successful!");
    }
    Err(e) => {
      println!("⚠️  Cleanup encountered issues: {}", e);
      println!("💡 This demonstrates the current architecture capabilities");
    }
  }

  println!("\n✅ Dynamic Subscription Demo Completed!");
  println!("🎯 Key Dynamic Features Demonstrated:");
  println!("   ✅ Runtime symbol addition across multiple batches");
  println!("   ✅ Runtime symbol removal with proper tracking");
  println!("   ✅ Dynamic mode changes for existing symbols");
  println!("   ✅ Real-time capacity distribution and monitoring");
  println!("   ✅ Independent message processing per connection");
  println!("   ✅ High-performance parser tasks with statistics");
  println!("   ✅ Complete subscription lifecycle management");

  Ok(())
}

async fn print_distribution(manager: &KiteTickerManager, context: &str) {
  let distribution = manager.get_symbol_distribution();
  println!("\n📈 Symbol Distribution ({}):", context);

  let mut total = 0;
  for (channel_id, symbols) in &distribution {
    println!("   {:?}: {} symbols", channel_id, symbols.len());
    total += symbols.len();
  }
  println!("   Total: {} symbols", total);
}

pub async fn unsubscribe_symbols( &mut self, symbols: &[u32], ) -> Result<(), String>

Unsubscribe from symbols

Examples found in repository

examples/basic/runtime_subscription_example.rs (line 87)

async fn runtime_subscription_demo(
  manager: &mut KiteTickerManager,
) -> Result<(), String> {
  println!("\n📡 Runtime Subscription Management Demo");
  println!("=====================================");

  // 1. Start with initial symbols
  println!("\n1️⃣ Initial subscription to base symbols");
  let initial_symbols = vec![256265, 265, 256777];
  manager
    .subscribe_symbols(&initial_symbols, Some(Mode::LTP))
    .await?;
  print_current_state(manager, "Initial subscription").await;

  sleep(Duration::from_secs(2)).await;

  // 2. Add more symbols dynamically
  println!("\n2️⃣ Adding symbols dynamically");
  let additional_symbols = vec![274441, 260105, 273929];
  manager
    .subscribe_symbols(&additional_symbols, Some(Mode::Quote))
    .await?;
  print_current_state(manager, "After adding symbols").await;

  sleep(Duration::from_secs(2)).await;

  // 3. Change mode for existing symbols
  println!("\n3️⃣ Changing subscription mode");
  let symbols_for_mode_change = vec![256265, 265];
  manager
    .change_mode(&symbols_for_mode_change, Mode::Full)
    .await?;
  print_current_state(manager, "After mode change").await;

  sleep(Duration::from_secs(2)).await;

  // 4. Remove some symbols
  println!("\n4️⃣ Removing symbols dynamically");
  let symbols_to_remove = vec![265, 274441];
  manager.unsubscribe_symbols(&symbols_to_remove).await?;
  print_current_state(manager, "After removing symbols").await;

  sleep(Duration::from_secs(2)).await;

  // 5. Add different symbols with different modes
  println!("\n5️⃣ Adding new symbols with Full mode");
  let full_mode_symbols = vec![257801, 258825];
  manager
    .subscribe_symbols(&full_mode_symbols, Some(Mode::Full))
    .await?;
  print_current_state(manager, "After adding Full mode symbols").await;

  // 6. Monitor live data for a short period
  println!("\n6️⃣ Monitoring live data (5 seconds)");
  monitor_live_data(manager, 5).await?;

  // 7. Complete cleanup
  println!("\n7️⃣ Complete cleanup");
  let all_symbols: Vec<u32> = manager
    .get_symbol_distribution()
    .values()
    .flat_map(|symbols| symbols.iter().cloned())
    .collect();

  if !all_symbols.is_empty() {
    manager.unsubscribe_symbols(&all_symbols).await?;
    print_current_state(manager, "After complete cleanup").await;
  }

  println!("\n✅ Runtime subscription demo completed!");
  Ok(())
}

examples/dynamic_subscription_demo.rs (line 425)

async fn demo_dynamic_subscription(
  manager: &mut KiteTickerManager,
) -> Result<(), String> {
  println!("\n🎯 True Dynamic Subscription Demo");
  println!("==================================");

  // Start with a small initial set
  let initial_symbols = vec![256265, 265, 256777]; // 3 symbols
  let additional_batch_1 = vec![274441, 260105, 273929]; // 3 more symbols
  let additional_batch_2 = vec![260617, 257033, 257289, 257545]; // 4 more symbols
  let symbols_to_remove = vec![265, 274441]; // Remove some symbols
  let final_batch = vec![257801, 258825]; // Add final symbols

  // Step 1: Initial subscription with small set
  println!(
    "\n📊 Step 1: Initial subscription to {} symbols",
    initial_symbols.len()
  );
  println!("Starting with: {:?}", initial_symbols);

  // Start listening for ticks BEFORE subscribing
  let channels_before_sub = manager.get_all_channels();
  let mut tick_listeners = Vec::new();

  for (channel_id, mut receiver) in channels_before_sub {
    let task = tokio::spawn(async move {
      let start = std::time::Instant::now();
      let mut tick_count = 0;

      // Listen for initial ticks for 5 seconds
      while start.elapsed() < Duration::from_secs(5) {
        match timeout(Duration::from_millis(500), receiver.recv()).await {
          Ok(Ok(message)) => {
            if let TickerMessage::Ticks(ticks) = message {
              tick_count += ticks.len();
              println!(
                "🎯 {:?}: Received {} ticks immediately after subscription!",
                channel_id,
                ticks.len()
              );

              // Debug: Show detailed tick information
              debug!("Processing {} ticks on {:?}", ticks.len(), channel_id);

              for (idx, tick) in ticks.iter().enumerate() {
                println!("  🔹 Tick {}: Symbol: {}, LTP: {:?}, Volume: {:?}, Change: {:?}",
                                    idx + 1,
                                    tick.instrument_token,
                                    tick.content.last_price,
                                    tick.content.volume_traded,
                                    tick.content.net_change
                                );

                // Debug: Show tick structure details
                debug!(
                  "Tick {} Details for symbol {}:",
                  idx + 1,
                  tick.instrument_token
                );
                debug!("  - Raw instrument_token: {}", tick.instrument_token);
                debug!("  - Raw last_price: {:?}", tick.content.last_price);
                debug!(
                  "  - Raw volume_traded: {:?}",
                  tick.content.volume_traded
                );
                debug!("  - Raw change: {:?}", tick.content.net_change);
                debug!("  - Raw last_quantity: <not available>");
                debug!(
                  "  - Raw average_price: {:?}",
                  tick.content.avg_traded_price
                );
                debug!(
                  "  - Raw buy_quantity: {:?}",
                  tick.content.total_buy_qty
                );
                debug!(
                  "  - Raw sell_quantity: {:?}",
                  tick.content.total_sell_qty
                );
                debug!("  - Raw oi: {:?}", tick.content.oi);
                debug!("  - Raw mode: {:?}", tick.content.mode);

                if let Some(ohlc) = &tick.content.ohlc {
                  debug!(
                    "  - OHLC available: O:{} H:{} L:{} C:{}",
                    ohlc.open, ohlc.high, ohlc.low, ohlc.close
                  );
                }

                if let Some(depth) = &tick.content.depth {
                  debug!(
                    "  - Market depth available: {} buy, {} sell",
                    depth.buy.len(),
                    depth.sell.len()
                  );
                }
              }
            }
          }
          _ => continue,
        }
      }
      (channel_id, tick_count)
    });
    tick_listeners.push(task);
  }

  // Now subscribe to symbols
  manager
    .subscribe_symbols(&initial_symbols, Some(Mode::LTP))
    .await?;

  println!("✅ Subscription sent, waiting for initial ticks...");

  // Wait for the listeners to finish
  for task in tick_listeners {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "📊 {:?}: Received {} total ticks during initial subscription",
        channel_id, count
      );
    }
  }

  print_distribution(manager, "After initial subscription").await;

  // Step 2: Wait and monitor initial data
  println!("\n⏳ Step 2: Monitoring initial data flow");
  monitor_ticks_briefly(manager, 5, "Initial Subscription Data").await;

  if let Ok(stats) = manager.get_stats().await {
    println!("✅ Current Statistics:");
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
  }

  // Step 3: Dynamic addition - Batch 1
  println!(
    "\n➕ Step 3: DYNAMIC ADDITION - Adding {} more symbols",
    additional_batch_1.len()
  );
  println!("Adding: {:?}", additional_batch_1);
  manager
    .subscribe_symbols(&additional_batch_1, Some(Mode::Quote))
    .await?;

  // Give time for new tick data to arrive
  sleep(Duration::from_secs(2)).await;

  print_distribution(manager, "After first dynamic addition").await;
  monitor_ticks_briefly(manager, 3, "After First Addition").await;

  // Step 4: Dynamic addition - Batch 2
  println!(
    "\n➕ Step 4: DYNAMIC ADDITION - Adding {} more symbols",
    additional_batch_2.len()
  );
  println!("Adding: {:?}", additional_batch_2);
  manager
    .subscribe_symbols(&additional_batch_2, Some(Mode::Full))
    .await?;

  // Give time for new tick data to arrive
  sleep(Duration::from_secs(2)).await;

  print_distribution(manager, "After second dynamic addition").await;
  monitor_ticks_briefly(manager, 3, "After Second Addition").await;

  // Step 5: Dynamic removal
  println!(
    "\n➖ Step 5: DYNAMIC REMOVAL - Removing {} symbols",
    symbols_to_remove.len()
  );
  println!("Removing: {:?}", symbols_to_remove);
  match manager.unsubscribe_symbols(&symbols_to_remove).await {
    Ok(()) => {
      print_distribution(manager, "After dynamic removal").await;
      println!("✅ Dynamic removal successful!");
    }
    Err(e) => {
      println!("⚠️  Dynamic removal failed: {}", e);
    }
  }
  monitor_ticks_briefly(manager, 3, "After Symbol Removal").await;

  // Step 6: Final addition and mode change demo
  println!(
    "\n➕ Step 6: FINAL ADDITION - Adding {} symbols",
    final_batch.len()
  );
  println!("Adding: {:?}", final_batch);
  manager
    .subscribe_symbols(&final_batch, Some(Mode::LTP))
    .await?;

  print_distribution(manager, "After final addition").await;

  // Step 7: Mode change demonstration
  println!("\n🔄 Step 7: MODE CHANGE - Changing subscription mode");
  let symbols_for_mode_change = vec![256265, 260105]; // Change some existing symbols to Full mode
  println!("Changing {:?} to Full mode", symbols_for_mode_change);
  match manager
    .change_mode(&symbols_for_mode_change, Mode::Full)
    .await
  {
    Ok(()) => println!("✅ Mode change successful!"),
    Err(e) => println!("⚠️  Mode change failed: {}", e),
  }

  // Step 8: Final statistics and monitoring
  println!("\n📈 Step 8: Final Statistics and Performance");
  sleep(Duration::from_secs(3)).await; // Let some data flow

  if let Ok(stats) = manager.get_stats().await {
    println!("✅ Final Manager Statistics:");
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
    println!("   Total errors: {}", stats.total_errors);

    for (i, conn_stats) in stats.connection_stats.iter().enumerate() {
      println!(
        "   Connection {}: {} symbols, {} messages",
        i + 1,
        conn_stats.symbol_count,
        conn_stats.messages_received
      );
    }
  }

  // Step 9: Show processor performance
  println!("\n⚡ Step 9: Final Parser Performance");
  let processor_stats = manager.get_processor_stats().await;
  for (channel_id, stats) in processor_stats {
    println!(
      "   {:?}: {:.1} msg/sec, {:?} avg latency",
      channel_id, stats.messages_per_second, stats.processing_latency_avg
    );
  }

  // Step 10: Monitor live data for a short period
  println!("\n📺 Step 10: Live Data Monitoring (10 seconds)");
  println!("Monitoring real-time tick data from all dynamically managed connections...");

  let channels = manager.get_all_channels();
  let mut tasks = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut count = 0;
      let start = std::time::Instant::now();

      while start.elapsed() < Duration::from_secs(10) {
        match timeout(Duration::from_secs(2), receiver.recv()).await {
          Ok(Ok(message)) => {
            count += 1;
            if let TickerMessage::Ticks(ticks) = message {
              println!("📋 {:?}: {} ticks received", channel_id, ticks.len());

              // Debug: Final monitoring with comprehensive details
              debug!(
                "Final monitoring - Message #{}, {} ticks on {:?}",
                count,
                ticks.len(),
                channel_id
              );

              for tick in &ticks {
                println!(
                  "  🔹 Symbol: {}, LTP: {:?}, Volume: {:?}, Change: {:?}",
                  tick.instrument_token,
                  tick.content.last_price,
                  tick.content.volume_traded,
                  tick.content.net_change
                );

                // Debug: Show complete tick analysis
                debug!(
                  "Final Debug Analysis for Symbol {}:",
                  tick.instrument_token
                );
                debug!(
                  "  - Timestamp: Now = {:?}",
                  std::time::SystemTime::now()
                );
                debug!("  - Data completeness check:");
                debug!("    * Last Price: {:?} (✅)", tick.content.last_price);
                debug!(
                  "    * Volume: {:?} ({})",
                  tick.content.volume_traded,
                  if tick.content.volume_traded.is_some() {
                    "✅"
                  } else {
                    "❌"
                  }
                );
                debug!(
                  "    * Change: {:?} ({})",
                  tick.content.net_change,
                  if tick.content.net_change.is_some() {
                    "✅"
                  } else {
                    "❌"
                  }
                );
                debug!("    * Mode: {:?} (✅)", tick.content.mode);

                // Validate mode-specific data
                match tick.content.mode {
                  Mode::Full => {
                    debug!("  - Full mode validation:");
                    debug!(
                      "    * OHLC: {}",
                      if tick.content.ohlc.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    debug!(
                      "    * Depth: {}",
                      if tick.content.depth.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    if let Some(ohlc) = &tick.content.ohlc {
                      debug!(
                        "    * OHLC Values: O:{} H:{} L:{} C:{}",
                        ohlc.open, ohlc.high, ohlc.low, ohlc.close
                      );
                    }
                    if let Some(depth) = &tick.content.depth {
                      debug!(
                        "    * Depth Levels: {} buy, {} sell",
                        depth.buy.len(),
                        depth.sell.len()
                      );
                    }
                  }
                  Mode::Quote => {
                    debug!("  - Quote mode validation:");
                    debug!(
                      "    * OHLC: {}",
                      if tick.content.ohlc.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    if let Some(ohlc) = &tick.content.ohlc {
                      debug!(
                        "    * OHLC Values: O:{} H:{} L:{} C:{}",
                        ohlc.open, ohlc.high, ohlc.low, ohlc.close
                      );
                    }
                  }
                  Mode::LTP => {
                    debug!("  - LTP mode validation: ✅ Basic data only");
                  }
                }
              }
            } else {
              println!("📋 {:?}: Non-tick message received", channel_id);
              debug!("Message type: {:?}", message);
            }
          }
          _ => continue,
        }
      }
      (channel_id, count)
    });
    tasks.push(task);
  }

  // Wait for monitoring to complete
  for task in tasks {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "📊 {:?}: {} total messages in 10 seconds",
        channel_id, count
      );
    }
  }

  // Final cleanup demonstration
  println!("\n🧹 Step 11: Final Cleanup Demonstration");
  let current_symbols: Vec<u32> = manager
    .get_symbol_distribution()
    .values()
    .flat_map(|symbols| symbols.iter().cloned())
    .collect();

  println!(
    "Attempting to unsubscribe from {} remaining symbols...",
    current_symbols.len()
  );

  match manager.unsubscribe_symbols(&current_symbols).await {
    Ok(()) => {
      print_distribution(manager, "After complete cleanup").await;
      println!("✅ Complete cleanup successful!");
    }
    Err(e) => {
      println!("⚠️  Cleanup encountered issues: {}", e);
      println!("💡 This demonstrates the current architecture capabilities");
    }
  }

  println!("\n✅ Dynamic Subscription Demo Completed!");
  println!("🎯 Key Dynamic Features Demonstrated:");
  println!("   ✅ Runtime symbol addition across multiple batches");
  println!("   ✅ Runtime symbol removal with proper tracking");
  println!("   ✅ Dynamic mode changes for existing symbols");
  println!("   ✅ Real-time capacity distribution and monitoring");
  println!("   ✅ Independent message processing per connection");
  println!("   ✅ High-performance parser tasks with statistics");
  println!("   ✅ Complete subscription lifecycle management");

  Ok(())
}

pub async fn change_mode( &mut self, symbols: &[u32], mode: Mode, ) -> Result<(), String>

Dynamically change subscription mode for existing symbols

Examples found in repository

examples/basic/runtime_subscription_example.rs (line 78)

async fn runtime_subscription_demo(
  manager: &mut KiteTickerManager,
) -> Result<(), String> {
  println!("\n📡 Runtime Subscription Management Demo");
  println!("=====================================");

  // 1. Start with initial symbols
  println!("\n1️⃣ Initial subscription to base symbols");
  let initial_symbols = vec![256265, 265, 256777];
  manager
    .subscribe_symbols(&initial_symbols, Some(Mode::LTP))
    .await?;
  print_current_state(manager, "Initial subscription").await;

  sleep(Duration::from_secs(2)).await;

  // 2. Add more symbols dynamically
  println!("\n2️⃣ Adding symbols dynamically");
  let additional_symbols = vec![274441, 260105, 273929];
  manager
    .subscribe_symbols(&additional_symbols, Some(Mode::Quote))
    .await?;
  print_current_state(manager, "After adding symbols").await;

  sleep(Duration::from_secs(2)).await;

  // 3. Change mode for existing symbols
  println!("\n3️⃣ Changing subscription mode");
  let symbols_for_mode_change = vec![256265, 265];
  manager
    .change_mode(&symbols_for_mode_change, Mode::Full)
    .await?;
  print_current_state(manager, "After mode change").await;

  sleep(Duration::from_secs(2)).await;

  // 4. Remove some symbols
  println!("\n4️⃣ Removing symbols dynamically");
  let symbols_to_remove = vec![265, 274441];
  manager.unsubscribe_symbols(&symbols_to_remove).await?;
  print_current_state(manager, "After removing symbols").await;

  sleep(Duration::from_secs(2)).await;

  // 5. Add different symbols with different modes
  println!("\n5️⃣ Adding new symbols with Full mode");
  let full_mode_symbols = vec![257801, 258825];
  manager
    .subscribe_symbols(&full_mode_symbols, Some(Mode::Full))
    .await?;
  print_current_state(manager, "After adding Full mode symbols").await;

  // 6. Monitor live data for a short period
  println!("\n6️⃣ Monitoring live data (5 seconds)");
  monitor_live_data(manager, 5).await?;

  // 7. Complete cleanup
  println!("\n7️⃣ Complete cleanup");
  let all_symbols: Vec<u32> = manager
    .get_symbol_distribution()
    .values()
    .flat_map(|symbols| symbols.iter().cloned())
    .collect();

  if !all_symbols.is_empty() {
    manager.unsubscribe_symbols(&all_symbols).await?;
    print_current_state(manager, "After complete cleanup").await;
  }

  println!("\n✅ Runtime subscription demo completed!");
  Ok(())
}

examples/mode_change_test.rs (line 131)

async fn test_mode_change_issue(
  manager: &mut KiteTickerManager,
) -> Result<(), String> {
  println!("\n🧪 Testing Mode Change Issue");
  println!("=============================");

  // Step 1: Subscribe to a symbol with LTP mode
  let test_symbol = 738561; // Test symbol provided by user
  println!("\n1️⃣ Subscribing to symbol {} with LTP mode", test_symbol);
  manager
    .subscribe_symbols(&[test_symbol], Some(Mode::LTP))
    .await?;

  // Start monitoring ticks to see the mode
  let channels = manager.get_all_channels();
  let mut tick_listeners = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut ticks_received = 0;
      let start = Instant::now();

      while start.elapsed() < Duration::from_secs(5) && ticks_received < 3 {
        match timeout(Duration::from_millis(500), receiver.recv()).await {
          Ok(Ok(message)) => {
            if let TickerMessage::Ticks(ticks) = message {
              for tick in &ticks {
                if tick.instrument_token == test_symbol {
                  ticks_received += 1;
                  println!(
                    "   📊 Received tick for {}: Mode={:?}, LTP={:?}",
                    tick.instrument_token,
                    tick.content.mode,
                    tick.content.last_price
                  );

                  // Check if OHLC data is present (should not be for LTP mode)
                  if tick.content.ohlc.is_some() {
                    println!(
                      "   ⚠️  OHLC data present in LTP mode - unexpected!"
                    );
                  } else {
                    println!("   ✅ No OHLC data in LTP mode - correct");
                  }
                }
              }
            }
          }
          _ => continue,
        }
      }
      (channel_id, ticks_received)
    });
    tick_listeners.push(task);
  }

  // Wait for initial ticks
  for task in tick_listeners {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "   📈 {:?}: Received {} ticks for initial LTP subscription",
        channel_id, count
      );
    }
  }

  sleep(Duration::from_secs(2)).await;

  // Step 2: Attempt to change mode to Full (this is where the issue should manifest)
  println!(
    "\n2️⃣ Attempting to change mode from LTP to Full for symbol {}",
    test_symbol
  );
  match manager.change_mode(&[test_symbol], Mode::Full).await {
    Ok(()) => {
      println!("   ✅ Mode change command sent successfully");
    }
    Err(e) => {
      println!("   ❌ Mode change command failed: {}", e);
      return Err(e);
    }
  }

  // Step 3: Monitor ticks after mode change to see if it actually worked
  println!("\n3️⃣ Monitoring ticks after mode change...");
  let channels = manager.get_all_channels();
  let mut post_change_listeners = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut ticks_received = 0;
      let mut ohlc_present = 0;
      let mut depth_present = 0;
      let start = Instant::now();

      while start.elapsed() < Duration::from_secs(10) && ticks_received < 5 {
        match timeout(Duration::from_millis(500), receiver.recv()).await {
          Ok(Ok(message)) => {
            if let TickerMessage::Ticks(ticks) = message {
              for tick in &ticks {
                if tick.instrument_token == test_symbol {
                  ticks_received += 1;
                  println!(
                    "   📊 Post-change tick for {}: Mode={:?}, LTP={:?}",
                    tick.instrument_token,
                    tick.content.mode,
                    tick.content.last_price
                  );

                  // Check if Full mode data is present
                  if let Some(ohlc) = &tick.content.ohlc {
                    ohlc_present += 1;
                    println!(
                      "   ✅ OHLC data present: O:{} H:{} L:{} C:{}",
                      ohlc.open, ohlc.high, ohlc.low, ohlc.close
                    );
                  } else {
                    println!("   ❌ OHLC data missing - mode change may not have worked!");
                  }

                  if let Some(depth) = &tick.content.depth {
                    depth_present += 1;
                    println!(
                      "   ✅ Market depth present: {} buy, {} sell orders",
                      depth.buy.len(),
                      depth.sell.len()
                    );
                  } else {
                    println!("   ❌ Market depth missing - mode change may not have worked!");
                  }

                  // Log the actual mode reported in the tick
                  info!("Tick mode reported: {:?}", tick.content.mode);
                }
              }
            }
          }
          _ => continue,
        }
      }
      (channel_id, ticks_received, ohlc_present, depth_present)
    });
    post_change_listeners.push(task);
  }

  // Wait for post-change ticks
  let mut total_ticks = 0;
  let mut total_ohlc = 0;
  let mut total_depth = 0;

  for task in post_change_listeners {
    if let Ok((channel_id, ticks, ohlc, depth)) = task.await {
      println!(
        "   📈 {:?}: {} ticks, {} with OHLC, {} with depth",
        channel_id, ticks, ohlc, depth
      );
      total_ticks += ticks;
      total_ohlc += ohlc;
      total_depth += depth;
    }
  }

  // Step 4: Analyze results
  println!("\n4️⃣ Test Results Analysis");
  println!("========================");

  if total_ticks == 0 {
    println!("   ⚠️  No ticks received after mode change - connection issue?");
  } else if total_ohlc == 0 && total_depth == 0 {
    println!("   ❌ ISSUE CONFIRMED: Mode change command sent but Full mode data not received");
    println!("   💡 This confirms that set_mode() alone doesn't work - need unsubscribe+resubscribe");
  } else if total_ohlc > 0 && total_depth > 0 {
    println!("   ✅ Mode change worked - Full mode data received");
  } else {
    println!(
      "   ⚠️  Partial mode change - some Full mode data received but not all"
    );
  }

  println!("\n📊 Summary:");
  println!("   Total ticks received: {}", total_ticks);
  println!("   Ticks with OHLC data: {}", total_ohlc);
  println!("   Ticks with market depth: {}", total_depth);

  Ok(())
}

examples/dynamic_subscription_demo.rs (line 453)

async fn demo_dynamic_subscription(
  manager: &mut KiteTickerManager,
) -> Result<(), String> {
  println!("\n🎯 True Dynamic Subscription Demo");
  println!("==================================");

  // Start with a small initial set
  let initial_symbols = vec![256265, 265, 256777]; // 3 symbols
  let additional_batch_1 = vec![274441, 260105, 273929]; // 3 more symbols
  let additional_batch_2 = vec![260617, 257033, 257289, 257545]; // 4 more symbols
  let symbols_to_remove = vec![265, 274441]; // Remove some symbols
  let final_batch = vec![257801, 258825]; // Add final symbols

  // Step 1: Initial subscription with small set
  println!(
    "\n📊 Step 1: Initial subscription to {} symbols",
    initial_symbols.len()
  );
  println!("Starting with: {:?}", initial_symbols);

  // Start listening for ticks BEFORE subscribing
  let channels_before_sub = manager.get_all_channels();
  let mut tick_listeners = Vec::new();

  for (channel_id, mut receiver) in channels_before_sub {
    let task = tokio::spawn(async move {
      let start = std::time::Instant::now();
      let mut tick_count = 0;

      // Listen for initial ticks for 5 seconds
      while start.elapsed() < Duration::from_secs(5) {
        match timeout(Duration::from_millis(500), receiver.recv()).await {
          Ok(Ok(message)) => {
            if let TickerMessage::Ticks(ticks) = message {
              tick_count += ticks.len();
              println!(
                "🎯 {:?}: Received {} ticks immediately after subscription!",
                channel_id,
                ticks.len()
              );

              // Debug: Show detailed tick information
              debug!("Processing {} ticks on {:?}", ticks.len(), channel_id);

              for (idx, tick) in ticks.iter().enumerate() {
                println!("  🔹 Tick {}: Symbol: {}, LTP: {:?}, Volume: {:?}, Change: {:?}",
                                    idx + 1,
                                    tick.instrument_token,
                                    tick.content.last_price,
                                    tick.content.volume_traded,
                                    tick.content.net_change
                                );

                // Debug: Show tick structure details
                debug!(
                  "Tick {} Details for symbol {}:",
                  idx + 1,
                  tick.instrument_token
                );
                debug!("  - Raw instrument_token: {}", tick.instrument_token);
                debug!("  - Raw last_price: {:?}", tick.content.last_price);
                debug!(
                  "  - Raw volume_traded: {:?}",
                  tick.content.volume_traded
                );
                debug!("  - Raw change: {:?}", tick.content.net_change);
                debug!("  - Raw last_quantity: <not available>");
                debug!(
                  "  - Raw average_price: {:?}",
                  tick.content.avg_traded_price
                );
                debug!(
                  "  - Raw buy_quantity: {:?}",
                  tick.content.total_buy_qty
                );
                debug!(
                  "  - Raw sell_quantity: {:?}",
                  tick.content.total_sell_qty
                );
                debug!("  - Raw oi: {:?}", tick.content.oi);
                debug!("  - Raw mode: {:?}", tick.content.mode);

                if let Some(ohlc) = &tick.content.ohlc {
                  debug!(
                    "  - OHLC available: O:{} H:{} L:{} C:{}",
                    ohlc.open, ohlc.high, ohlc.low, ohlc.close
                  );
                }

                if let Some(depth) = &tick.content.depth {
                  debug!(
                    "  - Market depth available: {} buy, {} sell",
                    depth.buy.len(),
                    depth.sell.len()
                  );
                }
              }
            }
          }
          _ => continue,
        }
      }
      (channel_id, tick_count)
    });
    tick_listeners.push(task);
  }

  // Now subscribe to symbols
  manager
    .subscribe_symbols(&initial_symbols, Some(Mode::LTP))
    .await?;

  println!("✅ Subscription sent, waiting for initial ticks...");

  // Wait for the listeners to finish
  for task in tick_listeners {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "📊 {:?}: Received {} total ticks during initial subscription",
        channel_id, count
      );
    }
  }

  print_distribution(manager, "After initial subscription").await;

  // Step 2: Wait and monitor initial data
  println!("\n⏳ Step 2: Monitoring initial data flow");
  monitor_ticks_briefly(manager, 5, "Initial Subscription Data").await;

  if let Ok(stats) = manager.get_stats().await {
    println!("✅ Current Statistics:");
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
  }

  // Step 3: Dynamic addition - Batch 1
  println!(
    "\n➕ Step 3: DYNAMIC ADDITION - Adding {} more symbols",
    additional_batch_1.len()
  );
  println!("Adding: {:?}", additional_batch_1);
  manager
    .subscribe_symbols(&additional_batch_1, Some(Mode::Quote))
    .await?;

  // Give time for new tick data to arrive
  sleep(Duration::from_secs(2)).await;

  print_distribution(manager, "After first dynamic addition").await;
  monitor_ticks_briefly(manager, 3, "After First Addition").await;

  // Step 4: Dynamic addition - Batch 2
  println!(
    "\n➕ Step 4: DYNAMIC ADDITION - Adding {} more symbols",
    additional_batch_2.len()
  );
  println!("Adding: {:?}", additional_batch_2);
  manager
    .subscribe_symbols(&additional_batch_2, Some(Mode::Full))
    .await?;

  // Give time for new tick data to arrive
  sleep(Duration::from_secs(2)).await;

  print_distribution(manager, "After second dynamic addition").await;
  monitor_ticks_briefly(manager, 3, "After Second Addition").await;

  // Step 5: Dynamic removal
  println!(
    "\n➖ Step 5: DYNAMIC REMOVAL - Removing {} symbols",
    symbols_to_remove.len()
  );
  println!("Removing: {:?}", symbols_to_remove);
  match manager.unsubscribe_symbols(&symbols_to_remove).await {
    Ok(()) => {
      print_distribution(manager, "After dynamic removal").await;
      println!("✅ Dynamic removal successful!");
    }
    Err(e) => {
      println!("⚠️  Dynamic removal failed: {}", e);
    }
  }
  monitor_ticks_briefly(manager, 3, "After Symbol Removal").await;

  // Step 6: Final addition and mode change demo
  println!(
    "\n➕ Step 6: FINAL ADDITION - Adding {} symbols",
    final_batch.len()
  );
  println!("Adding: {:?}", final_batch);
  manager
    .subscribe_symbols(&final_batch, Some(Mode::LTP))
    .await?;

  print_distribution(manager, "After final addition").await;

  // Step 7: Mode change demonstration
  println!("\n🔄 Step 7: MODE CHANGE - Changing subscription mode");
  let symbols_for_mode_change = vec![256265, 260105]; // Change some existing symbols to Full mode
  println!("Changing {:?} to Full mode", symbols_for_mode_change);
  match manager
    .change_mode(&symbols_for_mode_change, Mode::Full)
    .await
  {
    Ok(()) => println!("✅ Mode change successful!"),
    Err(e) => println!("⚠️  Mode change failed: {}", e),
  }

  // Step 8: Final statistics and monitoring
  println!("\n📈 Step 8: Final Statistics and Performance");
  sleep(Duration::from_secs(3)).await; // Let some data flow

  if let Ok(stats) = manager.get_stats().await {
    println!("✅ Final Manager Statistics:");
    println!("   Active connections: {}", stats.active_connections);
    println!("   Total symbols: {}", stats.total_symbols);
    println!("   Total messages: {}", stats.total_messages_received);
    println!("   Total errors: {}", stats.total_errors);

    for (i, conn_stats) in stats.connection_stats.iter().enumerate() {
      println!(
        "   Connection {}: {} symbols, {} messages",
        i + 1,
        conn_stats.symbol_count,
        conn_stats.messages_received
      );
    }
  }

  // Step 9: Show processor performance
  println!("\n⚡ Step 9: Final Parser Performance");
  let processor_stats = manager.get_processor_stats().await;
  for (channel_id, stats) in processor_stats {
    println!(
      "   {:?}: {:.1} msg/sec, {:?} avg latency",
      channel_id, stats.messages_per_second, stats.processing_latency_avg
    );
  }

  // Step 10: Monitor live data for a short period
  println!("\n📺 Step 10: Live Data Monitoring (10 seconds)");
  println!("Monitoring real-time tick data from all dynamically managed connections...");

  let channels = manager.get_all_channels();
  let mut tasks = Vec::new();

  for (channel_id, mut receiver) in channels {
    let task = tokio::spawn(async move {
      let mut count = 0;
      let start = std::time::Instant::now();

      while start.elapsed() < Duration::from_secs(10) {
        match timeout(Duration::from_secs(2), receiver.recv()).await {
          Ok(Ok(message)) => {
            count += 1;
            if let TickerMessage::Ticks(ticks) = message {
              println!("📋 {:?}: {} ticks received", channel_id, ticks.len());

              // Debug: Final monitoring with comprehensive details
              debug!(
                "Final monitoring - Message #{}, {} ticks on {:?}",
                count,
                ticks.len(),
                channel_id
              );

              for tick in &ticks {
                println!(
                  "  🔹 Symbol: {}, LTP: {:?}, Volume: {:?}, Change: {:?}",
                  tick.instrument_token,
                  tick.content.last_price,
                  tick.content.volume_traded,
                  tick.content.net_change
                );

                // Debug: Show complete tick analysis
                debug!(
                  "Final Debug Analysis for Symbol {}:",
                  tick.instrument_token
                );
                debug!(
                  "  - Timestamp: Now = {:?}",
                  std::time::SystemTime::now()
                );
                debug!("  - Data completeness check:");
                debug!("    * Last Price: {:?} (✅)", tick.content.last_price);
                debug!(
                  "    * Volume: {:?} ({})",
                  tick.content.volume_traded,
                  if tick.content.volume_traded.is_some() {
                    "✅"
                  } else {
                    "❌"
                  }
                );
                debug!(
                  "    * Change: {:?} ({})",
                  tick.content.net_change,
                  if tick.content.net_change.is_some() {
                    "✅"
                  } else {
                    "❌"
                  }
                );
                debug!("    * Mode: {:?} (✅)", tick.content.mode);

                // Validate mode-specific data
                match tick.content.mode {
                  Mode::Full => {
                    debug!("  - Full mode validation:");
                    debug!(
                      "    * OHLC: {}",
                      if tick.content.ohlc.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    debug!(
                      "    * Depth: {}",
                      if tick.content.depth.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    if let Some(ohlc) = &tick.content.ohlc {
                      debug!(
                        "    * OHLC Values: O:{} H:{} L:{} C:{}",
                        ohlc.open, ohlc.high, ohlc.low, ohlc.close
                      );
                    }
                    if let Some(depth) = &tick.content.depth {
                      debug!(
                        "    * Depth Levels: {} buy, {} sell",
                        depth.buy.len(),
                        depth.sell.len()
                      );
                    }
                  }
                  Mode::Quote => {
                    debug!("  - Quote mode validation:");
                    debug!(
                      "    * OHLC: {}",
                      if tick.content.ohlc.is_some() {
                        "✅ Present"
                      } else {
                        "❌ Missing"
                      }
                    );
                    if let Some(ohlc) = &tick.content.ohlc {
                      debug!(
                        "    * OHLC Values: O:{} H:{} L:{} C:{}",
                        ohlc.open, ohlc.high, ohlc.low, ohlc.close
                      );
                    }
                  }
                  Mode::LTP => {
                    debug!("  - LTP mode validation: ✅ Basic data only");
                  }
                }
              }
            } else {
              println!("📋 {:?}: Non-tick message received", channel_id);
              debug!("Message type: {:?}", message);
            }
          }
          _ => continue,
        }
      }
      (channel_id, count)
    });
    tasks.push(task);
  }

  // Wait for monitoring to complete
  for task in tasks {
    if let Ok((channel_id, count)) = task.await {
      println!(
        "📊 {:?}: {} total messages in 10 seconds",
        channel_id, count
      );
    }
  }

  // Final cleanup demonstration
  println!("\n🧹 Step 11: Final Cleanup Demonstration");
  let current_symbols: Vec<u32> = manager
    .get_symbol_distribution()
    .values()
    .flat_map(|symbols| symbols.iter().cloned())
    .collect();

  println!(
    "Attempting to unsubscribe from {} remaining symbols...",
    current_symbols.len()
  );

  match manager.unsubscribe_symbols(&current_symbols).await {
    Ok(()) => {
      print_distribution(manager, "After complete cleanup").await;
      println!("✅ Complete cleanup successful!");
    }
    Err(e) => {
      println!("⚠️  Cleanup encountered issues: {}", e);
      println!("💡 This demonstrates the current architecture capabilities");
    }
  }

  println!("\n✅ Dynamic Subscription Demo Completed!");
  println!("🎯 Key Dynamic Features Demonstrated:");
  println!("   ✅ Runtime symbol addition across multiple batches");
  println!("   ✅ Runtime symbol removal with proper tracking");
  println!("   ✅ Dynamic mode changes for existing symbols");
  println!("   ✅ Real-time capacity distribution and monitoring");
  println!("   ✅ Independent message processing per connection");
  println!("   ✅ High-performance parser tasks with statistics");
  println!("   ✅ Complete subscription lifecycle management");

  Ok(())
}

pub async fn stop(&mut self) -> Result<(), String>

Stop the manager and all connections

Examples found in repository

examples/basic/runtime_subscription_example.rs (line 44)

async fn main() -> Result<(), String> {
  // Initialize with your API credentials
  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️ Please set KITE_API_KEY and KITE_ACCESS_TOKEN environment variables"
    );
    return Err("Missing API credentials".to_string());
  }

  // Create configuration
  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 3,
    connection_buffer_size: 5000,
    parser_buffer_size: 10000,
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP,
  };

  // Start the manager
  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  println!("🚀 KiteTicker Manager started!");

  // Example of runtime subscription management
  runtime_subscription_demo(&mut manager).await?;

  // Stop the manager
  manager.stop().await?;
  Ok(())
}

examples/mode_change_test.rs (line 52)

async fn main() -> Result<(), String> {
  // Initialize logging
  env_logger::init();

  println!("🔄 KiteTicker Mode Change Test");
  println!("═══════════════════════════════");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️  KITE_API_KEY and KITE_ACCESS_TOKEN environment variables not set"
    );
    demonstrate_mode_change_issue().await;
    return Ok(());
  }

  // Create configuration
  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 3,
    connection_buffer_size: 5000,
    parser_buffer_size: 10000,
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP,
  };

  println!("🔧 Starting manager...");
  let mut manager = KiteTickerManager::new(api_key, access_token, config);

  manager.start().await?;
  println!("✅ Manager started successfully");

  // Test the mode change issue
  test_mode_change_issue(&mut manager).await?;

  // Stop the manager
  println!("\n🛑 Stopping manager...");
  manager.stop().await?;

  println!("🏁 Mode change test completed!");
  Ok(())
}

examples/performance/market_scanner.rs (line 83)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("🔍 Market Scanner Example");
  println!("=========================");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    return Err("Please set KITE_API_KEY and KITE_ACCESS_TOKEN".to_string());
  }

  // High-performance configuration for scanning
  let config = KiteManagerConfig {
    max_connections: 3,
    max_symbols_per_connection: 3000,
    connection_buffer_size: 20000, // Large buffer for high volume
    parser_buffer_size: 50000,     // Even larger for parsed data
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP, // LTP mode for scanning (minimal bandwidth)
    ..Default::default()
  };

  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Large symbol set for market scanning
  let large_symbol_set = generate_symbol_list(8000); // 8000 symbols across 3 connections

  println!(
    "📊 Scanning {} symbols across {} connections",
    large_symbol_set.len(),
    3
  );

  let start_time = Instant::now();
  manager
    .subscribe_symbols(&large_symbol_set, Some(Mode::LTP))
    .await?;
  println!(
    "✅ Subscribed to {} symbols in {:?}",
    large_symbol_set.len(),
    start_time.elapsed()
  );

  // Get all channels and start parallel processing
  let channels = manager.get_all_channels();
  let mut handles = Vec::new();

  for (channel_id, mut receiver) in channels {
    let handle = tokio::spawn(async move {
      let mut scanner = MarketScanner::new(channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          scanner.process_ticks(ticks).await;
        }
      }

      scanner.print_summary();
    });

    handles.push(handle);
  }

  // Run scanner for specified duration
  println!("🔄 Scanning market for 30 seconds...");
  sleep(Duration::from_secs(30)).await;

  // Stop manager
  manager.stop().await?;

  // Wait for all scanners to complete
  for handle in handles {
    let _ = handle.await;
  }

  println!("🏁 Market scanning completed");
  Ok(())
}

examples/performance/high_frequency.rs (line 88)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("⚡ High-Frequency Trading Example");
  println!("=================================");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    return Err("Please set KITE_API_KEY and KITE_ACCESS_TOKEN".to_string());
  }

  // Ultra-high performance configuration for HFT
  let config = KiteManagerConfig {
    max_connections: 3,
    max_symbols_per_connection: 1000, // Focused symbol set for HFT
    connection_buffer_size: 100000,   // Maximum possible buffer
    parser_buffer_size: 200000,       // Ultra-large parser buffer
    enable_dedicated_parsers: true,
    default_mode: Mode::Full, // Full market depth data
    ..Default::default()
  };

  println!("🚀 Optimized for:");
  println!("   Sub-microsecond latency");
  println!("   Maximum throughput");
  println!("   Real-time order book analysis");
  println!("   Ultra-low garbage collection");

  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Focus on highly liquid instruments for HFT
  let hft_symbols = vec![
    256265, // NIFTY 50 - highly liquid
    408065, // HDFC Bank - large volume
    738561, // Reliance - most traded
    884737, // ICICI Bank
    341249, // TCS
    492033, // ITC
    779521, // Kotak Bank
  ];

  println!("📊 Subscribing to {} liquid instruments", hft_symbols.len());
  manager
    .subscribe_symbols(&hft_symbols, Some(Mode::Full))
    .await?;

  // Setup high-frequency processing engines
  let channels = manager.get_all_channels();
  let mut hft_engines = Vec::new();

  for (channel_id, mut receiver) in channels {
    let engine = tokio::spawn(async move {
      let mut hft_processor = HFTProcessor::new(channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          hft_processor.process_ultra_fast(ticks).await;
        }
      }

      hft_processor.print_performance_metrics();
    });

    hft_engines.push(engine);
  }

  // Run HFT simulation
  println!("⚡ Starting high-frequency processing...");
  sleep(Duration::from_secs(60)).await;

  // Stop processing
  manager.stop().await?;

  // Wait for all engines to complete
  for engine in hft_engines {
    let _ = engine.await;
  }

  println!("🏁 High-frequency trading simulation completed");
  Ok(())
}

examples/simple_tick_test.rs (line 107)

async fn main() -> Result<(), String> {
  env_logger::init();

  println!("🔄 Simple Tick Test - Capturing Initial Ticks");
  println!("═══════════════════════════════════════════");

  let api_key = std::env::var("KITE_API_KEY").unwrap_or_default();
  let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap_or_default();

  if api_key.is_empty() || access_token.is_empty() {
    println!(
      "⚠️  KITE_API_KEY and KITE_ACCESS_TOKEN environment variables not set"
    );
    return Ok(());
  }

  let config = KiteManagerConfig {
    max_symbols_per_connection: 3000,
    max_connections: 1, // Use only 1 connection for simplicity
    connection_buffer_size: 5000,
    parser_buffer_size: 10000,
    connection_timeout: Duration::from_secs(30),
    health_check_interval: Duration::from_secs(5),
    max_reconnect_attempts: 5,
    reconnect_delay: Duration::from_secs(2),
    enable_dedicated_parsers: true,
    default_mode: Mode::LTP,
  };

  let mut manager = KiteTickerManager::new(api_key, access_token, config);

  // Start manager
  println!("📡 Starting manager...");
  manager.start().await?;
  println!("✅ Manager started");

  // Get channel BEFORE subscribing
  println!("🎯 Getting channels...");
  let channels = manager.get_all_channels();
  let (channel_id, mut receiver) = channels.into_iter().next().unwrap();

  // Start listener in background
  let listener_handle = tokio::spawn(async move {
    println!("👂 Listener started for {:?}", channel_id);

    let mut tick_count = 0;
    loop {
      match timeout(Duration::from_secs(30), receiver.recv()).await {
        Ok(Ok(message)) => match message {
          TickerMessage::Ticks(ticks) => {
            tick_count += ticks.len();
            println!(
              "🎯 CAPTURED TICKS! {:?}: {} ticks (total: {})",
              channel_id,
              ticks.len(),
              tick_count
            );

            for tick in &ticks {
              println!("🔹 FULL TICK DEBUG:");
              println!("{:#?}", tick);
              println!("─────────────────────────────────────");
            }
          }
          TickerMessage::Error(err) => {
            println!("❌ Error: {}", err);
          }
          _ => {
            println!("📨 Other message: {:?}", message);
          }
        },
        Ok(Err(e)) => {
          println!("❌ Receive error: {}", e);
          break;
        }
        Err(_) => {
          println!("⏱️  Listener timeout");
          break;
        }
      }
    }
    println!("👂 Listener stopped");
  });

  // Give listener time to start
  tokio::time::sleep(Duration::from_millis(200)).await;

  // Now subscribe to a symbol
  println!("📊 Subscribing to symbol 256265...");
  manager
    .subscribe_symbols(&[128083204], Some(Mode::Full))
    .await?;
  println!("✅ Subscription sent");

  // Wait for ticks
  println!("⏳ Waiting 10 seconds for ticks...");
  tokio::time::sleep(Duration::from_secs(10)).await;

  // Stop
  manager.stop().await?;
  listener_handle.abort();

  println!("🏁 Test completed");
  Ok(())
}

examples/basic/portfolio_monitor.rs (line 147)

async fn main() -> Result<(), String> {
  // Initialize logging
  env_logger::init();

  println!("📊 Portfolio Monitor Example");
  println!("============================");

  // Get credentials
  let api_key = std::env::var("KITE_API_KEY")
    .map_err(|_| "Please set KITE_API_KEY environment variable")?;
  let access_token = std::env::var("KITE_ACCESS_TOKEN")
    .map_err(|_| "Please set KITE_ACCESS_TOKEN environment variable")?;

  // Define portfolio
  let portfolio = vec![
    (256265, "NIFTY 50"),
    (408065, "HDFC Bank"),
    (738561, "Reliance"),
    (5633, "TCS"),
    (884737, "Asian Paints"),
  ];

  println!("📈 Monitoring Portfolio:");
  for (symbol, name) in &portfolio {
    println!("   • {} ({})", name, symbol);
  }
  println!();

  // Create manager with optimized configuration for portfolio monitoring
  let config = KiteManagerConfig {
    max_connections: 1, // Single connection for small portfolio
    max_symbols_per_connection: 100,
    connection_buffer_size: 2000,
    parser_buffer_size: 5000,
    enable_dedicated_parsers: true,
    default_mode: Mode::Quote, // Quote mode for price + volume
    ..Default::default()
  };

  // Start manager
  let mut manager = KiteTickerManager::new(api_key, access_token, config);
  manager.start().await?;

  // Subscribe to portfolio symbols
  let symbols: Vec<u32> = portfolio.iter().map(|(symbol, _)| *symbol).collect();
  manager
    .subscribe_symbols(&symbols, Some(Mode::Quote))
    .await?;

  println!("✅ Subscribed to {} symbols", symbols.len());

  // Create portfolio tracking
  let mut portfolio_data: HashMap<u32, StockInfo> = HashMap::new();
  for (symbol, name) in portfolio {
    portfolio_data.insert(
      symbol,
      StockInfo {
        symbol,
        name: name.to_string(),
        current_price: 0.0,
        volume: 0,
        last_update: Instant::now(),
      },
    );
  }

  // Get data channels
  let channels = manager.get_all_channels();

  // Start data processing
  for (channel_id, mut receiver) in channels {
    let mut portfolio_clone = portfolio_data.clone();

    tokio::spawn(async move {
      println!("📡 Started monitoring channel {:?}", channel_id);

      while let Ok(message) = receiver.recv().await {
        if let TickerMessage::Ticks(ticks) = message {
          for tick in ticks {
            if let Some(stock) = portfolio_clone.get_mut(&tick.instrument_token)
            {
              // Update stock data
              if let Some(price) = tick.content.last_price {
                stock.current_price = price;
              }
              if let Some(volume) = tick.content.volume_traded {
                stock.volume = volume;
              }
              stock.last_update = Instant::now();

              // Display update
              println!(
                "📈 {} ({}): ₹{:.2} | Vol: {} | {}",
                stock.name,
                stock.symbol,
                stock.current_price,
                stock.volume,
                format_time_ago(stock.last_update)
              );
            }
          }
        }
      }
    });
  }

  // Periodic portfolio summary
  let summary_portfolio = portfolio_data.clone();
  tokio::spawn(async move {
    let mut interval = tokio::time::interval(Duration::from_secs(30));

    loop {
      interval.tick().await;
      print_portfolio_summary(&summary_portfolio);
    }
  });

  // Monitor for demo duration
  println!("📊 Monitoring portfolio for 2 minutes...\n");
  sleep(Duration::from_secs(120)).await;

  // Final summary
  println!("\n🏁 Demo completed!");
  print_portfolio_summary(&portfolio_data);

  // Stop manager
  manager.stop().await?;
  println!("✅ Portfolio monitor stopped");

  Ok(())
}

Additional examples can be found in:

• examples/dynamic_subscription_demo.rs
• examples/advanced/manager_demo.rs

Trait Implementations

impl Debug for KiteTickerManager

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.