mirror of
https://github.com/ruvnet/RuView
synced 2026-07-31 18:51:42 +00:00
feat: vendor midstream and sublinear-time-solver libraries
Add ruvnet/midstream (AIMDS real-time inference) and ruvnet/sublinear-time-solver (sublinear optimization algorithms) as vendored dependencies under vendor/. Co-Authored-By: claude-flow <ruv@ruv.net>
This commit is contained in:
ruv
+144
@@ -0,0 +1,144 @@
|
||||
//! Example: Lean Agentic Stream Learning with MidStream
|
||||
//!
|
||||
//! This example demonstrates the revolutionary Lean Agentic Learning System
|
||||
//! integrated with MidStream for real-time LLM streaming with:
|
||||
//! - Formal verification of agent actions
|
||||
//! - Autonomous decision-making (Plan-Act-Observe-Learn loop)
|
||||
//! - Online learning and adaptation
|
||||
//! - Dynamic knowledge graph evolution
|
||||
//!
|
||||
//! Run with: cargo run --example lean_agentic_streaming
|
||||
|
||||
use midstream::{
|
||||
LeanAgenticSystem, LeanAgenticConfig, AgentContext,
|
||||
Midstream, HyprSettings, HyprServiceImpl, StreamProcessor, LLMClient,
|
||||
};
|
||||
use futures::stream::{BoxStream, iter};
|
||||
use tokio;
|
||||
|
||||
/// Example LLM client that simulates streaming responses
|
||||
struct SimulatedLLMClient {
|
||||
messages: Vec<String>,
|
||||
}
|
||||
|
||||
impl SimulatedLLMClient {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
messages: vec![
|
||||
"Hello! I can help you with weather information.".to_string(),
|
||||
"Let me learn your preferences.".to_string(),
|
||||
"What would you like to know?".to_string(),
|
||||
"I'm getting better at understanding you!".to_string(),
|
||||
],
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl LLMClient for SimulatedLLMClient {
|
||||
fn stream(&self) -> BoxStream<'static, String> {
|
||||
Box::pin(iter(self.messages.clone()))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
println!("🚀 Lean Agentic Stream Learning System\n");
|
||||
println!("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n");
|
||||
|
||||
// 1. Initialize Lean Agentic System
|
||||
println!("📚 Initializing Lean Agentic System...");
|
||||
let config = LeanAgenticConfig {
|
||||
enable_formal_verification: true,
|
||||
learning_rate: 0.01,
|
||||
max_planning_depth: 5,
|
||||
action_threshold: 0.7,
|
||||
enable_multi_agent: true,
|
||||
kg_update_freq: 100,
|
||||
};
|
||||
|
||||
let lean_system = LeanAgenticSystem::new(config);
|
||||
println!("✓ System initialized with formal verification enabled\n");
|
||||
|
||||
// 2. Initialize MidStream
|
||||
println!("🌊 Setting up MidStream...");
|
||||
let settings = HyprSettings::new()?;
|
||||
let hypr_service = HyprServiceImpl::new(&settings).await?;
|
||||
let llm_client = SimulatedLLMClient::new();
|
||||
|
||||
let midstream = Midstream::new(
|
||||
Box::new(llm_client),
|
||||
Box::new(hypr_service),
|
||||
);
|
||||
println!("✓ MidStream ready\n");
|
||||
|
||||
// 3. Process stream with lean agentic learning
|
||||
println!("🔄 Processing stream with agentic learning...\n");
|
||||
|
||||
let messages = midstream.process_stream().await?;
|
||||
|
||||
// Process each message through the lean agentic system
|
||||
let mut context = AgentContext::new("session_001".to_string());
|
||||
|
||||
for (i, msg) in messages.iter().enumerate() {
|
||||
println!(" Message #{}: {}", i + 1, msg.content);
|
||||
|
||||
// Process with lean agentic system
|
||||
let result = lean_system.process_stream_chunk(
|
||||
&msg.content,
|
||||
context.clone(),
|
||||
).await?;
|
||||
|
||||
println!(" → Action: {}", result.action.description);
|
||||
println!(" → Reward: {:.2}", result.reward);
|
||||
println!(" → Verified: {}", if result.verified { "✓" } else { "✗" });
|
||||
|
||||
// Update context
|
||||
context.add_message(msg.content.clone());
|
||||
println!();
|
||||
}
|
||||
|
||||
// 4. Display system statistics
|
||||
println!("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n");
|
||||
println!("📊 System Statistics:\n");
|
||||
|
||||
let stats = lean_system.get_stats().await;
|
||||
|
||||
println!(" Knowledge Graph:");
|
||||
println!(" - Entities: {}", stats.total_entities);
|
||||
println!(" - Theorems: {}", stats.total_theorems);
|
||||
|
||||
println!("\n Learning:");
|
||||
println!(" - Iterations: {}", stats.learning_iterations);
|
||||
println!(" - Actions: {}", stats.total_actions);
|
||||
println!(" - Avg Reward: {:.3}", stats.average_reward);
|
||||
|
||||
println!("\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
|
||||
|
||||
// 5. Demonstrate advanced features
|
||||
println!("\n🎯 Advanced Features Demonstration:\n");
|
||||
|
||||
// Test formal reasoning
|
||||
println!(" 1. Formal Reasoning:");
|
||||
let reasoner = lean_system.reasoner.read().await;
|
||||
println!(" - Axioms loaded: {}", reasoner.theorem_count());
|
||||
drop(reasoner);
|
||||
|
||||
// Test knowledge graph
|
||||
println!("\n 2. Knowledge Graph:");
|
||||
let kg = lean_system.knowledge.read().await;
|
||||
println!(" - Entities tracked: {}", kg.entity_count());
|
||||
println!(" - Relations: {}", kg.relation_count());
|
||||
drop(kg);
|
||||
|
||||
// Test online learning
|
||||
println!("\n 3. Online Learning:");
|
||||
let learner = lean_system.learner.read().await;
|
||||
let learning_stats = learner.get_stats();
|
||||
println!(" - Model parameters: {}", learning_stats.model_parameters);
|
||||
println!(" - Experience buffer: {}", learning_stats.buffer_size);
|
||||
drop(learner);
|
||||
|
||||
println!("\n✨ Lean Agentic Stream Learning Complete!");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
+151
@@ -0,0 +1,151 @@
|
||||
use midstream::{Midstream, HyprSettings, HyprServiceImpl, StreamProcessor, LLMClient};
|
||||
use futures::stream::{BoxStream, StreamExt};
|
||||
use reqwest::Client;
|
||||
use serde_json::{json, Value};
|
||||
use std::time::Duration;
|
||||
use eventsource_stream::Eventsource;
|
||||
use dotenv::dotenv;
|
||||
|
||||
struct OpenRouterClient {
|
||||
client: Client,
|
||||
api_key: String,
|
||||
}
|
||||
|
||||
impl OpenRouterClient {
|
||||
fn new(api_key: String) -> Self {
|
||||
Self {
|
||||
client: Client::new(),
|
||||
api_key,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl LLMClient for OpenRouterClient {
|
||||
fn stream(&self) -> BoxStream<'static, String> {
|
||||
let prompt = "Tell me a short story about a robot learning to paint. Make it emotional and stream it word by word.".to_string();
|
||||
let client = self.client.clone();
|
||||
let api_key = self.api_key.clone();
|
||||
|
||||
Box::pin(async_stream::stream! {
|
||||
let url = "https://openrouter.ai/api/v1/chat/completions";
|
||||
let referer = std::env::var("OPENROUTER_REFERER").unwrap_or_else(|_| "http://localhost:3000".to_string());
|
||||
let model = std::env::var("OPENROUTER_MODEL").unwrap_or_else(|_| "anthropic/claude-2".to_string());
|
||||
|
||||
println!("Sending request to OpenRouter API...");
|
||||
println!("Model: {}", model);
|
||||
|
||||
let payload = json!({
|
||||
"model": model,
|
||||
"messages": [
|
||||
{
|
||||
"role": "user",
|
||||
"content": prompt
|
||||
}
|
||||
],
|
||||
"stream": true
|
||||
});
|
||||
|
||||
let response = client
|
||||
.post(url)
|
||||
.header("Authorization", format!("Bearer {}", api_key))
|
||||
.header("HTTP-Referer", referer)
|
||||
.json(&payload)
|
||||
.send()
|
||||
.await
|
||||
.expect("Failed to send request");
|
||||
|
||||
println!("Response status: {}", response.status());
|
||||
|
||||
let mut stream = response
|
||||
.bytes_stream()
|
||||
.eventsource()
|
||||
.map(|event| {
|
||||
match event {
|
||||
Ok(event) => {
|
||||
println!("Received event: {}", event.data);
|
||||
if event.data == "[DONE]" {
|
||||
String::new()
|
||||
} else {
|
||||
match serde_json::from_str::<Value>(&event.data) {
|
||||
Ok(value) => {
|
||||
let content = value["choices"][0]["delta"]["content"]
|
||||
.as_str()
|
||||
.unwrap_or("");
|
||||
if !content.is_empty() {
|
||||
println!("Content: {}", content);
|
||||
}
|
||||
content.to_string()
|
||||
}
|
||||
Err(e) => {
|
||||
println!("Failed to parse JSON: {}", e);
|
||||
String::new()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
println!("Stream error: {}", e);
|
||||
format!("Error: {}", e)
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
while let Some(s) = stream.next().await {
|
||||
if !s.is_empty() {
|
||||
yield s.trim().to_string();
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
// Load environment variables
|
||||
dotenv().ok();
|
||||
|
||||
// Get API key from environment
|
||||
let api_key = std::env::var("OPENROUTER_API_KEY")
|
||||
.expect("OPENROUTER_API_KEY must be set in .env file");
|
||||
|
||||
// Initialize settings
|
||||
let settings = HyprSettings::new()?;
|
||||
|
||||
// Create hyprstream service
|
||||
let hypr_service = HyprServiceImpl::new(&settings).await?;
|
||||
|
||||
// Create OpenRouter client
|
||||
let llm_client = OpenRouterClient::new(api_key);
|
||||
|
||||
// Initialize Midstream
|
||||
let midstream = Midstream::new(
|
||||
Box::new(llm_client),
|
||||
Box::new(hypr_service),
|
||||
);
|
||||
|
||||
println!("\nStreaming story from Claude-2...\n");
|
||||
|
||||
// Process stream
|
||||
let messages = midstream.process_stream().await?;
|
||||
|
||||
println!("\nFinal story:");
|
||||
for msg in &messages {
|
||||
print!("{}", msg.content);
|
||||
}
|
||||
println!("\n");
|
||||
|
||||
// Get metrics
|
||||
let metrics = midstream.get_metrics().await;
|
||||
println!("\nMetrics collected:");
|
||||
for metric in &metrics {
|
||||
println!("- Token count: {}", metric.value);
|
||||
println!(" Labels: {:?}", metric.labels);
|
||||
println!();
|
||||
}
|
||||
|
||||
// Get average sentiment for last 5 minutes
|
||||
let avg = midstream.get_average_sentiment(Duration::from_secs(300)).await?;
|
||||
println!("\nAverage tokens per message: {:.2}", avg);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
+150
@@ -0,0 +1,150 @@
|
||||
/// Example demonstrating the pattern detection APIs in temporal-compare
|
||||
///
|
||||
/// This example shows how to use:
|
||||
/// 1. find_similar() - Find similar patterns in time series
|
||||
/// 2. detect_pattern() - Detect if a pattern exists
|
||||
/// 3. Advanced APIs for recurring and fuzzy pattern detection
|
||||
|
||||
use midstreamer_temporal_compare::{TemporalComparator, Pattern};
|
||||
|
||||
fn main() {
|
||||
println!("=== Temporal-Compare Pattern Detection Demo ===\n");
|
||||
|
||||
// Create a comparator with cache size 100 and max sequence length 1000
|
||||
let comparator: TemporalComparator<f64> = TemporalComparator::new(100, 1000);
|
||||
|
||||
// Example 1: find_similar() - Find exact matches
|
||||
println!("Example 1: Finding similar patterns with find_similar()");
|
||||
println!("---------------------------------------------------");
|
||||
let series1 = vec![1.0, 2.0, 3.0, 4.0, 5.0, 3.0, 4.0, 5.0, 6.0];
|
||||
let pattern1 = vec![3.0, 4.0, 5.0];
|
||||
|
||||
println!("Series: {:?}", series1);
|
||||
println!("Pattern: {:?}", pattern1);
|
||||
|
||||
let matches = comparator.find_similar(&series1, &pattern1, 0.5);
|
||||
println!("Found {} matches:", matches.len());
|
||||
for (idx, distance) in &matches {
|
||||
println!(" - Index {}: distance = {:.4}", idx, distance);
|
||||
}
|
||||
println!();
|
||||
|
||||
// Example 2: detect_pattern() - Simple boolean detection
|
||||
println!("Example 2: Detecting pattern existence with detect_pattern()");
|
||||
println!("-------------------------------------------------------------");
|
||||
let series2 = vec![10.0, 20.0, 30.0, 40.0, 50.0];
|
||||
let pattern2a = vec![30.0, 40.0, 50.0];
|
||||
let pattern2b = vec![100.0, 200.0, 300.0];
|
||||
|
||||
println!("Series: {:?}", series2);
|
||||
println!("Pattern A: {:?}", pattern2a);
|
||||
let found_a = comparator.detect_pattern(&series2, &pattern2a, 1.0);
|
||||
println!("Pattern A detected: {}", found_a);
|
||||
|
||||
println!("Pattern B: {:?}", pattern2b);
|
||||
let found_b = comparator.detect_pattern(&series2, &pattern2b, 1.0);
|
||||
println!("Pattern B detected: {}", found_b);
|
||||
println!();
|
||||
|
||||
// Example 3: Approximate matching with threshold
|
||||
println!("Example 3: Approximate matching with different thresholds");
|
||||
println!("----------------------------------------------------------");
|
||||
let series3 = vec![1.0, 2.0, 3.1, 4.2, 4.9, 6.0];
|
||||
let pattern3 = vec![3.0, 4.0, 5.0];
|
||||
|
||||
println!("Series: {:?}", series3);
|
||||
println!("Pattern: {:?}", pattern3);
|
||||
|
||||
// Strict threshold
|
||||
let strict_matches = comparator.find_similar(&series3, &pattern3, 0.5);
|
||||
println!("Strict threshold (0.5): {} matches", strict_matches.len());
|
||||
|
||||
// Loose threshold
|
||||
let loose_matches = comparator.find_similar(&series3, &pattern3, 2.0);
|
||||
println!("Loose threshold (2.0): {} matches", loose_matches.len());
|
||||
for (idx, distance) in &loose_matches {
|
||||
println!(" - Index {}: distance = {:.4}", idx, distance);
|
||||
}
|
||||
println!();
|
||||
|
||||
// Example 4: Generic API with integers
|
||||
println!("Example 4: Generic API with integer sequences");
|
||||
println!("----------------------------------------------");
|
||||
let comparator_int: TemporalComparator<i32> = TemporalComparator::new(100, 1000);
|
||||
|
||||
let haystack = vec![1, 2, 3, 4, 5, 3, 4, 5, 6, 7];
|
||||
let needle = vec![3, 4, 5];
|
||||
|
||||
println!("Haystack: {:?}", haystack);
|
||||
println!("Needle: {:?}", needle);
|
||||
|
||||
let matches = comparator_int.find_similar_generic(&haystack, &needle, 0.1).unwrap();
|
||||
println!("Found {} matches:", matches.len());
|
||||
for m in &matches {
|
||||
println!(" - Index {}: similarity = {:.4}, distance = {:.4}",
|
||||
m.start_index, m.similarity, m.distance);
|
||||
}
|
||||
println!();
|
||||
|
||||
// Example 5: Detect recurring patterns
|
||||
println!("Example 5: Automatic recurring pattern detection");
|
||||
println!("------------------------------------------------");
|
||||
let comparator_char: TemporalComparator<char> = TemporalComparator::new(100, 1000);
|
||||
|
||||
let sequence = vec!['a', 'b', 'c', 'a', 'b', 'c', 'd', 'e', 'd', 'e'];
|
||||
|
||||
println!("Sequence: {:?}", sequence);
|
||||
|
||||
let patterns = comparator_char.detect_recurring_patterns(&sequence, 2, 3).unwrap();
|
||||
println!("Found {} recurring patterns:", patterns.len());
|
||||
for (i, pattern) in patterns.iter().enumerate() {
|
||||
println!(" Pattern {}: {:?}", i + 1, pattern.sequence);
|
||||
println!(" Frequency: {}", pattern.frequency());
|
||||
println!(" Confidence: {:.4}", pattern.confidence);
|
||||
println!(" Occurrences at: {:?}", pattern.occurrences);
|
||||
}
|
||||
println!();
|
||||
|
||||
// Example 6: Fuzzy pattern detection
|
||||
println!("Example 6: Fuzzy pattern detection (groups similar patterns)");
|
||||
println!("-------------------------------------------------------------");
|
||||
let comparator_fuzzy: TemporalComparator<i32> = TemporalComparator::new(100, 1000);
|
||||
|
||||
let sequence = vec![1, 2, 3, 1, 2, 4, 1, 2, 3, 5, 6, 7, 5, 6, 8];
|
||||
|
||||
println!("Sequence: {:?}", sequence);
|
||||
|
||||
let fuzzy_patterns = comparator_fuzzy.detect_fuzzy_patterns(&sequence, 3, 3, 0.7).unwrap();
|
||||
println!("Found {} fuzzy pattern groups:", fuzzy_patterns.len());
|
||||
for (i, pattern) in fuzzy_patterns.iter().enumerate() {
|
||||
println!(" Pattern Group {}: {:?}", i + 1, pattern.sequence);
|
||||
println!(" Frequency: {} (includes variations)", pattern.frequency());
|
||||
println!(" Confidence: {:.4}", pattern.confidence);
|
||||
}
|
||||
println!();
|
||||
|
||||
// Example 7: Cache performance
|
||||
println!("Example 7: Cache performance demonstration");
|
||||
println!("------------------------------------------");
|
||||
|
||||
// Clear cache first
|
||||
comparator.clear_cache();
|
||||
|
||||
// Run same query multiple times
|
||||
for i in 1..=5 {
|
||||
let _ = comparator.find_similar(&series1, &pattern1, 0.5);
|
||||
let stats = comparator.cache_stats();
|
||||
println!(" Iteration {}: hits = {}, misses = {}, hit rate = {:.2}%",
|
||||
i, stats.hits, stats.misses, stats.hit_rate() * 100.0);
|
||||
}
|
||||
println!();
|
||||
|
||||
println!("=== Demo Complete ===");
|
||||
println!("\nKey Takeaways:");
|
||||
println!("1. find_similar() uses DTW to find pattern matches with configurable threshold");
|
||||
println!("2. detect_pattern() provides simple boolean detection");
|
||||
println!("3. Generic APIs work with any comparable type (f64, i32, char, etc.)");
|
||||
println!("4. Automatic pattern discovery finds recurring patterns");
|
||||
println!("5. Fuzzy matching groups similar pattern variations");
|
||||
println!("6. Built-in caching improves performance for repeated queries");
|
||||
}
|
||||
+300
@@ -0,0 +1,300 @@
|
||||
//! Production-Ready QUIC Multi-Stream Server Example
|
||||
//!
|
||||
//! This example demonstrates a comprehensive QUIC server implementation using
|
||||
//! the midstream-quic crate with support for:
|
||||
//! - Multiple concurrent bidirectional streams
|
||||
//! - Stream prioritization
|
||||
//! - TLS certificate generation (self-signed for demo)
|
||||
//! - Connection statistics and monitoring
|
||||
//! - Graceful shutdown handling
|
||||
//! - Performance metrics logging
|
||||
//!
|
||||
//! # Usage
|
||||
//!
|
||||
//! ```bash
|
||||
//! cargo run --example quic_server
|
||||
//! ```
|
||||
//!
|
||||
//! # Testing with Client
|
||||
//!
|
||||
//! You can test this server with a QUIC client on port 4433:
|
||||
//! ```bash
|
||||
//! # Example with curl (if built with HTTP/3 support)
|
||||
//! curl --http3 https://localhost:4433 --insecure
|
||||
//! ```
|
||||
|
||||
use midstream_quic::{QuicMultiStream, QuicConfig, StreamPriority};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::signal;
|
||||
use tokio::time::{Duration, Instant};
|
||||
use tracing::{info, warn, error, debug};
|
||||
|
||||
/// Server configuration constants
|
||||
const SERVER_PORT: u16 = 4433;
|
||||
const SERVER_ADDR: &str = "0.0.0.0";
|
||||
const MAX_CONCURRENT_STREAMS: u64 = 100;
|
||||
const IDLE_TIMEOUT_MS: u64 = 30_000;
|
||||
const KEEP_ALIVE_INTERVAL_MS: u64 = 5_000;
|
||||
|
||||
/// Connection statistics tracker
|
||||
#[derive(Default)]
|
||||
struct ServerStats {
|
||||
total_connections: AtomicU64,
|
||||
active_connections: AtomicU64,
|
||||
total_streams: AtomicU64,
|
||||
bytes_received: AtomicU64,
|
||||
bytes_sent: AtomicU64,
|
||||
}
|
||||
|
||||
impl ServerStats {
|
||||
fn log_stats(&self) {
|
||||
info!(
|
||||
"Server Stats - Connections: {} active, {} total | Streams: {} total | Data: {} bytes RX, {} bytes TX",
|
||||
self.active_connections.load(Ordering::Relaxed),
|
||||
self.total_connections.load(Ordering::Relaxed),
|
||||
self.total_streams.load(Ordering::Relaxed),
|
||||
self.bytes_received.load(Ordering::Relaxed),
|
||||
self.bytes_sent.load(Ordering::Relaxed),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Generate self-signed TLS certificate for demo purposes
|
||||
fn generate_self_signed_cert() -> Result<(Vec<u8>, Vec<u8>), Box<dyn std::error::Error>> {
|
||||
use rcgen::{Certificate, CertificateParams, DistinguishedName};
|
||||
|
||||
let mut params = CertificateParams::new(vec!["localhost".to_string()]);
|
||||
params.distinguished_name = DistinguishedName::new();
|
||||
params.distinguished_name.push(
|
||||
rcgen::DnType::CommonName,
|
||||
"Midstream QUIC Server".to_string(),
|
||||
);
|
||||
|
||||
let cert = Certificate::from_params(params)?;
|
||||
let cert_pem = cert.serialize_pem()?;
|
||||
let key_pem = cert.serialize_private_key_pem();
|
||||
|
||||
info!("Generated self-signed certificate for demo");
|
||||
|
||||
Ok((cert_pem.into_bytes(), key_pem.into_bytes()))
|
||||
}
|
||||
|
||||
/// Handle individual QUIC stream with echo functionality
|
||||
async fn handle_stream(
|
||||
mut stream: quinn::SendStream,
|
||||
mut recv: quinn::RecvStream,
|
||||
stream_id: u64,
|
||||
stats: Arc<ServerStats>,
|
||||
) -> Result<(), Box<dyn std::error::Error>> {
|
||||
let start_time = Instant::now();
|
||||
stats.total_streams.fetch_add(1, Ordering::Relaxed);
|
||||
|
||||
debug!("Stream {} opened", stream_id);
|
||||
|
||||
let mut buffer = vec![0u8; 8192];
|
||||
let mut total_bytes = 0u64;
|
||||
|
||||
loop {
|
||||
match recv.read(&mut buffer).await? {
|
||||
Some(bytes_read) => {
|
||||
total_bytes += bytes_read as u64;
|
||||
stats.bytes_received.fetch_add(bytes_read as u64, Ordering::Relaxed);
|
||||
|
||||
debug!("Stream {} received {} bytes", stream_id, bytes_read);
|
||||
|
||||
// Echo back the data
|
||||
stream.write_all(&buffer[..bytes_read]).await?;
|
||||
stats.bytes_sent.fetch_add(bytes_read as u64, Ordering::Relaxed);
|
||||
|
||||
// Check for special commands
|
||||
if let Ok(msg) = std::str::from_utf8(&buffer[..bytes_read]) {
|
||||
if msg.trim() == "STATS" {
|
||||
let stats_msg = format!(
|
||||
"Stream {} - Duration: {:?}, Bytes: {}\n",
|
||||
stream_id,
|
||||
start_time.elapsed(),
|
||||
total_bytes
|
||||
);
|
||||
stream.write_all(stats_msg.as_bytes()).await?;
|
||||
} else if msg.trim() == "CLOSE" {
|
||||
info!("Stream {} received close command", stream_id);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
debug!("Stream {} reached EOF", stream_id);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
stream.finish().await?;
|
||||
|
||||
info!(
|
||||
"Stream {} closed - Duration: {:?}, Total bytes: {}",
|
||||
stream_id,
|
||||
start_time.elapsed(),
|
||||
total_bytes
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Handle QUIC connection with multiple streams
|
||||
async fn handle_connection(
|
||||
conn: quinn::Connection,
|
||||
stats: Arc<ServerStats>,
|
||||
) -> Result<(), Box<dyn std::error::Error>> {
|
||||
let conn_id = stats.total_connections.fetch_add(1, Ordering::Relaxed);
|
||||
stats.active_connections.fetch_add(1, Ordering::Relaxed);
|
||||
|
||||
let remote_addr = conn.remote_address();
|
||||
info!("Connection {} accepted from {}", conn_id, remote_addr);
|
||||
|
||||
// Spawn task to handle incoming streams
|
||||
let stream_stats = stats.clone();
|
||||
let stream_handler = tokio::spawn(async move {
|
||||
let mut stream_count = 0u64;
|
||||
|
||||
loop {
|
||||
match conn.accept_bi().await {
|
||||
Ok((send, recv)) => {
|
||||
stream_count += 1;
|
||||
let stream_id = stream_count;
|
||||
let stats_clone = stream_stats.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = handle_stream(send, recv, stream_id, stats_clone).await {
|
||||
error!("Stream {} error: {}", stream_id, e);
|
||||
}
|
||||
});
|
||||
}
|
||||
Err(quinn::ConnectionError::ApplicationClosed(_)) => {
|
||||
info!("Connection {} closed by peer", conn_id);
|
||||
break;
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Connection {} error accepting stream: {}", conn_id, e);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Wait for connection to close
|
||||
let result = stream_handler.await;
|
||||
|
||||
stats.active_connections.fetch_sub(1, Ordering::Relaxed);
|
||||
info!("Connection {} terminated", conn_id);
|
||||
|
||||
result?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Main server function
|
||||
async fn run_server() -> Result<(), Box<dyn std::error::Error>> {
|
||||
// Initialize tracing
|
||||
tracing_subscriber::fmt()
|
||||
.with_max_level(tracing::Level::INFO)
|
||||
.with_target(false)
|
||||
.init();
|
||||
|
||||
info!("Starting Midstream QUIC Multi-Stream Server");
|
||||
|
||||
// Generate self-signed certificate
|
||||
let (cert_pem, key_pem) = generate_self_signed_cert()?;
|
||||
|
||||
// Configure QUIC server
|
||||
let mut server_config = quinn::ServerConfig::with_single_cert(
|
||||
vec![rustls::Certificate(cert_pem)],
|
||||
rustls::PrivateKey(key_pem),
|
||||
)?;
|
||||
|
||||
// Configure transport settings
|
||||
let mut transport = quinn::TransportConfig::default();
|
||||
transport.max_concurrent_bidi_streams(MAX_CONCURRENT_STREAMS.try_into()?);
|
||||
transport.max_concurrent_uni_streams(MAX_CONCURRENT_STREAMS.try_into()?);
|
||||
transport.max_idle_timeout(Some(Duration::from_millis(IDLE_TIMEOUT_MS).try_into()?));
|
||||
transport.keep_alive_interval(Some(Duration::from_millis(KEEP_ALIVE_INTERVAL_MS)));
|
||||
|
||||
server_config.transport_config(Arc::new(transport));
|
||||
|
||||
// Bind server endpoint
|
||||
let bind_addr = format!("{}:{}", SERVER_ADDR, SERVER_PORT);
|
||||
let endpoint = quinn::Endpoint::server(server_config, bind_addr.parse()?)?;
|
||||
|
||||
info!("Server listening on {}", bind_addr);
|
||||
info!("Configuration:");
|
||||
info!(" - Max concurrent streams: {}", MAX_CONCURRENT_STREAMS);
|
||||
info!(" - Idle timeout: {}ms", IDLE_TIMEOUT_MS);
|
||||
info!(" - Keep-alive interval: {}ms", KEEP_ALIVE_INTERVAL_MS);
|
||||
|
||||
let stats = Arc::new(ServerStats::default());
|
||||
|
||||
// Spawn statistics logger
|
||||
let stats_clone = stats.clone();
|
||||
tokio::spawn(async move {
|
||||
let mut interval = tokio::time::interval(Duration::from_secs(10));
|
||||
loop {
|
||||
interval.tick().await;
|
||||
stats_clone.log_stats();
|
||||
}
|
||||
});
|
||||
|
||||
// Handle graceful shutdown
|
||||
let shutdown = signal::ctrl_c();
|
||||
tokio::pin!(shutdown);
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
Some(incoming) = endpoint.accept() => {
|
||||
let stats_clone = stats.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
match incoming.await {
|
||||
Ok(conn) => {
|
||||
if let Err(e) = handle_connection(conn, stats_clone).await {
|
||||
error!("Connection handler error: {}", e);
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Incoming connection error: {}", e);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
_ = &mut shutdown => {
|
||||
info!("Shutdown signal received");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Graceful shutdown
|
||||
info!("Shutting down server...");
|
||||
endpoint.close(0u32.into(), b"Server shutdown");
|
||||
|
||||
// Wait for active connections to close (max 5 seconds)
|
||||
let shutdown_start = Instant::now();
|
||||
while stats.active_connections.load(Ordering::Relaxed) > 0
|
||||
&& shutdown_start.elapsed() < Duration::from_secs(5)
|
||||
{
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
|
||||
stats.log_stats();
|
||||
info!("Server shutdown complete");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
if let Err(e) = run_server().await {
|
||||
eprintln!("Server error: {}", e);
|
||||
std::process::exit(1);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user