server/game2006rust
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
azw 2024-05-12 22:36:16 +08:00
parent 5726c3ad04
commit 13204031ca
2 changed files with 58 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
server/stat/src/listener/wslistener.rs
View File

@ -15,7 +15,7 @@ use bytes::{BufMut, BytesMut};
use crate::app::UserApp; use crate::app::UserApp;
use crate::constant; use crate::constant;
use tokio::runtime::Runtime; use tokio::runtime::Runtime;
use crate::common::types::{DownStreamPack, DownStreamMessage, UpStreamPack}; use crate::common::types::{DownStreamMessage};
use std::sync::OnceLock; use std::sync::OnceLock;
use std::sync::atomic::{AtomicU16, Ordering}; use std::sync::atomic::{AtomicU16, Ordering};

63
server/stat/src/upstream/upstream.rs
View File

@ -21,35 +21,18 @@ pub struct UpStream {
async fn upstream_enter(stream_arc: Arc<tokio::sync::Mutex<TcpStream>>, down_pack_queue: Arc<std::sync::Mutex::<r9::QueueLock<DownStreamPack>>>, async fn upstream_enter(stream_arc: Arc<tokio::sync::Mutex<TcpStream>>, down_pack_queue: Arc<std::sync::Mutex::<r9::QueueLock<DownStreamPack>>>,
up_pack_queue: Arc<std::sync::Mutex::<r9::QueueLock<UpStreamPack>>>) { up_pack_queue: Arc<std::sync::Mutex::<r9::QueueLock<UpStreamPack>>>) {
tokio::spawn(async move { let on_read = |len: usize,
println!("upstream_enter2"); recv_buf_len: &mut usize,
let mut recv_buf = BytesMut::with_capacity(1024 * 64 * 3); recv_buf: &mut BytesMut,
let mut recv_buf_len = 0; data: &Vec<u8>,
{ up_pack_queue: &Arc<std::sync::Mutex::<r9::QueueLock<UpStreamPack>>>| {
for i in 0..recv_buf.capacity() {
recv_buf.put_i8(0);
}
}
loop {
tokio::time::sleep(Duration::from_millis(10)).await;
let mut stream = stream_arc.lock().await;
let ready = stream.ready(Interest::READABLE | Interest::WRITABLE).await;
match ready {
Ok(r) => {
{
if r.is_readable() {
// Try to read data, this may still fail with `WouldBlock`
// if the readiness event is a false positive.
let mut data = vec![0; 1024 * 64];
match stream.try_read(&mut data) {
Ok(len) => {
println!("read {0} bytes recv_buf_len:{1}",len, recv_buf_len); println!("read {0} bytes recv_buf_len:{1}",len, recv_buf_len);
for i in 0..len { for i in 0..len {
recv_buf[recv_buf_len + i] = data[i]; recv_buf[*recv_buf_len + i] = data[i];
//println!("sendmsg xxxxx idx:{0} val:{1} buf_val:{2} recv_buf_len:{3}", i, data[i], recv_buf[recv_buf_len + 1], recv_buf_len); //println!("sendmsg xxxxx idx:{0} val:{1} buf_val:{2} recv_buf_len:{3}", i, data[i], recv_buf[recv_buf_len + 1], recv_buf_len);
} }
recv_buf_len += len; *recv_buf_len += len;
while recv_buf_len >= 16 { while *recv_buf_len >= 16 {
let pack_len = (recv_buf[0] as u16) + ((recv_buf[1] as u16) << 8); let pack_len = (recv_buf[0] as u16) + ((recv_buf[1] as u16) << 8);
let msg_id = (recv_buf[2] as u16) + ((recv_buf[3] as u16) << 8); let msg_id = (recv_buf[2] as u16) + ((recv_buf[3] as u16) << 8);
let seq_id = (recv_buf[4] as i32) + ((recv_buf[5] as i32) << 8) + let seq_id = (recv_buf[4] as i32) + ((recv_buf[5] as i32) << 8) +
@ -58,7 +41,7 @@ async fn upstream_enter(stream_arc: Arc<tokio::sync::Mutex<TcpStream>>, down_pac
let rpc_error_code = (recv_buf[10] as u16) + ((recv_buf[11] as u16) << 8); let rpc_error_code = (recv_buf[10] as u16) + ((recv_buf[11] as u16) << 8);
let socket_handle = (recv_buf[12] as u16) + ((recv_buf[13] as u16) << 8); let socket_handle = (recv_buf[12] as u16) + ((recv_buf[13] as u16) << 8);
let ext_len = (recv_buf[14] as u16) + ((recv_buf[15] as u16) << 8); let ext_len = (recv_buf[14] as u16) + ((recv_buf[15] as u16) << 8);
if ((pack_len + 16) as usize) > recv_buf_len { if ((pack_len + 16) as usize) > *recv_buf_len {
break; break;
} }
let mut tmp_bytes = BytesMut::with_capacity((pack_len + 16) as usize); let mut tmp_bytes = BytesMut::with_capacity((pack_len + 16) as usize);
@ -88,11 +71,35 @@ async fn upstream_enter(stream_arc: Arc<tokio::sync::Mutex<TcpStream>>, down_pac
ext_len); ext_len);
} }
} }
for i in (pack_len + 16) as usize..recv_buf_len { for i in (pack_len + 16) as usize..*recv_buf_len {
recv_buf[i - (pack_len + 16) as usize] = recv_buf[i]; recv_buf[i - (pack_len + 16) as usize] = recv_buf[i];
} }
recv_buf_len -= (pack_len + 16) as usize; *recv_buf_len -= (pack_len + 16) as usize;
} }
};
tokio::spawn(async move {
println!("upstream_enter2");
let mut recv_buf = BytesMut::with_capacity(1024 * 64 * 3);
let mut recv_buf_len = 0;
{
for i in 0..recv_buf.capacity() {
recv_buf.put_i8(0);
}
}
loop {
tokio::time::sleep(Duration::from_millis(10)).await;
let mut stream = stream_arc.lock().await;
let ready = stream.ready(Interest::READABLE | Interest::WRITABLE).await;
match ready {
Ok(r) => {
{
if r.is_readable() {
// Try to read data, this may still fail with `WouldBlock`
// if the readiness event is a false positive.
let mut data = vec![0; 1024 * 64];
match stream.try_read(&mut data) {
Ok(len) => {
on_read(len, &mut recv_buf_len, &mut recv_buf, &data, &up_pack_queue);
} }
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
println!("read error 1"); println!("read error 1");