jmug/stoatchat
1
0
Fork 1
Code Issues Packages Projects Releases Wiki Activity

Run cargo fmt.

Browse Source
This commit is contained in:
Paul Makles
2021-01-18 20:26:26 +00:00
parent bd789b6825
commit a1a921bbcb
19 changed files with 141 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/notifications/events.rs
View File

@@ -33,7 +33,7 @@ pub enum ClientboundNotification {
Authenticated,
Ready {
users: Vec<User>,
channels: Vec<Channel>
channels: Vec<Channel>,
},
Message(Message),
@@ -93,7 +93,6 @@ pub enum ClientboundNotification {
GuildDelete {
id: String,
},*/
UserRelationship {
id: String,
user: String,
@@ -102,8 +101,8 @@ pub enum ClientboundNotification {
UserPresence {
id: String,
online: bool
}
online: bool,
},
}
impl ClientboundNotification {

29
src/notifications/payload.rs
View File

@@ -4,7 +4,10 @@ use crate::{
util::result::{Error, Result},
};
use futures::StreamExt;
use mongodb::{bson::{Bson, doc, from_bson, from_document}, options::FindOptions};
use mongodb::{
bson::{doc, from_bson, from_document, Bson},
options::FindOptions,
};
use super::websocket::is_online;
@@ -36,11 +39,10 @@ pub async fn generate_ready(mut user: User) -> Result<ClientboundNotification> {
while let Some(result) = cursor.next().await {
if let Ok(doc) = result {
let mut user: User =
from_document(doc).map_err(|_| Error::DatabaseError {
operation: "from_document",
with: "user",
})?;
let mut user: User = from_document(doc).map_err(|_| Error::DatabaseError {
operation: "from_document",
with: "user",
})?;
user.relationship = Some(
relationships
@@ -77,24 +79,21 @@ pub async fn generate_ready(mut user: User) -> Result<ClientboundNotification> {
}
]
},
None
None,
)
.await
.map_err(|_| Error::DatabaseError {
operation: "find",
with: "channels",
})?;
let mut channels = vec![];
while let Some(result) = cursor.next().await {
if let Ok(doc) = result {
channels.push(
from_document(doc)
.map_err(|_| Error::DatabaseError {
operation: "from_document",
with: "channel",
})?
);
channels.push(from_document(doc).map_err(|_| Error::DatabaseError {
operation: "from_document",
with: "channel",
})?);
}
}

12
src/notifications/subscriptions.rs
View File

@@ -1,9 +1,9 @@
use crate::database::*;
use futures::StreamExt;
use mongodb::bson::doc;
use hive_pubsub::PubSub;
use super::hive::get_hive;
use futures::StreamExt;
use hive_pubsub::PubSub;
use mongodb::bson::doc;
use mongodb::options::FindOptions;
pub async fn generate_subscriptions(user: &User) -> Result<(), String> {
@@ -35,13 +35,11 @@ pub async fn generate_subscriptions(user: &User) -> Result<(), String> {
}
]
},
FindOptions::builder()
.projection(doc! { "_id": 1 })
.build()
FindOptions::builder().projection(doc! { "_id": 1 }).build(),
)
.await
.map_err(|_| "Failed to fetch channels.".to_string())?;
while let Some(result) = cursor.next().await {
if let Ok(doc) = result {
hive.subscribe(user.id.clone(), doc.get_str("_id").unwrap().to_string())?;

37
src/notifications/websocket.rs
View File

@@ -64,7 +64,7 @@ async fn accept(stream: TcpStream) {
};
let session: Arc<Mutex<Option<Session>>> = Arc::new(Mutex::new(None));
let mutex_generator = || { session.clone() };
let mutex_generator = || session.clone();
let fwd = rx.map(Ok).forward(write);
let incoming = read.try_for_each(async move |msg| {
let mutex = mutex_generator();
@@ -79,22 +79,17 @@ async fn accept(stream: TcpStream) {
send(ClientboundNotification::Error(
WebSocketError::AlreadyAuthenticated,
));
return Ok(())
return Ok(());
}
}
if let Ok(validated_session) = Auth::new(get_collection("accounts"))
.verify_session(new_session)
.await {
.await
{
let id = validated_session.user_id.clone();
if let Ok(user) = (
Ref {
id: id.clone()
}
)
.fetch_user()
.await {
if let Ok(user) = (Ref { id: id.clone() }).fetch_user().await {
let was_online = is_online(&id);
{
match USERS.write() {
@@ -103,10 +98,12 @@ async fn accept(stream: TcpStream) {
}
Err(_) => {
send(ClientboundNotification::Error(
WebSocketError::InternalError { at: "Writing users map.".to_string() },
WebSocketError::InternalError {
at: "Writing users map.".to_string(),
},
));
return Ok(())
return Ok(());
}
}
}
@@ -115,10 +112,12 @@ async fn accept(stream: TcpStream) {
if let Err(_) = subscriptions::generate_subscriptions(&user).await {
send(ClientboundNotification::Error(
WebSocketError::InternalError { at: "Generating subscriptions.".to_string() },
WebSocketError::InternalError {
at: "Generating subscriptions.".to_string(),
},
));
return Ok(())
return Ok(());
}
send(ClientboundNotification::Authenticated);
@@ -130,7 +129,7 @@ async fn accept(stream: TcpStream) {
if !was_online {
ClientboundNotification::UserPresence {
id: id.clone(),
online: true
online: true,
}
.publish(id)
.await
@@ -139,10 +138,12 @@ async fn accept(stream: TcpStream) {
}
Err(_) => {
send(ClientboundNotification::Error(
WebSocketError::InternalError { at: "Generating payload.".to_string() },
WebSocketError::InternalError {
at: "Generating payload.".to_string(),
},
));
return Ok(())
return Ok(());
}
}
} else {