bonkmaykr/ojrc
1
0
Fork 0
mirror of https://git.ngram.ca/OpenJam/rc-servers synced 2026-08-23 23:08:52 +00:00
Code Activity

Implement lobby parts of custom games #38

Browse Source
This commit is contained in:
NG (Graham)
2026-03-17 21:52:15 -04:00
parent af04c4093c
commit 6a953bc613
32 changed files with 1045 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
rc_lobby_room/src/events/handler.rs Normal file
View File

@@ -0,0 +1,74 @@
//use oj_rc_core::persist::user::IntercomListener;
use oj_rc_core::persist::user::intercom::IntercomLobbyStateMessage;
const RETRY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);
pub struct IntercomHandler<U: oj_rc_core::persist::user::Userless> {
userless: std::sync::Arc<U>,
lobby: std::sync::Arc<crate::QueueHandler>,
}
impl <U: oj_rc_core::persist::user::Userless + 'static> IntercomHandler<U> {
pub fn new(
userless: std::sync::Arc<U>,
lobby: std::sync::Arc<crate::QueueHandler>,
) -> Self {
Self {
userless,
lobby,
}
}
async fn run_loop(
userless: std::sync::Arc<U>,
lobby: std::sync::Arc<crate::QueueHandler>,
) {
loop {
let listener = match userless.lobby_state_listener().await {
Ok(listener) => {
log::debug!("Connected lobby state intercom listener");
listener
},
Err(e) => {
log::error!("Failed to connect to lobby state intercom: {} (retrying in {}s)", e, RETRY_TIMEOUT.as_secs());
tokio::time::sleep(RETRY_TIMEOUT).await;
continue;
}
};
use futures::StreamExt;
let mut listener = listener.listen().await;
while let Some(msg) = listener.next().await {
match msg {
Ok(msg) => {
match msg {
IntercomLobbyStateMessage::CustomGame(state) => {
if state.users.is_empty() {
// disband
lobby.remove_custom_queue(&state.session_id).await;
} else {
let is_create = lobby.update_custom_queue(
&state.session_id,
state.users.iter().map(|user| (user.public_id.clone(), user.team)),
state.config,
).await;
if is_create {
log::debug!("Created custom game {} lobby data to {} members", state.session_id, state.users.len());
} else {
log::debug!("Updated custom game {} lobby data to {} members", state.session_id, state.users.len());
}
}
},
}
},
Err(e) => {
log::error!("Bad intercom message received: {}", e);
}
}
}
}
}
pub fn run(self) -> tokio::task::JoinHandle<()> {
tokio::spawn(Self::run_loop(self.userless, self.lobby))
}
}

3
rc_lobby_room/src/events/mod.rs
View File

@@ -1,3 +1,6 @@
pub mod battle_found;
pub mod battle_enter;
pub mod enqueue_error;
mod handler;
pub use handler::IntercomHandler;

230
rc_lobby_room/src/lobby.rs
View File

