anonymous calls
This commit is contained in:
parent
309342b730
commit
8747657999
12 changed files with 712 additions and 39 deletions
547
src/anonymous_clients/anonymous_client_connection.rs
Normal file
547
src/anonymous_clients/anonymous_client_connection.rs
Normal file
|
|
@ -0,0 +1,547 @@
|
||||||
|
use async_tungstenite::tungstenite::Message;
|
||||||
|
use async_tungstenite::{WebSocketReceiver, WebSocketSender};
|
||||||
|
use json::JsonValue;
|
||||||
|
use json::number::Number;
|
||||||
|
use std::str::FromStr;
|
||||||
|
use std::sync::Arc;
|
||||||
|
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||||
|
use tokio::sync::RwLock;
|
||||||
|
use tokio_util::compat::Compat;
|
||||||
|
use tungstenite::Utf8Bytes;
|
||||||
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
use crate::calls::call_manager;
|
||||||
|
use crate::data::{
|
||||||
|
communication::{CommunicationType, CommunicationValue, DataTypes},
|
||||||
|
user::User,
|
||||||
|
};
|
||||||
|
use crate::omega::omega_connection::{WAITING_TASKS, get_omega_connection};
|
||||||
|
use crate::rho::rho_manager;
|
||||||
|
use crate::util::logger::PrintType;
|
||||||
|
use crate::{log_in, log_out};
|
||||||
|
|
||||||
|
pub struct AnonymousClientConnection {
|
||||||
|
pub sender: Arc<RwLock<WebSocketSender<Compat<tokio::net::TcpStream>>>>,
|
||||||
|
pub receiver: Arc<RwLock<WebSocketReceiver<Compat<tokio::net::TcpStream>>>>,
|
||||||
|
pub user_id: Arc<RwLock<i64>>,
|
||||||
|
pub ping: Arc<RwLock<i64>>,
|
||||||
|
pub interested_users: Arc<RwLock<Vec<i64>>>,
|
||||||
|
is_open: Arc<RwLock<bool>>,
|
||||||
|
pub user_name: Arc<RwLock<String>>,
|
||||||
|
pub display_name: Arc<RwLock<String>>,
|
||||||
|
pub avatar: Arc<RwLock<String>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl AnonymousClientConnection {
|
||||||
|
/// Create a new AnonymousClientConnection
|
||||||
|
pub fn new(
|
||||||
|
sender: WebSocketSender<Compat<tokio::net::TcpStream>>,
|
||||||
|
receiver: WebSocketReceiver<Compat<tokio::net::TcpStream>>,
|
||||||
|
) -> Arc<Self> {
|
||||||
|
Arc::new(Self {
|
||||||
|
sender: Arc::new(RwLock::new(sender)),
|
||||||
|
receiver: Arc::new(RwLock::new(receiver)),
|
||||||
|
user_id: Arc::new(RwLock::new(
|
||||||
|
SystemTime::now()
|
||||||
|
.duration_since(UNIX_EPOCH)
|
||||||
|
.unwrap()
|
||||||
|
.as_millis() as i64,
|
||||||
|
)),
|
||||||
|
ping: Arc::new(RwLock::new(-1)),
|
||||||
|
interested_users: Arc::new(RwLock::new(Vec::new())),
|
||||||
|
is_open: Arc::new(RwLock::new(true)),
|
||||||
|
user_name: Arc::new(RwLock::new(String::new())),
|
||||||
|
display_name: Arc::new(RwLock::new(String::new())),
|
||||||
|
avatar: Arc::new(RwLock::new(String::new())),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get the user ID
|
||||||
|
pub async fn get_user_id(&self) -> i64 {
|
||||||
|
*self.user_id.read().await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get the user name
|
||||||
|
pub async fn get_user_name(&self) -> String {
|
||||||
|
self.user_name.read().await.clone()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get the display name
|
||||||
|
pub async fn get_display_name(&self) -> String {
|
||||||
|
self.display_name.read().await.clone()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get the avatar
|
||||||
|
pub async fn get_avatar(&self) -> String {
|
||||||
|
self.avatar.read().await.clone()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Send a string message to the client
|
||||||
|
pub async fn send_message_str(self: Arc<Self>, message: &str) {
|
||||||
|
let mut session = self.sender.write().await;
|
||||||
|
if let Err(e) = session
|
||||||
|
.send(Message::Text(Utf8Bytes::from(message.to_string())))
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
log_out!(
|
||||||
|
PrintType::Client,
|
||||||
|
"Failed to send message to anonymous client: {}",
|
||||||
|
e,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Send a CommunicationValue to the client
|
||||||
|
pub async fn send_message(self: Arc<Self>, cv: &CommunicationValue) {
|
||||||
|
if !*self.is_open.read().await {
|
||||||
|
log_out!(
|
||||||
|
PrintType::Client,
|
||||||
|
"Attempted to send message to a closed connection."
|
||||||
|
);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if !cv.is_type(CommunicationType::pong) {
|
||||||
|
log_out!(PrintType::Client, "{}", &cv.to_json().to_string());
|
||||||
|
}
|
||||||
|
self.send_message_str(&cv.to_json().to_string()).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Handle incoming message from client
|
||||||
|
pub async fn handle_message(self: Arc<Self>, message: Utf8Bytes) {
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let cv = CommunicationValue::from_json(&message);
|
||||||
|
if cv.is_type(CommunicationType::ping) {
|
||||||
|
self.handle_ping(cv).await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
log_in!(
|
||||||
|
PrintType::Client,
|
||||||
|
"Anonymous: {}",
|
||||||
|
&cv.to_json().to_string()
|
||||||
|
);
|
||||||
|
|
||||||
|
if cv.is_type(CommunicationType::identification) {
|
||||||
|
let call_id = Uuid::parse_str(
|
||||||
|
cv.get_data(DataTypes::call_id)
|
||||||
|
.unwrap_or(&JsonValue::Null)
|
||||||
|
.as_str()
|
||||||
|
.unwrap_or(""),
|
||||||
|
)
|
||||||
|
.unwrap_or(Uuid::new_v4());
|
||||||
|
|
||||||
|
let call = if let Some(call) = call_manager::get_call(call_id).await {
|
||||||
|
if call.is_anonymous().await {
|
||||||
|
call
|
||||||
|
} else {
|
||||||
|
self.send_error_response(
|
||||||
|
&cv.get_id(),
|
||||||
|
CommunicationType::error_not_authenticated,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
self.send_error_response(
|
||||||
|
&cv.get_id(),
|
||||||
|
CommunicationType::error_not_authenticated,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
|
||||||
|
let mut invited = JsonValue::new_array();
|
||||||
|
for call_invitee in call.members.read().await.clone() {
|
||||||
|
let call_invitee_cv = get_omega_connection()
|
||||||
|
.await_response(
|
||||||
|
&CommunicationValue::new(CommunicationType::get_user_data).add_data(
|
||||||
|
DataTypes::user_id,
|
||||||
|
JsonValue::from(call_invitee.user_id),
|
||||||
|
),
|
||||||
|
Some(Duration::from_secs(2)),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let mut json_invitee = JsonValue::new_object();
|
||||||
|
let _ = json_invitee.insert(
|
||||||
|
"user_id",
|
||||||
|
call_invitee_cv
|
||||||
|
.get_data(DataTypes::user_id)
|
||||||
|
.unwrap_or(&JsonValue::Null)
|
||||||
|
.clone(),
|
||||||
|
);
|
||||||
|
let _ = json_invitee.insert(
|
||||||
|
"username",
|
||||||
|
call_invitee_cv
|
||||||
|
.get_data(DataTypes::username)
|
||||||
|
.unwrap_or(&JsonValue::Null)
|
||||||
|
.clone(),
|
||||||
|
);
|
||||||
|
let _ = json_invitee.insert(
|
||||||
|
"display",
|
||||||
|
call_invitee_cv
|
||||||
|
.get_data(DataTypes::display)
|
||||||
|
.unwrap_or(&JsonValue::Null)
|
||||||
|
.clone(),
|
||||||
|
);
|
||||||
|
let _ = json_invitee.insert(
|
||||||
|
"avatar",
|
||||||
|
call_invitee_cv
|
||||||
|
.get_data(DataTypes::avatar)
|
||||||
|
.unwrap_or(&JsonValue::Null)
|
||||||
|
.clone(),
|
||||||
|
);
|
||||||
|
|
||||||
|
let _ = invited.push(json_invitee);
|
||||||
|
}
|
||||||
|
|
||||||
|
let token = call.create_anonymous_token(self.get_user_id().await).await;
|
||||||
|
|
||||||
|
let mut serialized = JsonValue::new_object();
|
||||||
|
let _ = serialized.insert("call_id", JsonValue::String(call_id.to_string()));
|
||||||
|
let _ = serialized.insert("call_invited", invited.clone());
|
||||||
|
let _ = serialized.insert("call_members", invited);
|
||||||
|
let _ = serialized.insert("call_token", JsonValue::String(token.unwrap()));
|
||||||
|
self.clone()
|
||||||
|
.send_message(
|
||||||
|
&&CommunicationValue::new(CommunicationType::identification_response)
|
||||||
|
.with_id(cv.get_id())
|
||||||
|
.add_data(
|
||||||
|
DataTypes::user_id,
|
||||||
|
JsonValue::from(self.get_user_id().await),
|
||||||
|
)
|
||||||
|
.add_data(
|
||||||
|
DataTypes::username,
|
||||||
|
JsonValue::String(self.clone().get_user_name().await),
|
||||||
|
)
|
||||||
|
.add_data(
|
||||||
|
DataTypes::display,
|
||||||
|
JsonValue::String(self.get_display_name().await),
|
||||||
|
)
|
||||||
|
.add_data(
|
||||||
|
DataTypes::avatar,
|
||||||
|
JsonValue::String(self.get_avatar().await),
|
||||||
|
)
|
||||||
|
.add_data(DataTypes::call_state, serialized),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Handle ping
|
||||||
|
if cv.is_type(CommunicationType::ping) {
|
||||||
|
self.handle_ping(cv).await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
// Handle client status changes
|
||||||
|
if cv.is_type(CommunicationType::client_changed) {
|
||||||
|
self.handle_client_changed(cv).await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Handle call invites
|
||||||
|
if cv.is_type(CommunicationType::call_invite) {
|
||||||
|
self.handle_call_invite(cv).await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Handle get call requests
|
||||||
|
if cv.is_type(CommunicationType::call_token) {
|
||||||
|
self.handle_get_call(cv).await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if cv.is_type(CommunicationType::call_disconnect_user) {
|
||||||
|
self.handle_call_disconnect_user(cv).await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if cv.is_type(CommunicationType::call_timeout_user) {
|
||||||
|
self.handle_call_timeout_user(cv).await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if cv.is_type(CommunicationType::change_user_data) {
|
||||||
|
// TODO
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if cv.is_type(CommunicationType::get_user_data)
|
||||||
|
|| cv.is_type(CommunicationType::get_iota_data)
|
||||||
|
|| cv.is_type(CommunicationType::delete_user)
|
||||||
|
{
|
||||||
|
self.handle_omega_forward(cv).await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
async fn handle_omega_forward(self: Arc<Self>, cv: CommunicationValue) {
|
||||||
|
let client_for_closure = self.clone();
|
||||||
|
WAITING_TASKS.insert(
|
||||||
|
cv.get_id(),
|
||||||
|
Box::new(move |_, response_cv| {
|
||||||
|
let client = client_for_closure.clone();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
client.send_message(&response_cv).await;
|
||||||
|
});
|
||||||
|
true
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
get_omega_connection()
|
||||||
|
.send_message(&cv.with_sender(*self.user_id.read().await))
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Handle ping message
|
||||||
|
async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) {
|
||||||
|
// Update our ping if provided
|
||||||
|
if let Some(last_ping) = cv.get_data(DataTypes::last_ping) {
|
||||||
|
if let Ok(ping_val) = last_ping.to_string().parse::<i64>() {
|
||||||
|
let mut ping_guard = self.ping.write().await;
|
||||||
|
*ping_guard = ping_val;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Send pong response
|
||||||
|
let response = CommunicationValue::new(CommunicationType::pong).with_id(cv.get_id());
|
||||||
|
|
||||||
|
self.send_message(&response).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Handle client status change
|
||||||
|
async fn handle_client_changed(self: Arc<Self>, _cv: CommunicationValue) {
|
||||||
|
/*let user_id = self.get_user_id().await;
|
||||||
|
if let Some(_status_str) = cv.get_data(DataTypes::user_state) {
|
||||||
|
let user_status = UserStatus::online;
|
||||||
|
}*/
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Handle call invite
|
||||||
|
async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) {
|
||||||
|
let receiver_id: i64 = cv
|
||||||
|
.get_data(DataTypes::receiver_id)
|
||||||
|
.unwrap_or(&json::JsonValue::Number(Number::from(0)))
|
||||||
|
.as_i64()
|
||||||
|
.unwrap_or(0);
|
||||||
|
if receiver_id == 0 {
|
||||||
|
self.send_error_response(&cv.get_id(), CommunicationType::error_no_user_id)
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
let call_id = match cv.get_data(DataTypes::call_id) {
|
||||||
|
Some(id_str) => match Uuid::parse_str(&id_str.to_string()) {
|
||||||
|
Ok(id) => id,
|
||||||
|
Err(_) => {
|
||||||
|
self.send_error_response(
|
||||||
|
&cv.get_id(),
|
||||||
|
CommunicationType::error_invalid_call_id,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
},
|
||||||
|
None => {
|
||||||
|
self.send_error_response(&cv.get_id(), CommunicationType::error_no_call_id)
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let invited =
|
||||||
|
call_manager::add_invite(call_id, *self.user_id.read().await, receiver_id).await;
|
||||||
|
if !invited {
|
||||||
|
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_call_id)
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Find target RhoConnection
|
||||||
|
let target_rho = match rho_manager::get_rho_con_for_user(receiver_id).await {
|
||||||
|
Some(rho) => rho,
|
||||||
|
None => {
|
||||||
|
self.send_error_response(&cv.get_id(), CommunicationType::error)
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
// Get sender user ID
|
||||||
|
let sender_id = self.get_user_id().await;
|
||||||
|
|
||||||
|
// Create and send call distribution message
|
||||||
|
let forward = CommunicationValue::new(CommunicationType::call_invite)
|
||||||
|
.with_receiver(receiver_id)
|
||||||
|
.with_sender(sender_id)
|
||||||
|
.add_data_str(DataTypes::call_id, call_id.to_string())
|
||||||
|
.add_data_str(DataTypes::receiver_id, receiver_id.to_string())
|
||||||
|
.add_data_str(DataTypes::sender_id, sender_id.to_string());
|
||||||
|
|
||||||
|
target_rho.message_to_client(forward).await;
|
||||||
|
|
||||||
|
let response = CommunicationValue::new(CommunicationType::success).with_id(cv.get_id());
|
||||||
|
self.send_message(&response).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Handle get call request
|
||||||
|
async fn handle_get_call(self: Arc<Self>, cv: CommunicationValue) {
|
||||||
|
let user_id = self.get_user_id().await;
|
||||||
|
|
||||||
|
let call_id = match cv.get_data(DataTypes::call_id) {
|
||||||
|
Some(id_str) => match Uuid::parse_str(&id_str.to_string()) {
|
||||||
|
Ok(id) => id,
|
||||||
|
Err(_) => {
|
||||||
|
self.send_error_response(&cv.get_id(), CommunicationType::error)
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
},
|
||||||
|
None => {
|
||||||
|
self.send_error_response(&cv.get_id(), CommunicationType::error)
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
if let Some(token) = call_manager::get_call_token(user_id, call_id).await {
|
||||||
|
let response = CommunicationValue::new(CommunicationType::call_token)
|
||||||
|
.with_id(cv.get_id())
|
||||||
|
.with_receiver(user_id)
|
||||||
|
.add_data_str(DataTypes::call_token, token);
|
||||||
|
self.send_message(&response).await;
|
||||||
|
} else {
|
||||||
|
self.send_error_response(&cv.get_id(), CommunicationType::error)
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
async fn handle_call_timeout_user(self: Arc<Self>, cv: CommunicationValue) {
|
||||||
|
let call_id = Uuid::from_str(
|
||||||
|
cv.get_data(DataTypes::call_id)
|
||||||
|
.unwrap_or(&JsonValue::Null)
|
||||||
|
.as_str()
|
||||||
|
.unwrap_or(""),
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
let user_id = cv
|
||||||
|
.get_data(DataTypes::user_id)
|
||||||
|
.unwrap_or(&JsonValue::Null)
|
||||||
|
.as_i64()
|
||||||
|
.unwrap_or(0);
|
||||||
|
let untill = cv
|
||||||
|
.get_data(DataTypes::untill)
|
||||||
|
.unwrap_or(&JsonValue::Null)
|
||||||
|
.as_i64()
|
||||||
|
.unwrap_or(0);
|
||||||
|
|
||||||
|
let call = call_manager::get_call(call_id).await;
|
||||||
|
if let Some(call) = call {
|
||||||
|
if call
|
||||||
|
.get_caller(self.get_user_id().await)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
.has_admin()
|
||||||
|
{
|
||||||
|
call.get_caller(user_id)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
.set_timeout(untill)
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
async fn handle_call_disconnect_user(self: Arc<Self>, cv: CommunicationValue) {
|
||||||
|
let call_id = Uuid::from_str(
|
||||||
|
cv.get_data(DataTypes::call_id)
|
||||||
|
.unwrap_or(&JsonValue::Null)
|
||||||
|
.as_str()
|
||||||
|
.unwrap_or(""),
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
let user_id = cv
|
||||||
|
.get_data(DataTypes::user_id)
|
||||||
|
.unwrap_or(&JsonValue::Null)
|
||||||
|
.as_i64()
|
||||||
|
.unwrap_or(0);
|
||||||
|
|
||||||
|
let call = call_manager::get_call(call_id).await;
|
||||||
|
if let Some(call) = call {
|
||||||
|
if call
|
||||||
|
.get_caller(self.get_user_id().await)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
.has_admin()
|
||||||
|
{
|
||||||
|
call.remove_caller(user_id).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Send error response
|
||||||
|
async fn send_error_response(
|
||||||
|
self: Arc<Self>,
|
||||||
|
message_id: &Uuid,
|
||||||
|
error_type: CommunicationType,
|
||||||
|
) {
|
||||||
|
let error = CommunicationValue::new(error_type).with_id(*message_id);
|
||||||
|
self.send_message(&error).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Close the connection
|
||||||
|
pub async fn close(&self) {
|
||||||
|
let mut is_open_guard = self.is_open.write().await;
|
||||||
|
if *is_open_guard {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
*is_open_guard = false;
|
||||||
|
|
||||||
|
let mut session = self.sender.write().await;
|
||||||
|
let _ = session.close(None).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Set interested users list
|
||||||
|
pub async fn set_interested_users(self: Arc<Self>, interested_ids: Vec<i64>) {
|
||||||
|
let mut interested_guard = self.interested_users.write().await;
|
||||||
|
*interested_guard = interested_ids;
|
||||||
|
}
|
||||||
|
pub async fn get_interested_users(self: Arc<Self>) -> Vec<i64> {
|
||||||
|
let interested_guard = self.interested_users.read().await;
|
||||||
|
interested_guard.clone()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Check if interested in a user and send notification
|
||||||
|
pub async fn are_you_interested(self: Arc<Self>, user: &User) {
|
||||||
|
let interested_guard = self.clone().get_interested_users().await;
|
||||||
|
if interested_guard.contains(&user.user_id) {
|
||||||
|
let notification = CommunicationValue::new(CommunicationType::client_changed)
|
||||||
|
.add_data_str(DataTypes::user_id, user.user_id.to_string())
|
||||||
|
.add_data_str(
|
||||||
|
DataTypes::user_state,
|
||||||
|
format!("{:?}", user.status.to_string()),
|
||||||
|
);
|
||||||
|
|
||||||
|
self.send_message(¬ification).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Handle connection close
|
||||||
|
pub async fn handle_close(&self) {
|
||||||
|
|
||||||
|
// TODO delete temp user
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Implement Clone to make it easier to work with Arc<AnonymousClientConnection>
|
||||||
|
impl Clone for AnonymousClientConnection {
|
||||||
|
fn clone(&self) -> Self {
|
||||||
|
Self {
|
||||||
|
sender: Arc::clone(&self.sender),
|
||||||
|
receiver: Arc::clone(&self.receiver),
|
||||||
|
user_id: Arc::clone(&self.user_id),
|
||||||
|
ping: Arc::clone(&self.ping),
|
||||||
|
interested_users: Arc::clone(&self.interested_users),
|
||||||
|
is_open: Arc::clone(&self.is_open),
|
||||||
|
user_name: Arc::clone(&self.user_name),
|
||||||
|
display_name: Arc::clone(&self.display_name),
|
||||||
|
avatar: Arc::clone(&self.avatar),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
21
src/anonymous_clients/anonymous_manager.rs
Normal file
21
src/anonymous_clients/anonymous_manager.rs
Normal file
|
|
@ -0,0 +1,21 @@
|
||||||
|
use std::sync::Arc;
|
||||||
|
|
||||||
|
use dashmap::DashMap;
|
||||||
|
use once_cell::sync::Lazy;
|
||||||
|
|
||||||
|
use crate::anonymous_clients::anonymous_client_connection::AnonymousClientConnection;
|
||||||
|
|
||||||
|
static ANONYMOUS_USERS: Lazy<DashMap<i64, Arc<AnonymousClientConnection>>> =
|
||||||
|
Lazy::new(|| DashMap::new());
|
||||||
|
|
||||||
|
pub async fn add_anonymous_user(connection: Arc<AnonymousClientConnection>) {
|
||||||
|
ANONYMOUS_USERS.insert(connection.get_user_id().await, connection);
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn remove_anonymous_user(user_id: i64) {
|
||||||
|
ANONYMOUS_USERS.remove(&user_id);
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn get_anonymous_user(user_id: i64) -> Option<Arc<AnonymousClientConnection>> {
|
||||||
|
ANONYMOUS_USERS.get(&user_id).map(|c| c.clone())
|
||||||
|
}
|
||||||
2
src/anonymous_clients/mod.rs
Normal file
2
src/anonymous_clients/mod.rs
Normal file
|
|
@ -0,0 +1,2 @@
|
||||||
|
pub mod anonymous_client_connection;
|
||||||
|
pub mod anonymous_manager;
|
||||||
|
|
@ -1,15 +1,20 @@
|
||||||
use std::sync::Arc;
|
use json::JsonValue;
|
||||||
|
use std::{env, sync::Arc, time::Duration};
|
||||||
use tokio::sync::RwLock;
|
use tokio::sync::RwLock;
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
use crate::calls::caller::Caller;
|
use crate::{
|
||||||
|
calls::{call_util, caller::Caller},
|
||||||
|
data::communication::{CommunicationType, CommunicationValue, DataTypes},
|
||||||
|
omega::omega_connection::get_omega_connection,
|
||||||
|
};
|
||||||
|
|
||||||
pub struct CallGroup {
|
pub struct CallGroup {
|
||||||
pub call_id: Uuid,
|
pub call_id: Uuid,
|
||||||
pub members: RwLock<Vec<Arc<Caller>>>,
|
pub members: RwLock<Vec<Arc<Caller>>>,
|
||||||
pub show: RwLock<bool>,
|
pub show: RwLock<bool>,
|
||||||
pub anonymous_joining: RwLock<bool>,
|
pub anonymous_joining: RwLock<bool>,
|
||||||
|
pub short_link: RwLock<Option<String>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl CallGroup {
|
impl CallGroup {
|
||||||
|
|
@ -19,6 +24,7 @@ impl CallGroup {
|
||||||
members: RwLock::new(vec![user]),
|
members: RwLock::new(vec![user]),
|
||||||
show: RwLock::new(true),
|
show: RwLock::new(true),
|
||||||
anonymous_joining: RwLock::new(false),
|
anonymous_joining: RwLock::new(false),
|
||||||
|
short_link: RwLock::new(None),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -31,8 +37,54 @@ impl CallGroup {
|
||||||
.cloned()
|
.cloned()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub async fn is_anonymous(&self) -> bool {
|
||||||
|
*self.anonymous_joining.read().await
|
||||||
|
}
|
||||||
|
|
||||||
pub async fn set_anonymous_joining(&self, enable: bool) {
|
pub async fn set_anonymous_joining(&self, enable: bool) {
|
||||||
*self.anonymous_joining.write().await = enable;
|
*self.anonymous_joining.write().await = enable;
|
||||||
|
|
||||||
|
if self.short_link.read().await.is_none() {
|
||||||
|
let long_link = format!(
|
||||||
|
"https://app.tensamin.net/call/anonymous?call_id={}&omikron_id={}",
|
||||||
|
self.call_id,
|
||||||
|
env::var("ID")
|
||||||
|
.unwrap_or("0".to_string())
|
||||||
|
.parse::<i64>()
|
||||||
|
.unwrap_or(0),
|
||||||
|
);
|
||||||
|
let response_cv = get_omega_connection()
|
||||||
|
.await_response(
|
||||||
|
&CommunicationValue::new(CommunicationType::shorten_link)
|
||||||
|
.add_data(DataTypes::link, JsonValue::from(long_link)),
|
||||||
|
Some(Duration::from_secs(20)),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
if let Ok(response) = response_cv {
|
||||||
|
*self.short_link.write().await = Some(
|
||||||
|
response
|
||||||
|
.get_data(DataTypes::link)
|
||||||
|
.unwrap()
|
||||||
|
.as_str()
|
||||||
|
.unwrap()
|
||||||
|
.to_string(),
|
||||||
|
);
|
||||||
|
log::info!(
|
||||||
|
"Shortened link for call {} is {}",
|
||||||
|
self.call_id,
|
||||||
|
self.short_link.read().await.as_ref().unwrap()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn create_anonymous_token(&self, user_id: i64) -> Option<String> {
|
||||||
|
if self.is_anonymous().await {
|
||||||
|
if let Ok(token) = call_util::create_token(user_id, self.call_id, false) {
|
||||||
|
return Some(token);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn remove_caller(&self, user_id: i64) {
|
pub async fn remove_caller(&self, user_id: i64) {
|
||||||
|
|
@ -41,4 +93,8 @@ impl CallGroup {
|
||||||
.await
|
.await
|
||||||
.retain(|caller| caller.user_id != user_id);
|
.retain(|caller| caller.user_id != user_id);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub async fn get_short_link(self: Arc<Self>) -> Option<String> {
|
||||||
|
self.short_link.read().await.clone()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -50,6 +50,9 @@ pub async fn get_call_token(user_id: i64, call_id: Uuid) -> Option<String> {
|
||||||
}
|
}
|
||||||
return Some(member.create_token());
|
return Some(member.create_token());
|
||||||
}
|
}
|
||||||
|
if cg.is_anonymous().await {
|
||||||
|
return cg.create_anonymous_token(user_id).await;
|
||||||
|
}
|
||||||
return None;
|
return None;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,7 +1,4 @@
|
||||||
use std::{
|
use std::time::{SystemTime, UNIX_EPOCH};
|
||||||
sync::Arc,
|
|
||||||
time::{SystemTime, UNIX_EPOCH},
|
|
||||||
};
|
|
||||||
|
|
||||||
use tokio::sync::RwLock;
|
use tokio::sync::RwLock;
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
|
||||||
|
|
@ -13,6 +13,9 @@ pub enum DataTypes {
|
||||||
accepted_ids,
|
accepted_ids,
|
||||||
uuid,
|
uuid,
|
||||||
register_id,
|
register_id,
|
||||||
|
|
||||||
|
link,
|
||||||
|
|
||||||
settings,
|
settings,
|
||||||
settings_name,
|
settings_name,
|
||||||
chat_partner_id,
|
chat_partner_id,
|
||||||
|
|
@ -42,7 +45,7 @@ pub enum DataTypes {
|
||||||
call_id,
|
call_id,
|
||||||
call_token,
|
call_token,
|
||||||
untill,
|
untill,
|
||||||
enable,
|
enabled,
|
||||||
start_date,
|
start_date,
|
||||||
end_date,
|
end_date,
|
||||||
receiver_id,
|
receiver_id,
|
||||||
|
|
@ -124,6 +127,9 @@ pub enum CommunicationType {
|
||||||
error_no_call_id,
|
error_no_call_id,
|
||||||
error_invalid_call_id,
|
error_invalid_call_id,
|
||||||
success,
|
success,
|
||||||
|
|
||||||
|
shorten_link,
|
||||||
|
|
||||||
settings_save,
|
settings_save,
|
||||||
settings_load,
|
settings_load,
|
||||||
settings_list,
|
settings_list,
|
||||||
|
|
|
||||||
37
src/main.rs
37
src/main.rs
|
|
@ -1,3 +1,4 @@
|
||||||
|
mod anonymous_clients;
|
||||||
mod calls;
|
mod calls;
|
||||||
mod data;
|
mod data;
|
||||||
mod omega;
|
mod omega;
|
||||||
|
|
@ -7,6 +8,7 @@ mod util;
|
||||||
use async_tungstenite::accept_hdr_async;
|
use async_tungstenite::accept_hdr_async;
|
||||||
use dotenv::dotenv;
|
use dotenv::dotenv;
|
||||||
use futures::StreamExt;
|
use futures::StreamExt;
|
||||||
|
use json::JsonValue;
|
||||||
use once_cell::sync::Lazy;
|
use once_cell::sync::Lazy;
|
||||||
use std::{env, sync::Arc};
|
use std::{env, sync::Arc};
|
||||||
use tokio::net::TcpListener;
|
use tokio::net::TcpListener;
|
||||||
|
|
@ -14,7 +16,9 @@ use tokio_util::compat::TokioAsyncReadCompatExt;
|
||||||
use tungstenite::handshake::server::{Request, Response};
|
use tungstenite::handshake::server::{Request, Response};
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
|
anonymous_clients::anonymous_client_connection::AnonymousClientConnection,
|
||||||
calls::call_manager::garbage_collect_calls,
|
calls::call_manager::garbage_collect_calls,
|
||||||
|
data::communication::{CommunicationType, CommunicationValue, DataTypes},
|
||||||
omega::omega_connection::OmegaConnection,
|
omega::omega_connection::OmegaConnection,
|
||||||
rho::{client_connection::ClientConnection, iota_connection::IotaConnection},
|
rho::{client_connection::ClientConnection, iota_connection::IotaConnection},
|
||||||
util::{
|
util::{
|
||||||
|
|
@ -100,6 +104,39 @@ async fn main() {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
} else if path == "/ws/anonymous_client/" {
|
||||||
|
log_in!(PrintType::Client, "New Anonymous Client connection");
|
||||||
|
let client_conn: Arc<AnonymousClientConnection> =
|
||||||
|
Arc::from(AnonymousClientConnection::new(sender, receiver));
|
||||||
|
loop {
|
||||||
|
let msg_result = {
|
||||||
|
let mut session_lock = client_conn.receiver.write().await;
|
||||||
|
session_lock.next().await
|
||||||
|
};
|
||||||
|
|
||||||
|
match msg_result {
|
||||||
|
Some(Ok(msg)) => {
|
||||||
|
if msg.is_text() {
|
||||||
|
let text = msg.into_text().unwrap();
|
||||||
|
client_conn.clone().handle_message(text).await;
|
||||||
|
} else if msg.is_close() {
|
||||||
|
log_in!(PrintType::Client, "Anonymous Client disconnected");
|
||||||
|
client_conn.handle_close().await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Some(Err(e)) => {
|
||||||
|
log_err!(PrintType::Client, "WebSocket error: {}", e);
|
||||||
|
client_conn.handle_close().await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
None => {
|
||||||
|
log_in!(PrintType::Client, "Anonymous Client stream ended");
|
||||||
|
client_conn.handle_close().await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
} else if path == "/ws/iota/" {
|
} else if path == "/ws/iota/" {
|
||||||
log_in!(PrintType::Iota, "New Iota connection");
|
log_in!(PrintType::Iota, "New Iota connection");
|
||||||
let iota_conn: Arc<IotaConnection> =
|
let iota_conn: Arc<IotaConnection> =
|
||||||
|
|
|
||||||
|
|
@ -273,12 +273,6 @@ impl ClientConnection {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Handle ping
|
|
||||||
if cv.is_type(CommunicationType::ping) {
|
|
||||||
self.handle_ping(cv).await;
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
log_in!(PrintType::Client, "{}", &cv.to_json().to_string());
|
|
||||||
// Handle client status changes
|
// Handle client status changes
|
||||||
if cv.is_type(CommunicationType::client_changed) {
|
if cv.is_type(CommunicationType::client_changed) {
|
||||||
self.handle_client_changed(cv).await;
|
self.handle_client_changed(cv).await;
|
||||||
|
|
@ -504,7 +498,11 @@ impl ClientConnection {
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.has_admin()
|
.has_admin()
|
||||||
{
|
{
|
||||||
call.get_caller(user_id).await.unwrap().set_timeout(untill);
|
call.get_caller(user_id)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
.set_timeout(untill)
|
||||||
|
.await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -543,12 +541,14 @@ impl ClientConnection {
|
||||||
)
|
)
|
||||||
.unwrap();
|
.unwrap();
|
||||||
let enable = cv
|
let enable = cv
|
||||||
.get_data(DataTypes::enable)
|
.get_data(DataTypes::enabled)
|
||||||
.unwrap_or(&JsonValue::Null)
|
.unwrap_or(&JsonValue::Null)
|
||||||
.as_bool()
|
.as_bool()
|
||||||
.unwrap_or(false);
|
.unwrap_or(true);
|
||||||
|
|
||||||
let call = call_manager::get_call(call_id).await;
|
let call = call_manager::get_call(call_id).await;
|
||||||
|
|
||||||
|
let mut short_link = None;
|
||||||
if let Some(call) = call {
|
if let Some(call) = call {
|
||||||
if call
|
if call
|
||||||
.get_caller(self.get_user_id().await)
|
.get_caller(self.get_user_id().await)
|
||||||
|
|
@ -558,7 +558,17 @@ impl ClientConnection {
|
||||||
{
|
{
|
||||||
call.set_anonymous_joining(enable).await;
|
call.set_anonymous_joining(enable).await;
|
||||||
}
|
}
|
||||||
|
short_link = call.get_short_link().await;
|
||||||
}
|
}
|
||||||
|
let mut response_cv =
|
||||||
|
CommunicationValue::new(CommunicationType::call_set_anonymous_joining)
|
||||||
|
.with_id(cv.get_id())
|
||||||
|
.add_data(DataTypes::call_id, JsonValue::String(call_id.to_string()))
|
||||||
|
.add_data(DataTypes::enabled, JsonValue::Boolean(enable));
|
||||||
|
if let Some(short_link) = short_link {
|
||||||
|
response_cv = response_cv.add_data(DataTypes::link, JsonValue::String(short_link));
|
||||||
|
}
|
||||||
|
self.send_message(&response_cv).await;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Forward message to Iota
|
/// Forward message to Iota
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,6 @@ use crate::calls::call_group::CallGroup;
|
||||||
use crate::calls::call_manager;
|
use crate::calls::call_manager;
|
||||||
use crate::get_private_key;
|
use crate::get_private_key;
|
||||||
use crate::get_public_key;
|
use crate::get_public_key;
|
||||||
use crate::log;
|
|
||||||
use crate::log_err;
|
use crate::log_err;
|
||||||
use crate::log_in;
|
use crate::log_in;
|
||||||
use crate::log_out;
|
use crate::log_out;
|
||||||
|
|
@ -514,22 +513,22 @@ impl IotaConnection {
|
||||||
for call in calls {
|
for call in calls {
|
||||||
for inviter in call.members.read().await.iter() {
|
for inviter in call.members.read().await.iter() {
|
||||||
let call_self = call.get_caller(receiver_id).await.unwrap();
|
let call_self = call.get_caller(receiver_id).await.unwrap();
|
||||||
|
let admin = call_self.has_admin();
|
||||||
let inviter_id = inviter.user_id;
|
let inviter_id = inviter.user_id;
|
||||||
|
let timeout = *call_self.timeout.read().await;
|
||||||
|
|
||||||
|
let mut call_obj = JsonValue::new_object();
|
||||||
|
let _ = call_obj.insert("call_id", JsonValue::String(call.call_id.to_string()));
|
||||||
|
if timeout > 0 {
|
||||||
|
let _ = call_obj.insert("timeout", JsonValue::from(timeout));
|
||||||
|
}
|
||||||
|
if admin {
|
||||||
|
let _ = call_obj.insert("admin", JsonValue::Boolean(admin));
|
||||||
|
}
|
||||||
|
|
||||||
if let Some(call_ids) = invites.get_mut(&inviter_id) {
|
if let Some(call_ids) = invites.get_mut(&inviter_id) {
|
||||||
let mut call_obj = JsonValue::new_object();
|
|
||||||
let _ = call_obj.insert("call_id", JsonValue::String(call.call_id.to_string()));
|
|
||||||
let timeout = *call_self.timeout.read().await;
|
|
||||||
if timeout > 0 {
|
|
||||||
let _ = call_obj.insert("timeout", JsonValue::from(timeout));
|
|
||||||
}
|
|
||||||
call_ids.push(call_obj);
|
call_ids.push(call_obj);
|
||||||
} else {
|
} else {
|
||||||
let mut call_obj = JsonValue::new_object();
|
|
||||||
let _ = call_obj.insert("call_id", JsonValue::String(call.call_id.to_string()));
|
|
||||||
let timeout = *call_self.timeout.read().await;
|
|
||||||
if timeout > 0 {
|
|
||||||
let _ = call_obj.insert("timeout", JsonValue::from(timeout));
|
|
||||||
}
|
|
||||||
invites.insert(inviter_id, vec![call_obj]);
|
invites.insert(inviter_id, vec![call_obj]);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1,13 +1,9 @@
|
||||||
use super::{client_connection::ClientConnection, iota_connection::IotaConnection, rho_manager};
|
use super::{client_connection::ClientConnection, iota_connection::IotaConnection, rho_manager};
|
||||||
use crate::omega::omega_connection::OmegaConnection;
|
use crate::data::{
|
||||||
use crate::util::logger::PrintType;
|
communication::{CommunicationType, CommunicationValue, DataTypes},
|
||||||
use crate::{
|
user::UserStatus,
|
||||||
data::{
|
|
||||||
communication::{CommunicationType, CommunicationValue, DataTypes},
|
|
||||||
user::UserStatus,
|
|
||||||
},
|
|
||||||
log,
|
|
||||||
};
|
};
|
||||||
|
use crate::omega::omega_connection::OmegaConnection;
|
||||||
use json::{JsonValue, number::Number};
|
use json::{JsonValue, number::Number};
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
|
||||||
|
|
@ -5,7 +5,6 @@ use aes_gcm::{
|
||||||
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STD};
|
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STD};
|
||||||
use hkdf::Hkdf;
|
use hkdf::Hkdf;
|
||||||
use sha2::{Digest, Sha256};
|
use sha2::{Digest, Sha256};
|
||||||
use std::fmt;
|
|
||||||
use x448::{PublicKey, Secret};
|
use x448::{PublicKey, Secret};
|
||||||
|
|
||||||
// --- Custom Errors ---
|
// --- Custom Errors ---
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue