Rust로 구현하는 고성능 MCP 서버: ZeroMQ를 활용한 비동기 메시징

[dependencies]
zmq = "0.10"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
tokio = { version = "1", features = ["full"] }

use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct McpMessage {
    pub source_id: String,
    pub target_topic: String,
    pub payload: String,
    pub timestamp: i64,
}

ouse std::time::SystemTime;
use zmq::{Context, Socket, SocketType};

struct McpRouter {
    context: Context,
    frontend: Socket, // Agents connect here
    // 필요시 다른 패턴의 소켓 추가 가능
}

impl McpRouter {
    fn new() -> Result<Self, zmq::Error> {
        let context = Context::new();
        let frontend = context.socket(SocketType::SUB)?;
        
        // 모든 메시지를 수신하도록 설정 (필터링 없음)
        frontend.set_subscribe(b"")?;
        
        Ok(McpRouter { context, frontend })
    }

    fn start(&self, endpoint: &str) -> Result<(), zmq::Error> {
        self.frontend.bind(endpoint)?;
        println!("[ZeroClaw Router] Listening on {}", endpoint);

        let mut msg = zmq::Message::new();
        loop {
            // Non-blocking receive를 위해 poll 사용 가능
            // 여기서는 간단한 blocking receive 예시
            match self.frontend.recv(&mut msg) {
                Ok(_) => {
                    let data = msg.as_str().unwrap();
                    // 메시지 파싱 및 라우팅 로직
                    if let Ok(mcp_msg) = serde_json::from_str::<McpMessage>(data) {
                        self.route_message(mcp_msg);
                    }
                }
                Err(e) => {
                    eprintln!("Receive Error: {}", e);
                }
            }
        }
    }

    fn route_message(&self, msg: McpMessage) {
        // 실제 라우팅 로직 구현
        println!("Routing from {} on topic {}", msg.source_id, msg.target_topic);
        // 예: 데이터베이스 저장, 다른 소켓으로 전달 등
    }
}

use std::thread;

fn spawn_publisher_agent(id: &str, endpoint: &str) {
    let context = Context::new();
    let sender = context.socket(SocketType::PUB).expect("Failed to create PUB socket");
    sender.connect(endpoint).expect("Failed to connect to Router");

    // 별도 스레드에서 실행하여 Tokio 런타임과 분리
    thread::spawn(move || {
        let counter = 0..100;
        for i in counter {
            let msg = McpMessage {
                source_id: id.to_string(),
                target_topic: "general".to_string(),
                payload: format!("Message #{} from agent {}", i, id),
                timestamp: SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs() as i64,
            };

            let json_str = serde_json::to_string(&msg).unwrap();
            // ZeroMQ는 전송 실패 시 자동으로 재시도하거나 큐에 쌓음 (High-water mark 설정 필요 시)
            sender.send(json_str.as_bytes(), 0).expect("Failed to send");
            
            thread::sleep(std::time::Duration::from_millis(100));
        }
    });
}

// TCP 대신 IPC 사용
let endpoint = "ipc:///tmp/mcp_router.ipc";
router.start(endpoint);