Bommels05/Modrinth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move ActiveSockets tuple into struct

Browse Source
This commit is contained in:
Josiah Glosson
2025-01-27 12:06:07 -06:00
parent b0883688bd
commit ab082e36ac
3 changed files with 44 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
apps/labrinth/src/queue/socket.rs
View File

@@ -4,13 +4,27 @@ use actix_ws::Session;
use dashmap::DashMap;
pub struct ActiveSockets {
pub auth_sockets: DashMap<UserId, (UserStatus, Session)>,
pub sockets: DashMap<UserId, ActiveSocket>,
}
impl Default for ActiveSockets {
fn default() -> Self {
Self {
auth_sockets: DashMap::new(),
sockets: DashMap::new(),
}
}
}
pub struct ActiveSocket {
pub status: UserStatus,
pub socket: Session,
}
impl ActiveSocket {
pub fn new(status: UserStatus, session: Session) -> Self {
Self {
status,
socket: session,
}
}
}

29
apps/labrinth/src/routes/internal/statuses.rs
View File

@@ -5,7 +5,7 @@ use crate::database::redis::RedisPool;
use crate::models::pats::Scopes;
use crate::models::users::User;
use crate::queue::session::AuthQueue;
use crate::queue::socket::ActiveSockets;
use crate::queue::socket::{ActiveSocket, ActiveSockets};
use crate::routes::ApiError;
use actix_web::web::{Data, Payload};
use actix_web::{get, web, HttpRequest, HttpResponse};
@@ -60,8 +60,9 @@ pub async fn ws_init(
let user = User::from_full(db_user);
if let Some((_, (_, session))) = db.auth_sockets.remove(&user.id) {
let _ = session.close(None).await;
if let Some((_, ActiveSocket { socket, .. })) = db.sockets.remove(&user.id)
{
let _ = socket.close(None).await;
}
let (res, mut session, msg_stream) = match actix_ws::handle(&req, body) {
@@ -83,7 +84,7 @@ pub async fn ws_init(
friends
.iter()
.filter_map(|x| {
db.auth_sockets.get(
db.sockets.get(
&if x.user_id == user.id.into() {
x.friend_id
} else {
@@ -92,7 +93,7 @@ pub async fn ws_init(
.into(),
)
})
.map(|x| x.value().0.clone())
.map(|x| x.value().status.clone())
.collect::<Vec<_>>()
} else {
Vec::new()
@@ -106,7 +107,8 @@ pub async fn ws_init(
)?)
.await;
db.auth_sockets.insert(user.id, (status.clone(), session));
db.sockets
.insert(user.id, ActiveSocket::new(status.clone(), session));
broadcast_friends(
user.id,
@@ -137,8 +139,8 @@ pub async fn ws_init(
}
Ok(Message::Ping(msg)) => {
if let Some(socket) = db.auth_sockets.get(&user.id) {
let (_, socket) = socket.value();
if let Some(socket) = db.sockets.get(&user.id) {
let ActiveSocket { socket, .. } = socket.value();
let _ = socket.clone().pong(&msg).await;
}
continue;
@@ -153,8 +155,8 @@ pub async fn ws_init(
match message.unwrap() {
ClientToServerMessage::StatusUpdate { profile_name } => {
if let Some(mut pair) = db.auth_sockets.get_mut(&user.id) {
let (status, _) = pair.value_mut();
if let Some(mut pair) = db.sockets.get_mut(&user.id) {
let ActiveSocket { status, .. } = pair.value_mut();
if status
.profile_name
@@ -219,8 +221,8 @@ pub async fn broadcast_friends(
};
if friend.accepted {
if let Some(socket) = sockets.auth_sockets.get(&friend_id.into()) {
let (_, socket) = socket.value();
if let Some(socket) = sockets.sockets.get(&friend_id.into()) {
let ActiveSocket { socket, .. } = socket.value();
// FIXME Probably shouldn't swallow sending errors
let _ = match message.serialize() {
@@ -242,7 +244,8 @@ pub async fn close_socket(
pool: &PgPool,
sockets: &ActiveSockets,
) -> Result<(), crate::database::models::DatabaseError> {
if let Some((_, (_, socket))) = sockets.auth_sockets.remove(&id) {
if let Some((_, ActiveSocket { socket, .. })) = sockets.sockets.remove(&id)
{
let _ = socket.close(None).await;
broadcast_friends(

22
apps/labrinth/src/routes/v3/friends.rs
View File

@@ -4,7 +4,7 @@ use crate::database::redis::RedisPool;
use crate::models::pats::Scopes;
use crate::models::users::UserFriend;
use crate::queue::session::AuthQueue;
use crate::queue::socket::ActiveSockets;
use crate::queue::socket::{ActiveSocket, ActiveSockets};
use crate::routes::internal::statuses::close_socket;
use crate::routes::ApiError;
use actix_web::{delete, get, post, web, HttpRequest, HttpResponse};
@@ -77,12 +77,14 @@ pub async fn add_friend(
friend_id: UserId,
sockets: &ActiveSockets,
) -> Result<(), ApiError> {
if let Some(pair) = sockets.auth_sockets.get(&user_id.into()) {
let (friend_status, _) = pair.value();
if let Some(socket) =
sockets.auth_sockets.get(&friend_id.into())
if let Some(pair) = sockets.sockets.get(&user_id.into()) {
let ActiveSocket {
status: friend_status,
..
} = pair.value();
if let Some(socket) = sockets.sockets.get(&friend_id.into())
{
let (_, socket) = socket.value();
let ActiveSocket { socket, .. } = socket.value();
let _ = socket
.clone()
@@ -122,8 +124,8 @@ pub async fn add_friend(
.insert(&mut transaction)
.await?;
if let Some(socket) = db.auth_sockets.get(&friend.id.into()) {
let (_, socket) = socket.value();
if let Some(socket) = db.sockets.get(&friend.id.into()) {
let ActiveSocket { socket, .. } = socket.value();
if socket
.clone()
@@ -179,8 +181,8 @@ pub async fn remove_friend(
)
.await?;
if let Some(socket) = db.auth_sockets.get(&friend.id.into()) {
let (_, socket) = socket.value();
if let Some(socket) = db.sockets.get(&friend.id.into()) {
let ActiveSocket { socket, .. } = socket.value();
let _ = socket
.clone()