Merge pull request #39 from cherish-ltt/v1.1.x

v1.1.14

Author: cherish-ltt

.gitignore

@@ -1,2 +1,3 @@
 /target
+*.DS_Store
 .DS_Store

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "lynn_tcp"
-version = "1.1.13"
+version = "1.1.14"
 edition = "2024"
 rust-version = "1.87"
 authors = ["lynn_tcp Contributors"]

Cargo.toml

@@ -38,21 +38,21 @@ Use `cargo add lynn_tcp` or:
 
 ```rust
 [dependencies]
-lynn_tcp = "=1.1.12"
+lynn_tcp = "=1.1.14"
 ```
 
 **server feature**
 
 ```rust
 [dependencies]
-lynn_tcp = { version = "=1.1.12" , features = "server" }
+lynn_tcp = { version = "=1.1.14" , features = "server" }
 ```
 
 **client feature**
 
 ```rust
 [dependencies]
-lynn_tcp = { version = "=1.1.12" , features = "client" }
+lynn_tcp = { version = "=1.1.14" , features = "client" }
 ```
 
 #### Server

README.md

@@ -6,7 +6,7 @@ use std::{
     time::{Duration, SystemTime},
 };
 
-use bytes::Bytes;
+use bytes::BytesMut;
 use tokio::{
     io::{AsyncReadExt, ReadHalf, split},
     net::TcpStream,
@@ -16,12 +16,11 @@ use tokio::{
 use tracing::{error, info, warn};
 
 use crate::{
-    app::ReactorEventSender,
+    app::{ReactorEventSender, event_api::event_api::ReactorEvent},
     const_config::{
         DEFAULT_MAX_RECEIVE_BYTES_SIZE, DEFAULT_MESSAGE_HEADER_MARK, DEFAULT_MESSAGE_TAIL_MARK,
         SERVER_MESSAGE_HEADER_MARK, SERVER_MESSAGE_TAIL_MARK,
     },
-    dto_factory::input_dto::{IHandlerCombinedTrait, MsgSelect},
     handler::{ClientsContext, HandlerContext},
     lynn_tcp_dependents::{HandlerResult, InputBufVO},
     vo_factory::big_buf::BigBufReader,
@@ -127,7 +126,7 @@ pub(crate) async fn check_handler_result(
 
 #[inline(always)]
 async fn send_response(
-    response: &Bytes,
+    response: &BytesMut,
     addrs: &Vec<SocketAddr>,
     clients: &ClientsStructType,
 ) -> Option<Vec<SocketAddr>> {
@@ -161,41 +160,26 @@ pub(crate) async fn input_dto_build(
     handler_method: Arc<AsyncFunc>,
     reactor_event_sender: ReactorEventSender,
 ) {
-    tokio::spawn(async move {
-        // Attempt to acquire a permit from the semaphore.
-        let result_permit = process_permit.try_acquire();
-        match result_permit {
-            Ok(permit) => {
-                // If the permit is acquired successfully, create a new `MsgSelect` instance and spawn a handler task.
-                let result = MsgSelect::new(
-                    addr,
-                    HandlerContext::new(
-                        input_buf_vo,
-                        ClientsContext::new(ClientsStruct(clients.clone())),
-                    ),
-                );
-                spawn_handler(result, clients, handler_method, reactor_event_sender).await;
-                // Release the permit after the handler task is completed.
-                drop(permit);
-            }
-            Err(_) => {
-                // If the permit cannot be acquired, log a warning.
-                warn!("addr:{} PROCESS_PERMIT_SIZE is full", addr)
-            }
+    // Attempt to acquire a permit from the semaphore.
+    let result_permit = process_permit.try_acquire();
+    match result_permit {
+        Ok(permit) => {
+            reactor_event_sender.push(ReactorEvent::crate_excute_task_event((
+                handler_method,
+                HandlerContext::new(
+                    input_buf_vo,
+                    ClientsContext::new(ClientsStruct(clients.clone())),
+                ),
+                clients,
+            )));
+            // Release the permit after the handler task is completed.
+            drop(permit);
         }
-    });
-}
-
-#[inline(always)]
-async fn spawn_handler(
-    mut result: MsgSelect,
-    clients: ClientsStructType,
-    handler_method: Arc<AsyncFunc>,
-    reactor_event_sender: ReactorEventSender,
-) {
-    result
-        .execute(clients, handler_method, reactor_event_sender)
-        .await;
+        Err(_) => {
+            // If the permit cannot be acquired, log a warning.
+            warn!("addr:{} PROCESS_PERMIT_SIZE is full", addr)
+        }
+    }
 }
 
 #[inline(always)]
@@ -260,13 +244,13 @@ pub(crate) async fn push_read_half(
                                     let guard = router_map_async.deref();
                                     if let Some(map) = guard {
                                         if map.contains_key(&method_id) {
-                                            let a = map.get(&method_id).unwrap();
+                                            let handler_method = map.get(&method_id).unwrap();
                                             input_dto_build(
                                                 addr,
                                                 input_buf_vo,
                                                 process_permit.clone(),
                                                 clients.clone(),
-                                                a.clone(),
+                                                handler_method.clone(),
                                                 reactor_event_sender.clone(),
                                             )
                                             .await;

src/app/common_api/mod.rs

@@ -2,8 +2,8 @@ use std::net::{SocketAddr, ToSocketAddrs};
 
 use crate::const_config::{
     DEFAULT_ADDR, DEFAULT_CHECK_HEART_INTERVAL, DEFAULT_CHECK_HEART_TIMEOUT_TIME,
-    DEFAULT_MAX_CONNECTIONS, DEFAULT_MAX_REACTOR_TASKPOOL_SIZE, DEFAULT_MAX_RECEIVE_BYTES_SIZE,
-    DEFAULT_MESSAGE_HEADER_MARK, DEFAULT_MESSAGE_TAIL_MARK, DEFAULT_PROCESS_PERMIT_SIZE,
+    DEFAULT_MAX_CONNECTIONS, DEFAULT_MAX_REACTOR_TASKPOOL_SIZE, DEFAULT_MESSAGE_HEADER_MARK,
+    DEFAULT_MESSAGE_TAIL_MARK, DEFAULT_PROCESS_PERMIT_SIZE,
 };
 
 /// Represents the configuration for the Lynn server.
@@ -18,8 +18,6 @@ pub struct LynnServerConfig<'a> {
     server_max_connections: Option<&'a usize>,
     // The maximum number of threads for the server.
     server_max_reactor_taskpool_size: &'a usize,
-    // The maximum number of bytes the server can receive.
-    server_max_receive_bytes_reader_size: &'a usize,
     // The permit size for a single process.
     server_single_processs_permit: &'a usize,
     // The interval for checking heartbeats.
@@ -36,50 +34,6 @@ pub struct LynnServerConfig<'a> {
 ///
 /// This implementation includes a constructor for the default configuration and methods to get the configuration parameters.
 impl<'a> LynnServerConfig<'a> {
-    /// Creates a new LynnServerConfig instance with the given parameters.
-    ///
-    /// # Parameters
-    ///
-    /// * `server_ipv4` - The IPv4 address of the server.
-    /// * `server_single_channel_size` - The size of a single channel.
-    /// * `server_max_connections` - The maximum number of connections for the server.
-    /// * `server_max_threadpool_size` - The maximum number of threads for the server.
-    /// * `server_max_receive_bytes_reader_size` - The maximum number of bytes the server can receive.
-    /// * `server_single_processs_permit` - The permit size for a single process.
-    /// * `server_check_heart_interval` - The interval for checking heartbeats.
-    /// * `server_check_heart_timeout_time` - The timeout time for checking heartbeats.
-    ///
-    /// # Returns
-    ///
-    /// A new LynnServerConfig instance.
-    fn new<T>(
-        server_addr: T,
-        server_max_connections: Option<&'a usize>,
-        server_max_reactor_taskpool_size: &'a usize,
-        server_max_receive_bytes_reader_size: &'a usize,
-        server_single_processs_permit: &'a usize,
-        server_check_heart_interval: &'a u64,
-        server_check_heart_timeout_time: &'a u64,
-        message_header_mark: &'a u16,
-        message_tail_mark: &'a u16,
-    ) -> Self
-    where
-        T: ToSocketAddrs,
-    {
-        let server_addr = server_addr.to_socket_addrs().unwrap().next().unwrap();
-        Self {
-            server_addr,
-            server_max_connections,
-            server_max_reactor_taskpool_size,
-            server_max_receive_bytes_reader_size,
-            server_single_processs_permit,
-            server_check_heart_interval,
-            server_check_heart_timeout_time,
-            message_header_mark,
-            message_tail_mark,
-        }
-    }
-
     /// Creates a default LynnServerConfig instance.
     ///
     /// # Returns
@@ -90,7 +44,6 @@ impl<'a> LynnServerConfig<'a> {
             server_addr: *DEFAULT_ADDR,
             server_max_connections: Some(&DEFAULT_MAX_CONNECTIONS),
             server_max_reactor_taskpool_size: &DEFAULT_MAX_REACTOR_TASKPOOL_SIZE,
-            server_max_receive_bytes_reader_size: &DEFAULT_MAX_RECEIVE_BYTES_SIZE,
             server_single_processs_permit: &DEFAULT_PROCESS_PERMIT_SIZE,
             server_check_heart_interval: &DEFAULT_CHECK_HEART_INTERVAL,
             server_check_heart_timeout_time: &DEFAULT_CHECK_HEART_TIMEOUT_TIME,
@@ -144,29 +97,10 @@ impl<'a> LynnServerConfig<'a> {
         self.server_max_connections
     }
 
-    /// Gets the maximum number of threads for the server.
-    ///
-    /// # Returns
-    ///
-    /// The maximum number of threads for the server.
-    #[deprecated(note = "use 'get_server_max_reactor_taskpool_size'", since = "1.1.12")]
-    pub(crate) fn get_server_max_threadpool_size(&self) -> &usize {
-        self.server_max_reactor_taskpool_size
-    }
-
     pub(crate) fn get_server_max_reactor_taskpool_size(&self) -> &usize {
         self.server_max_reactor_taskpool_size
     }
 
-    /// Gets the maximum number of bytes the server can receive.
-    ///
-    /// # Returns
-    ///
-    /// The maximum number of bytes the server can receive.
-    pub(crate) fn get_server_max_receive_bytes_reader_size(&self) -> &usize {
-        self.server_max_receive_bytes_reader_size
-    }
-
     /// Gets the mark for the message header.
     ///
     /// # Returns
@@ -343,24 +277,6 @@ impl<'a> LynnServerConfigBuilder<'a> {
         self
     }
 
-    /// Sets the maximum number of bytes the server can receive.
-    ///
-    /// # Parameters
-    ///
-    /// * `server_max_receive_bytes_reader_size` - The maximum number of bytes the server can receive.
-    ///
-    /// # Returns
-    ///
-    /// The updated `LynnServerConfigBuilder` instance.
-    pub fn with_server_max_receive_bytes_reader_size(
-        mut self,
-        server_max_receive_bytes_reader_size: &'a usize,
-    ) -> Self {
-        self.lynn_config.server_max_receive_bytes_reader_size =
-            server_max_receive_bytes_reader_size;
-        self
-    }
-
     /// Builds the `LynnServerConfig` instance.
     ///
     /// # Returns
@@ -379,7 +295,7 @@ impl<'a> LynnServerConfigBuilder<'a> {
     /// # Returns
     ///
     /// The updated `LynnServerConfigBuilder` instance.
-    pub(crate) fn with_message_header_mark(mut self, msg_header_mark: &'a u16) -> Self {
+    pub fn with_message_header_mark(mut self, msg_header_mark: &'a u16) -> Self {
         self.lynn_config.message_header_mark = msg_header_mark;
         self
     }
@@ -393,7 +309,7 @@ impl<'a> LynnServerConfigBuilder<'a> {
     /// # Returns
     ///
     /// The updated `LynnServerConfigBuilder` instance.
-    pub(crate) fn with_message_tail_mark(mut self, msg_tail_mark: &'a u16) -> Self {
+    pub fn with_message_tail_mark(mut self, msg_tail_mark: &'a u16) -> Self {
         self.lynn_config.message_tail_mark = msg_tail_mark;
         self
     }

src/app/lynn_server_config.rs

@@ -1,6 +1,6 @@
 use std::{sync::Arc, time::SystemTime};
 
-use bytes::Bytes;
+use bytes::BytesMut;
 use tokio::{
     io::{AsyncWriteExt, WriteHalf},
     net::TcpStream,
@@ -15,8 +15,6 @@ use tracing::error;
 pub(crate) struct LynnUser {
     /// The write_half used to send data to the client.
     write_half: *mut WriteHalf<TcpStream>,
-    /// An optional user ID.
-    user_id: Option<u64>,
     /// The last time the user communicated.
     last_communicate_time: Arc<RwLock<SystemTime>>,
     mutex: Mutex<()>,
@@ -45,7 +43,6 @@ impl LynnUser {
     ) -> Self {
         Self {
             write_half: Box::into_raw(Box::new(write_half)),
-            user_id: None,
             last_communicate_time,
             mutex: Mutex::new(()),
         }
@@ -60,7 +57,7 @@ impl LynnUser {
         self.last_communicate_time.clone()
     }
 
-    pub(crate) async fn send_response(&self, response: &Bytes) {
+    pub(crate) async fn send_response(&self, response: &BytesMut) {
         let _lock = self.mutex.lock().await;
         if !self.write_half.is_null() {
             if let Some(write_half) = unsafe { self.write_half.as_mut() } {

src/app/lynn_server_user.rs

@@ -6,7 +6,6 @@ mod tcp_reactor;
 use std::{
     collections::HashMap,
     net::{SocketAddr, ToSocketAddrs},
-    ops::DerefMut,
     sync::Arc,
 };
 
@@ -233,19 +232,6 @@ impl<'a> LynnServer<'a> {
         self.router_maps.0 = None;
     }
 
-    /// Removes a client from the server.
-    ///
-    /// # Parameters
-    ///
-    /// * `addr` - The address of the client to remove.
-    async fn remove_client(&mut self, addr: SocketAddr) {
-        let mut clients = self.clients.0.write().await;
-        let guard = clients.deref_mut();
-        if guard.contains_key(&addr) {
-            guard.remove(&addr);
-        }
-    }
-
     /// Checks the heartbeat of connected clients and removes those that have not sent messages for a long time.
     async fn check_heart(&self) {
         let clients = self.clients.0.clone();

src/app/mod.rs

@@ -131,6 +131,7 @@ impl EventManager {
     }
 }
 
+#[inline(always)]
 fn get_event(
     local_queue: &Worker<ReactorEvent>,
     global_queue: &ReactorEventSender,

src/app/tcp_reactor/event.rs

@@ -31,8 +31,8 @@ pub(crate) type NewSocketEventSender = Sender<(
 )>;
 
 pub(super) struct TcpReactor {
-    pub(crate) core_reactor: CoreReactor,
-    pub(crate) event_manager: EventManager,
+    core_reactor: CoreReactor,
+    event_manager: EventManager,
 }
 
 impl TcpReactor {