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-11 09:24:41 +08:00
parent 8ed90029b3
commit da9b5aff3f
2 changed files with 22 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
server/stat/src/app/user_app.rs
View File

@ -15,8 +15,8 @@ use r9::xtimer::TimerList;
pub struct DownStreamPack {
holder: Option<Rc::<RefCell::<DownStreamPack>>>,
socket_handle: u16,
data: bytes::BytesMut,
entry: Rc::<RefCell::<r9::ListHead<DownStreamPack>>>,
pub data: bytes::BytesMut,
pub entry: Rc::<RefCell::<r9::ListHead<DownStreamPack>>>,
}
unsafe impl Send for DownStreamPack {}

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

@ -27,6 +27,7 @@ use crate::{AppStateWithCounter, GSResponse, MAX_PACKET_LEN};
use tokio::net::TcpStream;
use tokio::runtime::Runtime;
use crate::app::user_app::DownStreamPack;
use tokio::io::{AsyncWriteExt, Ready};
#[derive(SharedFromSelf)]
#[derive(Singleton)]
@ -176,8 +177,10 @@ async fn index(req: HttpRequest, stream: web::Payload) -> Result<HttpResponse, E
resp
}
async fn upstream_enter(stream: TcpStream, down_pack_queue: Arc<Mutex::<r9::Queue<DownStreamPack>>>,) {
async fn upstream_enter(mut stream: TcpStream, down_pack_queue: Arc<Mutex::<r9::Queue<DownStreamPack>>>,) {
println!("upstream_enter1");
tokio::spawn(async move {
println!("upstream_enter2");
loop {
tokio::time::sleep(Duration::from_millis(10)).await;
let ready = stream.ready(Interest::READABLE | Interest::WRITABLE).await;
@ -211,18 +214,22 @@ async fn upstream_enter(stream: TcpStream, down_pack_queue: Arc<Mutex::<r9::Queu
{
if r.is_writable() {
/*
while Arc::downgrade(&mut send_queue).upgrade().unwrap().lock().unwrap().borrow().len() > 0 {
//Arc::downgrade(&mut send_queue).upgrade().unwrap().lock().unwrap().borrow_mut()[0]
let bytes = Arc::downgrade(&mut send_queue).upgrade().unwrap().lock().unwrap().borrow_mut().pop();
let len = bytes.clone().unwrap().clone().len();
{
//println!("sendmsg len:{0}", len);
let mut data = bytes::BytesMut::new();
{
down_pack_queue.lock().unwrap().fetch();
let work_list = down_pack_queue.lock().unwrap().work_list.lock().unwrap().clone();
while !work_list.borrow().empty() {
let node = &work_list.borrow().first_entry();
node.upgrade().unwrap().borrow_mut().entry.borrow_mut().del_init();
data = node.upgrade().unwrap().borrow().data.clone();
//stream.write_all(node.upgrade().unwrap().borrow().data.as_ref()).await;
//stream.write_all(data.as_ref()).await;
}
{
down_stream.write_all(bytes.unwrap().as_ref()).await;
}
}*/
}
if data.len() > 0 {
println!("write_all len:{0}", data.len());
stream.write_all(data.as_ref()).await;
}
}
}
}
@ -286,7 +293,7 @@ impl WsListener {
Ok(v) => {
//upstream.lock().unwrap().replace(v);
println!("connect ok");
upstream_enter(v, down_pack_queue);
upstream_enter(v, down_pack_queue).await;
},
Err(e) => {
println!("connect err")