feat & chore: clean up and push current progress on splatfest matchmaking

This commit is contained in:
DJMrTV 2025-05-14 09:52:24 +02:00
commit 7703aafe3c
32 changed files with 436 additions and 1181 deletions

View file

@ -97,7 +97,7 @@ impl Auth for AuthHandler {
source_login_data.0,
ticket.into(),
connection_data,
self.build_name.to_owned(),
format!("{}; Rust NEX Version {} by DJMrTV", self.build_name, env!("CARGO_PKG_VERSION")),
))
}

View file

@ -1,4 +1,5 @@
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::{Arc, Weak};
use std::sync::atomic::AtomicU32;
use std::sync::atomic::Ordering::{Relaxed, Release};
@ -7,7 +8,11 @@ use tokio::sync::{Mutex, RwLock};
use crate::kerberos::KerberosDateTime;
use crate::nex::user::User;
use crate::rmc::protocols::notifications::{NotificationEvent, RemoteNotification};
use crate::rmc::structures::matchmake::{Gathering, MatchmakeParam, MatchmakeSession};
use crate::rmc::protocols::notifications::notification_types::{HOST_CHANGED, OWNERSHIP_CHANGED};
use crate::rmc::response::ErrorCode;
use crate::rmc::response::ErrorCode::{Core_InvalidArgument, RendezVous_SessionVoid};
use crate::rmc::structures::matchmake::{Gathering, MatchmakeParam, MatchmakeSession, MatchmakeSessionSearchCriteria};
use crate::rmc::structures::matchmake::gathering_flags::PERSISTENT_GATHERING;
use crate::rmc::structures::variant::Variant;
pub struct MatchmakeManager{
@ -25,6 +30,19 @@ impl MatchmakeManager{
pub fn next_cid(&self) -> u32{
self.rv_cid_counter.fetch_add(1, Relaxed)
}
pub async fn get_session(&self, gid: u32) -> Result<Arc<Mutex<ExtendedMatchmakeSession>>, ErrorCode>{
let sessions = self.sessions.read().await;
let Some(session) = sessions.get(&gid) else {
return Err(RendezVous_SessionVoid);
};
let session = session.clone();
drop(sessions);
Ok(session)
}
}
@ -34,7 +52,34 @@ pub struct ExtendedMatchmakeSession{
pub connected_players: Vec<Weak<User>>,
}
fn read_bounds_string<T: FromStr>(str: &str) -> Option<(T,T)>{
let bounds = str.split_once(",")?;
Some((T::from_str(bounds.0).ok()?, T::from_str(bounds.1).ok()?))
}
fn check_bounds_str<T: FromStr + PartialOrd>(compare: T, str: &str) -> Option<bool>{
let bounds: (T, T) = read_bounds_string(str)?;
Some(bounds.0 <= compare && compare <= bounds.1)
}
pub async fn broadcast_notification<T: AsRef<User>>(players: &[T], notification_event: &NotificationEvent){
for player in players{
let player = player.as_ref();
player.remote.process_notification_event(notification_event.clone()).await;
}
}
impl ExtendedMatchmakeSession{
pub fn get_active_players(&self) -> Vec<Arc<User>>{
self.connected_players.iter().filter_map(|u| u.upgrade()).collect()
}
pub async fn broadcast_notification(&self, notification_event: &NotificationEvent){
broadcast_notification(&self.get_active_players(), notification_event).await;
}
pub async fn from_matchmake_session(gid: u32, session: MatchmakeSession, host: &Weak<User>) -> Self{
let Some(host) = host.upgrade() else{
return Default::default();
@ -66,18 +111,40 @@ impl ExtendedMatchmakeSession{
}
}
pub async fn add_player(&mut self, conn: Weak<User>, join_msg: String) {
let Some(arc_conn) = conn.upgrade() else {
pub async fn add_players(&mut self, conns: &[Weak<User>], join_msg: String) {
let Some(initiating_user) = conns[0].upgrade() else {
return
};
let joining_pid = arc_conn.pid;
let initiating_pid = initiating_user.pid;
let old_particip = self.connected_players.clone();
self.connected_players.push(conn);
for conn in conns {
self.connected_players.push(conn.clone());
}
self.session.participation_count = self.connected_players.len() as u32;
for other_connection in &self.connected_players[1..]{
let Some(other_conn) = other_connection.upgrade() else {
continue;
};
let other_pid = other_conn.pid;
/*if other_pid == self.session.gathering.owner_pid &&
joining_pid == self.session.gathering.owner_pid{
continue;
}*/
other_conn.remote.process_notification_event(NotificationEvent{
pid_source: initiating_pid,
notif_type: 122000,
param_1: self.session.gathering.self_gid,
param_2: other_pid,
str_param: "".into(),
param_3: 0
}).await;
}
for other_connection in &self.connected_players{
let Some(other_conn) = other_connection.upgrade() else {
@ -92,7 +159,7 @@ impl ExtendedMatchmakeSession{
}*/
other_conn.remote.process_notification_event(NotificationEvent{
pid_source: joining_pid,
pid_source: initiating_pid,
notif_type: 3001,
param_1: self.session.gathering.self_gid,
param_2: other_pid,
@ -109,8 +176,8 @@ impl ExtendedMatchmakeSession{
let older_pid = old_conns.pid;
arc_conn.remote.process_notification_event(NotificationEvent{
pid_source: joining_pid,
initiating_user.remote.process_notification_event(NotificationEvent{
pid_source: initiating_pid,
notif_type: 3001,
param_1: self.session.gathering.self_gid,
param_2: older_pid,
@ -119,4 +186,150 @@ impl ExtendedMatchmakeSession{
}).await;
}
}
#[inline]
pub fn is_reachable(&self) -> bool{
if self.session.gathering.flags & PERSISTENT_GATHERING != 0{
if !self.connected_players.is_empty(){
true
} else {
self.session.open_participation
}
} else {
!self.connected_players.is_empty()
}
}
#[inline]
pub fn is_joinable(&self) -> bool{
self.is_reachable() && self.session.open_participation
}
pub fn matches_criteria(&self, search_criteria: &MatchmakeSessionSearchCriteria) -> Result<bool, ErrorCode>{
// todo: implement the rest of the search criteria
if search_criteria.vacant_only {
if (self.connected_players.len() as u16 + search_criteria.vacant_participants) > self.session.gathering.maximum_participants{
return Ok(false);
}
}
if search_criteria.exclude_locked{
if !self.session.open_participation{
return Ok(false);
}
}
if search_criteria.exclude_system_password_set{
if self.session.system_password_enabled{
return Ok(false);
}
}
if search_criteria.exclude_user_password_set{
if self.session.user_password_enabled{
return Ok(false);
}
}
if !check_bounds_str(self.session.gathering.minimum_participants, &search_criteria.minimum_participants).ok_or(Core_InvalidArgument)? {
return Ok(false);
}
if !check_bounds_str(self.session.gathering.maximum_participants, &search_criteria.maximum_participants).ok_or(Core_InvalidArgument)? {
return Ok(false);
}
let game_mode: u32 = search_criteria.game_mode.parse().map_err(|_| Core_InvalidArgument)?;
if self.session.gamemode != game_mode{
return Ok(false);
}
let mm_sys_type: u32 = search_criteria.matchmake_system_type.parse().map_err(|_| Core_InvalidArgument)?;
if self.session.matchmake_system_type != mm_sys_type{
return Ok(false);
}
;
if search_criteria.attribs.get(0).map(|str| str.parse().ok()).flatten() != self.session.attributes.get(0).map(|v| *v){
return Ok(false);
}
if search_criteria.attribs.get(2).map(|str| str.parse().ok()).flatten() != self.session.attributes.get(2).map(|v| *v){
return Ok(false);
}
if search_criteria.attribs.get(3).map(|str| str.parse().ok()).flatten() != self.session.attributes.get(3).map(|v| *v){
return Ok(false);
}
Ok(true)
}
pub async fn migrate_ownership(&mut self, initiator_pid: u32) -> Result<(), ErrorCode>{
let players: Vec<_> = self.connected_players.iter().filter_map(|p| p.upgrade()).collect();
let Some(new_owner) = players.iter().find(|p| p.pid != self.session.gathering.owner_pid) else {
self.session.gathering.owner_pid = 0;
return Ok(());
};
self.session.gathering.owner_pid = new_owner.pid;
self.broadcast_notification(&NotificationEvent{
pid_source: initiator_pid,
notif_type: OWNERSHIP_CHANGED,
param_1: self.session.gathering.self_gid,
param_2: new_owner.pid,
..Default::default()
}).await;
Ok(())
}
pub async fn migrate_host(&mut self, initiator_pid: u32) -> Result<(), ErrorCode>{
let players: Vec<_> = self.connected_players.iter().filter_map(|p| p.upgrade()).collect();
self.session.gathering.host_pid = self.session.gathering.owner_pid;
self.broadcast_notification(&NotificationEvent{
pid_source: initiator_pid,
notif_type: HOST_CHANGED,
param_1: self.session.gathering.self_gid,
..Default::default()
}).await;
Ok(())
}
pub async fn remove_player_from_session(&mut self, pid: u32, message: &str) -> Result<(), ErrorCode>{
self.connected_players.retain(|u| u.upgrade().is_some_and(|u| u.pid != pid));
self.session.participation_count = (self.connected_players.len() & u32::MAX as usize) as u32;
if pid == self.session.gathering.owner_pid {
self.migrate_ownership(pid).await?;
}
if pid == self.session.gathering.host_pid {
self.migrate_host(pid).await?;
}
// todo: support DisconnectChangeOwner
// todo: finish the rest of this
for player in self.connected_players.iter().filter_map(|p| p.upgrade()){
player.remote.process_notification_event(NotificationEvent{
notif_type: 3008,
pid_source: pid,
param_1: self.session.gathering.self_gid,
param_2: pid,
str_param: message.to_owned(),
.. Default::default()
}).await;
}
Ok(())
}
}

View file

@ -19,8 +19,10 @@ use crate::rmc::protocols::nat_traversal::{
NatTraversal, RawNatTraversal, RawNatTraversalInfo, RemoteNatTraversal,
};
use crate::rmc::protocols::secure::{RawSecure, RawSecureInfo, RemoteSecure, Secure};
use crate::rmc::protocols::matchmake_ext::{MatchmakeExt, RawMatchmakeExt, RawMatchmakeExtInfo, RemoteMatchmakeExt};
use crate::rmc::response::ErrorCode;
use crate::rmc::structures::matchmake::{AutoMatchmakeParam, CreateMatchmakeSessionParam, JoinMatchmakeSessionParam, MatchmakeSession};
use crate::rmc::structures::qresult::QResult;
use macros::rmc_struct;
use std::net::{Ipv4Addr, SocketAddrV4};
@ -29,12 +31,14 @@ use log::{error, info};
use rocket::http::ext::IntoCollection;
use tokio::sync::{Mutex, RwLock};
use crate::prudp::station_url::nat_types::PUBLIC;
use crate::rmc::protocols::notifications::{NotificationEvent, RemoteNotification};
use crate::rmc::response::ErrorCode::{Core_Exception, Core_InvalidArgument, RendezVous_AccountExpired, RendezVous_SessionVoid};
define_rmc_proto!(
proto UserProtocol{
Secure,
MatchmakeExtension,
MatchmakeExt,
Matchmake,
NatTraversal
}
@ -140,10 +144,12 @@ impl Secure for User {
let mut lock = self.station_url.write().await;
*lock = vec![
public_station.clone(),
private_station
//private_station.clone()
];
drop(lock);
let result = QResult::success(ErrorCode::Core_Unknown);
@ -181,19 +187,22 @@ impl Secure for User {
}
impl MatchmakeExtension for User {
async fn close_participation(&self, gid: u32) -> Result<(), ErrorCode> {
let session = self.matchmake_manager.get_session(gid).await?;
let mut session = session.lock().await;
session.session.open_participation = false;
Ok(())
}
async fn get_playing_session(&self, pids: Vec<u32>) -> Result<Vec<()>, ErrorCode> {
Ok(Vec::new())
}
async fn update_progress_score(&self, gid: u32, progress: u8) -> Result<(), ErrorCode> {
let mut sessions = self.matchmake_manager.sessions.read().await;
let Some(session) = sessions.get(&gid) else {
return Err(RendezVous_SessionVoid);
};
let session = session.clone();
drop(sessions);
let session = self.matchmake_manager.get_session(gid).await?;
let mut session = session.lock().await;
@ -204,22 +213,34 @@ impl MatchmakeExtension for User {
async fn create_matchmake_session_with_param(
&self,
session: CreateMatchmakeSessionParam,
create_session_param: CreateMatchmakeSessionParam,
) -> Result<MatchmakeSession, ErrorCode> {
println!("{:?}", session);
println!("{:?}", create_session_param);
let gid = self.matchmake_manager.next_gid();
let mut new_session = ExtendedMatchmakeSession::from_matchmake_session(
gid,
session.matchmake_session,
create_session_param.matchmake_session,
&self.this.clone(),
)
.await;
new_session.session.participation_count = session.participation_count as u32;
let mut joining_players = vec![self.this.clone()];
let users = self.matchmake_manager.users.read().await;
for pid in create_session_param.additional_participants{
if let Some(user) = users.get(&pid){
joining_players.push(user.clone());
}
}
drop(users);
new_session.session.participation_count = create_session_param.participation_count as u32;
new_session
.add_player(self.this.clone(), session.join_message)
.add_players(&joining_players, create_session_param.join_message)
.await;
let session = new_session.session.clone();
@ -235,21 +256,26 @@ impl MatchmakeExtension for User {
&self,
join_session_param: JoinMatchmakeSessionParam,
) -> Result<MatchmakeSession, ErrorCode> {
let mut sessions = self.matchmake_manager.sessions.read().await;
let Some(session) = sessions.get(&join_session_param.gid) else {
return Err(ErrorCode::RendezVous_SessionVoid);
};
let session = session.clone();
drop(sessions);
let session = self.matchmake_manager.get_session(join_session_param.gid).await?;
let mut session = session.lock().await;
session.connected_players.retain(|v| v.upgrade().is_some_and(|v| v.pid != self.pid));
let mut joining_players = vec![self.this.clone()];
let users = self.matchmake_manager.users.read().await;
for pid in join_session_param.additional_participants{
if let Some(user) = users.get(&pid){
joining_players.push(user.clone());
}
}
drop(users);
session
.add_player(self.this.clone(), join_session_param.join_message)
.add_players(&joining_players, join_session_param.join_message)
.await;
let mm_session = session.session.clone();
@ -257,8 +283,51 @@ impl MatchmakeExtension for User {
Ok(mm_session)
}
async fn auto_matchmake_with_param_postpone(&self, session: AutoMatchmakeParam) -> Result<MatchmakeSession, ErrorCode> {
println!("{:?}", session.search_criteria);
async fn auto_matchmake_with_param_postpone(&self, param: AutoMatchmakeParam) -> Result<MatchmakeSession, ErrorCode> {
println!("{:?}", param);
let mut joining_players = vec![self.this.clone()];
let users = self.matchmake_manager.users.read().await;
for pid in &param.additional_participants{
if let Some(user) = users.get(pid){
joining_players.push(user.clone());
}
}
drop(users);
let sessions = self.matchmake_manager.sessions.read().await;
for session in sessions.values(){
let mut session = session.lock().await;
println!("checking session!");
if !session.is_joinable(){
continue;
}
let mut bool_matched_criteria = false;
for criteria in &param.search_criteria{
if session.matches_criteria(criteria)?{
bool_matched_criteria = true;
}
}
if bool_matched_criteria {
session.add_players(&joining_players, param.join_message).await;
return Ok(session.session.clone());
}
}
drop(sessions);
println!("making new session!");
let AutoMatchmakeParam{
join_message,
@ -267,7 +336,7 @@ impl MatchmakeExtension for User {
matchmake_session,
additional_participants,
..
} = session;
} = param;
self.create_matchmake_session_with_param(CreateMatchmakeSessionParam{
join_message,
@ -278,6 +347,13 @@ impl MatchmakeExtension for User {
additional_participants
}).await
}
async fn find_matchmake_session_by_gathering_id_detail(&self, gid: u32) -> Result<MatchmakeSession, ErrorCode> {
let session = self.matchmake_manager.get_session(gid).await?;
let session = session.lock().await;
Ok(session.session.clone())
}
}
impl Matchmake for User {
@ -285,15 +361,7 @@ impl Matchmake for User {
Ok(true)
}
async fn get_session_urls(&self, gid: u32) -> Result<Vec<StationUrl>, ErrorCode> {
let sessions = self.matchmake_manager.sessions.read().await;
let Some(session) = sessions.get(&gid) else {
return Err(ErrorCode::RendezVous_SessionVoid);
};
let session = session.clone();
drop(sessions);
let session = self.matchmake_manager.get_session(gid).await?;
let session = session.lock().await;
@ -315,6 +383,61 @@ impl Matchmake for User {
Ok(urls)
}
async fn update_session_host(&self, gid: u32, change_session_owner: bool) -> Result<(), ErrorCode> {
let session = self.matchmake_manager.get_session(gid).await?;
let mut session = session.lock().await;
session.session.gathering.host_pid = self.pid;
for player in &session.connected_players{
let Some(player) = player.upgrade() else {
continue;
};
player.remote.process_notification_event(NotificationEvent{
notif_type: 3008,
pid_source: self.pid,
param_1: gid,
param_2: self.pid,
param_3: 0,
str_param: "".to_string(),
}).await;
}
if change_session_owner{
session.session.gathering.owner_pid = self.pid;
for player in &session.connected_players{
let Some(player) = player.upgrade() else {
continue;
};
player.remote.process_notification_event(NotificationEvent{
notif_type: 4000,
pid_source: self.pid,
param_1: gid,
param_2: self.pid,
param_3: 0,
str_param: "".to_string(),
}).await;
}
}
Ok(())
}
}
impl MatchmakeExt for User {
async fn end_participation(&self, gid: u32, message: String) -> Result<bool, ErrorCode> {
let session = self.matchmake_manager.get_session(gid).await?;
let mut session = session.lock().await;
session.remove_player_from_session(self.pid, &message).await?;
Ok(true)
}
}
impl NatTraversal for User {
@ -340,6 +463,10 @@ impl NatTraversal for User {
Ok(())
}
async fn report_nat_traversal_result(&self, cid: u32, result: bool, rtt: u32) -> Result<(), ErrorCode> {
Ok(())
}
async fn request_probe_initiation(&self, station_to_probe: String) -> Result<(), ErrorCode> {
info!("NO!");
Err(RendezVous_AccountExpired)