LemmyNet/lemmy
2
0
Fork 0
mirror of https://github.com/LemmyNet/lemmy.git synced 2024-11-12 15:34:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fixing broken websocket sends. Removing WebSocketInfo (#1098)

Browse source
This commit is contained in:
Dessalines 2020-08-24 07:58:24 -04:00 committed by GitHub
parent 2080534744
commit 2d5a50e80f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 342 additions and 420 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

122
server/src/api/comment.rs
View file

@ -13,8 +13,8 @@ use crate::{
websocket::{
server::{JoinCommunityRoom, SendComment},
UserOperation,
WebsocketInfo,
},
ConnectionId,
DbPool,
LemmyContext,
LemmyError,
@ -129,7 +129,7 @@ impl Perform for CreateComment {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<CommentResponse, LemmyError> {
let data: &CreateComment = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -226,17 +226,15 @@ impl Perform for CreateComment {
form_id: data.form_id.to_owned(),
};
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendComment {
op: UserOperation::CreateComment,
comment: res.clone(),
my_id: ws.id,
});
context.chat_server().do_send(SendComment {
op: UserOperation::CreateComment,
comment: res.clone(),
websocket_id,
});
// strip out the recipient_ids, so that
// users don't get double notifs
res.recipient_ids = Vec::new();
}
// strip out the recipient_ids, so that
// users don't get double notifs
res.recipient_ids = Vec::new();
Ok(res)
}
@ -249,7 +247,7 @@ impl Perform for EditComment {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<CommentResponse, LemmyError> {
let data: &EditComment = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -311,17 +309,15 @@ impl Perform for EditComment {
form_id: data.form_id.to_owned(),
};
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendComment {
op: UserOperation::EditComment,
comment: res.clone(),
my_id: ws.id,
});
context.chat_server().do_send(SendComment {
op: UserOperation::EditComment,
comment: res.clone(),
websocket_id,
});
// strip out the recipient_ids, so that
// users don't get double notifs
res.recipient_ids = Vec::new();
}
// strip out the recipient_ids, so that
// users don't get double notifs
res.recipient_ids = Vec::new();
Ok(res)
}
@ -334,7 +330,7 @@ impl Perform for DeleteComment {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<CommentResponse, LemmyError> {
let data: &DeleteComment = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -398,17 +394,15 @@ impl Perform for DeleteComment {
form_id: None,
};
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendComment {
op: UserOperation::DeleteComment,
comment: res.clone(),
my_id: ws.id,
});
context.chat_server().do_send(SendComment {
op: UserOperation::DeleteComment,
comment: res.clone(),
websocket_id,
});
// strip out the recipient_ids, so that
// users don't get double notifs
res.recipient_ids = Vec::new();
}
// strip out the recipient_ids, so that
// users don't get double notifs
res.recipient_ids = Vec::new();
Ok(res)
}
@ -421,7 +415,7 @@ impl Perform for RemoveComment {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<CommentResponse, LemmyError> {
let data: &RemoveComment = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -495,17 +489,15 @@ impl Perform for RemoveComment {
form_id: None,
};
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendComment {
op: UserOperation::RemoveComment,
comment: res.clone(),
my_id: ws.id,
});
context.chat_server().do_send(SendComment {
op: UserOperation::RemoveComment,
comment: res.clone(),
websocket_id,
});
// strip out the recipient_ids, so that
// users don't get double notifs
res.recipient_ids = Vec::new();
}
// strip out the recipient_ids, so that
// users don't get double notifs
res.recipient_ids = Vec::new();
Ok(res)
}
@ -518,7 +510,7 @@ impl Perform for MarkCommentAsRead {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<CommentResponse, LemmyError> {
let data: &MarkCommentAsRead = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -590,7 +582,7 @@ impl Perform for SaveComment {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<CommentResponse, LemmyError> {
let data: &SaveComment = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -634,7 +626,7 @@ impl Perform for CreateCommentLike {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<CommentResponse, LemmyError> {
let data: &CreateCommentLike = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -726,17 +718,15 @@ impl Perform for CreateCommentLike {
form_id: None,
};
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendComment {
op: UserOperation::CreateCommentLike,
comment: res.clone(),
my_id: ws.id,
});
context.chat_server().do_send(SendComment {
op: UserOperation::CreateCommentLike,
comment: res.clone(),
websocket_id,
});
// strip out the recipient_ids, so that
// users don't get double notifs
res.recipient_ids = Vec::new();
}
// strip out the recipient_ids, so that
// users don't get double notifs
res.recipient_ids = Vec::new();
Ok(res)
}
@ -749,7 +739,7 @@ impl Perform for GetComments {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<GetCommentsResponse, LemmyError> {
let data: &GetComments = &self;
let user = get_user_from_jwt_opt(&data.auth, context.pool()).await?;
@ -777,17 +767,15 @@ impl Perform for GetComments {
Err(_) => return Err(APIError::err("couldnt_get_comments").into()),
};
if let Some(ws) = websocket_info {
if let Some(id) = websocket_id {
// You don't need to join the specific community room, bc this is already handled by
// GetCommunity
if data.community_id.is_none() {
if let Some(id) = ws.id {
// 0 is the "all" community
ws.chatserver.do_send(JoinCommunityRoom {
community_id: 0,
id,
});
}
// 0 is the "all" community
context.chat_server().do_send(JoinCommunityRoom {
community_id: 0,
id,
});
}
}

108
server/src/api/community.rs
View file

@ -6,8 +6,8 @@ use crate::{
websocket::{
server::{GetCommunityUsersOnline, JoinCommunityRoom, SendCommunityRoomMessage},
UserOperation,
WebsocketInfo,
},
ConnectionId,
};
use anyhow::Context;
use lemmy_db::{
@ -166,7 +166,7 @@ impl Perform for GetCommunity {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<GetCommunityResponse, LemmyError> {
let data: &GetCommunity = &self;
let user = get_user_from_jwt_opt(&data.auth, context.pool()).await?;
@ -205,20 +205,17 @@ impl Perform for GetCommunity {
Err(_e) => return Err(APIError::err("couldnt_find_community").into()),
};
let online = if let Some(ws) = websocket_info {
if let Some(id) = ws.id {
ws.chatserver.do_send(JoinCommunityRoom {
community_id: community.id,
id,
});
}
ws.chatserver
.send(GetCommunityUsersOnline { community_id })
.await
.unwrap_or(1)
} else {
0
};
if let Some(id) = websocket_id {
context
.chat_server()
.do_send(JoinCommunityRoom { community_id, id });
}
let online = context
.chat_server()
.send(GetCommunityUsersOnline { community_id })
.await
.unwrap_or(1);
let res = GetCommunityResponse {
community: community_view,
@ -238,7 +235,7 @@ impl Perform for CreateCommunity {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<CommunityResponse, LemmyError> {
let data: &CreateCommunity = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -333,7 +330,7 @@ impl Perform for EditCommunity {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<CommunityResponse, LemmyError> {
let data: &EditCommunity = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -403,7 +400,7 @@ impl Perform for EditCommunity {
community: community_view,
};
send_community_websocket(&res, websocket_info, UserOperation::EditCommunity);
send_community_websocket(&res, context, websocket_id, UserOperation::EditCommunity);
Ok(res)
}
@ -416,7 +413,7 @@ impl Perform for DeleteCommunity {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<CommunityResponse, LemmyError> {
let data: &DeleteCommunity = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -459,7 +456,7 @@ impl Perform for DeleteCommunity {
community: community_view,
};
send_community_websocket(&res, websocket_info, UserOperation::DeleteCommunity);
send_community_websocket(&res, context, websocket_id, UserOperation::DeleteCommunity);
Ok(res)
}
@ -472,7 +469,7 @@ impl Perform for RemoveCommunity {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<CommunityResponse, LemmyError> {
let data: &RemoveCommunity = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -527,7 +524,7 @@ impl Perform for RemoveCommunity {
community: community_view,
};
send_community_websocket(&res, websocket_info, UserOperation::RemoveCommunity);
send_community_websocket(&res, context, websocket_id, UserOperation::RemoveCommunity);
Ok(res)
}
@ -540,7 +537,7 @@ impl Perform for ListCommunities {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<ListCommunitiesResponse, LemmyError> {
let data: &ListCommunities = &self;
let user = get_user_from_jwt_opt(&data.auth, context.pool()).await?;
@ -582,7 +579,7 @@ impl Perform for FollowCommunity {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<CommunityResponse, LemmyError> {
let data: &FollowCommunity = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -643,7 +640,7 @@ impl Perform for GetFollowedCommunities {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<GetFollowedCommunitiesResponse, LemmyError> {
let data: &GetFollowedCommunities = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -670,7 +667,7 @@ impl Perform for BanFromCommunity {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<BanFromCommunityResponse, LemmyError> {
let data: &BanFromCommunity = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -757,14 +754,12 @@ impl Perform for BanFromCommunity {
banned: data.ban,
};
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendCommunityRoomMessage {
op: UserOperation::BanFromCommunity,
response: res.clone(),
community_id: data.community_id,
my_id: ws.id,
});
}
context.chat_server().do_send(SendCommunityRoomMessage {
op: UserOperation::BanFromCommunity,
response: res.clone(),
community_id,
websocket_id,
});
Ok(res)
}
@ -777,7 +772,7 @@ impl Perform for AddModToCommunity {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<AddModToCommunityResponse, LemmyError> {
let data: &AddModToCommunity = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -824,14 +819,12 @@ impl Perform for AddModToCommunity {
let res = AddModToCommunityResponse { moderators };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendCommunityRoomMessage {
op: UserOperation::AddModToCommunity,
response: res.clone(),
community_id: data.community_id,
my_id: ws.id,
});
}
context.chat_server().do_send(SendCommunityRoomMessage {
op: UserOperation::AddModToCommunity,
response: res.clone(),
community_id,
websocket_id,
});
Ok(res)
}
@ -844,7 +837,7 @@ impl Perform for TransferCommunity {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<GetCommunityResponse, LemmyError> {
let data: &TransferCommunity = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -957,20 +950,19 @@ impl Perform for TransferCommunity {
pub fn send_community_websocket(
res: &CommunityResponse,
websocket_info: Option<WebsocketInfo>,
context: &Data<LemmyContext>,
websocket_id: Option<ConnectionId>,
op: UserOperation,
) {
if let Some(ws) = websocket_info {
// Strip out the user id and subscribed when sending to others
let mut res_sent = res.clone();
res_sent.community.user_id = None;
res_sent.community.subscribed = None;
// Strip out the user id and subscribed when sending to others
let mut res_sent = res.clone();
res_sent.community.user_id = None;
res_sent.community.subscribed = None;
ws.chatserver.do_send(SendCommunityRoomMessage {
op,
response: res_sent,
community_id: res.community.id,
my_id: ws.id,
});
}
context.chat_server().do_send(SendCommunityRoomMessage {
op,
response: res_sent,
community_id: res.community.id,
websocket_id,
});
}

11
server/src/api/mod.rs
View file

@ -1,11 +1,4 @@
use crate::{
api::claims::Claims,
blocking,
websocket::WebsocketInfo,
DbPool,
LemmyContext,
LemmyError,
};
use crate::{api::claims::Claims, blocking, ConnectionId, DbPool, LemmyContext, LemmyError};
use actix_web::web::Data;
use lemmy_db::{
community::*,
@ -48,7 +41,7 @@ pub trait Perform {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<Self::Response, LemmyError>;
}

146
server/src/api/post.rs
View file

@ -15,8 +15,8 @@ use crate::{
websocket::{
server::{GetPostUsersOnline, JoinCommunityRoom, JoinPostRoom, SendPost},
UserOperation,
WebsocketInfo,
},
ConnectionId,
LemmyContext,
LemmyError,
};
@ -146,7 +146,7 @@ impl Perform for CreatePost {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<PostResponse, LemmyError> {
let data: &CreatePost = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -247,13 +247,11 @@ impl Perform for CreatePost {
let res = PostResponse { post: post_view };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendPost {
op: UserOperation::CreatePost,
post: res.clone(),
my_id: ws.id,
});
}
context.chat_server().do_send(SendPost {
op: UserOperation::CreatePost,
post: res.clone(),
websocket_id,
});
Ok(res)
}
@ -266,7 +264,7 @@ impl Perform for GetPost {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<GetPostResponse, LemmyError> {
let data: &GetPost = &self;
let user = get_user_from_jwt_opt(&data.auth, context.pool()).await?;
@ -304,20 +302,18 @@ impl Perform for GetPost {
})
.await??;
let online = if let Some(ws) = websocket_info {
if let Some(id) = ws.id {
ws.chatserver.do_send(JoinPostRoom {
post_id: data.id,
id,
});
}
ws.chatserver
.send(GetPostUsersOnline { post_id: data.id })
.await
.unwrap_or(1)
} else {
0
};
if let Some(id) = websocket_id {
context.chat_server().do_send(JoinPostRoom {
post_id: data.id,
id,
});
}
let online = context
.chat_server()
.send(GetPostUsersOnline { post_id: data.id })
.await
.unwrap_or(1);
// Return the jwt
Ok(GetPostResponse {
@ -337,7 +333,7 @@ impl Perform for GetPosts {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<GetPostsResponse, LemmyError> {
let data: &GetPosts = &self;
let user = get_user_from_jwt_opt(&data.auth, context.pool()).await?;
@ -377,17 +373,15 @@ impl Perform for GetPosts {
Err(_e) => return Err(APIError::err("couldnt_get_posts").into()),
};
if let Some(ws) = websocket_info {
if let Some(id) = websocket_id {
// You don't need to join the specific community room, bc this is already handled by
// GetCommunity
if data.community_id.is_none() {
if let Some(id) = ws.id {
// 0 is the "all" community
ws.chatserver.do_send(JoinCommunityRoom {
community_id: 0,
id,
});
}
// 0 is the "all" community
context.chat_server().do_send(JoinCommunityRoom {
community_id: 0,
id,
});
}
}
@ -402,7 +396,7 @@ impl Perform for CreatePostLike {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<PostResponse, LemmyError> {
let data: &CreatePostLike = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -465,13 +459,11 @@ impl Perform for CreatePostLike {
let res = PostResponse { post: post_view };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendPost {
op: UserOperation::CreatePostLike,
post: res.clone(),
my_id: ws.id,
});
}
context.chat_server().do_send(SendPost {
op: UserOperation::CreatePostLike,
post: res.clone(),
websocket_id,
});
Ok(res)
}
@ -484,7 +476,7 @@ impl Perform for EditPost {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<PostResponse, LemmyError> {
let data: &EditPost = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -560,13 +552,11 @@ impl Perform for EditPost {
let res = PostResponse { post: post_view };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendPost {
op: UserOperation::EditPost,
post: res.clone(),
my_id: ws.id,
});
}
context.chat_server().do_send(SendPost {
op: UserOperation::EditPost,
post: res.clone(),
websocket_id,
});
Ok(res)
}
@ -579,7 +569,7 @@ impl Perform for DeletePost {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<PostResponse, LemmyError> {
let data: &DeletePost = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -618,13 +608,11 @@ impl Perform for DeletePost {
let res = PostResponse { post: post_view };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendPost {
op: UserOperation::DeletePost,
post: res.clone(),
my_id: ws.id,
});
}
context.chat_server().do_send(SendPost {
op: UserOperation::DeletePost,
post: res.clone(),
websocket_id,
});
Ok(res)
}
@ -637,7 +625,7 @@ impl Perform for RemovePost {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<PostResponse, LemmyError> {
let data: &RemovePost = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -687,13 +675,11 @@ impl Perform for RemovePost {
let res = PostResponse { post: post_view };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendPost {
op: UserOperation::RemovePost,
post: res.clone(),
my_id: ws.id,
});
}
context.chat_server().do_send(SendPost {
op: UserOperation::RemovePost,
post: res.clone(),
websocket_id,
});
Ok(res)
}
@ -706,7 +692,7 @@ impl Perform for LockPost {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<PostResponse, LemmyError> {
let data: &LockPost = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -747,13 +733,11 @@ impl Perform for LockPost {
let res = PostResponse { post: post_view };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendPost {
op: UserOperation::LockPost,
post: res.clone(),
my_id: ws.id,
});
}
context.chat_server().do_send(SendPost {
op: UserOperation::LockPost,
post: res.clone(),
websocket_id,
});
Ok(res)
}
@ -766,7 +750,7 @@ impl Perform for StickyPost {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<PostResponse, LemmyError> {
let data: &StickyPost = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -811,13 +795,11 @@ impl Perform for StickyPost {
let res = PostResponse { post: post_view };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendPost {
op: UserOperation::StickyPost,
post: res.clone(),
my_id: ws.id,
});
}
context.chat_server().do_send(SendPost {
op: UserOperation::StickyPost,
post: res.clone(),
websocket_id,
});
Ok(res)
}
@ -830,7 +812,7 @@ impl Perform for SavePost {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<PostResponse, LemmyError> {
let data: &SavePost = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;

46
server/src/api/site.rs
View file

@ -15,8 +15,8 @@ use crate::{
websocket::{
server::{GetUsersOnline, SendAllMessage},
UserOperation,
WebsocketInfo,
},
ConnectionId,
LemmyContext,
LemmyError,
};
@ -167,7 +167,7 @@ impl Perform for ListCategories {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<ListCategoriesResponse, LemmyError> {
let _data: &ListCategories = &self;
@ -185,7 +185,7 @@ impl Perform for GetModlog {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<GetModlogResponse, LemmyError> {
let data: &GetModlog = &self;
@ -259,7 +259,7 @@ impl Perform for CreateSite {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<SiteResponse, LemmyError> {
let data: &CreateSite = &self;
@ -300,7 +300,7 @@ impl Perform for EditSite {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<SiteResponse, LemmyError> {
let data: &EditSite = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -337,13 +337,11 @@ impl Perform for EditSite {
let res = SiteResponse { site: site_view };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendAllMessage {
op: UserOperation::EditSite,
response: res.clone(),
my_id: ws.id,
});
}
context.chat_server().do_send(SendAllMessage {
op: UserOperation::EditSite,
response: res.clone(),
websocket_id,
});
Ok(res)
}
@ -356,7 +354,7 @@ impl Perform for GetSite {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<GetSiteResponse, LemmyError> {
let data: &GetSite = &self;
@ -375,7 +373,7 @@ impl Perform for GetSite {
captcha_uuid: None,
captcha_answer: None,
};
let login_response = register.perform(context, websocket_info.clone()).await?;
let login_response = register.perform(context, websocket_id).await?;
info!("Admin {} created", setup.admin_username);
let create_site = CreateSite {
@ -388,7 +386,7 @@ impl Perform for GetSite {
enable_nsfw: true,
auth: login_response.jwt,
};
create_site.perform(context, websocket_info.clone()).await?;
create_site.perform(context, websocket_id).await?;
info!("Site {} created", setup.site_name);
Some(blocking(context.pool(), move |conn| SiteView::read(conn)).await??)
} else {
@ -410,11 +408,11 @@ impl Perform for GetSite {
let banned = blocking(context.pool(), move |conn| UserView::banned(conn)).await??;
let online = if let Some(ws) = websocket_info {
ws.chatserver.send(GetUsersOnline).await.unwrap_or(1)
} else {
0
};
let online = context
.chat_server()
.send(GetUsersOnline)
.await
.unwrap_or(1);
let my_user = get_user_from_jwt_opt(&data.auth, context.pool())
.await?
@ -444,7 +442,7 @@ impl Perform for Search {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<SearchResponse, LemmyError> {
let data: &Search = &self;
@ -608,7 +606,7 @@ impl Perform for TransferSite {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<GetSiteResponse, LemmyError> {
let data: &TransferSite = &self;
let mut user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -671,7 +669,7 @@ impl Perform for GetSiteConfig {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<GetSiteConfigResponse, LemmyError> {
let data: &GetSiteConfig = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -692,7 +690,7 @@ impl Perform for SaveSiteConfig {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<GetSiteConfigResponse, LemmyError> {
let data: &SaveSiteConfig = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;

178
server/src/api/user.rs
View file

@ -14,8 +14,8 @@ use crate::{
websocket::{
server::{CaptchaItem, CheckCaptcha, JoinUserRoom, SendAllMessage, SendUserRoomMessage},
UserOperation,
WebsocketInfo,
},
ConnectionId,
LemmyContext,
LemmyError,
};
@ -303,7 +303,7 @@ impl Perform for Login {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<LoginResponse, LemmyError> {
let data: &Login = &self;
@ -338,7 +338,7 @@ impl Perform for Register {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<LoginResponse, LemmyError> {
let data: &Register = &self;
@ -357,27 +357,22 @@ impl Perform for Register {
// If its not the admin, check the captcha
if !data.admin && Settings::get().captcha.enabled {
match websocket_info {
Some(ws) => {
let check = ws
.chatserver
.send(CheckCaptcha {
uuid: data
.captcha_uuid
.to_owned()
.unwrap_or_else(|| "".to_string()),
answer: data
.captcha_answer
.to_owned()
.unwrap_or_else(|| "".to_string()),
})
.await?;
if !check {
return Err(APIError::err("captcha_incorrect").into());
}
}
None => return Err(APIError::err("captcha_incorrect").into()),
};
let check = context
.chat_server()
.send(CheckCaptcha {
uuid: data
.captcha_uuid
.to_owned()
.unwrap_or_else(|| "".to_string()),
answer: data
.captcha_answer
.to_owned()
.unwrap_or_else(|| "".to_string()),
})
.await?;
if !check {
return Err(APIError::err("captcha_incorrect").into());
}
}
check_slurs(&data.username)?;
@ -515,8 +510,8 @@ impl Perform for GetCaptcha {
async fn perform(
&self,
_context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
context: &Data<LemmyContext>,
_websocket_id: Option<ConnectionId>,
) -> Result<Self::Response, LemmyError> {
let captcha_settings = Settings::get().captcha;
@ -547,9 +542,8 @@ impl Perform for GetCaptcha {
expires: naive_now() + Duration::minutes(10), // expires in 10 minutes
};
if let Some(ws) = websocket_info {
ws.chatserver.do_send(captcha_item);
}
// Stores the captcha item on the queue
context.chat_server().do_send(captcha_item);
Ok(GetCaptchaResponse {
ok: Some(CaptchaResponse { png, uuid, wav }),
@ -564,7 +558,7 @@ impl Perform for SaveUserSettings {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<LoginResponse, LemmyError> {
let data: &SaveUserSettings = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -690,7 +684,7 @@ impl Perform for GetUserDetails {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<GetUserDetailsResponse, LemmyError> {
let data: &GetUserDetails = &self;
let user = get_user_from_jwt_opt(&data.auth, context.pool()).await?;
@ -788,7 +782,7 @@ impl Perform for AddAdmin {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<AddAdminResponse, LemmyError> {
let data: &AddAdmin = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -827,13 +821,11 @@ impl Perform for AddAdmin {
let res = AddAdminResponse { admins };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendAllMessage {
op: UserOperation::AddAdmin,
response: res.clone(),
my_id: ws.id,
});
}
context.chat_server().do_send(SendAllMessage {
op: UserOperation::AddAdmin,
response: res.clone(),
websocket_id,
});
Ok(res)
}
@ -846,7 +838,7 @@ impl Perform for BanUser {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<BanUserResponse, LemmyError> {
let data: &BanUser = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -909,13 +901,11 @@ impl Perform for BanUser {
banned: data.ban,
};
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendAllMessage {
op: UserOperation::BanUser,
response: res.clone(),
my_id: ws.id,
});
}
context.chat_server().do_send(SendAllMessage {
op: UserOperation::BanUser,
response: res.clone(),
websocket_id,
});
Ok(res)
}
@ -928,7 +918,7 @@ impl Perform for GetReplies {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<GetRepliesResponse, LemmyError> {
let data: &GetReplies = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -960,7 +950,7 @@ impl Perform for GetUserMentions {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<GetUserMentionsResponse, LemmyError> {
let data: &GetUserMentions = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -992,7 +982,7 @@ impl Perform for MarkUserMentionAsRead {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<UserMentionResponse, LemmyError> {
let data: &MarkUserMentionAsRead = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -1034,7 +1024,7 @@ impl Perform for MarkAllAsRead {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<GetRepliesResponse, LemmyError> {
let data: &MarkAllAsRead = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -1086,7 +1076,7 @@ impl Perform for DeleteAccount {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<LoginResponse, LemmyError> {
let data: &DeleteAccount = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -1123,7 +1113,7 @@ impl Perform for PasswordReset {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<PasswordResetResponse, LemmyError> {
let data: &PasswordReset = &self;
@ -1171,7 +1161,7 @@ impl Perform for PasswordChange {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<LoginResponse, LemmyError> {
let data: &PasswordChange = &self;
@ -1212,7 +1202,7 @@ impl Perform for CreatePrivateMessage {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<PrivateMessageResponse, LemmyError> {
let data: &CreatePrivateMessage = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -1290,14 +1280,12 @@ impl Perform for CreatePrivateMessage {
let res = PrivateMessageResponse { message };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendUserRoomMessage {
op: UserOperation::CreatePrivateMessage,
response: res.clone(),
recipient_id: recipient_user.id,
my_id: ws.id,
});
}
context.chat_server().do_send(SendUserRoomMessage {
op: UserOperation::CreatePrivateMessage,
response: res.clone(),
recipient_id,
websocket_id,
});
Ok(res)
}
@ -1310,7 +1298,7 @@ impl Perform for EditPrivateMessage {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<PrivateMessageResponse, LemmyError> {
let data: &EditPrivateMessage = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -1349,14 +1337,12 @@ impl Perform for EditPrivateMessage {
let res = PrivateMessageResponse { message };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendUserRoomMessage {
op: UserOperation::EditPrivateMessage,
response: res.clone(),
recipient_id,
my_id: ws.id,
});
}
context.chat_server().do_send(SendUserRoomMessage {
op: UserOperation::EditPrivateMessage,
response: res.clone(),
recipient_id,
websocket_id,
});
Ok(res)
}
@ -1369,7 +1355,7 @@ impl Perform for DeletePrivateMessage {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<PrivateMessageResponse, LemmyError> {
let data: &DeletePrivateMessage = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -1414,14 +1400,12 @@ impl Perform for DeletePrivateMessage {
let res = PrivateMessageResponse { message };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendUserRoomMessage {
op: UserOperation::DeletePrivateMessage,
response: res.clone(),
recipient_id,
my_id: ws.id,
});
}
context.chat_server().do_send(SendUserRoomMessage {
op: UserOperation::DeletePrivateMessage,
response: res.clone(),
recipient_id,
websocket_id,
});
Ok(res)
}
@ -1434,7 +1418,7 @@ impl Perform for MarkPrivateMessageAsRead {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<PrivateMessageResponse, LemmyError> {
let data: &MarkPrivateMessageAsRead = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -1472,14 +1456,12 @@ impl Perform for MarkPrivateMessageAsRead {
let res = PrivateMessageResponse { message };
if let Some(ws) = websocket_info {
ws.chatserver.do_send(SendUserRoomMessage {
op: UserOperation::MarkPrivateMessageAsRead,
response: res.clone(),
recipient_id,
my_id: ws.id,
});
}
context.chat_server().do_send(SendUserRoomMessage {
op: UserOperation::MarkPrivateMessageAsRead,
response: res.clone(),
recipient_id,
websocket_id,
});
Ok(res)
}
@ -1492,7 +1474,7 @@ impl Perform for GetPrivateMessages {
async fn perform(
&self,
context: &Data<LemmyContext>,
_websocket_info: Option<WebsocketInfo>,
_websocket_id: Option<ConnectionId>,
) -> Result<PrivateMessagesResponse, LemmyError> {
let data: &GetPrivateMessages = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
@ -1521,18 +1503,16 @@ impl Perform for UserJoin {
async fn perform(
&self,
context: &Data<LemmyContext>,
websocket_info: Option<WebsocketInfo>,
websocket_id: Option<ConnectionId>,
) -> Result<UserJoinResponse, LemmyError> {
let data: &UserJoin = &self;
let user = get_user_from_jwt(&data.auth, context.pool()).await?;
if let Some(ws) = websocket_info {
if let Some(id) = ws.id {
ws.chatserver.do_send(JoinUserRoom {
user_id: user.id,
id,
});
}
if let Some(ws_id) = websocket_id {
context.chat_server().do_send(JoinUserRoom {
user_id: user.id,
id: ws_id,
});
}
Ok(UserJoinResponse { user_id: user.id })

4
server/src/apub/inbox/activities/create.rs
View file

@ -75,7 +75,7 @@ async fn receive_create_post(
context.chat_server().do_send(SendPost {
op: UserOperation::CreatePost,
post: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(create, &user, context).await?;
@ -128,7 +128,7 @@ async fn receive_create_comment(
context.chat_server().do_send(SendComment {
op: UserOperation::CreateComment,
comment: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(create, &user, context).await?;

6
server/src/apub/inbox/activities/delete.rs
View file

@ -100,7 +100,7 @@ async fn receive_delete_post(
context.chat_server().do_send(SendPost {
op: UserOperation::EditPost,
post: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(delete, &user, context).await?;
@ -158,7 +158,7 @@ async fn receive_delete_comment(
context.chat_server().do_send(SendComment {
op: UserOperation::EditComment,
comment: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(delete, &user, context).await?;
@ -222,7 +222,7 @@ async fn receive_delete_community(
op: UserOperation::EditCommunity,
response: res,
community_id,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(delete, &user, context).await?;

4
server/src/apub/inbox/activities/dislike.rs
View file

@ -85,7 +85,7 @@ async fn receive_dislike_post(
context.chat_server().do_send(SendPost {
op: UserOperation::CreatePostLike,
post: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(dislike, &user, context).await?;
@ -142,7 +142,7 @@ async fn receive_dislike_comment(
context.chat_server().do_send(SendComment {
op: UserOperation::CreateCommentLike,
comment: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(dislike, &user, context).await?;

4
server/src/apub/inbox/activities/like.rs
View file

@ -76,7 +76,7 @@ async fn receive_like_post(like: Like, context: &LemmyContext) -> Result<HttpRes
context.chat_server().do_send(SendPost {
op: UserOperation::CreatePostLike,
post: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(like, &user, context).await?;
@ -127,7 +127,7 @@ async fn receive_like_comment(
context.chat_server().do_send(SendComment {
op: UserOperation::CreateCommentLike,
comment: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(like, &user, context).await?;

6
server/src/apub/inbox/activities/remove.rs
View file

@ -107,7 +107,7 @@ async fn receive_remove_post(
context.chat_server().do_send(SendPost {
op: UserOperation::EditPost,
post: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(remove, &mod_, context).await?;
@ -165,7 +165,7 @@ async fn receive_remove_comment(
context.chat_server().do_send(SendComment {
op: UserOperation::EditComment,
comment: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(remove, &mod_, context).await?;
@ -229,7 +229,7 @@ async fn receive_remove_community(
op: UserOperation::EditCommunity,
response: res,
community_id,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(remove, &mod_, context).await?;

20
server/src/apub/inbox/activities/undo.rs
View file

@ -202,7 +202,7 @@ async fn receive_undo_delete_comment(
context.chat_server().do_send(SendComment {
op: UserOperation::EditComment,
comment: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(undo, &user, context).await?;
@ -261,7 +261,7 @@ async fn receive_undo_remove_comment(
context.chat_server().do_send(SendComment {
op: UserOperation::EditComment,
comment: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(undo, &mod_, context).await?;
@ -321,7 +321,7 @@ async fn receive_undo_delete_post(
context.chat_server().do_send(SendPost {
op: UserOperation::EditPost,
post: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(undo, &user, context).await?;
@ -381,7 +381,7 @@ async fn receive_undo_remove_post(
context.chat_server().do_send(SendPost {
op: UserOperation::EditPost,
post: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(undo, &mod_, context).await?;
@ -446,7 +446,7 @@ async fn receive_undo_delete_community(
op: UserOperation::EditCommunity,
response: res,
community_id,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(undo, &user, context).await?;
@ -511,7 +511,7 @@ async fn receive_undo_remove_community(
op: UserOperation::EditCommunity,
response: res,
community_id,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(undo, &mod_, context).await?;
@ -556,7 +556,7 @@ async fn receive_undo_like_comment(
context.chat_server().do_send(SendComment {
op: UserOperation::CreateCommentLike,
comment: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(undo, &user, context).await?;
@ -595,7 +595,7 @@ async fn receive_undo_like_post(
context.chat_server().do_send(SendPost {
op: UserOperation::CreatePostLike,
post: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(undo, &user, context).await?;
@ -646,7 +646,7 @@ async fn receive_undo_dislike_comment(
context.chat_server().do_send(SendComment {
op: UserOperation::CreateCommentLike,
comment: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(undo, &user, context).await?;
@ -691,7 +691,7 @@ async fn receive_undo_dislike_post(
context.chat_server().do_send(SendPost {
op: UserOperation::CreatePostLike,
post: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(undo, &user, context).await?;

4
server/src/apub/inbox/activities/update.rs
View file

@ -81,7 +81,7 @@ async fn receive_update_post(
context.chat_server().do_send(SendPost {
op: UserOperation::EditPost,
post: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(update, &user, context).await?;
@ -136,7 +136,7 @@ async fn receive_update_comment(
context.chat_server().do_send(SendComment {
op: UserOperation::EditComment,
comment: res,
my_id: None,
websocket_id: None,
});
announce_if_community_is_local(update, &user, context).await?;

8
server/src/apub/inbox/user_inbox.rs
View file

@ -152,7 +152,7 @@ async fn receive_create_private_message(
op: UserOperation::CreatePrivateMessage,
response: res,
recipient_id,
my_id: None,
websocket_id: None,
});
Ok(HttpResponse::Ok().finish())
@ -201,7 +201,7 @@ async fn receive_update_private_message(
op: UserOperation::EditPrivateMessage,
response: res,
recipient_id,
my_id: None,
websocket_id: None,
});
Ok(HttpResponse::Ok().finish())
@ -262,7 +262,7 @@ async fn receive_delete_private_message(
op: UserOperation::EditPrivateMessage,
response: res,
recipient_id,
my_id: None,
websocket_id: None,
});
Ok(HttpResponse::Ok().finish())
@ -323,7 +323,7 @@ async fn receive_undo_delete_private_message(
op: UserOperation::EditPrivateMessage,
response: res,
recipient_id,
my_id: None,
websocket_id: None,
});
Ok(HttpResponse::Ok().finish())

7
server/src/main.rs
View file

@ -74,11 +74,12 @@ async fn main() -> Result<(), LemmyError> {
settings.bind, settings.port
);
let chat_server =
ChatServer::startup(pool.clone(), rate_limiter.clone(), Client::default()).start();
// Create Http server with websocket support
HttpServer::new(move || {
let chat_server =
ChatServer::startup(pool.clone(), rate_limiter.clone(), Client::default()).start();
let context = LemmyContext::create(pool.clone(), chat_server, Client::default());
let context = LemmyContext::create(pool.clone(), chat_server.to_owned(), Client::default());
let settings = Settings::get();
let rate_limiter = rate_limiter.clone();
App::new()

8
server/src/routes/api.rs
View file

@ -1,7 +1,6 @@
use crate::{
api::{comment::*, community::*, post::*, site::*, user::*, Perform},
rate_limit::RateLimit,
websocket::WebsocketInfo,
LemmyContext,
};
use actix_web::{error::ErrorBadRequest, *};
@ -182,13 +181,8 @@ where
Request: Perform,
Request: Send + 'static,
{
let ws_info = WebsocketInfo {
chatserver: context.chat_server().to_owned(),
id: None,
};
let res = data
.perform(&context, Some(ws_info))
.perform(&context, None)
.await
.map(|json| HttpResponse::Ok().json(json))
.map_err(ErrorBadRequest)?;

8
server/src/websocket/mod.rs
View file

@ -1,6 +1,5 @@
pub mod server;
use crate::ConnectionId;
use actix::prelude::*;
use diesel::{
r2d2::{ConnectionManager, Pool},
@ -10,7 +9,6 @@ use log::{error, info};
use rand::{rngs::ThreadRng, Rng};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use server::ChatServer;
use std::{
collections::{HashMap, HashSet},
str::FromStr,
@ -77,9 +75,3 @@ pub enum UserOperation {
GetSiteConfig,
SaveSiteConfig,
}
#[derive(Clone)]
pub struct WebsocketInfo {
pub chatserver: Addr<ChatServer>,
pub id: Option<ConnectionId>,
}

72
server/src/websocket/server.rs
View file

@ -58,7 +58,7 @@ pub struct StandardMessage {
pub struct SendAllMessage<Response> {
pub op: UserOperation,
pub response: Response,
pub my_id: Option<ConnectionId>,
pub websocket_id: Option<ConnectionId>,
}
#[derive(Message)]
@ -67,7 +67,7 @@ pub struct SendUserRoomMessage<Response> {
pub op: UserOperation,
pub response: Response,
pub recipient_id: UserId,
pub my_id: Option<ConnectionId>,
pub websocket_id: Option<ConnectionId>,
}
#[derive(Message)]
@ -76,7 +76,7 @@ pub struct SendCommunityRoomMessage<Response> {
pub op: UserOperation,
pub response: Response,
pub community_id: CommunityId,
pub my_id: Option<ConnectionId>,
pub websocket_id: Option<ConnectionId>,
}
#[derive(Message)]
@ -84,7 +84,7 @@ pub struct SendCommunityRoomMessage<Response> {
pub struct SendPost {
pub op: UserOperation,
pub post: PostResponse,
pub my_id: Option<ConnectionId>,
pub websocket_id: Option<ConnectionId>,
}
#[derive(Message)]
@ -92,7 +92,7 @@ pub struct SendPost {
pub struct SendComment {
pub op: UserOperation,
pub comment: CommentResponse,
pub my_id: Option<ConnectionId>,
pub websocket_id: Option<ConnectionId>,
}
#[derive(Message)]
@ -285,7 +285,7 @@ impl ChatServer {
op: &UserOperation,
response: &Response,
post_id: PostId,
my_id: Option<ConnectionId>,
websocket_id: Option<ConnectionId>,
) -> Result<(), LemmyError>
where
Response: Serialize,
@ -293,7 +293,7 @@ impl ChatServer {
let res_str = &to_json_string(op, response)?;
if let Some(sessions) = self.post_rooms.get(&post_id) {
for id in sessions {
if let Some(my_id) = my_id {
if let Some(my_id) = websocket_id {
if *id == my_id {
continue;
}
@ -309,7 +309,7 @@ impl ChatServer {
op: &UserOperation,
response: &Response,
community_id: CommunityId,
my_id: Option<ConnectionId>,
websocket_id: Option<ConnectionId>,
) -> Result<(), LemmyError>
where
Response: Serialize,
@ -317,7 +317,7 @@ impl ChatServer {
let res_str = &to_json_string(op, response)?;
if let Some(sessions) = self.community_rooms.get(&community_id) {
for id in sessions {
if let Some(my_id) = my_id {
if let Some(my_id) = websocket_id {
if *id == my_id {
continue;
}
@ -332,14 +332,14 @@ impl ChatServer {
&self,
op: &UserOperation,
response: &Response,
my_id: Option<ConnectionId>,
websocket_id: Option<ConnectionId>,
) -> Result<(), LemmyError>
where
Response: Serialize,
{
let res_str = &to_json_string(op, response)?;
for id in self.sessions.keys() {
if let Some(my_id) = my_id {
if let Some(my_id) = websocket_id {
if *id == my_id {
continue;
}
@ -354,7 +354,7 @@ impl ChatServer {
op: &UserOperation,
response: &Response,
recipient_id: UserId,
my_id: Option<ConnectionId>,
websocket_id: Option<ConnectionId>,
) -> Result<(), LemmyError>
where
Response: Serialize,
@ -362,7 +362,7 @@ impl ChatServer {
let res_str = &to_json_string(op, response)?;
if let Some(sessions) = self.user_rooms.get(&recipient_id) {
for id in sessions {
if let Some(my_id) = my_id {
if let Some(my_id) = websocket_id {
if *id == my_id {
continue;
}
@ -377,7 +377,7 @@ impl ChatServer {
&self,
user_operation: &UserOperation,
comment: &CommentResponse,
my_id: Option<ConnectionId>,
websocket_id: Option<ConnectionId>,
) -> Result<(), LemmyError> {
let mut comment_reply_sent = comment.clone();
comment_reply_sent.comment.my_vote = None;
@ -391,21 +391,26 @@ impl ChatServer {
user_operation,
&comment_post_sent,
comment_post_sent.comment.post_id,
my_id,
websocket_id,
)?;
// Send it to the recipient(s) including the mentioned users
for recipient_id in &comment_reply_sent.recipient_ids {
self.send_user_room_message(user_operation, &comment_reply_sent, *recipient_id, my_id)?;
self.send_user_room_message(
user_operation,
&comment_reply_sent,
*recipient_id,
websocket_id,
)?;
}
// Send it to the community too
self.send_community_room_message(user_operation, &comment_post_sent, 0, my_id)?;
self.send_community_room_message(user_operation, &comment_post_sent, 0, websocket_id)?;
self.send_community_room_message(
user_operation,
&comment_post_sent,
comment.comment.community_id,
my_id,
websocket_id,
)?;
Ok(())
@ -415,7 +420,7 @@ impl ChatServer {
&self,
user_operation: &UserOperation,
post: &PostResponse,
my_id: Option<ConnectionId>,
websocket_id: Option<ConnectionId>,
) -> Result<(), LemmyError> {
let community_id = post.post.community_id;
@ -425,11 +430,11 @@ impl ChatServer {
post_sent.post.user_id = None;
// Send it to /c/all and that community
self.send_community_room_message(user_operation, &post_sent, 0, my_id)?;
self.send_community_room_message(user_operation, &post_sent, community_id, my_id)?;
self.send_community_room_message(user_operation, &post_sent, 0, websocket_id)?;
self.send_community_room_message(user_operation, &post_sent, community_id, websocket_id)?;
// Send it to the post room
self.send_post_room_message(user_operation, &post_sent, post.post.id, my_id)?;
self.send_post_room_message(user_operation, &post_sent, post.post.id, websocket_id)?;
Ok(())
}
@ -471,7 +476,7 @@ impl ChatServer {
client,
};
let args = Args {
context: &context,
context,
rate_limiter,
id: msg.id,
ip,
@ -565,7 +570,7 @@ impl ChatServer {
}
struct Args<'a> {
context: &'a LemmyContext,
context: LemmyContext,
rate_limiter: RateLimit,
id: ConnectionId,
ip: IPAddr,
@ -587,18 +592,13 @@ where
data,
} = args;
let ws_info = WebsocketInfo {
chatserver: context.chat_server().to_owned(),
id: Some(id),
};
let data = data.to_string();
let op2 = op.clone();
let fut = async move {
let parsed_data: Data = serde_json::from_str(&data)?;
let res = parsed_data
.perform(&web::Data::new(context.to_owned()), Some(ws_info))
.perform(&web::Data::new(context), Some(id))
.await?;
to_json_string(&op, &res)
};
@ -692,7 +692,7 @@ where
fn handle(&mut self, msg: SendAllMessage<Response>, _: &mut Context<Self>) {
self
.send_all_message(&msg.op, &msg.response, msg.my_id)
.send_all_message(&msg.op, &msg.response, msg.websocket_id)
.ok();
}
}
@ -705,7 +705,7 @@ where
fn handle(&mut self, msg: SendUserRoomMessage<Response>, _: &mut Context<Self>) {
self
.send_user_room_message(&msg.op, &msg.response, msg.recipient_id, msg.my_id)
.send_user_room_message(&msg.op, &msg.response, msg.recipient_id, msg.websocket_id)
.ok();
}
}
@ -718,7 +718,7 @@ where
fn handle(&mut self, msg: SendCommunityRoomMessage<Response>, _: &mut Context<Self>) {
self
.send_community_room_message(&msg.op, &msg.response, msg.community_id, msg.my_id)
.send_community_room_message(&msg.op, &msg.response, msg.community_id, msg.websocket_id)
.ok();
}
}
@ -727,7 +727,7 @@ impl Handler<SendPost> for ChatServer {
type Result = ();
fn handle(&mut self, msg: SendPost, _: &mut Context<Self>) {
self.send_post(&msg.op, &msg.post, msg.my_id).ok();
self.send_post(&msg.op, &msg.post, msg.websocket_id).ok();
}
}
@ -735,7 +735,9 @@ impl Handler<SendComment> for ChatServer {
type Result = ();
fn handle(&mut self, msg: SendComment, _: &mut Context<Self>) {
self.send_comment(&msg.op, &msg.comment, msg.my_id).ok();
self
.send_comment(&msg.op, &msg.comment, msg.websocket_id)
.ok();
}
}