Loading Logging etc
This commit is contained in:
Alex Emmet
parent
a1d8bcbeff
commit
bba3f548ab
6 changed files with 520 additions and 264 deletions
|
|
@ -1,30 +1,36 @@
|
|||
use futures_util::{SinkExt, StreamExt};
|
||||
use std::collections::HashMap;
|
||||
use std::str::FromStr;
|
||||
use std::sync::Arc;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
use json::JsonValue;
|
||||
use tokio::sync::Mutex;
|
||||
use tokio::net::TcpStream;
|
||||
use tokio::time::{sleep, Duration};
|
||||
use tokio_tungstenite::{
|
||||
connect_async,
|
||||
tungstenite::protocol::{Message},
|
||||
MaybeTlsStream,
|
||||
WebSocketStream,
|
||||
};
|
||||
use uuid::{Uuid};
|
||||
use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes};
|
||||
use crate::users::contact::Contact;
|
||||
use crate::users::user_community_util::UserCommunityUtil;
|
||||
use crate::util::chat_files::ChatFiles;
|
||||
use crate::util::chats_util::ChatsUtil;
|
||||
use crate::util::chats_util::{get_user, get_users, mod_user};
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use json::JsonValue;
|
||||
use std::collections::HashMap;
|
||||
use std::str::FromStr;
|
||||
use std::sync::Arc;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
use tokio::net::TcpStream;
|
||||
use tokio::sync::Mutex;
|
||||
use tokio::time::{Duration, sleep};
|
||||
use tokio_tungstenite::{
|
||||
MaybeTlsStream, WebSocketStream, connect_async, tungstenite::protocol::Message,
|
||||
};
|
||||
use uuid::Uuid;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct OmikronConnection {
|
||||
pub(crate) writer: Arc<Mutex<Option<futures_util::stream::SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>>>>,
|
||||
pub(crate) writer: Arc<
|
||||
Mutex<
|
||||
Option<
|
||||
futures_util::stream::SplitSink<
|
||||
WebSocketStream<MaybeTlsStream<TcpStream>>,
|
||||
Message,
|
||||
>,
|
||||
>,
|
||||
>,
|
||||
>,
|
||||
waiting: Arc<Mutex<HashMap<Uuid, Box<dyn Fn(CommunicationValue) + Send>>>>, // waiting for responses
|
||||
pingpong: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>, // ping-pong handler
|
||||
pingpong: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>, // ping-pong handler
|
||||
}
|
||||
|
||||
impl OmikronConnection {
|
||||
|
|
@ -66,11 +72,13 @@ impl OmikronConnection {
|
|||
/// Listener for all incoming messages
|
||||
fn spawn_listener(
|
||||
&self,
|
||||
mut read_half: futures_util::stream::SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
|
||||
mut read_half: futures_util::stream::SplitStream<
|
||||
WebSocketStream<MaybeTlsStream<TcpStream>>,
|
||||
>,
|
||||
) {
|
||||
let waiting = self.waiting.clone();
|
||||
let writer = self.writer.clone();
|
||||
|
||||
|
||||
tokio::spawn(async move {
|
||||
while let Some(msg) = read_half.next().await {
|
||||
match msg {
|
||||
|
|
@ -92,97 +100,151 @@ impl OmikronConnection {
|
|||
let sender_id = &cv.get_sender();
|
||||
let receiver_id = &cv.get_receiver();
|
||||
ChatFiles::add_message(
|
||||
|
||||
cv.get_data(DataTypes::SendTime).unwrap().as_i64().unwrap_or(-1),
|
||||
cv.get_data(DataTypes::SendTime)
|
||||
.unwrap()
|
||||
.as_i64()
|
||||
.unwrap_or(-1),
|
||||
false,
|
||||
receiver_id.unwrap(),
|
||||
sender_id.unwrap(),
|
||||
cv.get_data(DataTypes::MessageContent).unwrap().as_str().unwrap(),
|
||||
cv.get_data(DataTypes::MessageContent)
|
||||
.unwrap()
|
||||
.as_str()
|
||||
.unwrap(),
|
||||
);
|
||||
let response = CommunicationValue::new(CommunicationType::MessageLive)
|
||||
.with_id(cv.get_id())
|
||||
.with_receiver(cv.get_receiver().unwrap())
|
||||
.add_data(DataTypes::SendTime, cv.get_data(DataTypes::SendTime).unwrap().clone())
|
||||
.add_data(DataTypes::Message, cv.get_data(DataTypes::MessageContent).unwrap().clone())
|
||||
.add_data(DataTypes::SenderId, JsonValue::String(cv.get_sender().unwrap().clone().to_string()));
|
||||
Self::send_message_static(&writer.clone(), response.to_json().to_string()).await;
|
||||
.add_data(
|
||||
DataTypes::SendTime,
|
||||
cv.get_data(DataTypes::SendTime).unwrap().clone(),
|
||||
)
|
||||
.add_data(
|
||||
DataTypes::Message,
|
||||
cv.get_data(DataTypes::MessageContent).unwrap().clone(),
|
||||
)
|
||||
.add_data(
|
||||
DataTypes::SenderId,
|
||||
JsonValue::String(cv.get_sender().unwrap().clone().to_string()),
|
||||
);
|
||||
Self::send_message_static(
|
||||
&writer.clone(),
|
||||
response.to_json().to_string(),
|
||||
)
|
||||
.await;
|
||||
continue;
|
||||
}
|
||||
|
||||
if cv.is_type(CommunicationType::Message) {
|
||||
let my_id = cv.get_sender();
|
||||
ChatFiles::add_message(SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as i64,
|
||||
true,
|
||||
my_id.unwrap(),
|
||||
Uuid::from_str(&*cv.get_data(DataTypes::ReceiverId).unwrap().to_string()).unwrap(),
|
||||
&*cv.get_data(DataTypes::MessageContent).unwrap().to_string()
|
||||
ChatFiles::add_message(
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as i64,
|
||||
true,
|
||||
my_id.unwrap(),
|
||||
Uuid::from_str(
|
||||
&*cv.get_data(DataTypes::ReceiverId).unwrap().to_string(),
|
||||
)
|
||||
.unwrap(),
|
||||
&*cv.get_data(DataTypes::MessageContent).unwrap().to_string(),
|
||||
);
|
||||
// ack
|
||||
let ack = CommunicationValue::ack_message(cv.get_id(), my_id);
|
||||
Self::send_message_static(&writer.clone(), ack.to_json().to_string()).await;
|
||||
Self::send_message_static(&writer.clone(), ack.to_json().to_string())
|
||||
.await;
|
||||
// forward
|
||||
let forward = CommunicationValue::forward_to_other_iota(&mut cv);
|
||||
Self::send_message_static(&writer.clone(), forward.to_json().to_string()).await;
|
||||
Self::send_message_static(
|
||||
&writer.clone(),
|
||||
forward.to_json().to_string(),
|
||||
)
|
||||
.await;
|
||||
continue;
|
||||
}
|
||||
|
||||
if cv.is_type(CommunicationType::MessageGet) {
|
||||
let my_id = cv.get_sender().unwrap();
|
||||
let partner_id = Uuid::from_str(&*cv.get_data(DataTypes::ChatPartnerId).unwrap().to_string()).unwrap();
|
||||
let offset = cv.get_data(DataTypes::LoadedMessages).unwrap().to_string().parse::<i64>().unwrap();
|
||||
let amount = cv.get_data(DataTypes::MessageAmount).unwrap().to_string().parse::<i64>().unwrap();
|
||||
let partner_id = Uuid::from_str(
|
||||
&*cv.get_data(DataTypes::ChatPartnerId).unwrap().to_string(),
|
||||
)
|
||||
.unwrap();
|
||||
let offset = cv
|
||||
.get_data(DataTypes::LoadedMessages)
|
||||
.unwrap()
|
||||
.to_string()
|
||||
.parse::<i64>()
|
||||
.unwrap();
|
||||
let amount = cv
|
||||
.get_data(DataTypes::MessageAmount)
|
||||
.unwrap()
|
||||
.to_string()
|
||||
.parse::<i64>()
|
||||
.unwrap();
|
||||
|
||||
let messages = ChatFiles::get_messages(my_id, partner_id, offset, amount); // needs ChatFiles
|
||||
let messages =
|
||||
ChatFiles::get_messages(my_id, partner_id, offset, amount); // needs ChatFiles
|
||||
let mut resp = CommunicationValue::new(CommunicationType::MessageChunk)
|
||||
.with_id(cv.get_id())
|
||||
.with_receiver(my_id);
|
||||
if !messages.is_empty() {
|
||||
resp = resp.add_data(DataTypes::MessageChunk, messages);
|
||||
}
|
||||
Self::send_message_static(&writer.clone(), resp.to_json().to_string()).await;
|
||||
Self::send_message_static(&writer.clone(), resp.to_json().to_string())
|
||||
.await;
|
||||
continue;
|
||||
}
|
||||
|
||||
if cv.is_type(CommunicationType::GetChats) {
|
||||
let user_id = cv.get_sender();
|
||||
let users = ChatsUtil::get_users(user_id.unwrap()); // needs ChatsUtil
|
||||
let users = get_users(user_id.unwrap()); // needs ChatsUtil
|
||||
let resp = CommunicationValue::new(CommunicationType::GetChats)
|
||||
.with_id(cv.get_id())
|
||||
.with_receiver(user_id.unwrap())
|
||||
.add_data(DataTypes::UserIds, users);
|
||||
Self::send_message_static(&writer.clone(), resp.to_json().to_string()).await;
|
||||
Self::send_message_static(&writer.clone(), resp.to_json().to_string())
|
||||
.await;
|
||||
continue;
|
||||
}
|
||||
|
||||
if cv.is_type(CommunicationType::AddChat) {
|
||||
let user_id = cv.get_sender();
|
||||
let other_id = Uuid::from_str(&*cv.get_data(DataTypes::UserId).unwrap().to_string()).unwrap();
|
||||
let mut contact = ChatsUtil::get_user(user_id.unwrap(), other_id).unwrap_or(Contact::new(other_id)); // needs ChatsUtil + Contact
|
||||
contact.set_last_message_at(SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as i64);
|
||||
ChatsUtil::mod_user(user_id.unwrap(), &contact);
|
||||
let other_id = Uuid::from_str(
|
||||
&*cv.get_data(DataTypes::UserId).unwrap().to_string(),
|
||||
)
|
||||
.unwrap();
|
||||
let mut contact = get_user(user_id.unwrap(), other_id)
|
||||
.unwrap_or(Contact::new(other_id)); // needs ChatsUtil + Contact
|
||||
contact.set_last_message_at(
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as i64,
|
||||
);
|
||||
mod_user(user_id.unwrap(), &contact);
|
||||
let resp = CommunicationValue::new(CommunicationType::AddChat)
|
||||
.with_id(cv.get_id())
|
||||
.with_receiver(user_id.unwrap());
|
||||
Self::send_message_static(&writer.clone(), resp.to_json().to_string()).await;
|
||||
Self::send_message_static(&writer.clone(), resp.to_json().to_string())
|
||||
.await;
|
||||
continue;
|
||||
}
|
||||
|
||||
if cv.is_type(CommunicationType::AddCommunity) {
|
||||
UserCommunityUtil::add_community( cv.get_sender().unwrap(),
|
||||
cv.get_data(DataTypes::CommunityAddress).unwrap().to_string(),
|
||||
cv.get_data(DataTypes::CommunityTitle).unwrap().to_string(),
|
||||
cv.get_data(DataTypes::Position).unwrap().to_string()
|
||||
UserCommunityUtil::add_community(
|
||||
cv.get_sender().unwrap(),
|
||||
cv.get_data(DataTypes::CommunityAddress)
|
||||
.unwrap()
|
||||
.to_string(),
|
||||
cv.get_data(DataTypes::CommunityTitle).unwrap().to_string(),
|
||||
cv.get_data(DataTypes::Position).unwrap().to_string(),
|
||||
);
|
||||
let resp = CommunicationValue::new(CommunicationType::AddCommunity)
|
||||
.with_id(cv.get_id())
|
||||
.with_receiver(cv.get_sender().unwrap());
|
||||
Self::send_message_static(&writer.clone(), resp.to_json().to_string()).await;
|
||||
Self::send_message_static(&writer.clone(), resp.to_json().to_string())
|
||||
.await;
|
||||
continue;
|
||||
}
|
||||
|
||||
|
|
@ -190,17 +252,27 @@ impl OmikronConnection {
|
|||
let resp = CommunicationValue::new(CommunicationType::GetCommunities)
|
||||
.with_id(cv.get_id())
|
||||
.with_receiver(cv.get_sender().unwrap())
|
||||
.add_data(DataTypes::Communities, UserCommunityUtil::get_communities(cv.get_sender().unwrap())); // needs UserCommunityUtil
|
||||
Self::send_message_static(&writer.clone(), resp.to_json().to_string()).await;
|
||||
.add_data(
|
||||
DataTypes::Communities,
|
||||
UserCommunityUtil::get_communities(cv.get_sender().unwrap()),
|
||||
); // needs UserCommunityUtil
|
||||
Self::send_message_static(&writer.clone(), resp.to_json().to_string())
|
||||
.await;
|
||||
continue;
|
||||
}
|
||||
|
||||
if cv.is_type(CommunicationType::RemoveCommunity) {
|
||||
UserCommunityUtil::remove_community(cv.get_sender().unwrap(), cv.get_data(DataTypes::CommunityAddress).unwrap().to_string()); // needs UserCommunityUtil
|
||||
UserCommunityUtil::remove_community(
|
||||
cv.get_sender().unwrap(),
|
||||
cv.get_data(DataTypes::CommunityAddress)
|
||||
.unwrap()
|
||||
.to_string(),
|
||||
); // needs UserCommunityUtil
|
||||
let resp = CommunicationValue::new(CommunicationType::RemoveCommunity)
|
||||
.with_id(cv.get_id())
|
||||
.with_receiver(cv.get_sender().unwrap());
|
||||
Self::send_message_static(&writer.clone(), resp.to_json().to_string()).await;
|
||||
Self::send_message_static(&writer.clone(), resp.to_json().to_string())
|
||||
.await;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
|
@ -217,7 +289,16 @@ impl OmikronConnection {
|
|||
Self::send_message_static(&self.writer, msg).await;
|
||||
}
|
||||
pub async fn send_message_static(
|
||||
writer: &Arc<Mutex<Option<futures_util::stream::SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>>>>,
|
||||
writer: &Arc<
|
||||
Mutex<
|
||||
Option<
|
||||
futures_util::stream::SplitSink<
|
||||
WebSocketStream<MaybeTlsStream<TcpStream>>,
|
||||
Message,
|
||||
>,
|
||||
>,
|
||||
>,
|
||||
>,
|
||||
msg: String,
|
||||
) -> Result<(), tokio_tungstenite::tungstenite::Error> {
|
||||
let mut guard = writer.lock().await;
|
||||
|
|
@ -229,7 +310,7 @@ impl OmikronConnection {
|
|||
Err(tokio_tungstenite::tungstenite::Error::ConnectionClosed)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
pub fn on_answer<F>(&self, message_id: Uuid, callback: F)
|
||||
where
|
||||
F: Fn(CommunicationValue) + Send + 'static,
|
||||
|
|
|
|||
Loading…
Reference in a new issue