tensamin/omikron
Watch
6
0
Fork
You've already forked omikron
 0
Code Issues Pull requests 3 Projects Releases Packages Wiki Activity Actions

Compare commits

base: tensamin:ab8bb2633ac795542e396a004eac276a3c5bd6c9
Branches Tags
tensamin:main
tensamin:renovate/livekit-api-0.x
tensamin:renovate/base64-0.x
tensamin:renovate/livekit-protocol-0.x
..
compare: tensamin:139c7cc6095c21d9c536b710545ff94a161eb1a2
Branches Tags
tensamin:renovate/livekit-api-0.x
tensamin:renovate/base64-0.x
tensamin:renovate/livekit-protocol-0.x
tensamin:main
15 changed files with 1190 additions and 965 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.cargo/config.toml
View file

@ -1,2 +0,0 @@
[env]
MTP_TYPE_MAPS = { value = "type-maps.yaml", relative = true }

831
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

3
Cargo.toml
View file

@ -4,7 +4,8 @@ version = "0.1.0"
edition = "2024" edition = "2024"
[dependencies] [dependencies]
mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = ["host", "client"] } ttp-core = { git = "https://git.methanium.net/Tensamin/TTP.git", package = "ttp-core" }
ttp-native = { git = "https://git.methanium.net/Tensamin/TTP.git", package = "ttp-native" }
ansi_term = "*" ansi_term = "*"
uuid = { version = "*", features = ["v4"] } uuid = { version = "*", features = ["v4"] }

214
src/anonymous_clients/anonymous_client_connection.rs
View file

