mirror of
https://github.com/clash-verge-rev/clash-verge-rev.git
synced 2026-10-21 01:32:08 +08:00
fix: clippy
This commit is contained in:
Tunglies
@@ -39,8 +39,6 @@ static INSTANCE_LOCK: OnceCell<std::fs::File> = OnceCell::new();
|
||||
const PAC_INITIAL_AVAILABLE: bool = false;
|
||||
static PAC_AVAILABLE: AtomicBool = AtomicBool::new(PAC_INITIAL_AVAILABLE);
|
||||
static COMMANDS_READY: AtomicBool = AtomicBool::new(false);
|
||||
#[cfg(feature = "verge-dev")]
|
||||
static DEV_QUIT_REQUESTED: AtomicBool = AtomicBool::new(false);
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum SingletonDisposition {
|
||||
@@ -144,55 +142,12 @@ fn instance_auth(token: String) -> impl warp::Filter<Extract = (), Error = warp:
|
||||
.untuple_one()
|
||||
}
|
||||
|
||||
#[cfg(feature = "verge-dev")]
|
||||
fn dev_quit_route<F>(
|
||||
token: String,
|
||||
commands_ready: &'static AtomicBool,
|
||||
quit_requested: &'static AtomicBool,
|
||||
trigger: F,
|
||||
) -> impl warp::Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone
|
||||
where
|
||||
F: Fn() + Clone + Send + Sync + 'static,
|
||||
{
|
||||
instance_auth(token)
|
||||
.and(warp::path!("commands" / "dev" / "quit"))
|
||||
.and(warp::post())
|
||||
.and_then(move || {
|
||||
let trigger = trigger.clone();
|
||||
async move {
|
||||
if !commands_ready.load(Ordering::Acquire) {
|
||||
return Ok::<_, warp::Rejection>(warp::reply::with_status(
|
||||
String::from("starting"),
|
||||
warp::http::StatusCode::SERVICE_UNAVAILABLE,
|
||||
));
|
||||
}
|
||||
if quit_requested
|
||||
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
|
||||
.is_ok()
|
||||
{
|
||||
trigger();
|
||||
}
|
||||
Ok(warp::reply::with_status(
|
||||
String::from("accepted"),
|
||||
warp::http::StatusCode::ACCEPTED,
|
||||
))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(feature = "verge-dev")]
|
||||
fn release_dev_quit_latch(quit_requested: &AtomicBool, outcome: clash_verge_signal::ShutdownOutcome) {
|
||||
if matches!(outcome, clash_verge_signal::ShutdownOutcome::Canceled) {
|
||||
quit_requested.store(false, Ordering::Release);
|
||||
}
|
||||
}
|
||||
|
||||
fn start_embedded_server(listener: tokio::net::TcpListener, token: String) {
|
||||
let (shutdown_tx, shutdown_rx) = oneshot::channel();
|
||||
let _ = SHUTDOWN_SENDER.set(Mutex::new(Some(shutdown_tx)));
|
||||
|
||||
#[cfg(feature = "verge-dev")]
|
||||
let auth = instance_auth(token.clone());
|
||||
let auth = instance_auth(token);
|
||||
#[cfg(not(feature = "verge-dev"))]
|
||||
let auth = instance_auth(token);
|
||||
|
||||
@@ -256,12 +211,6 @@ fn start_embedded_server(listener: tokio::net::TcpListener, token: String) {
|
||||
});
|
||||
|
||||
let commands = auth.clone().and(visible).or(auth.and(scheme)).or(pac);
|
||||
#[cfg(feature = "verge-dev")]
|
||||
let commands = commands.or(dev_quit_route(token, &COMMANDS_READY, &DEV_QUIT_REQUESTED, || {
|
||||
AsyncHandler::spawn(|| async {
|
||||
release_dev_quit_latch(&DEV_QUIT_REQUESTED, crate::feat::quit().await);
|
||||
});
|
||||
}));
|
||||
AsyncHandler::spawn(move || async move {
|
||||
warp::serve(commands)
|
||||
.incoming(listener)
|
||||
@@ -432,113 +381,10 @@ fn write_instance_record(path: &Path, record: &InstanceRecord) -> Result<()> {
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
InstanceRecord, PAC_INITIAL_AVAILABLE, bind_primary_listener, read_instance_record, write_instance_record,
|
||||
};
|
||||
use super::{InstanceRecord, bind_primary_listener, read_instance_record, write_instance_record};
|
||||
#[cfg(unix)]
|
||||
use super::{open_instance_lock, try_lock_instance};
|
||||
|
||||
#[cfg(feature = "verge-dev")]
|
||||
use super::{INSTANCE_TOKEN_HEADER, dev_quit_route, release_dev_quit_latch};
|
||||
#[cfg(feature = "verge-dev")]
|
||||
use std::sync::atomic::Ordering;
|
||||
#[cfg(feature = "verge-dev")]
|
||||
use std::sync::{
|
||||
Arc,
|
||||
atomic::{AtomicBool, AtomicUsize},
|
||||
};
|
||||
#[test]
|
||||
fn pac_is_fail_closed_before_core_readiness() {
|
||||
// Asserts the default rather than the live flag: PAC availability is now derived from
|
||||
// the Running Mode by `core::runstate`, so any test that drives a Core transition
|
||||
// legitimately moves the live flag. That the derivation itself can never disagree with
|
||||
// the Running Mode is covered in `core::runstate` against a fake environment.
|
||||
const { assert!(!PAC_INITIAL_AVAILABLE) }
|
||||
}
|
||||
|
||||
#[cfg(feature = "verge-dev")]
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
||||
async fn dev_quit_route_requires_private_auth_ready_state_and_dispatches_once() -> anyhow::Result<()> {
|
||||
static READY: AtomicBool = AtomicBool::new(false);
|
||||
static REQUESTED: AtomicBool = AtomicBool::new(false);
|
||||
READY.store(false, Ordering::Release);
|
||||
REQUESTED.store(false, Ordering::Release);
|
||||
let dispatches = Arc::new(AtomicUsize::new(0));
|
||||
let dispatches_for_route = Arc::clone(&dispatches);
|
||||
let token = "ab".repeat(32);
|
||||
let route = dev_quit_route(token.clone(), &READY, &REQUESTED, move || {
|
||||
dispatches_for_route.fetch_add(1, Ordering::AcqRel);
|
||||
});
|
||||
|
||||
let missing = warp::test::request()
|
||||
.method("POST")
|
||||
.path("/commands/dev/quit")
|
||||
.reply(&route)
|
||||
.await;
|
||||
assert_eq!(missing.status(), warp::http::StatusCode::NOT_FOUND);
|
||||
|
||||
let wrong_token = warp::test::request()
|
||||
.method("POST")
|
||||
.path("/commands/dev/quit")
|
||||
.header(INSTANCE_TOKEN_HEADER, "cd".repeat(32))
|
||||
.reply(&route)
|
||||
.await;
|
||||
assert_eq!(wrong_token.status(), warp::http::StatusCode::NOT_FOUND);
|
||||
|
||||
let not_ready = warp::test::request()
|
||||
.method("POST")
|
||||
.path("/commands/dev/quit")
|
||||
.header(INSTANCE_TOKEN_HEADER, &token)
|
||||
.reply(&route)
|
||||
.await;
|
||||
assert_eq!(not_ready.status(), warp::http::StatusCode::SERVICE_UNAVAILABLE);
|
||||
assert!(!REQUESTED.load(Ordering::Acquire));
|
||||
|
||||
let wrong_method = warp::test::request()
|
||||
.method("GET")
|
||||
.path("/commands/dev/quit")
|
||||
.header(INSTANCE_TOKEN_HEADER, &token)
|
||||
.reply(&route)
|
||||
.await;
|
||||
assert_eq!(wrong_method.status(), warp::http::StatusCode::METHOD_NOT_ALLOWED);
|
||||
assert!(!REQUESTED.load(Ordering::Acquire));
|
||||
|
||||
READY.store(true, Ordering::Release);
|
||||
let barrier = Arc::new(tokio::sync::Barrier::new(8));
|
||||
let accepted = futures::future::join_all((0..8).map(|_| {
|
||||
let barrier = Arc::clone(&barrier);
|
||||
let route = route.clone();
|
||||
let token = token.clone();
|
||||
tokio::spawn(async move {
|
||||
barrier.wait().await;
|
||||
warp::test::request()
|
||||
.method("POST")
|
||||
.path("/commands/dev/quit")
|
||||
.header(INSTANCE_TOKEN_HEADER, token)
|
||||
.reply(&route)
|
||||
.await
|
||||
})
|
||||
}))
|
||||
.await;
|
||||
for accepted in accepted {
|
||||
let accepted = accepted?;
|
||||
assert_eq!(accepted.status(), warp::http::StatusCode::ACCEPTED);
|
||||
}
|
||||
assert_eq!(dispatches.load(Ordering::Acquire), 1);
|
||||
|
||||
release_dev_quit_latch(&REQUESTED, clash_verge_signal::ShutdownOutcome::Canceled);
|
||||
assert!(!REQUESTED.load(Ordering::Acquire));
|
||||
let retry = warp::test::request()
|
||||
.method("POST")
|
||||
.path("/commands/dev/quit")
|
||||
.header(INSTANCE_TOKEN_HEADER, &token)
|
||||
.reply(&route)
|
||||
.await;
|
||||
assert_eq!(retry.status(), warp::http::StatusCode::ACCEPTED);
|
||||
assert_eq!(dispatches.load(Ordering::Acquire), 2);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn retained_record_supplies_the_next_primarys_preferred_port() -> anyhow::Result<()> {
|
||||
let root = std::env::temp_dir().join(format!("singleton-sticky-port-{}", std::process::id()));
|
||||
|
||||
Reference in New Issue
Block a user