@@ -63,8 +63,21 @@ struct QueueUser {
user: std::sync::Arc<Box<dyn oj_rc_core::persist::user::User<()> + Send + Sync>>,
}
struct CustomGameQueue {
config: oj_rc_core::persist::user::intercom::IntercomLobbyCustomGameConfig,
users: Vec<CustomGameQueueUser>,
}
struct CustomGameQueueUser {
public_id: String,
team: u8,
queue_user: Option<QueueUser>, // if None, the user is not enqueued
}
pub struct QueueHandler {
users_in_queue: std::sync::Arc<tokio::sync::Mutex<HashMap<QueueKey, Vec<QueueUser>>>>,
users_in_custom_games_queue: std::sync::Arc<tokio::sync::Mutex<HashMap<String, CustomGameQueue>>>,
custom_game_for_user: std::sync::Arc<tokio::sync::RwLock<HashMap<String, String>>>,
users_per_game: usize,
is_enabled: bool,
hostname: String,
@@ -84,6 +97,8 @@ impl QueueHandler {
let mp_settings = oj_rc_core::ConfigProvider::<()>::multiplayer_settings(conf);
Self {
users_in_queue: std::sync::Arc::new(tokio::sync::Mutex::new(HashMap::new())),
users_in_custom_games_queue: std::sync::Arc::new(tokio::sync::Mutex::new(HashMap::new())),
custom_game_for_user: std::sync::Arc::new(tokio::sync::RwLock::new(HashMap::new())),
users_per_game: oj_rc_core::ConfigProvider::<()>::players_per_game(conf),
is_enabled: mp_settings.is_enabled,
hostname: domain.to_owned(),
@@ -220,6 +235,7 @@ impl QueueHandler {
is_ranked: false,
is_custom: false,
is_complete: false,
overrides: None,
};
let team_picker = user.team_chooser(&game_desc).await;
/*let team_picker = match key.mode {
@@ -272,7 +288,88 @@ impl QueueHandler {
}
}
}
async fn enter_custom_match(
&self,
_session_id: &str,
session: &mut CustomGameQueue,
user: &(dyn oj_rc_core::persist::user::LobbyUser + Send + Sync),
) {
let mode = match session.config.game_mode {
oj_rc_core::persist::user::intercom::CustomGameMode::BattleArena => oj_rc_core::data::game_mode::GameMode::BattleArena,
oj_rc_core::persist::user::intercom::CustomGameMode::TeamDeathmatch => oj_rc_core::data::game_mode::GameMode::TeamDeathmatch,
oj_rc_core::persist::user::intercom::CustomGameMode::Pit => oj_rc_core::data::game_mode::GameMode::Pit,
oj_rc_core::persist::user::intercom::CustomGameMode::SuddenDeath => oj_rc_core::data::game_mode::GameMode::SuddenDeath,
};
let visibility = match session.config.map_visibility {
oj_rc_core::persist::user::intercom::CustomGameVisibility::Good => oj_rc_core::data::game_mode::MapVisibility::Good,
oj_rc_core::persist::user::intercom::CustomGameVisibility::Poor => oj_rc_core::data::game_mode::MapVisibility::Poor,
oj_rc_core::persist::user::intercom::CustomGameVisibility::Bad => oj_rc_core::data::game_mode::MapVisibility::Bad,
};
let key = QueueKey {
map: session.config.map.clone(),
mode,
visibility,
auto_heal: session.config.health_regen,
};
let guid_str = key.unique_guid();
let game_desc = oj_rc_core::persist::user::GameDescriptor {
guid: guid_str.clone(),
map: session.config.map.clone(),
mode,
visibility,
auto_heal: session.config.health_regen,
is_ranked: false,
is_custom: true,
is_complete: false,
overrides: Some(session.config.as_core()),
};
let player_descs = session.users.iter()
.map(|user| {
let q_user = user.queue_user.as_ref().unwrap();
oj_rc_core::persist::user::PlayerLobbyDescriptor {
user_id: q_user.user_id,
team: user.team as i32,
group: None,
public_id: q_user.player.name.clone(),
display_name: q_user.player.display_name.clone(),
}
})
.collect();
match user.start_custom_game(game_desc, player_descs).await {
Ok(_) => {
let player_datas = session.users.iter()
.map(|x| x.queue_user.as_ref().unwrap().player.clone())
.collect();
let enter_battle_ev = crate::events::battle_enter::BattleEnter {
host: self.hostname.clone(),
port: self.hostport,
map: session.config.map.clone(),
mode,
guid: guid_str.clone(),
is_ranked: false,
is_custom: true,
visibility: Some(key.visibility),
auto_heal: key.auto_heal,
player_datas,
network_config: self.network_conf.clone(),
};
let arc_event = std::sync::Arc::new(enter_battle_ev);
for user in session.users.iter_mut() {
let player = user.queue_user.take().unwrap();
tokio::spawn(Self::send_events_to_player(arc_event.clone(), player.emitter.clone()));
}
log::info!("{} players are entering custom match {}; {}", session.users.len(), guid_str, key.short());
},
Err(e) => {
if let Some(msg) = e.error_msg() {
log::error!("Cannot send enter battle events to players since LobbyUser.start_game(...) failed: {} ({})", msg, e.error_code());
} else {
log::error!("Cannot send enter battle events to players since LobbyUser.start_game(...) failed ({})", e.error_code());
}
}
}
}
async fn send_events_to_player(enter_event: std::sync::Arc<crate::events::battle_enter::BattleEnter>, sender: polariton_server::events::EventEmitter) {
@@ -283,6 +380,118 @@ impl QueueHandler {
}
}
pub async fn join_custom_queue(&self, user: std::sync::Arc<Box<dyn oj_rc_core::persist::user::User<()> + Send + Sync>>, event_emitter: polariton_server::events::EventEmitter) {
if !self.is_enabled {
event_emitter.emit(crate::events::enqueue_error::QueueJoinError {
code: oj_rc_core::data::error_codes::LobbyReasonCode::NoSuitableLobbyFound as i16,
text: "Multiplayer is not enabled".to_owned(),
});
return;
}
let public_id = user.public_id();
if let Some(session_id) = self.custom_game_for_user.read().await.get(public_id) {
let mut lock = self.users_in_custom_games_queue.lock().await;
let session = lock.get_mut(session_id).unwrap();
let lobby_user = user.as_ref().as_ref();
match lobby_user.player_data(&self.cpu_counter).await {
Ok(mut player_data) => {
let target_queuer = session.users.iter_mut()
.find(|user| user.public_id == public_id)
.unwrap();
player_data.team = target_queuer.team as i32;
let new_player = QueueUser {
emitter: event_emitter,
player: player_data,
user_id: oj_rc_core::persist::user::LobbyUser::user_id(lobby_user),
enqueued_at: chrono::Utc::now(),
user: user.clone(),
};
target_queuer.queue_user = Some(new_player);
},
Err(e) => {
event_emitter.emit(crate::events::enqueue_error::QueueJoinError {
code: e.error_code(),
text: e.error_msg().map(|x| x.to_owned()).unwrap_or_else(|| "Unknown queue join error".to_owned()),
});
return;
}
}
let is_all_enqueued = session.users.iter().all(|user| user.queue_user.is_some());
if is_all_enqueued {
self.enter_custom_match(session_id, session, lobby_user).await;
}
} else {
log::debug!("Rejecting join queue for unknown custom game for user {}", public_id);
event_emitter.emit(crate::events::enqueue_error::QueueJoinError {
code: oj_rc_core::data::error_codes::LobbyReasonCode::NoSuitableLobbyFound as i16,
text: "User is not in a custom game".to_owned(),
});
}
}
pub async fn remove_custom_queue(&self, session_id: &str) -> bool {
log::debug!("Disbanding custom game {}", session_id);
self.custom_game_for_user.write().await.retain(|_key, val| val != session_id);
let mut lock = self.users_in_custom_games_queue.lock().await;
lock.remove(session_id).is_some()
}
pub async fn update_custom_queue(&self, session_id: &str, members: impl std::iter::Iterator<Item = (String, u8)>, config: oj_rc_core::persist::user::intercom::IntercomLobbyCustomGameConfig) -> bool {
let mut lock = self.users_in_custom_games_queue.lock().await;
let members: std::collections::HashMap<_, _> = members.collect();
let mut user_map_lock = self.custom_game_for_user.write().await;
if let Some(session) = lock.get_mut(session_id) {
log::debug!("Updating existing custom game {}", session_id);
let member_ids: std::collections::HashSet<_> = members.keys().collect();
// remove users who are no longer part of the custom game
user_map_lock.retain(|key, val| (members.contains_key(key) && val == session_id) || (!members.contains_key(key) && val != session_id));
session.users.retain(|user| member_ids.contains(&user.public_id));
// update team assignments
session.users.iter_mut()
.for_each(|user| {
let new_team = *members.get(&user.public_id).unwrap();
user.team = new_team;
if let Some(q_user) = &mut user.queue_user {
q_user.player.team = new_team as i32;
}
});
// collect users which are still members
let existing_ids: std::collections::HashSet<_> = session.users.iter()
.map(|mem| mem.public_id.clone())
.collect();
// add new members
for (id, team) in members.iter() {
if !existing_ids.contains(id) {
session.users.push(CustomGameQueueUser {
public_id: id.to_owned(),
team: *team,
queue_user: None,
});
user_map_lock.insert(id.to_owned(), session_id.to_owned());
}
}
// update config overrides
session.config = config;
false
} else {
log::debug!("Creating new custom game {}", session_id);
lock.insert(session_id.to_owned(), CustomGameQueue {
config,
users: members.iter().map(|(mem_id, team)| {
user_map_lock.insert(mem_id.to_owned(), session_id.to_owned());
CustomGameQueueUser {
public_id: mem_id.to_owned(),
team: *team,
queue_user: None,
}
})
.collect()
});
true
}
}
pub async fn join_queue(&self, map: String, mode: oj_rc_core::data::game_mode::GameMode, visibility: oj_rc_core::data::game_mode::MapVisibility, auto_heal: bool, user: std::sync::Arc<Box<dyn oj_rc_core::persist::user::User<()> + Send + Sync>>, event_emitter: polariton_server::events::EventEmitter) {
if !self.is_enabled {
event_emitter.emit(crate::events::enqueue_error::QueueJoinError {
@@ -378,12 +587,23 @@ impl QueueHandler {
}
pub async fn leave_queue(&self, user: std::sync::Arc<Box<dyn oj_rc_core::persist::user::User<()> + Send + Sync>>) {
let public_id = user.public_id();
let user_id = oj_rc_core::persist::user::LobbyUser::user_id(user.as_ref().as_ref());
for (queue_key, queue) in self.users_in_queue.lock().await.iter_mut() {
if let Some((i, _)) = queue.iter().enumerate().find(|(_, user)| user.user_id == user_id) {
queue.remove(i);
log::info!("User {} was removed from queue {}", user_id, queue_key.short());
break;
if let Some(session_id) = self.custom_game_for_user.read().await.get(public_id) {
// player is in custom game session
let mut lock = self.users_in_custom_games_queue.lock().await;
let session = lock.get_mut(session_id).unwrap();
let target = session.users.iter_mut().find(|user| user.public_id == public_id).unwrap();
target.queue_user = None;
log::info!("User {} was removed from custom game {} queue", user_id, session_id);
} else {
// fallback to regular multiplayer
for (queue_key, queue) in self.users_in_queue.lock().await.iter_mut() {
if let Some((i, _)) = queue.iter().enumerate().find(|(_, user)| user.user_id == user_id) {
queue.remove(i);
log::info!("User {} was removed from queue {}", user_id, queue_key.short());
break;
}
}
}
}

5
rc_lobby_room/src/main.rs
View File

@@ -46,6 +46,11 @@ async fn main() -> std::io::Result<()> {
let server = std::sync::Arc::new(polariton_server::Server::new(operations::handler(&init_ctx), polariton_server::events::EventsHandler::new()));
crate::events::IntercomHandler::new(
init_ctx.users.clone(),
init_ctx.queue.clone(),
).run();
let ip_addr: std::net::IpAddr = args.ip.parse().expect("Invalid IP address");
let listener = net::TcpListener::bind(std::net::SocketAddr::new(ip_addr, args.port)).await?;

53
rc_lobby_room/src/operations/join_queue.rs
View File

@@ -30,24 +30,49 @@ impl <C: Send + 'static> SimpleOperation<C> for QueueJoinProvider {
if let Some(Typed::Bool(is_leader)) = params.remove(&IS_GROUP_LEADER_PARAM_KEY) {
if let Some(Typed::Int(lobby_ty)) = params.remove(&LOBBY_TY_PARAM_KEY) {
let lobby_ty = oj_rc_core::data::lobby::LobbyType::from_int(lobby_ty)?;
if let Some(Typed::Str(event_to_join)) = params.remove(&EVENT_TO_JOIN_PARAM_KEY) {
log::debug!("Got lobby join queue request of platoon {} ({} players, is_leader:{}) slot {} lobby {:?} event {}", group_id.string, group_size, is_leader, slot_id, lobby_ty, event_to_join.string);
params.insert(ESTIMATED_QUEUE_TIME_PARAM_KEY, Typed::Int(42));
params.insert(PERSONAL_RANKING_PARAM_KEY, Typed::Double(42.0));
let events = user.event_sender();
let user_info = user.user()?;
if let Some(current_lobby) = user_info.current_game_event_setter().get_multiplayer().await {
self.queue_handler.join_queue(
current_lobby.map,
current_lobby.mode,
current_lobby.visibility,
current_lobby.auto_heal,
log::debug!("Got lobby join queue request of platoon {} ({} players, is_leader:{}) slot {} lobby {:?}", group_id.string, group_size, is_leader, slot_id, lobby_ty);
match lobby_ty {
oj_rc_core::data::lobby::LobbyType::None => {
log::warn!("Cannot join queue for None mode (???)");
},
oj_rc_core::data::lobby::LobbyType::CustomGame => {
log::debug!("Joining platoon {} to custom game lobby", group_id.string);
params.insert(ESTIMATED_QUEUE_TIME_PARAM_KEY, Typed::Int(42));
params.insert(PERSONAL_RANKING_PARAM_KEY, Typed::Double(42.0));
let events = user.event_sender();
let user_info = user.user()?;
self.queue_handler.join_custom_queue(
user_info.clone(),
events.to_owned(),
).await;
}
},
oj_rc_core::data::lobby::LobbyType::QuickPlay => {
// regular multiplayer
if let Some(Typed::Str(event_to_join)) = params.remove(&EVENT_TO_JOIN_PARAM_KEY) {
log::debug!("Joining platoon {} to multiplayer lobby with event {}", group_id.string, event_to_join.string);
params.insert(ESTIMATED_QUEUE_TIME_PARAM_KEY, Typed::Int(42));
params.insert(PERSONAL_RANKING_PARAM_KEY, Typed::Double(42.0));
let events = user.event_sender();
let user_info = user.user()?;
if let Some(current_lobby) = user_info.current_game_event_setter().get_multiplayer().await {
self.queue_handler.join_queue(
current_lobby.map,
current_lobby.mode,
current_lobby.visibility,
current_lobby.auto_heal,
user_info.clone(),
events.to_owned(),
).await;
}
} else {
log::warn!("Missing multiplayer event, not joining platoon {} to multiplayer lobby queue", group_id.string);
}
},
oj_rc_core::data::lobby::LobbyType::Solo => {
log::warn!("Cannot join queue for solo mode (it's singleplayer!)");
},
}
}
}
}