@ -1,9 +1,9 @@
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue, TypeMap};
use mtp::transport::{Receiver, Sender};
use std::str::FromStr; use std::str::FromStr;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tokio::sync::RwLock; use tokio::sync::RwLock;
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use ttp_native::{Receiver, Sender};
use uuid::Uuid; use uuid::Uuid;
use crate::anonymous_clients::anonymous_manager::{self, generate_username}; use crate::anonymous_clients::anonymous_manager::{self, generate_username};
@ -88,7 +88,7 @@ impl AnonymousClientConnection {
); );
return; return;
} }
if !cv.is_type(CommunicationType::Pong) { if !cv.is_type(CommunicationType::pong) {
log_cv_out!(PrintType::Client, &cv); log_cv_out!(PrintType::Client, &cv);
} }
if let Err(e) = self.sender.send(&cv).await { if let Err(e) = self.sender.send(&cv).await {
@ -104,15 +104,16 @@ impl AnonymousClientConnection {
/// Handle incoming message from client /// Handle incoming message from client
pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) { pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) {
tokio::spawn(async move { tokio::spawn(async move {
if cv.is_type(CommunicationType::Ping) { if cv.is_type(CommunicationType::ping) {
self.handle_ping(cv).await; self.handle_ping(cv).await;
return; return;
} }
log_cv_in!(PrintType::Client, &cv); log_cv_in!(PrintType::Client, &cv);
if cv.is_type(CommunicationType::Identification) { if cv.is_type(CommunicationType::identification) {
let call_id = Uuid::parse_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")) let call_id =
.unwrap_or(Uuid::new_v4()); Uuid::parse_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or(""))
.unwrap_or(Uuid::new_v4());
let call = if let Some(call) = call_manager::get_call(call_id).await { let call = if let Some(call) = call_manager::get_call(call_id).await {
if call.is_anonymous().await { if call.is_anonymous().await {
@ -120,7 +121,7 @@ impl AnonymousClientConnection {
} else { } else {
self.send_error_response( self.send_error_response(
&cv.get_id(), &cv.get_id(),
CommunicationType::ErrorNotAuthenticated, CommunicationType::error_not_authenticated,
) )
.await; .await;
return; return;
@ -128,7 +129,7 @@ impl AnonymousClientConnection {
} else { } else {
self.send_error_response( self.send_error_response(
&cv.get_id(), &cv.get_id(),
CommunicationType::ErrorNotAuthenticated, CommunicationType::error_not_authenticated,
) )
.await; .await;
return; return;
@ -138,9 +139,9 @@ impl AnonymousClientConnection {
for call_invitee in call.members.read().await.clone() { for call_invitee in call.members.read().await.clone() {
let call_invitee_cv = get_omega_connection() let call_invitee_cv = get_omega_connection()
.await_response( .await_response(
&CommunicationValue::new(CommunicationType::GetUserData).add_typed_default( &CommunicationValue::new(CommunicationType::get_user_data).add_data(
DataType::UserId, DataTypes::user_id,
DataValue::SignedNumber(call_invitee.user_id.into()), DataValue::Number(call_invitee.user_id as i64),
), ),
Some(Duration::from_secs(2)), Some(Duration::from_secs(2)),
) )
@ -148,122 +149,120 @@ impl AnonymousClientConnection {
.unwrap(); .unwrap();
let mut json_invitee = Vec::new(); let mut json_invitee = Vec::new();
let _ = json_invitee.push(( let _ = json_invitee.push((
DataType::UserId, DataTypes::user_id,
call_invitee_cv.get_data(DataType::UserId).clone(), call_invitee_cv.get_data(DataTypes::user_id).clone(),
)); ));
let _ = json_invitee.push(( let _ = json_invitee.push((
DataType::Username, DataTypes::username,
call_invitee_cv.get_data(DataType::Username).clone(), call_invitee_cv.get_data(DataTypes::username).clone(),
)); ));
let _ = json_invitee.push(( let _ = json_invitee.push((
DataType::Display, DataTypes::display,
call_invitee_cv.get_data(DataType::Display).clone(), call_invitee_cv.get_data(DataTypes::display).clone(),
)); ));
let _ = json_invitee.push(( let _ = json_invitee.push((
DataType::Avatar, DataTypes::avatar,
call_invitee_cv.get_data(DataType::Avatar).clone(), call_invitee_cv.get_data(DataTypes::avatar).clone(),
)); ));
let _ = invited.push(DataValue::Container(json_invitee.iter().map(|(k, v)| (k.to_id(&TypeMap::latest()), v.clone())).collect())); let _ = invited.push(DataValue::Container(json_invitee));
} }
let token = call.create_anonymous_token(self.get_user_id()).await; let token = call.create_anonymous_token(self.get_user_id()).await;
let mut serialized = Vec::new(); let mut serialized = Vec::new();
let _ = serialized.push((DataType::CallId, DataValue::Str(call_id.to_string()))); let _ = serialized.push((DataTypes::call_id, DataValue::Str(call_id.to_string())));
let _ = serialized.push((DataType::CallInvited, DataValue::Array(invited.clone()))); let _ =
let _ = serialized.push((DataType::CallMembers, DataValue::Array(invited))); serialized.push((DataTypes::call_invited, DataValue::Array(invited.clone())));
let _ = serialized.push((DataType::CallToken, DataValue::Str(token.unwrap()))); let _ = serialized.push((DataTypes::call_members, DataValue::Array(invited)));
let _ = serialized.push((DataTypes::call_token, DataValue::Str(token.unwrap())));
self.clone() self.clone()
.send_message( .send_message(
&&CommunicationValue::new(CommunicationType::IdentificationResponse) &&CommunicationValue::new(CommunicationType::identification_response)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default( .add_data(DataTypes::user_id, DataValue::Number(self.user_id as i64))
DataType::UserId, .add_data(
DataValue::SignedNumber(self.user_id.into()), DataTypes::username,
)
.add_typed_default(
DataType::Username,
DataValue::Str(self.clone().get_user_name().await), DataValue::Str(self.clone().get_user_name().await),
) )
.add_typed_default( .add_data(
DataType::Display, DataTypes::display,
DataValue::Str(self.get_display_name().await), DataValue::Str(self.get_display_name().await),
) )
.add_typed_default(DataType::Avatar, DataValue::Str(self.get_avatar().await)) .add_data(DataTypes::avatar, DataValue::Str(self.get_avatar().await))
.add_typed_default(DataType::CallState, DataValue::Container(serialized.iter().map(|(k, v)| (k.to_id(&TypeMap::latest()), v.clone())).collect())), .add_data(DataTypes::call_state, DataValue::Container(serialized)),
) )
.await; .await;
} }
// Handle ping // Handle ping
if cv.is_type(CommunicationType::Ping) { if cv.is_type(CommunicationType::ping) {
self.handle_ping(cv).await; self.handle_ping(cv).await;
return; return;
} }
// Handle client status changes // Handle client status changes
if cv.is_type(CommunicationType::ClientChanged) { if cv.is_type(CommunicationType::client_changed) {
self.handle_client_changed(cv).await; self.handle_client_changed(cv).await;
return; return;
} }
// Handle call invites // Handle call invites
if cv.is_type(CommunicationType::CallInvite) { if cv.is_type(CommunicationType::call_invite) {
self.handle_call_invite(cv).await; self.handle_call_invite(cv).await;
return; return;
} }
// Handle get call requests // Handle get call requests
if cv.is_type(CommunicationType::CallToken) { if cv.is_type(CommunicationType::call_token) {
self.handle_get_call(cv).await; self.handle_get_call(cv).await;
return; return;
} }
if cv.is_type(CommunicationType::CallDisconnectUser) { if cv.is_type(CommunicationType::call_disconnect_user) {
self.handle_call_disconnect_user(cv).await; self.handle_call_disconnect_user(cv).await;
return; return;
} }
if cv.is_type(CommunicationType::CallTimeoutUser) { if cv.is_type(CommunicationType::call_timeout_user) {
self.handle_call_timeout_user(cv).await; self.handle_call_timeout_user(cv).await;
return; return;
} }
if cv.is_type(CommunicationType::ChangeUserData) { if cv.is_type(CommunicationType::change_user_data) {
if let Some(display_name) = cv.get_data(DataType::Display).as_str() { if let Some(display_name) = cv.get_data(DataTypes::display).as_str() {
let _ = self.set_display_name(display_name.to_string()).await; let _ = self.set_display_name(display_name.to_string()).await;
} }
return; return;
} }
if cv.is_type(CommunicationType::GetUserData) { if cv.is_type(CommunicationType::get_user_data) {
if let Some(anonymous) = { if let Some(anonymous) = {
if let Some(user_id) = cv.get_data(DataType::UserId).as_signed_number() { if let Some(user_id) = cv.get_data(DataTypes::user_id).as_number() {
anonymous_manager::get_anonymous_user(user_id as u64).await anonymous_manager::get_anonymous_user(user_id as u64).await
} else if let Some(username) = cv.get_data(DataType::Username).as_str() { } else if let Some(username) = cv.get_data(DataTypes::username).as_str() {
anonymous_manager::get_anonymous_user_by_name(username.to_string()).await anonymous_manager::get_anonymous_user_by_name(username.to_string()).await
} else { } else {
None None
} }
} { } {
let response = CommunicationValue::new(CommunicationType::GetUserData) let response = CommunicationValue::new(CommunicationType::get_user_data)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default( .add_data(
DataType::Username, DataTypes::username,
DataValue::Str(anonymous.get_user_name().await), DataValue::Str(anonymous.get_user_name().await),
) )
.add_typed_default( .add_data(
DataType::UserId, DataTypes::user_id,
DataValue::SignedNumber(anonymous.user_id.into()), DataValue::Number(anonymous.user_id as i64),
) )
.add_typed_default( .add_data(
DataType::Display, DataTypes::display,
DataValue::Str(anonymous.get_display_name().await), DataValue::Str(anonymous.get_display_name().await),
) )
.add_typed_default(DataType::UserState, DataValue::Str("online".to_string())) .add_data(DataTypes::user_state, DataValue::Str("online".to_string()))
.add_typed_default( .add_data(
DataType::Avatar, DataTypes::avatar,
DataValue::Str(anonymous.get_avatar().await), DataValue::Str(anonymous.get_avatar().await),
); );
@ -273,9 +272,9 @@ impl AnonymousClientConnection {
} }
} }
if cv.is_type(CommunicationType::GetUserData) if cv.is_type(CommunicationType::get_user_data)
|| cv.is_type(CommunicationType::GetIotaData) || cv.is_type(CommunicationType::get_iota_data)
|| cv.is_type(CommunicationType::DeleteUser) || cv.is_type(CommunicationType::delete_user)
{ {
self.handle_omega_forward(cv).await; self.handle_omega_forward(cv).await;
return; return;
@ -297,7 +296,7 @@ impl AnonymousClientConnection {
/// Handle ping message /// Handle ping message
async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) { async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) {
// Update our ping if provided // Update our ping if provided
if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) { if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) {
if let Ok(ping_val) = last_ping.to_string().parse::<i64>() { if let Ok(ping_val) = last_ping.to_string().parse::<i64>() {
let mut ping_guard = self.ping.write().await; let mut ping_guard = self.ping.write().await;
*ping_guard = ping_val; *ping_guard = ping_val;
@ -305,14 +304,14 @@ impl AnonymousClientConnection {
} }
// Send pong response // Send pong response
let response = CommunicationValue::new(CommunicationType::Pong).with_id(cv.get_id()); let response = CommunicationValue::new(CommunicationType::pong).with_id(cv.get_id());
self.send_message(&response).await; self.send_message(&response).await;
} }
/// Handle client status change /// Handle client status change
async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) { async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) {
if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) { if let DataValue::Str(status_str) = cv.get_data(DataTypes::user_state) {
let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online); let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online);
OmegaConnection::client_changed(self.user_id as i64, self.user_id as i64, user_status) OmegaConnection::client_changed(self.user_id as i64, self.user_id as i64, user_status)
.await; .await;
@ -321,37 +320,40 @@ impl AnonymousClientConnection {
/// Handle call invite /// Handle call invite
async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) {
let receiver_id: i64 = cv.get_data(DataType::ReceiverId).as_signed_number().unwrap_or(0) as i64; let receiver_id: i64 = cv.get_data(DataTypes::receiver_id).as_number().unwrap_or(0);
if receiver_id == 0 { if receiver_id == 0 {
self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoUserId) self.send_error_response(&cv.get_id(), CommunicationType::error_no_user_id)
.await; .await;
return; return;
} }
let call_id = match cv.get_data(DataType::CallId) { let call_id = match cv.get_data(DataTypes::call_id) {
DataValue::Str(id_str) => match Uuid::parse_str(&id_str.to_string()) { DataValue::Str(id_str) => match Uuid::parse_str(&id_str.to_string()) {
Ok(id) => id, Ok(id) => id,
Err(_) => { Err(_) => {
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidCallId) self.send_error_response(
.await; &cv.get_id(),
CommunicationType::error_invalid_call_id,
)
.await;
return; return;
} }
}, },
_ => { _ => {
self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoCallId) self.send_error_response(&cv.get_id(), CommunicationType::error_no_call_id)
.await; .await;
return; return;
} }
}; };
let secret = cv let secret = cv
.get_data(DataType::CallSecret) .get_data(DataTypes::call_secret)
.as_str() .as_str()
.map(|s| s.to_string()); .map(|s| s.to_string());
let invited = let invited =
call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret).await; call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret).await;
if !invited { if !invited {
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidCallId) self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_call_id)
.await; .await;
return; return;
} }
@ -364,15 +366,12 @@ impl AnonymousClientConnection {
let sender_id = self.get_user_id(); let sender_id = self.get_user_id();
// User is offline - send push notification for call invite // User is offline - send push notification for call invite
let push_cv = CommunicationValue::new(CommunicationType::PushNotification) let push_cv = CommunicationValue::new(CommunicationType::push_notification)
.with_receiver(receiver_id as u64) .with_receiver(receiver_id as u64)
.add_typed_default( .add_data(DataTypes::sender_id, DataValue::Number(sender_id as i64))
DataType::SenderId, .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
DataValue::SignedNumber(sender_id.into()), .add_data(
) DataTypes::notifications,
.add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string()))
.add_typed_default(
DataType::Notifications,
DataValue::Str("call_invite".to_string()), DataValue::Str("call_invite".to_string()),
); );
@ -382,9 +381,9 @@ impl AnonymousClientConnection {
let _ = omega_conn.send_message(&push_cv).await; let _ = omega_conn.send_message(&push_cv).await;
}); });
let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) let error_cv = CommunicationValue::new(CommunicationType::error_not_found)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::ReceiverId, DataValue::SignedNumber(receiver_id.into())); .add_data(DataTypes::receiver_id, DataValue::Number(receiver_id));
self.send_message(&error_cv).await; self.send_message(&error_cv).await;
return; return;
} }
@ -394,19 +393,19 @@ impl AnonymousClientConnection {
let sender_id = self.get_user_id(); let sender_id = self.get_user_id();
// Create and send call distribution message // Create and send call distribution message
let forward = CommunicationValue::new(CommunicationType::CallInvite) let forward = CommunicationValue::new(CommunicationType::call_invite)
.with_receiver(receiver_id as u64) .with_receiver(receiver_id as u64)
.with_sender(sender_id) .with_sender(sender_id)
.add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())) .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
.add_typed_default( .add_data(
DataType::ReceiverId, DataTypes::receiver_id,
DataValue::Str(receiver_id.to_string()), DataValue::Str(receiver_id.to_string()),
) )
.add_typed_default(DataType::SenderId, DataValue::Str(sender_id.to_string())); .add_data(DataTypes::sender_id, DataValue::Str(sender_id.to_string()));
target_rho.message_to_client(forward).await; target_rho.message_to_client(forward).await;
let response = CommunicationValue::new(CommunicationType::Success).with_id(cv.get_id()); let response = CommunicationValue::new(CommunicationType::success).with_id(cv.get_id());
self.send_message(&response).await; self.send_message(&response).await;
} }
@ -414,40 +413,44 @@ impl AnonymousClientConnection {
async fn handle_get_call(self: Arc<Self>, cv: CommunicationValue) { async fn handle_get_call(self: Arc<Self>, cv: CommunicationValue) {
let user_id = self.get_user_id(); let user_id = self.get_user_id();
let call_id = match cv.get_data(DataType::CallId) { let call_id = match cv.get_data(DataTypes::call_id) {
DataValue::Str(id_str) => match Uuid::parse_str(&id_str.to_string()) { DataValue::Str(id_str) => match Uuid::parse_str(&id_str.to_string()) {
Ok(id) => id, Ok(id) => id,
Err(_) => { Err(_) => {
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidCallId) self.send_error_response(
.await; &cv.get_id(),
CommunicationType::error_invalid_call_id,
)
.await;
return; return;
} }
}, },
_ => { _ => {
self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoCallId) self.send_error_response(&cv.get_id(), CommunicationType::error_no_call_id)
.await; .await;
return; return;
} }
}; };
if let Some(token) = call_manager::get_call_token(user_id, call_id).await { if let Some(token) = call_manager::get_call_token(user_id, call_id).await {
let response = CommunicationValue::new(CommunicationType::CallToken) let response = CommunicationValue::new(CommunicationType::call_token)
.with_id(cv.get_id()) .with_id(cv.get_id())
.with_receiver(user_id) .with_receiver(user_id)
.add_typed_default(DataType::CallToken, DataValue::Str(token.to_string())); .add_data(DataTypes::call_token, DataValue::Str(token.to_string()));
self.send_message(&response).await; self.send_message(&response).await;
} else { } else {
let error_cv = CommunicationValue::new(CommunicationType::ErrorNoCallId) let error_cv = CommunicationValue::new(CommunicationType::error_no_call_id)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())); .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()));
self.send_message(&error_cv).await; self.send_message(&error_cv).await;
return; return;
} }
} }
async fn handle_call_timeout_user(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_timeout_user(self: Arc<Self>, cv: CommunicationValue) {
let call_id = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")).unwrap(); let call_id =
let user_id = cv.get_data(DataType::UserId).as_signed_number().unwrap_or(0); Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap();
let untill = cv.get_data(DataType::Untill).as_signed_number().unwrap_or(0); let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
let untill = cv.get_data(DataTypes::untill).as_number().unwrap_or(0);
let call = call_manager::get_call(call_id).await; let call = call_manager::get_call(call_id).await;
if let Some(call) = call { if let Some(call) = call {
@ -460,14 +463,15 @@ impl AnonymousClientConnection {
call.get_caller(user_id as u64) call.get_caller(user_id as u64)
.await .await
.unwrap() .unwrap()
.set_timeout(untill.try_into().unwrap()) .set_timeout(untill)
.await; .await;
} }
} }
} }
async fn handle_call_disconnect_user(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_disconnect_user(self: Arc<Self>, cv: CommunicationValue) {
let call_id = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")).unwrap(); let call_id =
let user_id = cv.get_data(DataType::UserId).as_signed_number().unwrap_or(0); Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap();
let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
let call = call_manager::get_call(call_id).await; let call = call_manager::get_call(call_id).await;
if let Some(call) = call { if let Some(call) = call {
@ -521,9 +525,9 @@ impl AnonymousClientConnection {
} else { } else {
user_status user_status
}; };
let notification = CommunicationValue::new(CommunicationType::ClientChanged) let notification = CommunicationValue::new(CommunicationType::client_changed)
.add_typed_default(DataType::UserId, DataValue::Str(user_id.to_string())) .add_data(DataTypes::user_id, DataValue::Str(user_id.to_string()))
.add_typed_default(DataType::UserState, DataValue::Str(status.to_string())); .add_data(DataTypes::user_state, DataValue::Str(status.to_string()));
self.send_message(&notification).await; self.send_message(&notification).await;
} }

8
src/calls/call_group.rs
View file

@ -1,5 +1,5 @@
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use serde_json::Map; use serde_json::Map;
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use std::{collections::BTreeMap, env, sync::Arc, time::Duration}; use std::{collections::BTreeMap, env, sync::Arc, time::Duration};
use tokio::sync::RwLock; use tokio::sync::RwLock;
@ -111,15 +111,15 @@ impl CallGroup {
); );
let response_cv = get_omega_connection() let response_cv = get_omega_connection()
.await_response( .await_response(
&CommunicationValue::new(CommunicationType::ShortenLink) &CommunicationValue::new(CommunicationType::shorten_link)
.add_typed_default(DataType::Link, DataValue::Str(long_link)), .add_data(DataTypes::link, DataValue::Str(long_link)),
Some(Duration::from_secs(20)), Some(Duration::from_secs(20)),
) )
.await; .await;
if let Ok(response) = response_cv { if let Ok(response) = response_cv {
*self.short_link.write().await = Some( *self.short_link.write().await = Some(
response response
.get_data(DataType::Link) .get_data(DataTypes::link)
.as_str() .as_str()
.unwrap() .unwrap()
.to_string(), .to_string(),

103
src/omega/omega_connection.rs
View file

@ -9,8 +9,6 @@ use crate::{
}, },
}; };
use dashmap::DashMap; use dashmap::DashMap;
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use mtp::transport::{Policy, Receiver, SendMode, Sender};
use once_cell::sync::Lazy; use once_cell::sync::Lazy;
use std::{collections::HashMap, env, sync::Arc, time::Duration}; use std::{collections::HashMap, env, sync::Arc, time::Duration};
use tokio::{ use tokio::{
@ -18,6 +16,8 @@ use tokio::{
task::JoinHandle, task::JoinHandle,
time::{Instant, sleep}, time::{Instant, sleep},
}; };
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue, rand_u32};
use ttp_native::{Policy, Receiver, SendMode, Sender};
use uuid::Uuid; use uuid::Uuid;
// ============================================================================ // ============================================================================
@ -264,7 +264,7 @@ impl OmegaConnection {
let addr_str = format!("https://{}:{}", self.host, self.port); let addr_str = format!("https://{}:{}", self.host, self.port);
let (sender, mut receiver) = mtp::transport::client::connect( let (sender, mut receiver) = ttp_native::client::connect(
&addr_str, &addr_str,
None, None,
Policy { Policy {
@ -347,22 +347,22 @@ impl OmegaConnection {
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
async fn send_identification(&self) { async fn send_identification(&self) {
let id = rand::random::<u32>(); let id = rand_u32();
let omikron_id = env::var("ID") let omikron_id = env::var("ID")
.unwrap_or("0".to_string()) .unwrap_or("0".to_string())
.parse::<i64>() .parse::<i64>()
.unwrap_or(0); .unwrap_or(0);
let identify_msg = CommunicationValue::new(CommunicationType::Identification) let identify_msg = CommunicationValue::new(CommunicationType::identification)
.with_id(id) .with_id(id)
.add_typed_default(DataType::OmikronId, DataValue::SignedNumber(omikron_id.into())); .add_data(DataTypes::omikron_id, DataValue::Number(omikron_id));
WAITING_TASKS.insert( WAITING_TASKS.insert(
id, id,
WaitingTask { WaitingTask {
task: Box::new(|selfc, cv| { task: Box::new(|selfc, cv| {
if cv.is_type(CommunicationType::ErrorNotFound) { if cv.is_type(CommunicationType::error_not_found) {
log_err!( log_err!(
0, 0,
PrintType::Omega, PrintType::Omega,
@ -370,7 +370,7 @@ impl OmegaConnection {
); );
return false; return false;
} }
if !cv.is_type(CommunicationType::Challenge) { if !cv.is_type(CommunicationType::challenge) {
return false; return false;
} }
@ -390,12 +390,12 @@ impl OmegaConnection {
async fn handle_challenge(&self, cv: CommunicationValue) -> Result<(), String> { async fn handle_challenge(&self, cv: CommunicationValue) -> Result<(), String> {
let challenge = cv let challenge = cv
.get_data(DataType::Challenge) .get_data(DataTypes::challenge)
.as_str() .as_str()
.ok_or("Challenge not found")?; .ok_or("Challenge not found")?;
let server_pub_key = cv let server_pub_key = cv
.get_data(DataType::PublicKey) .get_data(DataTypes::public_key)
.as_str() .as_str()
.ok_or("Public key not found")?; .ok_or("Public key not found")?;
@ -406,9 +406,9 @@ impl OmegaConnection {
) )
.map_err(|e| format!("Decryption failed: {:?}", e))?; .map_err(|e| format!("Decryption failed: {:?}", e))?;
let response_msg = CommunicationValue::new(CommunicationType::ChallengeResponse) let response_msg = CommunicationValue::new(CommunicationType::challenge_response)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::Challenge, DataValue::Str(decrypted_challenge)); .add_data(DataTypes::challenge, DataValue::Str(decrypted_challenge));
let response_id = response_msg.get_id(); let response_id = response_msg.get_id();
@ -416,13 +416,13 @@ impl OmegaConnection {
response_id, response_id,
WaitingTask { WaitingTask {
task: Box::new(|selfc, final_cv| { task: Box::new(|selfc, final_cv| {
if !final_cv.is_type(CommunicationType::IdentificationResponse) { if !final_cv.is_type(CommunicationType::identification_response) {
log_err!(0, PrintType::Omega, "Expected identification_response"); log_err!(0, PrintType::Omega, "Expected identification_response");
return false; return false;
} }
let accepted = final_cv let accepted = final_cv
.get_data(DataType::Accepted) .get_data(DataTypes::accepted)
.as_bool() .as_bool()
.unwrap_or(false); .unwrap_or(false);
@ -459,25 +459,23 @@ impl OmegaConnection {
let rho_connections_reader = RHO_CONNECTIONS.read().await; let rho_connections_reader = RHO_CONNECTIONS.read().await;
for iota_id in rho_connections_reader.keys() { for iota_id in rho_connections_reader.keys() {
connected_iota_ids.push(DataValue::SignedNumber((*iota_id).into())); connected_iota_ids.push(DataValue::Number(*iota_id));
} }
for rho in rho_connections_reader.values() { for rho in rho_connections_reader.values() {
for client_conn in rho.get_client_connections().await { for client_conn in rho.get_client_connections().await {
connected_user_ids.push(DataValue::SignedNumber( connected_user_ids.push(DataValue::Number(client_conn.get_user_id().await as i64));
client_conn.get_user_id().await.into(),
));
} }
} }
drop(rho_connections_reader); drop(rho_connections_reader);
let sync_msg = CommunicationValue::new(CommunicationType::SyncClientIotaStatus) let sync_msg = CommunicationValue::new(CommunicationType::sync_client_iota_status)
.add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids)) .add_data(DataTypes::iota_ids, DataValue::Array(connected_iota_ids))
.add_typed_default(DataType::UserIds, DataValue::Array(connected_user_ids)) .add_data(DataTypes::user_ids, DataValue::Array(connected_user_ids))
.add_typed_default( .add_data(
DataType::RhoConnections, DataTypes::rho_connections,
DataValue::SignedNumber(connection_count().await as i128), DataValue::Number(connection_count().await as i64),
); );
self.send_message(&sync_msg).await; self.send_message(&sync_msg).await;
@ -490,7 +488,7 @@ impl OmegaConnection {
async fn read_loop( async fn read_loop(
self: Arc<Self>, self: Arc<Self>,
receiver: &mut Receiver, receiver: &mut Receiver,
sender_handle: Arc<mtp::transport::ConnectionHandle>, sender_handle: Arc<ttp_native::ConnectionHandle>,
) { ) {
// Monitor both receiver and sender handle for close // Monitor both receiver and sender handle for close
let mut close_rx = sender_handle.subscribe_close(); let mut close_rx = sender_handle.subscribe_close();
@ -500,11 +498,11 @@ impl OmegaConnection {
result = receiver.receive() => { result = receiver.receive() => {
match result { match result {
Ok(cv) => { Ok(cv) => {
if !cv.is_type(CommunicationType::Pong) && !cv.is_type(CommunicationType::Ping) { if !cv.is_type(CommunicationType::pong) && !cv.is_type(CommunicationType::ping) {
log_cv_in!(PrintType::Omega, &cv); log_cv_in!(PrintType::Omega, &cv);
} }
if cv.is_type(CommunicationType::Pong) || cv.is_type(CommunicationType::Ping) { if cv.is_type(CommunicationType::pong) || cv.is_type(CommunicationType::ping) {
self.handle_pong(&cv).await; self.handle_pong(&cv).await;
continue; continue;
} }
@ -516,17 +514,17 @@ impl OmegaConnection {
} }
} }
if cv.is_type(CommunicationType::IotaUserData) { if cv.is_type(CommunicationType::iota_user_data) {
if let DataValue::Array(users) = cv.get_data(DataType::UserIds) { if let DataValue::Array(users) = cv.get_data(DataTypes::user_ids) {
let mut user_ids: Vec<u64> = Vec::new(); let mut user_ids: Vec<u64> = Vec::new();
for value in users { for value in users {
if let DataValue::SignedNumber(user_id) = value { if let DataValue::Number(user_id) = value {
user_ids.push(*user_id as u64); user_ids.push(*user_id as u64);
} }
} }
let connections = crate::rho::rho_manager::RHO_CONNECTIONS.read().await; let connections = crate::rho::rho_manager::RHO_CONNECTIONS.read().await;
if let Some(iota_id) = cv.get_data(DataType::IotaId).as_number() { if let Some(iota_id) = cv.get_data(DataTypes::iota_id).as_number() {
if let Some(rho) = connections.get(&(iota_id as i64)) { if let Some(rho) = connections.get(&iota_id) {
rho.get_iota_connection().set_user_ids(user_ids).await; rho.get_iota_connection().set_user_ids(user_ids).await;
} }
} else { } else {
@ -580,13 +578,13 @@ impl OmegaConnection {
} }
async fn send_ping(&self) { async fn send_ping(&self) {
let ping = CommunicationValue::new(CommunicationType::Ping).add_typed_default( let ping = CommunicationValue::new(CommunicationType::ping).add_data(
DataType::SendTime, DataTypes::send_time,
DataValue::SignedNumber( DataValue::Number(
std::time::SystemTime::now() std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs() as i128, .as_secs() as i64,
), ),
); );
self.send_message(&ping).await; self.send_message(&ping).await;
@ -594,16 +592,16 @@ impl OmegaConnection {
async fn handle_pong(&self, cv: &CommunicationValue) { async fn handle_pong(&self, cv: &CommunicationValue) {
let timestamp = cv let timestamp = cv
.get_data(DataType::SendTime) .get_data(DataTypes::send_time)
.as_number() .as_number()
.unwrap_or_else(|| { .unwrap_or_else(|| {
std::time::SystemTime::now() std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs() as i128 .as_secs() as i64
}); });
*self.last_ping.lock().await = timestamp.try_into().unwrap(); *self.last_ping.lock().await = timestamp;
} }
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
@ -611,7 +609,7 @@ impl OmegaConnection {
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
pub async fn send_message(&self, cv: &CommunicationValue) { pub async fn send_message(&self, cv: &CommunicationValue) {
if !cv.is_type(CommunicationType::Pong) && !cv.is_type(CommunicationType::Ping) { if !cv.is_type(CommunicationType::pong) && !cv.is_type(CommunicationType::ping) {
log_cv_out!(PrintType::Omega, &cv); log_cv_out!(PrintType::Omega, &cv);
} }
@ -712,33 +710,30 @@ impl OmegaConnection {
#[allow(dead_code)] #[allow(dead_code)]
pub async fn close_iota(iota_id: i64) { pub async fn close_iota(iota_id: i64) {
let cv = CommunicationValue::new(CommunicationType::IotaDisconnected) let cv = CommunicationValue::new(CommunicationType::iota_disconnected)
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())); .add_data(DataTypes::iota_id, DataValue::Number(iota_id));
OMEGA_CONNECTION.send_message(&cv).await; OMEGA_CONNECTION.send_message(&cv).await;
} }
pub async fn client_changed(_iota_id: i64, user_id: i64, state: UserStatus) { pub async fn client_changed(_iota_id: i64, user_id: i64, state: UserStatus) {
let msg_type = match state { let msg_type = match state {
UserStatus::iota_offline => CommunicationType::UserDisconnected, UserStatus::iota_offline => CommunicationType::user_disconnected,
UserStatus::user_offline => CommunicationType::UserDisconnected, UserStatus::user_offline => CommunicationType::user_disconnected,
UserStatus::user_invisible => CommunicationType::UserDisconnected, UserStatus::user_invisible => CommunicationType::user_disconnected,
_ => CommunicationType::UserConnected, _ => CommunicationType::user_connected,
}; };
let cv = CommunicationValue::new(msg_type) let cv = CommunicationValue::new(msg_type)
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) .add_data(DataTypes::user_id, DataValue::Number(user_id))
.add_typed_default(DataType::UserState, DataValue::Str(state.to_string())); .add_data(DataTypes::user_state, DataValue::Str(state.to_string()));
OMEGA_CONNECTION.send_message(&cv).await; OMEGA_CONNECTION.send_message(&cv).await;
} }
pub async fn user_states(user_id: i64, user_ids: Vec<i64>) { pub async fn user_states(user_id: i64, user_ids: Vec<i64>) {
let user_ids = user_ids let user_ids = user_ids.iter().map(|v| DataValue::Number(*v)).collect();
.iter()
.map(|v| DataValue::SignedNumber((*v).into()))
.collect();
let cv = CommunicationValue::new(CommunicationType::GetStates) let cv = CommunicationValue::new(CommunicationType::get_states)
.add_typed_default(DataType::UserIds, DataValue::Array(user_ids)); .add_data(DataTypes::user_ids, DataValue::Array(user_ids));
let msg_id = cv.get_id(); let msg_id = cv.get_id();
WAITING_TASKS.insert( WAITING_TASKS.insert(

10
src/omega/ping_pong_task.rs
View file

@ -1,6 +1,6 @@
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use std::time::Duration; use std::time::Duration;
use tokio::time::Instant; use tokio::time::Instant;
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue, rand_u32};
use crate::omega::omega_connection::OmegaConnection; use crate::omega::omega_connection::OmegaConnection;
@ -8,7 +8,7 @@ const PING_TIMEOUT: Duration = Duration::from_secs(30);
impl OmegaConnection { impl OmegaConnection {
pub async fn send_ping(&self) { pub async fn send_ping(&self) {
let id = rand::random::<u32>(); let id = rand_u32();
let send_time = Instant::now(); let send_time = Instant::now();
let mut message_send_times = self.message_send_times.lock().await; let mut message_send_times = self.message_send_times.lock().await;
@ -19,11 +19,11 @@ impl OmegaConnection {
} }
pub async fn send_ping_message(&self, id: u32) { pub async fn send_ping_message(&self, id: u32) {
let ping_message = CommunicationValue::new(CommunicationType::Ping) let ping_message = CommunicationValue::new(CommunicationType::ping)
.with_id(id) .with_id(id)
.add_data( .add_data(
DataType::LastPing, DataTypes::last_ping,
DataValue::SignedNumber(self.last_ping.lock().await.unwrap()), DataValue::Number(self.last_ping.lock().await.unwrap()),
); );
self.send_message(&ping_message).await; self.send_message(&ping_message).await;

81
src/rho/app_connection.rs
View file

@ -4,18 +4,17 @@ use crate::rho::connection::GeneralConnection;
use crate::rho::{rho_connection::RhoConnection, rho_manager}; use crate::rho::{rho_connection::RhoConnection, rho_manager};
use crate::util::logger::PrintType; use crate::util::logger::PrintType;
use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out}; use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out};
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use mtp::transport::{Receiver, Sender};
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH}; use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::RwLock; use tokio::sync::RwLock;
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use ttp_native::{Receiver, Sender};
use uuid::Uuid; use uuid::Uuid;
pub struct AppConnection { pub struct AppConnection {
pub user_id: u64, pub user_id: u64,
pub app_identifier: String, pub app_identifier: String,
pub app_session: Uuid, pub app_session: Uuid,
pub client_version: String,
pub sender: Arc<Sender>, pub sender: Arc<Sender>,
pub receiver: Arc<Receiver>, pub receiver: Arc<Receiver>,
@ -37,7 +36,6 @@ impl AppConnection {
user_id: user_id, user_id: user_id,
app_identifier: general.app_identifier.read().await.clone().unwrap(), app_identifier: general.app_identifier.read().await.clone().unwrap(),
app_session: general.app_session.read().await.clone().unwrap(), app_session: general.app_session.read().await.clone().unwrap(),
client_version: general.client_version.read().await.clone(),
}) })
} }
pub fn start(self: Arc<Self>) { pub fn start(self: Arc<Self>) {
@ -54,7 +52,7 @@ impl AppConnection {
tokio::time::sleep(Duration::from_millis(50)).await; tokio::time::sleep(Duration::from_millis(50)).await;
if self_clone2.get_rho_connection().await.is_none() { if self_clone2.get_rho_connection().await.is_none() {
self_clone2 self_clone2
.send_error_response(0, CommunicationType::ErrorNoIota) .send_error_response(0, CommunicationType::error_no_iota)
.await; .await;
} }
}); });
@ -86,7 +84,7 @@ impl AppConnection {
); );
return; return;
} }
if !cv.is_type(CommunicationType::Pong) && !cv.is_type(CommunicationType::Ping) { if !cv.is_type(CommunicationType::pong) && !cv.is_type(CommunicationType::ping) {
log_cv_out!(PrintType::App, &cv); log_cv_out!(PrintType::App, &cv);
} }
let _ = self.sender.send(&cv).await; let _ = self.sender.send(&cv).await;
@ -95,41 +93,41 @@ impl AppConnection {
/// Handle incoming message from app /// Handle incoming message from app
pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) { pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) {
tokio::spawn(async move { tokio::spawn(async move {
if cv.is_type(CommunicationType::Ping) { if cv.is_type(CommunicationType::ping) {
self.handle_ping(cv).await; self.handle_ping(cv).await;
return; return;
} }
log_cv_in!(PrintType::App, cv); log_cv_in!(PrintType::App, cv);
if cv.is_type(CommunicationType::GetUserData) { if cv.is_type(CommunicationType::get_user_data) {
if let Some(anonymous) = { if let Some(anonymous) = {
if let Some(user_id) = cv.get_data(DataType::UserId).as_number() { if let Some(user_id) = cv.get_data(DataTypes::user_id).as_number() {
anonymous_manager::get_anonymous_user(user_id as u64).await anonymous_manager::get_anonymous_user(user_id as u64).await
} else if let Some(username) = cv.get_data(DataType::Username).as_str() { } else if let Some(username) = cv.get_data(DataTypes::username).as_str() {
anonymous_manager::get_anonymous_user_by_name(username.to_string()).await anonymous_manager::get_anonymous_user_by_name(username.to_string()).await
} else { } else {
None None
} }
} { } {
let response = CommunicationValue::new(CommunicationType::GetUserData) let response = CommunicationValue::new(CommunicationType::get_user_data)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default( .add_data(
DataType::Username, DataTypes::username,
DataValue::Str(anonymous.get_user_name().await), DataValue::Str(anonymous.get_user_name().await),
) )
.add_typed_default( .add_data(
DataType::UserId, DataTypes::user_id,
DataValue::SignedNumber(anonymous.get_user_id().into()), DataValue::Number(anonymous.get_user_id() as i64),
) )
.add_typed_default( .add_data(
DataType::Display, DataTypes::display,
DataValue::Str(anonymous.get_display_name().await), DataValue::Str(anonymous.get_display_name().await),
) )
.add_typed_default( .add_data(
DataType::Avatar, DataTypes::avatar,
DataValue::Str(anonymous.get_avatar().await), DataValue::Str(anonymous.get_avatar().await),
) )
.add_typed_default(DataType::UserState, DataValue::Str("online".to_string())); .add_data(DataTypes::user_state, DataValue::Str("online".to_string()));
self.send_message(&response).await; self.send_message(&response).await;
@ -137,8 +135,8 @@ impl AppConnection {
} }
} }
if cv.is_type(CommunicationType::GetUserData) if cv.is_type(CommunicationType::get_user_data)
|| cv.is_type(CommunicationType::GetIotaData) || cv.is_type(CommunicationType::get_iota_data)
{ {
let sender = self.get_user_id().await; let sender = self.get_user_id().await;
self.handle_omega_forward(cv.with_sender(sender as u64)) self.handle_omega_forward(cv.with_sender(sender as u64))
@ -164,13 +162,13 @@ impl AppConnection {
/// Handle ping message /// Handle ping message
async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) { async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) {
// Update our ping if provided // Update our ping if provided
if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) { if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) {
let current = SystemTime::now() let current = SystemTime::now()
.duration_since(UNIX_EPOCH) .duration_since(UNIX_EPOCH)
.unwrap() .unwrap()
.as_millis(); .as_millis();
let mut ping_guard = self.ping.write().await; let mut ping_guard = self.ping.write().await;
*ping_guard = (current as i128 - *last_ping) as i64; *ping_guard = current as i64 - last_ping;
} }
// Get Iota ping from RhoConnection // Get Iota ping from RhoConnection
@ -181,9 +179,9 @@ impl AppConnection {
}; };
// Send pong response // Send pong response
let response = CommunicationValue::new(CommunicationType::Pong) let response = CommunicationValue::new(CommunicationType::pong)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::PingIota, DataValue::SignedNumber(iota_ping.into())); .add_data(DataTypes::ping_iota, DataValue::Number(iota_ping));
self.send_message(&response).await; self.send_message(&response).await;
} }
@ -204,11 +202,14 @@ impl AppConnection {
cv.get_receiver() cv.get_receiver()
); );
if cv.is_type(CommunicationType::AddConversation) if cv.is_type(CommunicationType::add_conversation)
&& cv.get_data(DataType::ChatPartnerId).as_number().is_none() && cv
.get_data(DataTypes::chat_partner_id)
.as_number()
.is_none()
{ {
let chat_partner_name = cv let chat_partner_name = cv
.get_data(DataType::ChatPartnerName) .get_data(DataTypes::chat_partner_name)
.as_str() .as_str()
.unwrap_or("") .unwrap_or("")
.to_string(); .to_string();
@ -217,17 +218,17 @@ impl AppConnection {
.await .await
.is_some() .is_some()
{ {
self.send_error_response(cv.get_id(), CommunicationType::ErrorAnonymous) self.send_error_response(cv.get_id(), CommunicationType::error_anonymous)
.await; .await;
return; return;
} }
let load_uuid_response = get_omega_connection() let load_uuid_response = get_omega_connection()
.await_response( .await_response(
&CommunicationValue::new(CommunicationType::GetUserData) &CommunicationValue::new(CommunicationType::get_user_data)
.with_id(cv.clone().get_id()) .with_id(cv.clone().get_id())
.add_typed_default( .add_data(
DataType::Username, DataTypes::username,
DataValue::Str(chat_partner_name.clone()), DataValue::Str(chat_partner_name.clone()),
), ),
Some(Duration::from_secs(20)), Some(Duration::from_secs(20)),
@ -235,7 +236,7 @@ impl AppConnection {
.await; .await;
let chat_partner_id = { let chat_partner_id = {
if let Ok(load_uuid_response) = load_uuid_response { if let Ok(load_uuid_response) = load_uuid_response {
load_uuid_response.get_data(DataType::UserId).clone() load_uuid_response.get_data(DataTypes::user_id).clone()
} else { } else {
DataValue::Null DataValue::Null
} }
@ -254,7 +255,7 @@ impl AppConnection {
let updated_cv = cv let updated_cv = cv
.with_sender(sender_user_id as u64) .with_sender(sender_user_id as u64)
.add_typed_default(DataType::ChatPartnerId, chat_partner_id); .add_data(DataTypes::chat_partner_id, chat_partner_id);
rho_conn.message_to_iota(updated_cv).await; rho_conn.message_to_iota(updated_cv).await;
} else { } else {
log_err!( log_err!(
@ -292,12 +293,9 @@ impl AppConnection {
msg_type, msg_type,
msg_id msg_id
); );
let error_cv = CommunicationValue::new(CommunicationType::ErrorNoIota) let error_cv = CommunicationValue::new(CommunicationType::error_no_iota)
.with_id(msg_id) .with_id(msg_id)
.add_typed_default( .add_data(DataTypes::user_id, DataValue::Number(sender_user_id as i64));
DataType::UserId,
DataValue::SignedNumber(sender_user_id.into()),
);
self.send_message(&error_cv).await; self.send_message(&error_cv).await;
} }
} }
@ -338,7 +336,6 @@ impl Clone for AppConnection {
user_id: self.user_id, user_id: self.user_id,
app_identifier: self.app_identifier.clone(), app_identifier: self.app_identifier.clone(),
app_session: self.app_session, app_session: self.app_session,
client_version: self.client_version.clone(),
ping: Arc::clone(&self.ping), ping: Arc::clone(&self.ping),
pub_key: Arc::clone(&self.pub_key), pub_key: Arc::clone(&self.pub_key),
rho_connection: Arc::clone(&self.rho_connection), rho_connection: Arc::clone(&self.rho_connection),

319
src/rho/client_connection.rs
View file

@ -6,18 +6,17 @@ use crate::rho::{rho_connection::RhoConnection, rho_manager};
use crate::util::logger::PrintType; use crate::util::logger::PrintType;
use crate::{data::user::UserStatus, omega::omega_connection::OmegaConnection}; use crate::{data::user::UserStatus, omega::omega_connection::OmegaConnection};
use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out}; use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out};
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use mtp::transport::{Receiver, Sender};
use std::str::FromStr; use std::str::FromStr;
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH}; use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::RwLock; use tokio::sync::RwLock;
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use ttp_native::{Receiver, Sender};
use uuid::Uuid; use uuid::Uuid;
pub struct ClientConnection { pub struct ClientConnection {
pub user_id: u64, pub user_id: u64,
pub session_id: u64, pub session_id: u64,
pub client_version: String,
pub sender: Arc<Sender>, pub sender: Arc<Sender>,
pub receiver: Arc<Receiver>, pub receiver: Arc<Receiver>,
@ -40,7 +39,6 @@ impl ClientConnection {
receiver: general.receiver.clone(), receiver: general.receiver.clone(),
user_id: user_id, user_id: user_id,
session_id: general.session_id.read().await.clone(), session_id: general.session_id.read().await.clone(),
client_version: general.client_version.read().await.clone(),
}) })
} }
pub fn start(self: Arc<Self>) { pub fn start(self: Arc<Self>) {
@ -57,7 +55,7 @@ impl ClientConnection {
tokio::time::sleep(Duration::from_millis(50)).await; tokio::time::sleep(Duration::from_millis(50)).await;
if self_clone2.get_rho_connection().await.is_none() { if self_clone2.get_rho_connection().await.is_none() {
self_clone2 self_clone2
.send_error_response(0, CommunicationType::ErrorNoIota) .send_error_response(0, CommunicationType::error_no_iota)
.await; .await;
} }
}); });
@ -88,7 +86,7 @@ impl ClientConnection {
); );
return; return;
} }
if !cv.is_type(CommunicationType::Pong) && !cv.is_type(CommunicationType::Ping) { if !cv.is_type(CommunicationType::pong) && !cv.is_type(CommunicationType::ping) {
log_cv_out!(PrintType::Client, &cv); log_cv_out!(PrintType::Client, &cv);
} }
let _ = self.sender.send(&cv).await; let _ = self.sender.send(&cv).await;
@ -97,7 +95,7 @@ impl ClientConnection {
/// Handle incoming message from client /// Handle incoming message from client
pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) { pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) {
tokio::spawn(async move { tokio::spawn(async move {
if cv.is_type(CommunicationType::Ping) { if cv.is_type(CommunicationType::ping) {
self.handle_ping(cv).await; self.handle_ping(cv).await;
return; return;
} }
@ -106,63 +104,75 @@ impl ClientConnection {
let mut cv = cv; let mut cv = cv;
// Handle client status changes // Handle client status changes
if cv.is_type(CommunicationType::ClientChanged) { if cv.is_type(CommunicationType::client_changed) {
self.handle_client_changed(cv).await; self.handle_client_changed(cv).await;
return; return;
} }
// Handle call invites // Handle call invites
if cv.is_type(CommunicationType::CallInvite) { if cv.is_type(CommunicationType::call_invite) {
self.handle_call_invite(cv).await; self.handle_call_invite(cv).await;
return; return;
} }
// Handle get call requests // Handle get call requests
if cv.is_type(CommunicationType::CallToken) { if cv.is_type(CommunicationType::call_token) {
self.handle_get_call(cv).await; self.handle_get_call(cv).await;
return; return;
} }
if cv.is_type(CommunicationType::CallData) { if cv.is_type(CommunicationType::call_data) {
self.handle_get_call_data(cv).await; self.handle_get_call_data(cv).await;
return; return;
} }
if cv.is_type(CommunicationType::CallDisconnectUser) { if cv.is_type(CommunicationType::call_disconnect_user) {
self.handle_call_disconnect_user(cv).await; self.handle_call_disconnect_user(cv).await;
return; return;
} }
if cv.is_type(CommunicationType::CallTimeoutUser) { if cv.is_type(CommunicationType::call_timeout_user) {
self.handle_call_timeout_user(cv).await; self.handle_call_timeout_user(cv).await;
return; return;
} }
if cv.is_type(CommunicationType::CallSetAnonymousJoining) { if cv.is_type(CommunicationType::call_set_anonymous_joining) {
self.handle_call_set_anonymous_joining(cv).await; self.handle_call_set_anonymous_joining(cv).await;
return; return;
} }
if cv.is_type(CommunicationType::LoadTxtRecord) { if cv.is_type(CommunicationType::load_txt_record) {
self.handle_load_txt_record(cv).await; self.handle_load_txt_record(cv).await;
return; return;
} }
if cv.is_type(CommunicationType::GetUserData) { if cv.is_type(CommunicationType::get_user_data) {
if let Some(anonymous) = { if let Some(anonymous) = {
if let Some(user_id) = cv.get_data(DataType::UserId).as_signed_number() { if let Some(user_id) = cv.get_data(DataTypes::user_id).as_number() {
anonymous_manager::get_anonymous_user(user_id as u64).await anonymous_manager::get_anonymous_user(user_id as u64).await
} else if let Some(username) = cv.get_data(DataType::Username).as_str() { } else if let Some(username) = cv.get_data(DataTypes::username).as_str() {
anonymous_manager::get_anonymous_user_by_name(username.to_string()).await anonymous_manager::get_anonymous_user_by_name(username.to_string()).await
} else { } else {
None None
} }
} { } {
let response = CommunicationValue::new(CommunicationType::GetUserData) let response = CommunicationValue::new(CommunicationType::get_user_data)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::Username, DataValue::Str(anonymous.get_user_name().await)) .add_data(
.add_typed_default(DataType::UserId, DataValue::SignedNumber(anonymous.get_user_id().into())) DataTypes::username,
.add_typed_default(DataType::Display, DataValue::Str(anonymous.get_display_name().await)) DataValue::Str(anonymous.get_user_name().await),
.add_typed_default(DataType::Avatar, DataValue::Str(anonymous.get_avatar().await)) )
.add_typed_default(DataType::UserState, DataValue::Str("online".to_string())); .add_data(
DataTypes::user_id,
DataValue::Number(anonymous.get_user_id() as i64),
)
.add_data(
DataTypes::display,
DataValue::Str(anonymous.get_display_name().await),
)
.add_data(
DataTypes::avatar,
DataValue::Str(anonymous.get_avatar().await),
)
.add_data(DataTypes::user_state, DataValue::Str("online".to_string()));
self.send_message(&response).await; self.send_message(&response).await;
@ -170,12 +180,12 @@ impl ClientConnection {
} }
} }
if cv.is_type(CommunicationType::ChangeUserData) if cv.is_type(CommunicationType::change_user_data)
|| cv.is_type(CommunicationType::ReadNotification) || cv.is_type(CommunicationType::read_notification)
|| cv.is_type(CommunicationType::GetNotifications) || cv.is_type(CommunicationType::get_notifications)
|| cv.is_type(CommunicationType::GetUserData) || cv.is_type(CommunicationType::get_user_data)
|| cv.is_type(CommunicationType::GetIotaData) || cv.is_type(CommunicationType::get_iota_data)
|| cv.is_type(CommunicationType::DeleteUser) || cv.is_type(CommunicationType::delete_user)
{ {
let sender = self.get_user_id().await; let sender = self.get_user_id().await;
self.handle_omega_forward(cv.with_sender(sender as u64)) self.handle_omega_forward(cv.with_sender(sender as u64))
@ -183,48 +193,72 @@ impl ClientConnection {
return; return;
} }
let is_per_device_settings = cv.is_type(CommunicationType::SettingsSave) let is_per_device_settings = cv.is_type(CommunicationType::settings_save)
|| cv.is_type(CommunicationType::SettingsLoad) || cv.is_type(CommunicationType::settings_load)
|| cv.is_type(CommunicationType::SettingsList); || cv.is_type(CommunicationType::settings_list);
let is_global_settings = cv.is_type(CommunicationType::GlobalSettingsSave) let is_global_settings = cv.is_type(CommunicationType::global_settings_save)
|| cv.is_type(CommunicationType::GlobalSettingsLoad); || cv.is_type(CommunicationType::global_settings_load);
if is_per_device_settings || is_global_settings { if is_per_device_settings || is_global_settings {
let expected_session_id: i128 = self.session_id.into(); let expected_session_id = self.session_id as i64;
let session_id = cv.get_data(DataType::SessionId).as_signed_number(); let session_id = cv.get_data(DataTypes::session_id).as_number();
if is_per_device_settings { if is_per_device_settings {
let Some(session_id) = session_id else { let Some(session_id) = session_id else {
let response = CommunicationValue::new(CommunicationType::ErrorInvalidData) let response =
.with_id(cv.get_id()) CommunicationValue::new(CommunicationType::error_invalid_data)
.with_receiver(self.user_id) .with_id(cv.get_id())
.add_typed_default(DataType::Message, DataValue::Str("Missing session_id".to_string())) .with_receiver(self.user_id)
.add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id)); .add_data(
DataTypes::message,
DataValue::Str("Missing session_id".to_string()),
)
.add_data(
DataTypes::session_id,
DataValue::Number(expected_session_id),
);
self.send_message(&response).await; self.send_message(&response).await;
return; return;
}; };
if session_id != expected_session_id { if session_id != expected_session_id {
let response = CommunicationValue::new(CommunicationType::ErrorInvalidData) let response =
.with_id(cv.get_id()) CommunicationValue::new(CommunicationType::error_invalid_data)
.with_receiver(self.user_id) .with_id(cv.get_id())
.add_typed_default(DataType::Message, DataValue::Str("session_id mismatch".to_string())) .with_receiver(self.user_id)
.add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id)); .add_data(
DataTypes::message,
DataValue::Str("session_id mismatch".to_string()),
)
.add_data(
DataTypes::session_id,
DataValue::Number(expected_session_id),
);
self.send_message(&response).await; self.send_message(&response).await;
return; return;
} }
} else if let Some(session_id) = session_id { } else if let Some(session_id) = session_id {
if session_id != expected_session_id { if session_id != expected_session_id {
let response = CommunicationValue::new(CommunicationType::ErrorInvalidData) let response =
.with_id(cv.get_id()) CommunicationValue::new(CommunicationType::error_invalid_data)
.with_receiver(self.user_id) .with_id(cv.get_id())
.add_typed_default(DataType::Message, DataValue::Str("session_id mismatch".to_string())) .with_receiver(self.user_id)
.add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id)); .add_data(
DataTypes::message,
DataValue::Str("session_id mismatch".to_string()),
)
.add_data(
DataTypes::session_id,
DataValue::Number(expected_session_id),
);
self.send_message(&response).await; self.send_message(&response).await;
return; return;
} }
} else { } else {
cv = cv.add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id)); cv = cv.add_data(
DataTypes::session_id,
DataValue::Number(expected_session_id),
);
} }
} }
@ -247,13 +281,13 @@ impl ClientConnection {
/// Handle ping message /// Handle ping message
async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) { async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) {
// Update our ping if provided // Update our ping if provided
if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) { if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) {
let current = SystemTime::now() let current = SystemTime::now()
.duration_since(UNIX_EPOCH) .duration_since(UNIX_EPOCH)
.unwrap() .unwrap()
.as_millis(); .as_millis();
let mut ping_guard = self.ping.write().await; let mut ping_guard = self.ping.write().await;
*ping_guard = (current as i128 - *last_ping) as i64; *ping_guard = current as i64 - last_ping;
} }
// Get Iota ping from RhoConnection // Get Iota ping from RhoConnection
@ -264,9 +298,9 @@ impl ClientConnection {
}; };
// Send pong response // Send pong response
let response = CommunicationValue::new(CommunicationType::Pong) let response = CommunicationValue::new(CommunicationType::pong)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::PingIota, DataValue::SignedNumber(iota_ping.into())); .add_data(DataTypes::ping_iota, DataValue::Number(iota_ping));
self.send_message(&response).await; self.send_message(&response).await;
} }
@ -274,7 +308,7 @@ impl ClientConnection {
/// Handle client status change /// Handle client status change
async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) { async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) {
let user_id = self.get_user_id().await; let user_id = self.get_user_id().await;
if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) { if let DataValue::Str(status_str) = cv.get_data(DataTypes::user_state) {
let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online); let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online);
if let Some(rho_conn) = self.get_rho_connection().await { if let Some(rho_conn) = self.get_rho_connection().await {
OmegaConnection::client_changed( OmegaConnection::client_changed(
@ -289,54 +323,57 @@ impl ClientConnection {
/// Handle call invite /// Handle call invite
async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) {
let receiver_id: i128 = cv.get_data(DataType::ReceiverId).as_signed_number().unwrap_or(0); let receiver_id: i64 = cv.get_data(DataTypes::receiver_id).as_number().unwrap_or(0);
if receiver_id == 0 { if receiver_id == 0 {
self.send_error_response(cv.get_id(), CommunicationType::ErrorNoUserId) self.send_error_response(cv.get_id(), CommunicationType::error_no_user_id)
.await; .await;
return; return;
} }
let call_id = match cv.get_data(DataType::CallId) { let call_id = match cv.get_data(DataTypes::call_id) {
DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) { DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) {
Ok(id) => id, Ok(id) => id,
Err(_) => { Err(_) => {
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
.await; .await;
return; return;
} }
}, },
_ => { _ => {
self.send_error_response(cv.get_id(), CommunicationType::ErrorNoCallId) self.send_error_response(cv.get_id(), CommunicationType::error_no_call_id)
.await; .await;
return; return;
} }
}; };
let secret = cv let secret = cv
.get_data(DataType::CallSecret) .get_data(DataTypes::call_secret)
.as_str() .as_str()
.map(|s| s.to_string()); .map(|s| s.to_string());
let invited = let invited =
call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret).await; call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret).await;
if !invited { if !invited {
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
.await; .await;
return; return;
} }
// Find target RhoConnection // Find target RhoConnection
let target_rho = match rho_manager::get_rho_con_for_user(receiver_id as i64).await { let target_rho = match rho_manager::get_rho_con_for_user(receiver_id).await {
Some(rho) => rho, Some(rho) => rho,
_ => { _ => {
// Get sender user ID // Get sender user ID
let sender_id = self.get_user_id().await; let sender_id = self.get_user_id().await;
// User is offline - send push notification for call invite // User is offline - send push notification for call invite
let push_cv = CommunicationValue::new(CommunicationType::PushNotification) let push_cv = CommunicationValue::new(CommunicationType::push_notification)
.with_receiver(receiver_id as u64) .with_receiver(receiver_id as u64)
.add_typed_default(DataType::SenderId, DataValue::SignedNumber(sender_id.into())) .add_data(DataTypes::sender_id, DataValue::Number(sender_id as i64))
.add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())) .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
.add_typed_default(DataType::Notifications, DataValue::Str("call_invite".to_string())); .add_data(
DataTypes::notifications,
DataValue::Str("call_invite".to_string()),
);
let omega_conn = get_omega_connection(); let omega_conn = get_omega_connection();
// Send fire-and-forget, don't await to avoid blocking // Send fire-and-forget, don't await to avoid blocking
@ -344,9 +381,9 @@ impl ClientConnection {
let _ = omega_conn.send_message(&push_cv).await; let _ = omega_conn.send_message(&push_cv).await;
}); });
let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) let error_cv = CommunicationValue::new(CommunicationType::error_not_found)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::ReceiverId, DataValue::SignedNumber(receiver_id.into())); .add_data(DataTypes::receiver_id, DataValue::Number(receiver_id));
self.send_message(&error_cv).await; self.send_message(&error_cv).await;
return; return;
} }
@ -356,17 +393,20 @@ impl ClientConnection {
let sender_id = self.get_user_id().await as i64; let sender_id = self.get_user_id().await as i64;
// Create and send call distribution message // Create and send call distribution message
let forward = CommunicationValue::new(CommunicationType::CallInvite) let forward = CommunicationValue::new(CommunicationType::call_invite)
.with_receiver(receiver_id as u64) .with_receiver(receiver_id as u64)
.with_sender(sender_id as u64) .with_sender(sender_id as u64)
.add_typed_default(DataType::CallSecret, cv.get_data(DataType::CallSecret).clone()) .add_data(
.add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())) DataTypes::call_secret,
.add_typed_default(DataType::ReceiverId, DataValue::SignedNumber(receiver_id.into())) cv.get_data(DataTypes::call_secret).clone(),
.add_typed_default(DataType::SenderId, DataValue::SignedNumber(sender_id.into())); )
.add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
.add_data(DataTypes::receiver_id, DataValue::Number(receiver_id))
.add_data(DataTypes::sender_id, DataValue::Number(sender_id));
target_rho.message_to_client(forward).await; target_rho.message_to_client(forward).await;
let response = CommunicationValue::new(CommunicationType::Success).with_id(cv.get_id()); let response = CommunicationValue::new(CommunicationType::success).with_id(cv.get_id());
self.send_message(&response).await; self.send_message(&response).await;
} }
@ -374,32 +414,32 @@ impl ClientConnection {
async fn handle_get_call(self: Arc<Self>, cv: CommunicationValue) { async fn handle_get_call(self: Arc<Self>, cv: CommunicationValue) {
let user_id = self.get_user_id().await; let user_id = self.get_user_id().await;
let call_id = match cv.get_data(DataType::CallId) { let call_id = match cv.get_data(DataTypes::call_id) {
DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) { DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) {
Ok(id) => id, Ok(id) => id,
Err(_) => { Err(_) => {
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
.await; .await;
return; return;
} }
}, },
_ => { _ => {
self.send_error_response(cv.get_id(), CommunicationType::ErrorNoCallId) self.send_error_response(cv.get_id(), CommunicationType::error_no_call_id)
.await; .await;
return; return;
} }
}; };
if let Some(token) = call_manager::get_call_token(user_id, call_id).await { if let Some(token) = call_manager::get_call_token(user_id, call_id).await {
let response = CommunicationValue::new(CommunicationType::CallToken) let response = CommunicationValue::new(CommunicationType::call_token)
.with_id(cv.get_id()) .with_id(cv.get_id())
.with_receiver(user_id as u64) .with_receiver(user_id as u64)
.add_typed_default(DataType::CallToken, DataValue::Str(token)); .add_data(DataTypes::call_token, DataValue::Str(token));
self.send_message(&response).await; self.send_message(&response).await;
} else { } else {
let error_cv = CommunicationValue::new(CommunicationType::ErrorNoCallId) let error_cv = CommunicationValue::new(CommunicationType::error_no_call_id)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())); .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()));
self.send_message(&error_cv).await; self.send_message(&error_cv).await;
return; return;
} }
@ -407,17 +447,17 @@ impl ClientConnection {
async fn handle_get_call_data(self: Arc<Self>, cv: CommunicationValue) { async fn handle_get_call_data(self: Arc<Self>, cv: CommunicationValue) {
let user_id = self.get_user_id().await; let user_id = self.get_user_id().await;
let call_id = match cv.get_data(DataType::CallId) { let call_id = match cv.get_data(DataTypes::call_id) {
DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) { DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) {
Ok(id) => id, Ok(id) => id,
Err(_) => { Err(_) => {
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
.await; .await;
return; return;
} }
}, },
_ => { _ => {
self.send_error_response(cv.get_id(), CommunicationType::ErrorNoCallId) self.send_error_response(cv.get_id(), CommunicationType::error_no_call_id)
.await; .await;
return; return;
} }
@ -429,75 +469,75 @@ impl ClientConnection {
let members = call.members.read().await.clone(); let members = call.members.read().await.clone();
for member in members { for member in members {
if member.user_id == user_id { if member.user_id == user_id {
user_ids.push(DataValue::SignedNumber(member.user_id.into())); user_ids.push(DataValue::Number(member.user_id as i64));
} }
} }
let response = CommunicationValue::new(CommunicationType::CallData) let response = CommunicationValue::new(CommunicationType::call_data)
.with_id(cv.get_id()) .with_id(cv.get_id())
.with_receiver(user_id as u64) .with_receiver(user_id as u64)
.add_typed_default(DataType::UserIds, DataValue::Array(user_ids)); .add_data(DataTypes::user_ids, DataValue::Array(user_ids));
self.send_message(&response).await; self.send_message(&response).await;
} else { } else {
let error_cv = CommunicationValue::new(CommunicationType::ErrorInvalidUserId) let error_cv = CommunicationValue::new(CommunicationType::error_invalid_user_id)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())); .add_data(DataTypes::user_id, DataValue::Number(user_id as i64));
self.send_message(&error_cv).await; self.send_message(&error_cv).await;
return; return;
} }
} else { } else {
let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) let error_cv = CommunicationValue::new(CommunicationType::error_not_found)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())); .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()));
self.send_message(&error_cv).await; self.send_message(&error_cv).await;
return; return;
} }
} }
async fn handle_call_timeout_user(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_timeout_user(self: Arc<Self>, cv: CommunicationValue) {
let Ok(call_id) = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")) let Ok(call_id) = Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or(""))
else { else {
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
.await; .await;
return; return;
}; };
let user_id = cv.get_data(DataType::UserId).as_signed_number().unwrap_or(0); let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
let untill = cv.get_data(DataType::Untill).as_signed_number().unwrap_or(0); let untill = cv.get_data(DataTypes::untill).as_number().unwrap_or(0);
let Some(call) = call_manager::get_call(call_id).await else { let Some(call) = call_manager::get_call(call_id).await else {
self.send_error_response(cv.get_id(), CommunicationType::ErrorNotFound) self.send_error_response(cv.get_id(), CommunicationType::error_not_found)
.await; .await;
return; return;
}; };
let Some(caller) = call.get_caller(self.get_user_id().await).await else { let Some(caller) = call.get_caller(self.get_user_id().await).await else {
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidUserId) self.send_error_response(cv.get_id(), CommunicationType::error_invalid_user_id)
.await; .await;
return; return;
}; };
if caller.has_admin() { if caller.has_admin() {
let _ = call_util::remove_participant(call_id, user_id as u64).await; let _ = call_util::remove_participant(call_id, user_id as u64).await;
if let Some(target) = call.get_caller(user_id as u64).await { if let Some(target) = call.get_caller(user_id as u64).await {
target.set_timeout(untill as i64).await; target.set_timeout(untill).await;
} }
} }
} }
async fn handle_call_disconnect_user(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_disconnect_user(self: Arc<Self>, cv: CommunicationValue) {
let Ok(call_id) = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")) let Ok(call_id) = Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or(""))
else { else {
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
.await; .await;
return; return;
}; };
let user_id = cv.get_data(DataType::UserId).as_signed_number().unwrap_or(0); let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
let Some(call) = call_manager::get_call(call_id).await else { let Some(call) = call_manager::get_call(call_id).await else {
self.send_error_response(cv.get_id(), CommunicationType::ErrorNotFound) self.send_error_response(cv.get_id(), CommunicationType::error_not_found)
.await; .await;
return; return;
}; };
let Some(caller) = call.get_caller(self.get_user_id().await).await else { let Some(caller) = call.get_caller(self.get_user_id().await).await else {
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidUserId) self.send_error_response(cv.get_id(), CommunicationType::error_invalid_user_id)
.await; .await;
return; return;
}; };
@ -506,13 +546,13 @@ impl ClientConnection {
} }
} }
async fn handle_call_set_anonymous_joining(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_set_anonymous_joining(self: Arc<Self>, cv: CommunicationValue) {
let Ok(call_id) = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")) let Ok(call_id) = Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or(""))
else { else {
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
.await; .await;
return; return;
}; };
let enable = cv.get_data(DataType::Enabled).as_bool().unwrap_or(true); let enable = cv.get_data(DataTypes::enabled).as_bool().unwrap_or(true);
let call = call_manager::get_call(call_id).await; let call = call_manager::get_call(call_id).await;
@ -525,18 +565,19 @@ impl ClientConnection {
} }
short_link = call.get_short_link().await; short_link = call.get_short_link().await;
} }
let mut response_cv = CommunicationValue::new(CommunicationType::CallSetAnonymousJoining) let mut response_cv =
.with_id(cv.get_id()) CommunicationValue::new(CommunicationType::call_set_anonymous_joining)
.add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())) .with_id(cv.get_id())
.add_typed_default(DataType::Enabled, DataValue::Bool(enable)); .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
.add_data(DataTypes::enabled, DataValue::Bool(enable));
if let Some(short_link) = short_link { if let Some(short_link) = short_link {
response_cv = response_cv.add_typed_default(DataType::Link, DataValue::Str(short_link)); response_cv = response_cv.add_data(DataTypes::link, DataValue::Str(short_link));
} }
self.send_message(&response_cv).await; self.send_message(&response_cv).await;
} }
async fn handle_load_txt_record(self: Arc<Self>, cv: CommunicationValue) { async fn handle_load_txt_record(self: Arc<Self>, cv: CommunicationValue) {
if let Some(path) = cv.get_data(DataType::Path).as_str() { if let Some(path) = cv.get_data(DataTypes::path).as_str() {
if let Ok(builder) = hickory_resolver::Resolver::builder_tokio() { if let Ok(builder) = hickory_resolver::Resolver::builder_tokio() {
let resolver = builder.build(); let resolver = builder.build();
if let Ok(lookup) = resolver.txt_lookup(path).await { if let Ok(lookup) = resolver.txt_lookup(path).await {
@ -544,9 +585,12 @@ impl ClientConnection {
for txt_data in record.txt_data() { for txt_data in record.txt_data() {
if let Ok(s) = std::str::from_utf8(txt_data) { if let Ok(s) = std::str::from_utf8(txt_data) {
let response = let response =
CommunicationValue::new(CommunicationType::LoadTxtRecord) CommunicationValue::new(CommunicationType::load_txt_record)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::Content, DataValue::Str(s.to_string())); .add_data(
DataTypes::content,
DataValue::Str(s.to_string()),
);
self.send_message(&response).await; self.send_message(&response).await;
return; return;
} }
@ -555,10 +599,10 @@ impl ClientConnection {
} }
} }
} }
let path_data = cv.get_data(DataType::Path).clone(); let path_data = cv.get_data(DataTypes::path).clone();
let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) let error_cv = CommunicationValue::new(CommunicationType::error_not_found)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::Path, path_data); .add_data(DataTypes::path, path_data);
self.send_message(&error_cv).await; self.send_message(&error_cv).await;
} }
@ -578,11 +622,14 @@ impl ClientConnection {
cv.get_receiver() cv.get_receiver()
); );
if cv.is_type(CommunicationType::AddConversation) if cv.is_type(CommunicationType::add_conversation)
&& cv.get_data(DataType::ChatPartnerId).as_signed_number().is_none() && cv
.get_data(DataTypes::chat_partner_id)
.as_number()
.is_none()
{ {
let chat_partner_name = cv let chat_partner_name = cv
.get_data(DataType::ChatPartnerName) .get_data(DataTypes::chat_partner_name)
.as_str() .as_str()
.unwrap_or("") .unwrap_or("")
.to_string(); .to_string();
@ -591,22 +638,25 @@ impl ClientConnection {
.await .await
.is_some() .is_some()
{ {
self.send_error_response(cv.get_id(), CommunicationType::ErrorAnonymous) self.send_error_response(cv.get_id(), CommunicationType::error_anonymous)
.await; .await;
return; return;
} }
let load_uuid_response = get_omega_connection() let load_uuid_response = get_omega_connection()
.await_response( .await_response(
&CommunicationValue::new(CommunicationType::GetUserData) &CommunicationValue::new(CommunicationType::get_user_data)
.with_id(cv.clone().get_id()) .with_id(cv.clone().get_id())
.add_typed_default(DataType::Username, DataValue::Str(chat_partner_name.clone())), .add_data(
DataTypes::username,
DataValue::Str(chat_partner_name.clone()),
),
Some(Duration::from_secs(20)), Some(Duration::from_secs(20)),
) )
.await; .await;
let chat_partner_id = { let chat_partner_id = {
if let Ok(load_uuid_response) = load_uuid_response { if let Ok(load_uuid_response) = load_uuid_response {
load_uuid_response.get_data(DataType::UserId).clone() load_uuid_response.get_data(DataTypes::user_id).clone()
} else { } else {
DataValue::Null DataValue::Null
} }
@ -625,7 +675,7 @@ impl ClientConnection {
let updated_cv = cv let updated_cv = cv
.with_sender(sender_user_id as u64) .with_sender(sender_user_id as u64)
.add_typed_default(DataType::ChatPartnerId, chat_partner_id); .add_data(DataTypes::chat_partner_id, chat_partner_id);
rho_conn.message_to_iota(updated_cv).await; rho_conn.message_to_iota(updated_cv).await;
} else { } else {
log_err!( log_err!(
@ -663,9 +713,9 @@ impl ClientConnection {
msg_type, msg_type,
msg_id msg_id
); );
let error_cv = CommunicationValue::new(CommunicationType::ErrorNoIota) let error_cv = CommunicationValue::new(CommunicationType::error_no_iota)
.with_id(msg_id) .with_id(msg_id)
.add_typed_default(DataType::UserId, DataValue::SignedNumber(sender_user_id.into())); .add_data(DataTypes::user_id, DataValue::Number(sender_user_id as i64));
self.send_message(&error_cv).await; self.send_message(&error_cv).await;
} }
} }
@ -708,9 +758,9 @@ impl ClientConnection {
} else { } else {
user_status user_status
}; };
let notification = CommunicationValue::new(CommunicationType::ClientChanged) let notification = CommunicationValue::new(CommunicationType::client_changed)
.add_typed_default(DataType::UserId, DataValue::Str(user_id.to_string())) .add_data(DataTypes::user_id, DataValue::Str(user_id.to_string()))
.add_typed_default(DataType::UserState, DataValue::Str(status.to_string())); .add_data(DataTypes::user_state, DataValue::Str(status.to_string()));
self.send_message(&notification).await; self.send_message(&notification).await;
} }
@ -735,7 +785,6 @@ impl Clone for ClientConnection {
receiver: Arc::clone(&self.receiver), receiver: Arc::clone(&self.receiver),
user_id: self.user_id, user_id: self.user_id,
session_id: self.session_id, session_id: self.session_id,
client_version: self.client_version.clone(),
ping: Arc::clone(&self.ping), ping: Arc::clone(&self.ping),
pub_key: Arc::clone(&self.pub_key), pub_key: Arc::clone(&self.pub_key),
rho_connection: Arc::clone(&self.rho_connection), rho_connection: Arc::clone(&self.rho_connection),

202
src/rho/connection.rs
View file

@ -1,8 +1,8 @@
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataTypeId, DataValue};
use mtp::transport::{Receiver, Sender};
use rand::{Rng, distributions::Alphanumeric}; use rand::{Rng, distributions::Alphanumeric};
use std::{collections::BTreeMap, collections::HashMap, sync::Arc, time::Duration}; use std::{collections::BTreeMap, collections::HashMap, sync::Arc, time::Duration};
use tokio::sync::RwLock; use tokio::sync::RwLock;
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use ttp_native::{Receiver, Sender};
use uuid::Uuid; use uuid::Uuid;
use crate::{ use crate::{
@ -45,7 +45,6 @@ pub struct GeneralConnection {
pub session_id: Arc<RwLock<u64>>, pub session_id: Arc<RwLock<u64>>,
pub app_identifier: Arc<RwLock<Option<String>>>, pub app_identifier: Arc<RwLock<Option<String>>>,
pub app_session: Arc<RwLock<Option<Uuid>>>, pub app_session: Arc<RwLock<Option<Uuid>>>,
pub client_version: Arc<RwLock<String>>,
pub_key: Arc<RwLock<Option<Vec<u8>>>>, pub_key: Arc<RwLock<Option<Vec<u8>>>>,
} }
@ -64,7 +63,6 @@ impl GeneralConnection {
session_id: Arc::new(RwLock::new(0)), session_id: Arc::new(RwLock::new(0)),
app_identifier: Arc::new(RwLock::new(None)), app_identifier: Arc::new(RwLock::new(None)),
app_session: Arc::new(RwLock::new(None)), app_session: Arc::new(RwLock::new(None)),
client_version: Arc::new(RwLock::new(mtp::codec::PROTOCOL_VERSION.to_string())),
pub_key: Arc::new(RwLock::new(None)), pub_key: Arc::new(RwLock::new(None)),
}) })
} }
@ -97,31 +95,26 @@ impl GeneralConnection {
log_out!(0, PrintType::General, "General connection handler stopped"); log_out!(0, PrintType::General, "General connection handler stopped");
} }
async fn handle_identification(self: &Arc<Self>, cv: CommunicationValue) { async fn handle_identification(self: &Arc<Self>, cv: CommunicationValue) {
if let Some(version) = cv.get_data(DataType::Version).as_str() { if cv.is_type(CommunicationType::register_iota) {
*self.client_version.write().await = version.to_string(); if let DataValue::Str(pub_key) = cv.get_data(DataTypes::public_key) {
} let msg = CommunicationValue::new(CommunicationType::complete_register_iota)
.add_data(DataTypes::public_key, DataValue::Str(pub_key.clone()));
if cv.is_type(CommunicationType::RegisterIota) {
if let DataValue::Str(pub_key) = cv.get_data(DataType::PublicKey) {
let msg = CommunicationValue::new(CommunicationType::CompleteRegisterIota)
.add_typed_default(DataType::PublicKey, DataValue::Str(pub_key.clone()));
let response = get_omega_connection() let response = get_omega_connection()
.await_response(&msg, Some(Duration::from_secs(20))) .await_response(&msg, Some(Duration::from_secs(20)))
.await; .await;
if let Ok(response_cv) = response { if let Ok(response_cv) = response {
if let DataValue::SignedNumber(iota_id) = response_cv.get_data(DataType::IotaId) if let DataValue::Number(iota_id) = response_cv.get_data(DataTypes::iota_id) {
{ let success_msg = CommunicationValue::new(CommunicationType::success)
let success_msg = CommunicationValue::new(CommunicationType::Success)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(*iota_id)); .add_data(DataTypes::iota_id, DataValue::Number(*iota_id));
log_cv_out!(success_msg); log_cv_out!(success_msg);
let _ = self.sender.send(&success_msg).await; let _ = self.sender.send(&success_msg).await;
} else { } else {
let err_msg = let err_msg =
CommunicationValue::new(CommunicationType::ErrorInvalidOmikronId) CommunicationValue::new(CommunicationType::error_invalid_omikron_id)
.with_id(cv.get_id()); .with_id(cv.get_id());
let _ = self.sender.send(&err_msg).await; let _ = self.sender.send(&err_msg).await;
} }
@ -130,16 +123,19 @@ impl GeneralConnection {
return; return;
} }
if cv.is_type(CommunicationType::AppIdentification) { if cv.is_type(CommunicationType::app_identification) {
let app_identifier = cv let app_identifier = cv
.get_data(DataType::AppIdentifier) .get_data(DataTypes::app_identifier)
.as_str() .as_str()
.unwrap_or("") .unwrap_or("")
.to_string(); .to_string();
let app_session_id_str = cv.get_data(DataType::AppSession).as_str().unwrap_or(""); let app_session_id_str = cv.get_data(DataTypes::app_session).as_str().unwrap_or("");
let app_session_id = Uuid::parse_str(app_session_id_str).unwrap_or(Uuid::new_v4()); let app_session_id = Uuid::parse_str(app_session_id_str).unwrap_or(Uuid::new_v4());
let pub_key_str = cv.get_data(DataType::AppPublicKey).as_str().unwrap_or(""); let pub_key_str = cv
let user_id = cv.get_data(DataType::UserId).as_number().unwrap_or(0); .get_data(DataTypes::app_public_key)
.as_str()
.unwrap_or("");
let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
*self.id.write().await = user_id as u64; *self.id.write().await = user_id as u64;
*self.app_identifier.write().await = Some(app_identifier); *self.app_identifier.write().await = Some(app_identifier);
@ -168,29 +164,29 @@ impl GeneralConnection {
.unwrap() .unwrap()
.export(DataFormat::Base64); .export(DataFormat::Base64);
let response = CommunicationValue::new(CommunicationType::AppChallenge) let response = CommunicationValue::new(CommunicationType::app_challenge)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default( .add_data(
DataType::PublicKey, DataTypes::public_key,
DataValue::Str(public_key_to_base64(&get_public_key())), DataValue::Str(public_key_to_base64(&get_public_key())),
) )
.add_typed_default(DataType::Challenge, DataValue::Str(encrypted_challenge)); .add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge));
log_cv_out!(response); log_cv_out!(response);
let _ = self.sender.send(&response).await; let _ = self.sender.send(&response).await;
return; return;
} }
if !cv.is_type(CommunicationType::Identification) { if !cv.is_type(CommunicationType::identification) {
return; return;
} }
if let DataValue::SignedNumber(iota_id) = cv.get_data(DataType::IotaId) { if let DataValue::Number(iota_id) = cv.get_data(DataTypes::iota_id) {
*self.id.write().await = *iota_id as u64; *self.id.write().await = *iota_id as u64;
*self.connection_kind.write().await = Some(ConnectionKind::Iota); *self.connection_kind.write().await = Some(ConnectionKind::Iota);
let get_pub_key_msg = CommunicationValue::new(CommunicationType::GetIotaData) let get_pub_key_msg = CommunicationValue::new(CommunicationType::get_iota_data)
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(*iota_id)); .add_data(DataTypes::iota_id, DataValue::Number(*iota_id));
let response_cv = get_omega_connection() let response_cv = get_omega_connection()
.await_response(&get_pub_key_msg, Some(Duration::from_secs(20))) .await_response(&get_pub_key_msg, Some(Duration::from_secs(20)))
@ -200,7 +196,7 @@ impl GeneralConnection {
Ok(r) => r, Ok(r) => r,
Err(_) => { Err(_) => {
log_err!( log_err!(
*iota_id as i64, *iota_id,
PrintType::Iota, PrintType::Iota,
"Failed to get Iota data from Omega for iota_id={}", "Failed to get Iota data from Omega for iota_id={}",
iota_id iota_id
@ -210,7 +206,7 @@ impl GeneralConnection {
}; };
let base64_pub = response_cv let base64_pub = response_cv
.get_data(DataType::PublicKey) .get_data(DataTypes::public_key)
.as_str() .as_str()
.unwrap_or(""); .unwrap_or("");
@ -218,7 +214,7 @@ impl GeneralConnection {
Some(pk) => pk, Some(pk) => pk,
None => { None => {
log_err!( log_err!(
*iota_id as i64, *iota_id,
PrintType::Iota, PrintType::Iota,
"Failed to load public key for iota_id={}", "Failed to load public key for iota_id={}",
iota_id iota_id
@ -245,48 +241,48 @@ impl GeneralConnection {
.unwrap() .unwrap()
.export(DataFormat::Base64); .export(DataFormat::Base64);
let response = CommunicationValue::new(CommunicationType::Challenge) let response = CommunicationValue::new(CommunicationType::challenge)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default( .add_data(
DataType::PublicKey, DataTypes::public_key,
DataValue::Str(public_key_to_base64(&get_public_key())), DataValue::Str(public_key_to_base64(&get_public_key())),
) )
.add_typed_default(DataType::Challenge, DataValue::Str(encrypted_challenge)); .add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge));
log_cv_out!(response); log_cv_out!(response);
let _ = self.sender.send(&response).await; let _ = self.sender.send(&response).await;
} else if let DataValue::SignedNumber(user_id) = cv.get_data(DataType::UserId) { } else if let DataValue::Number(user_id) = cv.get_data(DataTypes::user_id) {
*self.id.write().await = *user_id as u64; *self.id.write().await = *user_id as u64;
*self.session_id.write().await = match cv.get_data(DataType::SessionId) { *self.session_id.write().await = match cv.get_data(DataTypes::session_id) {
DataValue::SignedNumber(s) => *s as u64, DataValue::Number(s) => *s as u64,
_ => cv.get_sender(), _ => cv.get_sender(),
}; };
#[allow(unused_assignments)] #[allow(unused_assignments)]
let mut base64_pub = String::new(); let mut base64_pub = String::new();
if cv.is_type(CommunicationType::AppIdentification) { if cv.is_type(CommunicationType::app_identification) {
*self.connection_kind.write().await = Some(ConnectionKind::Phi); *self.connection_kind.write().await = Some(ConnectionKind::Phi);
if let DataValue::Str(app_id) = cv.get_data(DataType::AppIdentifier) { if let DataValue::Str(app_id) = cv.get_data(DataTypes::app_identifier) {
*self.app_identifier.write().await = Some(app_id.clone()); *self.app_identifier.write().await = Some(app_id.clone());
} }
if let DataValue::Str(app_sess) = cv.get_data(DataType::AppSession) { if let DataValue::Str(app_sess) = cv.get_data(DataTypes::app_session) {
if let Ok(uuid) = uuid::Uuid::parse_str(&app_sess) { if let Ok(uuid) = uuid::Uuid::parse_str(&app_sess) {
*self.app_session.write().await = Some(uuid); *self.app_session.write().await = Some(uuid);
} }
} }
base64_pub = cv base64_pub = cv
.get_data(DataType::AppPublicKey) .get_data(DataTypes::app_public_key)
.as_str() .as_str()
.unwrap_or("") .unwrap_or("")
.to_string(); .to_string();
} else { } else {
*self.connection_kind.write().await = Some(ConnectionKind::Client); *self.connection_kind.write().await = Some(ConnectionKind::Client);
let get_pub_key_msg = CommunicationValue::new(CommunicationType::GetUserData) let get_pub_key_msg = CommunicationValue::new(CommunicationType::get_user_data)
.add_typed_default(DataType::UserId, DataValue::SignedNumber(*user_id)); .add_data(DataTypes::user_id, DataValue::Number(*user_id));
let response_cv = get_omega_connection() let response_cv = get_omega_connection()
.await_response(&get_pub_key_msg, Some(Duration::from_secs(20))) .await_response(&get_pub_key_msg, Some(Duration::from_secs(20)))
@ -300,7 +296,7 @@ impl GeneralConnection {
}; };
base64_pub = response_cv base64_pub = response_cv
.get_data(DataType::PublicKey) .get_data(DataTypes::public_key)
.as_str() .as_str()
.unwrap_or("") .unwrap_or("")
.to_string(); .to_string();
@ -331,20 +327,20 @@ impl GeneralConnection {
.unwrap() .unwrap()
.export(DataFormat::Base64); .export(DataFormat::Base64);
let challenge_type = if cv.is_type(CommunicationType::AppIdentification) { let challenge_type = if cv.is_type(CommunicationType::app_identification) {
CommunicationType::AppChallenge CommunicationType::app_challenge
} else { } else {
CommunicationType::Challenge CommunicationType::challenge
}; };
let response = CommunicationValue::new(challenge_type) let response = CommunicationValue::new(challenge_type)
.with_id(cv.get_id()) .with_id(cv.get_id())
.with_receiver(*self.session_id.read().await) .with_receiver(*self.session_id.read().await)
.add_typed_default( .add_data(
DataType::PublicKey, DataTypes::public_key,
DataValue::Str(public_key_to_base64(&get_public_key())), DataValue::Str(public_key_to_base64(&get_public_key())),
) )
.add_typed_default(DataType::Challenge, DataValue::Str(encrypted_challenge)); .add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge));
log_cv_out!(response); log_cv_out!(response);
let _ = self.sender.send(&response).await; let _ = self.sender.send(&response).await;
@ -353,13 +349,13 @@ impl GeneralConnection {
async fn handle_challenge_response(self: &Arc<Self>, cv: CommunicationValue) { async fn handle_challenge_response(self: &Arc<Self>, cv: CommunicationValue) {
let id = *self.id.read().await as i64; let id = *self.id.read().await as i64;
if !cv.is_type(CommunicationType::ChallengeResponse) if !cv.is_type(CommunicationType::challenge_response)
&& !cv.is_type(CommunicationType::AppChallengeResponse) && !cv.is_type(CommunicationType::app_challenge_response)
{ {
return; return;
} }
if let DataValue::Str(response) = cv.get_data(DataType::Challenge) { if let DataValue::Str(response) = cv.get_data(DataTypes::challenge) {
let expected = self.challenge.read().await.clone(); let expected = self.challenge.read().await.clone();
if *response == expected { if *response == expected {
@ -373,10 +369,10 @@ impl GeneralConnection {
expected, expected,
response response
); );
let err_msg = CommunicationValue::new(CommunicationType::ErrorInvalidChallenge) let err_msg = CommunicationValue::new(CommunicationType::error_invalid_challenge)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default( .add_data(
DataType::Message, DataTypes::message,
DataValue::Str( DataValue::Str(
"Challenge response mismatch; authentication rejected".to_string(), "Challenge response mismatch; authentication rejected".to_string(),
), ),
@ -403,8 +399,8 @@ impl GeneralConnection {
match kind { match kind {
ConnectionKind::Client => { ConnectionKind::Client => {
let notify = CommunicationValue::new(CommunicationType::UserConnected) let notify = CommunicationValue::new(CommunicationType::user_connected)
.add_typed_default(DataType::UserId, DataValue::SignedNumber((id as i64).into())); .add_data(DataTypes::user_id, DataValue::Number(id as i64));
get_omega_connection().send_message(&notify).await; get_omega_connection().send_message(&notify).await;
let user_id = id as i64; let user_id = id as i64;
@ -414,18 +410,18 @@ impl GeneralConnection {
let mut rho = rho_manager::get_rho_con_for_user(user_id).await; let mut rho = rho_manager::get_rho_con_for_user(user_id).await;
if rho.is_none() { if rho.is_none() {
let get_user_msg = CommunicationValue::new(CommunicationType::GetUserData) let get_user_msg = CommunicationValue::new(CommunicationType::get_user_data)
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())); .add_data(DataTypes::user_id, DataValue::Number(user_id));
if let Ok(user_data_cv) = get_omega_connection() if let Ok(user_data_cv) = get_omega_connection()
.await_response(&get_user_msg, Some(Duration::from_secs(20))) .await_response(&get_user_msg, Some(Duration::from_secs(20)))
.await .await
{ {
if let DataValue::SignedNumber(iota_id) = if let DataValue::Number(iota_id) =
user_data_cv.get_data(DataType::IotaId) user_data_cv.get_data(DataTypes::iota_id)
{ {
if let Some(bound_rho) = if let Some(bound_rho) =
rho_manager::bind_user_to_iota(user_id, *iota_id as i64).await rho_manager::bind_user_to_iota(user_id, *iota_id).await
{ {
bound_rho.bind_user_id(user_id).await; bound_rho.bind_user_id(user_id).await;
rho = Some(bound_rho); rho = Some(bound_rho);
@ -438,9 +434,9 @@ impl GeneralConnection {
if let Some(rho_conn) = rho { if let Some(rho_conn) = rho {
let session_id = *self.session_id.read().await as i64; let session_id = *self.session_id.read().await as i64;
let iota_msg = CommunicationValue::new(CommunicationType::ClientConnected) let iota_msg = CommunicationValue::new(CommunicationType::client_connected)
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) .add_data(DataTypes::user_id, DataValue::Number(user_id))
.add_typed_default(DataType::SessionId, DataValue::SignedNumber(session_id.into())); .add_data(DataTypes::session_id, DataValue::Number(session_id));
if let Ok(resp) = rho_conn if let Ok(resp) = rho_conn
.get_iota_connection() .get_iota_connection()
@ -449,11 +445,10 @@ impl GeneralConnection {
.await .await
{ {
let mut ident_resp = let mut ident_resp =
CommunicationValue::new(CommunicationType::IdentificationResponse) CommunicationValue::new(CommunicationType::identification_response)
.with_id(*self.challenge_cv_id.read().await); .with_id(*self.challenge_cv_id.read().await);
let tm = mtp::codec::TypeMap::latest(); for (k, v) in resp.get_data_container() {
for (k, v) in resp.iter_typed_data() { let value_to_add = if k == &DataTypes::contacts {
let value_to_add = if k == Some(DataType::Contacts) {
if let Some(contacts) = v.as_array() { if let Some(contacts) = v.as_array() {
let call_groups = let call_groups =
call_manager::get_call_groups(user_id as u64).await; call_manager::get_call_groups(user_id as u64).await;
@ -473,30 +468,30 @@ impl GeneralConnection {
// List of all members in the call // List of all members in the call
let member_ids: Vec<DataValue> = members let member_ids: Vec<DataValue> = members
.iter() .iter()
.map(|m| DataValue::SignedNumber(m.user_id.into())) .map(|m| DataValue::Number(m.user_id as i64))
.collect(); .collect();
let mut base_call_map: BTreeMap<DataTypeId, DataValue> = let mut base_call_map: BTreeMap<DataTypes, DataValue> =
BTreeMap::new(); BTreeMap::new();
base_call_map.insert( base_call_map.insert(
DataType::CallId.to_id(&tm), DataTypes::call_id,
DataValue::Str(call.call_id.to_string()), DataValue::Str(call.call_id.to_string()),
); );
base_call_map.insert( base_call_map.insert(
DataType::CallMembers.to_id(&tm), DataTypes::call_members,
DataValue::Array(member_ids), DataValue::Array(member_ids),
); );
if timeout > 0 { if timeout > 0 {
base_call_map.insert( base_call_map.insert(
DataType::Timeout.to_id(&tm), DataTypes::timeout,
DataValue::SignedNumber((timeout as i64).into()), DataValue::Number(timeout as i64),
); );
} }
if admin { if admin {
base_call_map.insert( base_call_map.insert(
DataType::HasAdmin.to_id(&tm), DataTypes::has_admin,
DataValue::Bool(true), DataValue::Bool(true),
); );
} }
@ -521,7 +516,7 @@ impl GeneralConnection {
.get(&(member_id, user_id as u64)) .get(&(member_id, user_id as u64))
{ {
contact_call_map.insert( contact_call_map.insert(
DataType::CallSecret.to_id(&tm), DataTypes::call_secret,
DataValue::Str(secret.clone()), DataValue::Str(secret.clone()),
); );
} }
@ -539,12 +534,12 @@ impl GeneralConnection {
let mut new_contacts: Vec<DataValue> = Vec::new(); let mut new_contacts: Vec<DataValue> = Vec::new();
for contact in contacts { for contact in contacts {
if let Some(mut contact_map) = contact.as_map() { if let Some(mut contact_map) = contact.as_map() {
if let Some(DataValue::SignedNumber(id)) = if let Some(DataValue::Number(id)) =
contact_map.get(&DataType::UserId.to_id(&tm)) contact_map.get(&DataTypes::user_id)
{ {
if let Some(call_list) = invites.get(&(*id as i64)) { if let Some(call_list) = invites.get(id) {
contact_map.insert( contact_map.insert(
DataType::Calls.to_id(&tm), DataTypes::calls,
DataValue::Array(call_list.clone()), DataValue::Array(call_list.clone()),
); );
} }
@ -557,7 +552,7 @@ impl GeneralConnection {
} }
ident_resp = ident_resp ident_resp = ident_resp
.add_typed_default(DataType::Calls, DataValue::Array(global_calls)); .add_data(DataTypes::calls, DataValue::Array(global_calls));
DataValue::Array(new_contacts) DataValue::Array(new_contacts)
} else { } else {
v.clone() v.clone()
@ -566,9 +561,7 @@ impl GeneralConnection {
v.clone() v.clone()
}; };
if let Some(dt) = k { ident_resp = ident_resp.add_data(k.clone(), value_to_add);
ident_resp = ident_resp.add_typed_default(dt, value_to_add);
}
} }
log_cv_out!(ident_resp); log_cv_out!(ident_resp);
let _ = self.sender.send(&ident_resp).await; let _ = self.sender.send(&ident_resp).await;
@ -587,8 +580,8 @@ impl GeneralConnection {
client.start(); client.start();
} }
ConnectionKind::Iota => { ConnectionKind::Iota => {
let notify = CommunicationValue::new(CommunicationType::IotaConnected) let notify = CommunicationValue::new(CommunicationType::iota_connected)
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(id.into())); .add_data(DataTypes::iota_id, DataValue::Number(id as i64));
get_omega_connection().send_message(&notify).await; get_omega_connection().send_message(&notify).await;
let iota = IotaConnection::from_general(self.clone(), id).await; let iota = IotaConnection::from_general(self.clone(), id).await;
@ -599,17 +592,17 @@ impl GeneralConnection {
rho_manager::add_rho(rho).await; rho_manager::add_rho(rho).await;
let get_iota_msg = CommunicationValue::new(CommunicationType::GetIotaData) let get_iota_msg = CommunicationValue::new(CommunicationType::get_iota_data)
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(id.into())); .add_data(DataTypes::iota_id, DataValue::Number(id as i64));
if let Ok(iota_data_cv) = get_omega_connection() if let Ok(iota_data_cv) = get_omega_connection()
.await_response(&get_iota_msg, Some(Duration::from_secs(20))) .await_response(&get_iota_msg, Some(Duration::from_secs(20)))
.await .await
{ {
if let DataValue::Array(users) = iota_data_cv.get_data(DataType::UserIds) { if let DataValue::Array(users) = iota_data_cv.get_data(DataTypes::user_ids) {
let mut user_ids: Vec<u64> = Vec::new(); let mut user_ids: Vec<u64> = Vec::new();
for value in users { for value in users {
if let DataValue::SignedNumber(user_id) = value { if let DataValue::Number(user_id) = value {
user_ids.push(*user_id as u64); user_ids.push(*user_id as u64);
} }
} }
@ -617,9 +610,10 @@ impl GeneralConnection {
} }
} }
let ident_resp = CommunicationValue::new(CommunicationType::IdentificationResponse) let ident_resp =
.with_id(*self.challenge_cv_id.read().await) CommunicationValue::new(CommunicationType::identification_response)
.add_typed_default(DataType::Accepted, DataValue::Bool(true)); .with_id(*self.challenge_cv_id.read().await)
.add_data(DataTypes::accepted, DataValue::Bool(true));
log_cv_out!(ident_resp); log_cv_out!(ident_resp);
let _ = self.sender.send(&ident_resp).await; let _ = self.sender.send(&ident_resp).await;
@ -635,18 +629,18 @@ impl GeneralConnection {
let mut rho = rho_manager::get_rho_con_for_user(user_id).await; let mut rho = rho_manager::get_rho_con_for_user(user_id).await;
if rho.is_none() { if rho.is_none() {
let get_user_msg = CommunicationValue::new(CommunicationType::GetUserData) let get_user_msg = CommunicationValue::new(CommunicationType::get_user_data)
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())); .add_data(DataTypes::user_id, DataValue::Number(user_id));
if let Ok(user_data_cv) = get_omega_connection() if let Ok(user_data_cv) = get_omega_connection()
.await_response(&get_user_msg, Some(Duration::from_secs(20))) .await_response(&get_user_msg, Some(Duration::from_secs(20)))
.await .await
{ {
if let DataValue::SignedNumber(iota_id) = if let DataValue::Number(iota_id) =
user_data_cv.get_data(DataType::IotaId) user_data_cv.get_data(DataTypes::iota_id)
{ {
if let Some(bound_rho) = if let Some(bound_rho) =
rho_manager::bind_user_to_iota(user_id, *iota_id as i64).await rho_manager::bind_user_to_iota(user_id, *iota_id).await
{ {
bound_rho.bind_user_id(user_id).await; bound_rho.bind_user_id(user_id).await;
rho = Some(bound_rho); rho = Some(bound_rho);
@ -665,7 +659,7 @@ impl GeneralConnection {
} }
let response = let response =
CommunicationValue::new(CommunicationType::AppIdentificationResponse) CommunicationValue::new(CommunicationType::app_identification_response)
.with_id(*self.challenge_cv_id.read().await) .with_id(*self.challenge_cv_id.read().await)
.with_receiver(*self.session_id.read().await); .with_receiver(*self.session_id.read().await);

102
src/rho/iota_connection.rs
View file

@ -8,18 +8,16 @@ use crate::omega::omega_connection::get_omega_connection;
use crate::rho::connection::GeneralConnection; use crate::rho::connection::GeneralConnection;
use crate::util::logger::PrintType; use crate::util::logger::PrintType;
use dashmap::DashMap; use dashmap::DashMap;
use mtp::codec::CommunicationType;
use mtp::codec::CommunicationValue;
use mtp::codec::DataType;
use mtp::codec::DataTypeId;
use mtp::codec::DataValue;
use mtp::codec::TypeMap;
use mtp::transport::Receiver;
use mtp::transport::Sender;
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::{collections::HashMap, sync::Arc, time::Duration}; use std::{collections::HashMap, sync::Arc, time::Duration};
use tokio::sync::RwLock; use tokio::sync::RwLock;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use ttp_core::CommunicationType;
use ttp_core::CommunicationValue;
use ttp_core::DataTypes;
use ttp_core::DataValue;
use ttp_native::Receiver;
use ttp_native::Sender;
use x448::PublicKey; use x448::PublicKey;
use super::{rho_connection::RhoConnection, rho_manager}; use super::{rho_connection::RhoConnection, rho_manager};
@ -28,7 +26,6 @@ use crate::omega::omega_connection::OmegaConnection;
#[allow(dead_code)] #[allow(dead_code)]
pub struct IotaConnection { pub struct IotaConnection {
pub iota_id: u64, pub iota_id: u64,
pub client_version: String,
pub sender: Arc<Sender>, pub sender: Arc<Sender>,
pub receiver: Arc<Receiver>, pub receiver: Arc<Receiver>,
pub user_ids: Arc<RwLock<Vec<u64>>>, pub user_ids: Arc<RwLock<Vec<u64>>>,
@ -49,7 +46,6 @@ impl IotaConnection {
sender: general.sender.clone(), sender: general.sender.clone(),
receiver: general.receiver.clone(), receiver: general.receiver.clone(),
iota_id: iota_id, iota_id: iota_id,
client_version: general.client_version.read().await.clone(),
waiting_tasks: DashMap::new(), waiting_tasks: DashMap::new(),
}) })
} }
@ -145,7 +141,7 @@ impl IotaConnection {
/// Send a CommunicationValue to the Iota /// Send a CommunicationValue to the Iota
pub async fn send_message(&self, cv: &CommunicationValue) { pub async fn send_message(&self, cv: &CommunicationValue) {
if !cv.is_type(CommunicationType::Pong) { if !cv.is_type(CommunicationType::pong) {
log_cv_out!(PrintType::Iota, cv); log_cv_out!(PrintType::Iota, cv);
} }
if let Err(e) = self.sender.send(&cv).await { if let Err(e) = self.sender.send(&cv).await {
@ -168,7 +164,7 @@ impl IotaConnection {
} }
// Handle ping // Handle ping
if cv.is_type(CommunicationType::Ping) || cv.is_type(CommunicationType::Pong) { if cv.is_type(CommunicationType::ping) || cv.is_type(CommunicationType::pong) {
self.handle_ping(cv).await; self.handle_ping(cv).await;
return; return;
} }
@ -176,7 +172,7 @@ impl IotaConnection {
log_cv_in!(PrintType::Iota, cv); log_cv_in!(PrintType::Iota, cv);
// Handle GET_CHATS // Handle GET_CHATS
if cv.is_type(CommunicationType::GetChats) { if cv.is_type(CommunicationType::get_chats) {
self.handle_get_chats(cv).await; self.handle_get_chats(cv).await;
return; return;
} }
@ -184,14 +180,14 @@ impl IotaConnection {
// Handle forwarding to other Iotas or clients // Handle forwarding to other Iotas or clients
let receiver_id = cv.get_receiver(); let receiver_id = cv.get_receiver();
if (receiver_id != 0 && !self.get_user_ids().await.contains(&(receiver_id as u64))) if (receiver_id != 0 && !self.get_user_ids().await.contains(&(receiver_id as u64)))
|| cv.is_type(CommunicationType::MessageOtherIota) || cv.is_type(CommunicationType::message_other_iota)
|| cv.is_type(CommunicationType::SendChat) || cv.is_type(CommunicationType::send_chat)
{ {
self.handle_forward_message(cv).await; self.handle_forward_message(cv).await;
return; return;
} }
if cv.is_type(CommunicationType::CompleteRegisterUser) { if cv.is_type(CommunicationType::complete_register_user) {
let response_cv = get_omega_connection() let response_cv = get_omega_connection()
.await_response( .await_response(
&cv.clone().with_sender(self.iota_id), &cv.clone().with_sender(self.iota_id),
@ -199,8 +195,8 @@ impl IotaConnection {
) )
.await; .await;
if let Ok(response_cv) = response_cv { if let Ok(response_cv) = response_cv {
if response_cv.is_type(CommunicationType::Success) { if response_cv.is_type(CommunicationType::success) {
if let Some(user_id) = cv.get_data(DataType::UserId).as_number() { if let Some(user_id) = cv.get_data(DataTypes::user_id).as_number() {
self.add_user_id(user_id as u64).await; self.add_user_id(user_id as u64).await;
} }
} }
@ -209,12 +205,12 @@ impl IotaConnection {
return; return;
} }
if cv.is_type(CommunicationType::ChangeIotaData) if cv.is_type(CommunicationType::change_iota_data)
|| cv.is_type(CommunicationType::PushNotification) || cv.is_type(CommunicationType::push_notification)
|| cv.is_type(CommunicationType::GetUserData) || cv.is_type(CommunicationType::get_user_data)
|| cv.is_type(CommunicationType::GetIotaData) || cv.is_type(CommunicationType::get_iota_data)
|| cv.is_type(CommunicationType::GetRegister) || cv.is_type(CommunicationType::get_register)
|| cv.is_type(CommunicationType::DeleteIota) || cv.is_type(CommunicationType::delete_iota)
{ {
let sender = self.get_iota_id().await; let sender = self.get_iota_id().await;
@ -247,7 +243,7 @@ impl IotaConnection {
} }
/// Handle ping message /// Handle ping message
async fn handle_ping(&self, cv: CommunicationValue) { async fn handle_ping(&self, cv: CommunicationValue) {
if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) { if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) {
if let Ok(ping_val) = last_ping.to_string().parse::<i64>() { if let Ok(ping_val) = last_ping.to_string().parse::<i64>() {
let mut ping_guard = self.ping.write().await; let mut ping_guard = self.ping.write().await;
*ping_guard = ping_val; *ping_guard = ping_val;
@ -260,21 +256,13 @@ impl IotaConnection {
HashMap::new() HashMap::new()
}; };
let tm = TypeMap::latest(); let pings: Vec<(DataTypes, DataValue)> = client_pings
let pings: Vec<DataValue> = client_pings
.into_iter() .into_iter()
.map(|(k, v)| { .map(|(k, v)| (DataTypes::parse(k), DataValue::Number(v)))
let mut map = BTreeMap::new();
if let Ok(uid) = k.parse::<i128>() {
map.insert(DataType::UserId.to_id(&tm), DataValue::SignedNumber(uid));
}
map.insert(DataType::LastPing.to_id(&tm), DataValue::SignedNumber(v.into()));
DataValue::container_from_map(&map)
})
.collect(); .collect();
let response = CommunicationValue::new(CommunicationType::Pong) let response = CommunicationValue::new(CommunicationType::pong)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_typed_default(DataType::PingClients, DataValue::Array(pings)); .add_data(DataTypes::ping_clients, DataValue::Container(pings));
self.send_message(&response).await; self.send_message(&response).await;
} }
@ -300,7 +288,7 @@ impl IotaConnection {
if let Some(target_rho) = rho_manager::get_rho_con_for_user(receiver_id as i64).await { if let Some(target_rho) = rho_manager::get_rho_con_for_user(receiver_id as i64).await {
target_rho.message_to_iota(cv).await; target_rho.message_to_iota(cv).await;
} else { } else {
let error = CommunicationValue::new(CommunicationType::ErrorNoIota) let error = CommunicationValue::new(CommunicationType::error_no_iota)
.with_id(cv.get_id()) .with_id(cv.get_id())
.with_sender(cv.get_sender()); .with_sender(cv.get_sender());
self.send_message(&error).await; self.send_message(&error).await;
@ -315,8 +303,8 @@ impl IotaConnection {
); );
self.send_message( self.send_message(
&CommunicationValue::new(CommunicationType::ErrorInvalidUserId).add_typed_default( &CommunicationValue::new(CommunicationType::error_invalid_user_id).add_data(
DataType::ErrorType, DataTypes::error_type,
DataValue::Str( DataValue::Str(
"You are sending to another User without authority.".to_string(), "You are sending to another User without authority.".to_string(),
), ),
@ -342,7 +330,6 @@ impl IotaConnection {
} }
let mut interested_ids: Vec<i64> = Vec::new(); let mut interested_ids: Vec<i64> = Vec::new();
let tm = TypeMap::latest();
// ============================ // ============================
// Load Calls // Load Calls
@ -366,21 +353,20 @@ impl IotaConnection {
// List of all members in the call // List of all members in the call
let member_ids: Vec<DataValue> = members let member_ids: Vec<DataValue> = members
.iter() .iter()
.map(|m| DataValue::SignedNumber(m.user_id.into())) .map(|m| DataValue::Number(m.user_id as i64))
.collect(); .collect();
// Build base call container // Build base call container
let mut base_call_map: BTreeMap<DataTypeId, DataValue> = BTreeMap::new(); let mut base_call_map: BTreeMap<DataTypes, DataValue> = BTreeMap::new();
base_call_map.insert(DataType::CallId.to_id(&tm), DataValue::Str(call.call_id.to_string())); base_call_map.insert(DataTypes::call_id, DataValue::Str(call.call_id.to_string()));
base_call_map.insert(DataType::CallMembers.to_id(&tm), DataValue::Array(member_ids)); base_call_map.insert(DataTypes::call_members, DataValue::Array(member_ids));
if timeout > 0 { if timeout > 0 {
base_call_map base_call_map.insert(DataTypes::timeout, DataValue::Number(timeout as i64));
.insert(DataType::Timeout.to_id(&tm), DataValue::SignedNumber(timeout.into()));
} }
if admin { if admin {
base_call_map.insert(DataType::HasAdmin.to_id(&tm), DataValue::Bool(true)); base_call_map.insert(DataTypes::has_admin, DataValue::Bool(true));
} }
// Add to global calls (without contact-specific secret) // Add to global calls (without contact-specific secret)
@ -398,7 +384,7 @@ impl IotaConnection {
// Add secret if it exists for this pairing // Add secret if it exists for this pairing
if let Some(secret) = call.secrets.read().await.get(&(member_id, user_id)) { if let Some(secret) = call.secrets.read().await.get(&(member_id, user_id)) {
contact_call_map contact_call_map
.insert(DataType::CallSecret.to_id(&tm), DataValue::Str(secret.clone())); .insert(DataTypes::call_secret, DataValue::Str(secret.clone()));
} }
invites invites
@ -413,27 +399,27 @@ impl IotaConnection {
// Enrich Contacts // Enrich Contacts
// ============================ // ============================
let enriched_contacts = if empty { let enriched_contacts = if empty {
match cv.get_data(DataType::UserIds) { match cv.get_data(DataTypes::user_ids) {
DataValue::Array(arr) => DataValue::Array(arr.clone()), DataValue::Array(arr) => DataValue::Array(arr.clone()),
_ => DataValue::Array(vec![]), _ => DataValue::Array(vec![]),
} }
} else { } else {
let mut enriched: Vec<DataValue> = Vec::new(); let mut enriched: Vec<DataValue> = Vec::new();
if let DataValue::Array(users) = cv.get_data(DataType::UserIds) { if let DataValue::Array(users) = cv.get_data(DataTypes::user_ids) {
for user_val in users { for user_val in users {
if let DataValue::Container(entries) = user_val { if let DataValue::Container(entries) = user_val {
let mut user_map: BTreeMap<DataTypeId, DataValue> = let mut user_map: BTreeMap<DataTypes, DataValue> =
entries.iter().cloned().collect(); entries.iter().cloned().collect();
if let Some(DataValue::SignedNumber(id)) = user_map.get(&DataType::UserId.to_id(&tm)) { if let Some(DataValue::Number(id)) = user_map.get(&DataTypes::user_id) {
interested_ids.push(*id as i64); interested_ids.push(*id);
if let Some(call_list) = invites.get(&(*id as i64)) if let Some(call_list) = invites.get(id)
&& !call_list.is_empty() && !call_list.is_empty()
{ {
user_map user_map
.insert(DataType::Calls.to_id(&tm), DataValue::Array(call_list.clone())); .insert(DataTypes::calls, DataValue::Array(call_list.clone()));
} }
} }
@ -463,8 +449,8 @@ impl IotaConnection {
// Forward to client // Forward to client
// ============================ // ============================
self.forward_to_client( self.forward_to_client(
cv.add_typed_default(DataType::UserIds, enriched_contacts) cv.add_data(DataTypes::user_ids, enriched_contacts)
.add_typed_default(DataType::Calls, DataValue::Array(global_calls)), .add_data(DataTypes::calls, DataValue::Array(global_calls)),
) )
.await; .await;
} }

10
src/rho/rho_connection.rs
View file

@ -2,10 +2,10 @@ use super::{client_connection::ClientConnection, iota_connection::IotaConnection
use crate::omega::omega_connection::OmegaConnection; use crate::omega::omega_connection::OmegaConnection;
use crate::{data::user::UserStatus, rho::app_connection::AppConnection}; use crate::{data::user::UserStatus, rho::app_connection::AppConnection};
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::Arc; use std::sync::Arc;
use tokio::sync::RwLock; use tokio::sync::RwLock;
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use uuid::Uuid; use uuid::Uuid;
pub struct RhoConnection { pub struct RhoConnection {
@ -125,8 +125,8 @@ impl RhoConnection {
let user_id = connection.user_id as i64; let user_id = connection.user_id as i64;
let session_id = connection.session_id as i64; let session_id = connection.session_id as i64;
let notification = CommunicationValue::new(CommunicationType::ClientConnected) let notification = CommunicationValue::new(CommunicationType::client_connected)
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())); .add_data(DataTypes::user_id, DataValue::Number(user_id));
self.iota_connection.send_message(&notification).await; self.iota_connection.send_message(&notification).await;
@ -216,14 +216,14 @@ impl RhoConnection {
pub async fn message_to_client(&self, cv: CommunicationValue) { pub async fn message_to_client(&self, cv: CommunicationValue) {
let connections = self.client_connections.read().await; let connections = self.client_connections.read().await;
let receiver_id = cv.get_receiver(); let receiver_id = cv.get_receiver();
let session_id = cv.get_data(DataType::SessionId).as_number(); let session_id = cv.get_data(DataTypes::session_id).as_number();
for connection in connections.iter() { for connection in connections.iter() {
if connection.user_id != receiver_id { if connection.user_id != receiver_id {
continue; continue;
} }
if let Some(session_id) = session_id { if let Some(session_id) = session_id {
if connection.session_id as i128 != session_id { if connection.session_id as i64 != session_id {
continue; continue;
} }
} }

4
src/rho/server.rs
View file

@ -1,4 +1,3 @@
use std::net::{IpAddr, Ipv4Addr};
use std::time::Duration; use std::time::Duration;
use crate::{ use crate::{
@ -6,7 +5,7 @@ use crate::{
rho::connection::GeneralConnection, rho::connection::GeneralConnection,
util::{file_util::load_file_vec, logger::PrintType}, util::{file_util::load_file_vec, logger::PrintType},
}; };
use mtp::transport::{Host, Policy, SendMode, host}; use ttp_native::{Host, Policy, SendMode, host};
pub async fn start(port: u16) -> Result<(), Box<dyn std::error::Error>> { pub async fn start(port: u16) -> Result<(), Box<dyn std::error::Error>> {
let cert_pem = load_file_vec("certs", "cert.pem").expect("Error loading Pemfile"); let cert_pem = load_file_vec("certs", "cert.pem").expect("Error loading Pemfile");
@ -14,7 +13,6 @@ pub async fn start(port: u16) -> Result<(), Box<dyn std::error::Error>> {
let key_pem = load_file_vec("certs", "key.pem").expect("Error loading Keyfile"); let key_pem = load_file_vec("certs", "key.pem").expect("Error loading Keyfile");
let mut host: Host = host( let mut host: Host = host(
IpAddr::V4(Ipv4Addr::UNSPECIFIED),
port, port,
cert_pem, cert_pem,
key_pem, key_pem,

27
src/util/logger.rs
View file

@ -1,4 +1,5 @@
use std::{ use std::{
collections::BTreeMap,
fs::{self, OpenOptions}, fs::{self, OpenOptions},
io::Write, io::Write,
sync::{OnceLock, mpsc}, sync::{OnceLock, mpsc},
@ -7,7 +8,7 @@ use std::{
}; };
use ansi_term::Color; use ansi_term::Color;
use mtp::codec::{CommunicationValue, DataTypeId, DataValue, Version}; use ttp_core::{CommunicationValue, DataTypes, DataValue};
static LOGGER: OnceLock<mpsc::Sender<LogMessage>> = OnceLock::new(); static LOGGER: OnceLock<mpsc::Sender<LogMessage>> = OnceLock::new();
@ -176,19 +177,17 @@ pub fn format_cv(cv: &CommunicationValue) -> String {
let comm_type = cv.get_type().to_string(); let comm_type = cv.get_type().to_string();
parts.push(format!("{}", comm_type)); parts.push(format!("{}", comm_type));
let data = cv.data(); let data: &BTreeMap<DataTypes, DataValue> = cv.get_data_container();
let formated_data = format_data_container( let formated_data =
data.iter().map(|(k, v)| (*k, v.clone())).collect(), format_data_container(data.iter().map(|(k, v)| (k.clone(), v.clone())).collect());
Version(1, 0),
);
parts.push(format!("{}", formated_data)); parts.push(format!("{}", formated_data));
parts.join(": ") parts.join(": ")
} }
fn format_data_container(data: Vec<(DataTypeId, DataValue)>, version: Version) -> String { fn format_data_container(data: Vec<(DataTypes, DataValue)>) -> String {
let parts: Vec<String> = data let parts: Vec<String> = data
.into_iter() .into_iter()
.map(|(key, value)| { .map(|(key, value)| {
@ -198,12 +197,12 @@ fn format_data_container(data: Vec<(DataTypeId, DataValue)>, version: Version) -
DataValue::Str(s) => format!("{}=\"{}\"", key_str, s), DataValue::Str(s) => format!("{}=\"{}\"", key_str, s),
DataValue::Container(inner) => { DataValue::Container(inner) => {
let inner_formatted = format_data_container(inner, version.clone()); let inner_formatted = format_data_container(inner);
format!("{}={{ {} }}", key_str, inner_formatted) format!("{}={{ {} }}", key_str, inner_formatted)
} }
DataValue::Array(arr) => { DataValue::Array(arr) => {
let arr_formatted = format_array(arr, version.clone()); let arr_formatted = format_array(arr);
format!("{}=[{}]", key_str, arr_formatted) format!("{}=[{}]", key_str, arr_formatted)
} }
@ -212,7 +211,7 @@ fn format_data_container(data: Vec<(DataTypeId, DataValue)>, version: Version) -
DataValue::BoolTrue => format!("{}=true", key_str), DataValue::BoolTrue => format!("{}=true", key_str),
DataValue::BoolFalse => format!("{}=false", key_str), DataValue::BoolFalse => format!("{}=false", key_str),
DataValue::SignedNumber(num) => format!("{}={}", key_str, num), DataValue::Number(num) => format!("{}={}", key_str, num),
_ => "".to_string(), _ => "".to_string(),
} }
@ -222,19 +221,19 @@ fn format_data_container(data: Vec<(DataTypeId, DataValue)>, version: Version) -
parts.join(", ") parts.join(", ")
} }
fn format_array(arr: Vec<DataValue>, version: Version) -> String { fn format_array(arr: Vec<DataValue>) -> String {
let parts: Vec<String> = arr let parts: Vec<String> = arr
.into_iter() .into_iter()
.map(|value| match value { .map(|value| match value {
DataValue::Str(s) => format!("\"{}\"", s), DataValue::Str(s) => format!("\"{}\"", s),
DataValue::Container(inner) => { DataValue::Container(inner) => {
let inner_formatted = format_data_container(inner, version.clone()); let inner_formatted = format_data_container(inner);
format!("{{ {} }}", inner_formatted) format!("{{ {} }}", inner_formatted)
} }
DataValue::Array(inner_arr) => { DataValue::Array(inner_arr) => {
let formatted = format_array(inner_arr, version.clone()); let formatted = format_array(inner_arr);
format!("[{}]", formatted) format!("[{}]", formatted)
} }
@ -243,7 +242,7 @@ fn format_array(arr: Vec<DataValue>, version: Version) -> String {
DataValue::BoolTrue => "true".to_string(), DataValue::BoolTrue => "true".to_string(),
DataValue::BoolFalse => "false".to_string(), DataValue::BoolFalse => "false".to_string(),
DataValue::SignedNumber(num) => num.to_string(), DataValue::Number(num) => num.to_string(),
_ => String::new(), _ => String::new(),
}) })

241
type-maps.yaml
View file

@ -1,241 +0,0 @@
# The version a Client should use
protocol_version: "1.0"
# Note that markers 0 to 31 are reserved for default use, manually working with them is not recommended
# Fixed CommunicationType markers are:
# Error: 0
# ErrorParsing: 1
# ErrorBadVersion: 2
# Disconnect: 3
# Redirect: 4
# Shutdown: 5
# BadRequest: 6
# Unauthorized: 7
# Forbidden: 8
# NotFound: 9
# TooManyRequests: 10
# InternalServerError: 11
# BadGateway: 12
# ServiceUnavailable: 13
# GatewayTimeout: 14
# Identification: 15
# IdentificationResponse: 16
# Register: 17
# RegisterResponse: 18
# Ping: 19
# Pong: 20
#
# Fixed Data Type markers are:
# Error: 0
# ErrorParsing: 1
# ErrorMessage: 2
# Version: 3
# Description: 4
# Timestamp: 5
# Id: 6
# ClientNonce: 7
# ServerNonce: 8
# PublicKeys: 9
# Signature: 10
# Connected: 11
#
# If a Type can't be used it will be mapped to 0
type_maps:
"1.0": # Protocol version 1.0
CommunicationTypes:
ErrorProtocol: 33
ErrorAnonymous: 34
ErrorInternal: 35
ErrorInvalidData: 36
ErrorInvalidUserId: 37
ErrorInvalidOmikronId: 38
ErrorNotFound: 39
ErrorNotAuthenticated: 40
ErrorNoIota: 41
ErrorInvalidChallenge: 42
ErrorInvalidSecret: 43
ErrorInvalidPrivateKey: 44
ErrorInvalidPublicKey: 45
ErrorNoUserId: 46
ErrorNoCallId: 47
ErrorInvalidCallId: 48
Success: 49
ShortenLink: 50
SettingsSave: 51
SettingsLoad: 52
SettingsList: 53
GlobalSettingsSave: 54
GlobalSettingsLoad: 55
Message: 56
MessageState: 57
MessageSend: 58
MessageLive: 59
MessageOtherIota: 60
MessageChunk: 61
MessagesGet: 62
PushNotification: 63
ReadNotification: 64
GetNotifications: 65
TauriIdentification: 66
ChangeConfirm: 67
ConfirmReceive: 68
ConfirmRead: 69
GetChats: 70
GetStates: 71
AddCommunity: 72
RemoveCommunity: 73
GetCommunities: 74
RegisterIota: 81
RegisterIotaSuccess: 82
AddConversation: 85
SendChat: 86
ClientChanged: 87
ClientConnected: 88
ClientDisconnected: 89
ClientClosed: 90
PublicKey: 91
PrivateKey: 92
WebrtcSdp: 93
WebrtcIce: 94
StartStream: 95
EndStream: 96
WatchStream: 97
CallToken: 98
CallInvite: 99
CallDisconnectUser: 100
CallTimeoutUser: 101
CallSetAnonymousJoining: 102
CallData: 103
EndCall: 104
Function: 105
Update: 106
CreateUser: 107
RhoUpdate: 108
UserConnected: 109
UserDisconnected: 110
IotaConnected: 111
IotaDisconnected: 112
SyncClientIotaStatus: 113
GetUserData: 114
GetIotaData: 115
IotaUserData: 116
ChangeUserData: 117
ChangeIotaData: 118
GetRegister: 119
CompleteRegisterUser: 120
CompleteRegisterIota: 121
DeleteUser: 122
DeleteIota: 123
StartRegister: 124
CompleteRegister: 125
GetApp: 126
CreateApp: 127
DeleteApp: 128
SaveAppData: 129
LoadAppData: 130
AppIdentification: 131
AppChallenge: 132
AppChallengeResponse: 133
AppIdentificationResponse: 134
LoadTxtRecord: 135
DataTypes:
ErrorType: 32
ErrorProtocol: 33
AcceptedIds: 34
Uuid: 35
RegisterId: 36
Link: 37
Settings: 38
SettingsName: 39
ChatPartnerId: 40
ChatPartnerName: 41
IotaId: 42
UserId: 43
UserIds: 44
IotaIds: 45
UserState: 46
UserStates: 47
UserPings: 48
CallState: 49
ScreenShare: 50
PrivateKeyHash: 51
Accepted: 52
AcceptedProfiles: 53
DeniedProfiles: 54
Content: 55
Messages: 56
Notifications: 57
SendTime: 58
GetTime: 59
GetVariant: 60
SharedSecretOwn: 61
SharedSecretOther: 62
SharedSecretSign: 63
SharedSecret: 64
CallId: 65
CallToken: 66
CallSecret: 67
Untill: 68
Enabled: 69
StartDate: 70
EndDate: 71
ReceiverId: 72
SenderId: 73
Signed: 75
Message: 76
MessageState: 77
LastPing: 78
PingIota: 79
PingClients: 80
Matches: 81
Omikron: 82
Offset: 83
Amount: 84
Position: 85
Name: 86
Path: 87
Codec: 88
Function: 89
Payload: 90
Result: 91
Interactables: 92
WantToWatch: 93
Watcher: 94
CreatedAt: 95
Username: 96
Display: 97
Avatar: 98
About: 99
Status: 100
PublicKey: 101
SubLevel: 102
SubEnd: 103
CommunityAddress: 104
CommunityTitle: 106
Communities: 107
RhoConnections: 108
User: 109
OnlineStatus: 110
OmikronId: 111
OmikronConnections: 112
ResetToken: 113
NewToken: 114
CallInvited: 115
CallMembers: 116
Calls: 117
Timeout: 118
HasAdmin: 119
LastMessageAt: 120
Height: 121
SentBySelf: 122
SessionId: 123
Contacts: 124
LastMessage: 125
AppIdentifier: 127
AppPrivateKey: 128
AppPublicKey: 129
AppSession: 130
AppData: 131
TauriToken: 132
Challenge: 133