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

[WIP] Webtransport

Browse source
This commit is contained in:
Alex Emmet 2026-03-03 22:13:54 +01:00
commit eed2674e44
3 changed files with 336 additions and 98 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/main.rs
View file

@ -9,6 +9,7 @@ use std::env;
use dotenv::dotenv;
use once_cell::sync::Lazy;
use rustls::crypto::aws_lc_rs;
use crate::{
calls::call_util::garbage_collect_calls,
@ -31,6 +32,7 @@ pub fn get_public_key() -> x448::PublicKey {
#[tokio::main]
async fn main() {
aws_lc_rs::default_provider().install_default();
dotenv().ok();
startup();

153
src/omega/omega_connection.rs
View file

@ -34,8 +34,8 @@ use uuid::Uuid;
// Configuration
// ============================================================================
const OMEGA_HOST_DEFAULT: &str = "omega.tensamin.net";
const OMEGA_PORT_DEFAULT: u16 = 443;
const OMEGA_HOST_DEFAULT: &str = "188.114.97.0";
const OMEGA_PORT_DEFAULT: u16 = 9187;
const RECONNECT_DELAY: Duration = Duration::from_secs(5);
const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(300);
const CONNECTION_TIMEOUT: Duration = Duration::from_secs(10);
@ -213,16 +213,21 @@ impl OmegaConnection {
let addr_str = format!("{}:{}", self.host, self.port);
// Resolve address (DNS lookup)
let remote_addr = tokio::net::lookup_host(&addr_str)
let addrs: Vec<SocketAddr> = tokio::net::lookup_host(&addr_str)
.await
.map_err(|e| format!("DNS lookup failed for {}: {}", addr_str, e))?
.next()
.ok_or_else(|| format!("No addresses found for {}", addr_str))?;
.collect();
let bind_addr: SocketAddr = "0.0.0.0:0".parse().unwrap();
let endpoint =
Endpoint::client(bind_addr).map_err(|e| format!("Failed to create endpoint: {}", e))?;
if addrs.is_empty() {
return Err(format!("No addresses found for {}", addr_str));
}
// Prefer IPv4 if available, otherwise use first available (IPv6 or IPv4)
let remote_addr = addrs
.iter()
.find(|a| a.is_ipv4())
.copied()
.unwrap_or_else(|| addrs[0]);
log_in!(
0,
@ -232,24 +237,19 @@ impl OmegaConnection {
remote_addr
);
// Connect to Omega server
let connection = endpoint
.connect(remote_addr, &self.host)
.map_err(|e| format!("Connect failed: {}", e))?
// Connect using epsilon_native wrapper
let (sender, receiver) = epsilon_native::client::connect(remote_addr)
.await
.map_err(|e| format!("Connection failed: {}", e))?;
log_in!(
0,
PrintType::Omega,
"QUIC connection established to {}",
addr_str
"QUIC connection established to {} (via {})",
addr_str,
remote_addr
);
// Create Sender and Receiver using your epsilon_native API
let sender = Sender::new(connection.clone());
let receiver = Receiver::new(connection);
// Store sender
*self.sender.write().await = Some(Arc::new(sender));
*self.state.write().await = ConnectionState::Connected { identified: false };
@ -287,21 +287,6 @@ impl OmegaConnection {
}
}
fn load_client_tls(&self) -> Result<RustlsClientConfig, Box<dyn std::error::Error>> {
let _ = aws_lc_rs::default_provider().install_default();
let mut root_store = rustls::RootCertStore::empty();
// Add webpki roots for system CA certificates
root_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
let config = RustlsClientConfig::builder()
.with_root_certificates(root_store)
.with_no_client_auth();
Ok(config)
}
// -------------------------------------------------------------------------
// Identification Handshake
// -------------------------------------------------------------------------
@ -601,6 +586,54 @@ impl OmegaConnection {
pub async fn is_identified(&self) -> bool {
self.state.read().await.is_identified()
}
pub async fn close_iota(iota_id: i64) {
let cv = CommunicationValue::new(CommunicationType::iota_disconnected)
.add_data(DataTypes::iota_id, DataValue::Number(iota_id));
OMEGA_CONNECTION.send_message(&cv).await;
}
pub async fn client_changed(_iota_id: i64, user_id: i64, state: UserStatus) {
let msg_type = match state {
UserStatus::iota_offline => CommunicationType::user_disconnected,
UserStatus::user_offline => CommunicationType::user_disconnected,
_ => CommunicationType::user_connected,
};
let cv = CommunicationValue::new(msg_type)
.add_data(DataTypes::user_id, DataValue::Number(user_id));
OMEGA_CONNECTION.send_message(&cv).await;
}
pub async fn user_states(user_id: i64, user_ids: Vec<i64>) {
let user_ids = user_ids.iter().map(|v| DataValue::Number(*v)).collect();
let cv = CommunicationValue::new(CommunicationType::get_states)
.add_data(DataTypes::user_ids, DataValue::Array(user_ids));
let msg_id = cv.get_id();
WAITING_TASKS.insert(
msg_id,
WaitingTask {
task: Box::new(
move |_: Arc<OmegaConnection>, response: CommunicationValue| {
tokio::spawn(async move {
let rho = rho_manager::get_rho_con_for_user(user_id).await;
if let Some(rho) = rho {
for client in rho.get_client_connections_for_user(user_id).await {
client.send_message(&response).await;
}
}
});
true
},
),
inserted_at: Instant::now(),
},
);
OMEGA_CONNECTION.send_message(&cv).await;
}
}
// ============================================================================
@ -624,55 +657,3 @@ static OMEGA_CONNECTION: Lazy<Arc<OmegaConnection>> = Lazy::new(|| {
pub fn get_omega_connection() -> Arc<OmegaConnection> {
OMEGA_CONNECTION.clone()
}
// ============================================================================
// Global Helpers
// ============================================================================
pub async fn close_iota(iota_id: i64) {
let cv = CommunicationValue::new(CommunicationType::iota_disconnected)
.add_data(DataTypes::iota_id, DataValue::Number(iota_id));
OMEGA_CONNECTION.send_message(&cv).await;
}
pub async fn client_changed(_iota_id: i64, user_id: i64, state: UserStatus) {
let msg_type = match state {
UserStatus::iota_offline => CommunicationType::user_disconnected,
UserStatus::user_offline => CommunicationType::user_disconnected,
_ => CommunicationType::user_connected,
};
let cv =
CommunicationValue::new(msg_type).add_data(DataTypes::user_id, DataValue::Number(user_id));
OMEGA_CONNECTION.send_message(&cv).await;
}
pub async fn user_states(user_id: i64, user_ids: Vec<i64>) {
let user_ids = user_ids.iter().map(|v| DataValue::Number(*v)).collect();
let cv = CommunicationValue::new(CommunicationType::get_states)
.add_data(DataTypes::user_ids, DataValue::Array(user_ids));
let msg_id = cv.get_id();
WAITING_TASKS.insert(
msg_id,
WaitingTask {
task: Box::new(
move |_: Arc<OmegaConnection>, response: CommunicationValue| {
tokio::spawn(async move {
let rho = rho_manager::get_rho_con_for_user(user_id).await;
if let Some(rho) = rho {
for client in rho.get_client_connections_for_user(user_id).await {
client.send_message(&response).await;
}
}
});
true
},
),
inserted_at: Instant::now(),
},
);
OMEGA_CONNECTION.send_message(&cv).await;
}