tamipes/mc-ingress
1
0
Fork 0
Code Issues 4 Pull requests Projects Releases Packages 1 Wiki Activity Actions

feat: impement kube manifest caching in kube_cache, also fix Server to request data from the cache, instead of saving it
All checks were successful
/ build (push) Successful in 6m33s
Details

Browse source
This commit is contained in:
Tamipes 2026-06-07 16:03:44 +02:00
parent 6db804cdc1
commit 46c3a994c6
3 changed files with 207 additions and 185 deletions
Download patch file Download diff file Expand all files Collapse all files

102
src/main.rs
View file

@ -21,6 +21,7 @@ mod packets;
mod types;
static BYE_MESSAGE: &str = concat!("§dTami§r with §d<3§r §8(rev: ", env!("COMMIT_HASH"), ")§r");
static OFFLINE_TIMER: std::time::Duration = Duration::from_secs(600);
#[tokio::main]
async fn main() {
@ -39,7 +40,7 @@ async fn main() {
let revision: &'static str = env!("COMMIT_HASH");
tracing::info!(revision);
let api = kube_cache::McApi::create().await.unwrap();
let (api, api_task) = kube_cache::McApi::create().await.unwrap();
tracing::info!("initialized kube api");
let config: Config = Default::default();
@ -47,53 +48,64 @@ async fn main() {
let listener = TcpListener::bind(config.bind_addr.clone()).await.unwrap();
tracing::info!(bind_addr = config.bind_addr, "started tcp server");
loop {
let (socket, addr) = listener.accept().await.unwrap();
let api = api.clone();
let conn_task = tokio::spawn(async move {
loop {
let (socket, addr) = listener.accept().await.unwrap();
let api = api.clone();
let config = config.clone();
tokio::spawn(async move {
tracing::debug!(
addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
"Client connected"
);
if let Err(e) = process_connection(socket, addr, api, config).await {
match e.level {
tracing::Level::ERROR => tracing::error!(
// addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
trace = %e.print_span_trace(),
err = format!("{}", e.context),
"Client disconnected"
),
tracing::Level::WARN => tracing::warn!(
// addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
trace = %e.print_span_trace(),
err = format!("{}", e.context),
"Client disconnected"
),
tracing::Level::INFO => tracing::info!(
// addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
trace = %e.print_span_trace(),
err = format!("{}", e.context),
"Client disconnected"
),
_ => {
tracing::error!(
let config = config.clone();
tokio::spawn(async move {
tracing::debug!(
addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
"Client connected"
);
if let Err(e) = process_connection(socket, addr, api, config).await {
match e.level {
tracing::Level::ERROR => tracing::error!(
// addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
trace = %e.print_span_trace(),
err = format!("{}", e.context),
actual_level = ?e.level,
"Client disconnected (bad level)"
)
"Client disconnected"
),
tracing::Level::WARN => tracing::warn!(
// addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
trace = %e.print_span_trace(),
err = format!("{}", e.context),
"Client disconnected"
),
tracing::Level::INFO => tracing::info!(
// addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
trace = %e.print_span_trace(),
err = format!("{}", e.context),
"Client disconnected"
),
_ => {
tracing::error!(
// addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
trace = %e.print_span_trace(),
err = format!("{}", e.context),
actual_level = ?e.level,
"Client disconnected (bad level)"
)
}
}
} else {
tracing::debug!(
addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
"Client disconnected"
);
}
} else {
tracing::debug!(
addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
"Client disconnected"
);
}
});
});
}
});
tokio::select! {
result = api_task => {
tracing::error!("The api tokio:spawn'ed task run to completion, which should not happen!");
}
result = conn_task => {
tracing::error!("The connection handling tokio:spawn'ed task run to completion, which should not happen!");
}
}
}
@ -299,8 +311,7 @@ where
tracing::debug!(msg = "server status", status = ?status);
match status {
ServerDeploymentStatus::Connectable(mut server_stream) => {
api.start_watch(server.clone(), Duration::from_secs(600))
.await?;
api.start_watch(server.clone(), OFFLINE_TIMER).await?;
// referenced from:
// https://github.com/hanyu-dev/tokio-splice2/blob/fc47199fffde8946b0acf867d1fa0b2222267a34/examples/proxy.rs
@ -350,8 +361,7 @@ where
}
ServerDeploymentStatus::Offline => {
server.start().await?;
api.start_watch(server.clone(), Duration::from_secs(600))
.await?;
api.start_watch(server.clone(), OFFLINE_TIMER).await?;
mc_server::send_disconnect(client_stream, format!("[\"\",{{\"text\":\"Okayy, §2starting§r the server!\n\n\"}},{{\"text\":\"{BYE_MESSAGE}\"}}]").as_str()).await?;
}
ServerDeploymentStatus::Unavailable(_) => unreachable!(),