msgtrans 1.0.7

Support for a variety of communication protocols such as TCP / QUIC / WebSocket, easy to create server and client network library.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

//! Load test server - high-performance echo server for benchmarking
//!
//! Usage:
//!   cargo run --example load_test_server --release
//!
//! This is an echo server using the Actor model for maximum throughput.
//! Each connection gets its own dedicated queue and worker task, eliminating
//! global contention and enabling natural backpressure.
//!
//! Architecture:
//!   Connection → mpsc(bounded) → SessionActor → EchoHandler
//!
//! Use with load_test client for benchmarking.

use async_trait::async_trait;
use msgtrans::{
    packet::{Packet, PacketType},
    protocol::QuicServerConfig,
    protocol::TcpServerConfig,
    protocol::WebSocketServerConfig,
    tokio,
    transport::{SessionHandler, SessionSender, TransportServerBuilder},
    SessionId,
};
use std::sync::Arc;

/// Echo handler - implements SessionHandler trait
///
/// This is a minimal echo handler that directly echoes back received messages.
/// No broadcast channel, no global contention, just pure per-connection processing.
struct EchoHandler;

#[async_trait]
impl SessionHandler for EchoHandler {
    async fn on_message(&self, _session_id: SessionId, packet: Packet, sender: SessionSender) {
        match packet.header.packet_type {
            PacketType::Request => {
                let _ = sender
                    .respond(
                        packet.header.message_id,
                        packet.header.biz_type,
                        packet.payload,
                    )
                    .await;
            }
            _ => {
                let _ = sender.send_data(packet.payload).await;
            }
        }
    }

    async fn on_connected(&self, session_id: SessionId) {
        tracing::debug!("[CONNECT] Session {} connected", session_id);
    }

    async fn on_disconnected(&self, session_id: SessionId, _reason: msgtrans::error::CloseReason) {
        tracing::debug!("[DISCONNECT] Session {} disconnected", session_id);
    }
}

#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Minimal logging - only errors
    tracing_subscriber::fmt()
        .with_max_level(tracing::Level::ERROR)
        .init();

    println!("Load Test Server - High-performance Echo Server");
    println!("================================================");
    println!();
    println!("Architecture: Connection -> mpsc(4096) -> SessionActor -> Handler");
    println!();

    let tcp_config = TcpServerConfig::new("127.0.0.1:8001")?;
    let websocket_config = WebSocketServerConfig::new("127.0.0.1:8002")?;
    let quic_config = QuicServerConfig::new("127.0.0.1:8003")?;

    // Create echo handler
    let handler = Arc::new(EchoHandler);

    // Build server with actor mode
    let transport = TransportServerBuilder::new()
        .max_connections(10000)
        .with_handler(handler) // Enable actor mode
        .actor_buffer_size(4096) // Buffer size per connection
        .with_protocol(tcp_config)
        .with_protocol(websocket_config)
        .with_protocol(quic_config)
        .build()
        .await?;

    println!("Listening on:");
    println!("  TCP:       127.0.0.1:8001");
    println!("  WebSocket: 127.0.0.1:8002");
    println!("  QUIC:      127.0.0.1:8003");
    println!();
    println!("Mode: Actor (per-connection mpsc)");
    println!("Server running... Press Ctrl+C to stop.");

    // In actor mode, we don't need to subscribe to events or spawn a dispatcher
    // Each connection has its own actor that calls our handler directly
    transport.serve().await?;

    Ok(())
}