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

[Fix]Connection Stability

Browse source
This commit is contained in:
Alex Emmet 2026-07-04 23:02:02 +02:00
commit d56e6c8a03
5 changed files with 73 additions and 142 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/rho/connection.rs
View file

@ -296,6 +296,8 @@ impl GeneralConnection {
client.start();
}
ConnectionKind::Iota => {
log_in!(id as i64, PrintType::Iota, "Iota {} connected", id);
let notify = CommunicationValue::new(CommunicationType::IotaConnected)
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(id.into()));
get_omega_connection().send_message(&notify).await;

2
src/rho/iota_connection.rs
View file

@ -4,6 +4,7 @@ use crate::log_cv_in;
use crate::log_cv_out;
use crate::log_err;
use crate::log_in;
use crate::log_out;
use crate::omega::omega_connection::get_omega_connection;
use crate::rho::connection::GeneralConnection;
use crate::util::logger::PrintType;
@ -493,6 +494,7 @@ impl IotaConnection {
}
pub async fn handle_close(&self) {
log_out!(self.iota_id as i64, PrintType::Iota, "Iota {} disconnected", self.iota_id);
if let Some(rho_conn) = self.get_rho_connection().await {
rho_conn.close_iota_connection().await;
}

24
src/rho/server.rs
View file

@ -27,8 +27,11 @@ pub async fn get_by_connector_id(
description: Option<String>,
) -> Option<PublicKeyBundle> {
let request = match description.as_deref() {
Some("iota") => CommunicationValue::new(CommunicationType::GetIotaData)
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(client_id as i128)),
Some("iota") => {
println!("Iota connection request for client_id: {}", client_id);
CommunicationValue::new(CommunicationType::GetIotaData)
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(client_id as i128))
}
Some("client") => CommunicationValue::new(CommunicationType::GetUserData)
.add_typed_default(DataType::UserId, DataValue::SignedNumber(client_id as i128)),
_ => return None,
@ -59,9 +62,11 @@ pub async fn get_by_connector_id(
/* Only Iota registration goes through mtp's Register flow; users are registered out of band. */
pub async fn complete_register(pub_key: PublicKeyBundle, description: Option<String>) -> u64 {
println!("Iota register start");
if description.as_deref() != Some("iota") {
return 0;
}
println!("Iota connection request");
let request = CommunicationValue::new(CommunicationType::CompleteRegisterIota)
.add_typed_default(
@ -126,16 +131,21 @@ pub async fn start(port: u16) -> Result<(), Box<dyn std::error::Error>> {
.with_authentication_policy(AuthenticationPolicy::AllowAuthentication);
let mut host: Host = Host::new(host_config).await?;
log!(0, PrintType::General, "Server listening on port {}", port);
log!(0, PrintType::General, "Server listening on port {}.", port);
loop {
let conn = match host.accept().await {
Ok(Some(conn)) => conn,
Ok(Some(conn)) => {
println!("Accepted connection");
conn
}
Ok(None) => break,
Err(e) => {
// A single client's failed/aborted handshake (bad auth, a
// probe, a mid-handshake disconnect) must not take down the
// whole listener - only that connection attempt is lost.
/*
* A single client's failed/aborted handshake (bad auth, a
* probe, a mid-handshake disconnect) must not take down the
* whole listener, only that connection attempt is lost.
*/
log_err!(0, PrintType::General, "Rejected connection: {}", e);
continue;
}