lynn_tcp 1.2.5

Lightweight asynchronous TCP framework
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

use std::{net::SocketAddr, sync::Arc, time::SystemTime};

use tokio::{
    io::ReadHalf,
    net::{TcpListener, TcpStream},
    sync::{RwLock, Semaphore, mpsc::Sender},
};

use crate::app::{
    ClientsStructType, LynnRouter, ReactorEventSender,
    tcp_reactor::{event::EventManager, reactor::CoreReactor},
};

mod event;
mod reactor;

/// TCP socket configuration.
pub(super) struct TcpSocketConfig {
    pub nodelay: bool,
    pub keepalive_enabled: bool,
    pub keepalive_time_secs: u64,
    pub recv_buffer_size: usize,
    pub send_buffer_size: usize,
}

pub(crate) mod event_api {
    pub(crate) use super::event::*;
}

pub(crate) type NewSocketEventSender = Sender<(
    ReadHalf<TcpStream>,
    Arc<Semaphore>,
    SocketAddr,
    ClientsStructType,
    u16,
    u16,
    Arc<LynnRouter>,
    ReactorEventSender,
    Arc<RwLock<SystemTime>>,
)>;

pub(super) struct TcpReactor {
    core_reactor: CoreReactor,
    event_manager: EventManager,
}

impl TcpReactor {
    pub(super) fn new() -> Self {
        Self {
            core_reactor: CoreReactor::new(),
            event_manager: EventManager::new(),
        }
    }

    pub(super) async fn start(
        &self,
        clients: ClientsStructType,
        server_single_processs_permit: &usize,
        message_header_mark: u16,
        message_tail_mark: u16,
        lynn_router: Arc<LynnRouter>,
        tcp_listener: TcpListener,
        alow_max_connections: Option<&usize>,
        server_max_reactor_taskpool_size: &usize,
        connection_limiter: Option<(
            &u64,                // rate_limit
            &usize,               // max_connections_per_ip
            Arc<crate::app::connection_limiter::ConnectionLimiter>,
        )>,
        tcp_config: TcpSocketConfig,
    ) {
        self.event_manager.run(
            clients.clone(),
            server_single_processs_permit,
            message_header_mark,
            message_tail_mark,
            lynn_router,
            self.event_manager.get_global_queue(),
            self.core_reactor.tx.clone(),
            server_max_reactor_taskpool_size,
        );

        // Spawn cleanup task for connection limiter if rate limiting is enabled
        if let Some((rate_limit, _, limiter)) = &connection_limiter {
            if **rate_limit > 0 {
                limiter.clone().spawn_cleanup_task();
            }
        }

        self.core_reactor
            .run(
                tcp_listener,
                clients,
                alow_max_connections,
                self.event_manager.get_global_queue(),
                connection_limiter,
                tcp_config,
            )
            .await;
    }
}