Merge pull request #68 from Watfaq/rp

Rp

clash/tests/data/config/rule-set.yaml

@@ -0,0 +1,2 @@
+payload:
+  - 'httpbin.yba.dev'

clash/tests/data/config/rules.yaml

@@ -195,9 +195,16 @@ proxy-providers:
       url: http://www.gstatic.com/generate_204
       interval: 300
 
+rule-providers:
+  file-provider:
+    type: file
+    path: ./rule-set.yaml
+    interval: 300
+    behavior: domain
+
 rules:
   - DOMAIN,ipinfo.io,relay
-  - DOMAIN-KEYWORD,httpbin,h2-vmess
+  - RULE-SET,file-provider,h2-vmess
   - GEOIP,CN,relay
   - DOMAIN-SUFFIX,facebook.com,REJECT
   - DOMAIN-KEYWORD,google,select

clash_lib/Cargo.toml

@@ -39,6 +39,7 @@ boring = { git = "https://github.com/Watfaq/boring.git", rev = "24c006f" }
 boring-sys = { git = "https://github.com/Watfaq/boring.git", rev = "24c006f" }
 hyper-boring = { git = "https://github.com/Watfaq/boring.git", rev = "24c006f" }
 tokio-boring = { git = "https://github.com/Watfaq/boring.git", rev = "24c006f" }
+ip_network_table-deps-treebitmap = "0.5.0"
 
 crc32fast = "1.3.2"
 brotli = "3.3.4"

clash_lib/src/app/api/handlers/provider.rs

@@ -11,11 +11,11 @@ use axum::{
 };
 use serde::Deserialize;
 
-use crate::app::{api::AppState, outbound::manager::ThreadSafeOutboundManager};
-use crate::{
-    app::proxy_manager::providers::proxy_provider::ThreadSafeProxyProvider,
-    proxy::AnyOutboundHandler,
+use crate::app::{
+    api::AppState, outbound::manager::ThreadSafeOutboundManager,
+    remote_content_manager::providers::proxy_provider::ThreadSafeProxyProvider,
 };
+use crate::proxy::AnyOutboundHandler;
 #[derive(Clone)]
 struct ProviderState {
     outbound_manager: ThreadSafeOutboundManager,

clash_lib/src/app/dispatcher/dispatcher.rs

@@ -104,7 +104,7 @@ impl Dispatcher {
         };
 
         let mode = self.mode.lock().await;
-        info!("dispatching {} with mode {}", sess, mode);
+        debug!("dispatching {} with mode {}", sess, mode);
         let (outbound_name, rule) = match *mode {
             RunMode::Global => (PROXY_GLOBAL, None),
             RunMode::Rule => self.router.match_route(&sess).await,
@@ -120,7 +120,7 @@ impl Dispatcher {
 
         match handler.connect_stream(&sess, self.resolver.clone()).await {
             Ok(rhs) => {
-                info!("remote connection established {}", sess);
+                debug!("remote connection established {}", sess);
                 let mut rhs = Box::new(
                     TrackedStream::new(rhs, self.manager.clone(), sess.clone(), rule).await,
                 );

clash_lib/src/app/mod.rs

@@ -5,5 +5,5 @@ pub mod inbound;
 pub mod logging;
 pub mod outbound;
 pub mod profile;
-pub mod proxy_manager;
+pub mod remote_content_manager;
 pub mod router;

clash_lib/src/app/outbound/manager.rs

@@ -5,21 +5,23 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::{Mutex, RwLock};
-use tracing::debug;
+use tracing::error;
+use tracing::warn;
+
 use tracing::info;
 
 use crate::app::dns::ThreadSafeDNSResolver;
 use crate::app::profile::ThreadSafeCacheFile;
-use crate::app::proxy_manager::healthcheck::HealthCheck;
-use crate::app::proxy_manager::providers::file_vehicle;
-use crate::app::proxy_manager::providers::http_vehicle;
-use crate::app::proxy_manager::providers::plain_provider::PlainProvider;
-use crate::app::proxy_manager::providers::proxy_provider::ThreadSafeProxyProvider;
-use crate::app::proxy_manager::providers::proxy_set_provider::ProxySetProvider;
-use crate::app::proxy_manager::ProxyManager;
-
+use crate::app::remote_content_manager::healthcheck::HealthCheck;
+use crate::app::remote_content_manager::providers::file_vehicle;
+use crate::app::remote_content_manager::providers::http_vehicle;
+use crate::app::remote_content_manager::ProxyManager;
+
+use crate::app::remote_content_manager::providers::proxy_provider::PlainProvider;
+use crate::app::remote_content_manager::providers::proxy_provider::ProxySetProvider;
+use crate::app::remote_content_manager::providers::proxy_provider::ThreadSafeProxyProvider;
 use crate::config::internal::proxy::PROXY_GLOBAL;
-use crate::config::internal::proxy::{OutboundProxyProvider, PROXY_DIRECT, PROXY_REJECT};
+use crate::config::internal::proxy::{OutboundProxyProviderDef, PROXY_DIRECT, PROXY_REJECT};
 use crate::proxy::fallback;
 use crate::proxy::loadbalance;
 use crate::proxy::selector;
@@ -52,7 +54,7 @@ impl OutboundManager {
     pub async fn new(
         outbounds: Vec<OutboundProxyProtocol>,
         outbound_groups: Vec<OutboundGroupProtocol>,
-        proxy_providers: HashMap<String, OutboundProxyProvider>,
+        proxy_providers: HashMap<String, OutboundProxyProviderDef>,
         proxy_names: Vec<String>,
         dns_resolver: ThreadSafeDNSResolver,
         cache_store: ThreadSafeCacheFile,
@@ -548,14 +550,14 @@ impl OutboundManager {
     }
 
     async fn load_proxy_providers(
-        proxy_providers: HashMap<String, OutboundProxyProvider>,
+        proxy_providers: HashMap<String, OutboundProxyProviderDef>,
         proxy_manager: ProxyManager,
         resolver: ThreadSafeDNSResolver,
         provider_registry: &mut HashMap<String, ThreadSafeProxyProvider>,
     ) -> Result<(), Error> {
         for (name, provider) in proxy_providers.into_iter() {
             match provider {
-                OutboundProxyProvider::Http(http) => {
+                OutboundProxyProviderDef::Http(http) => {
                     let vehicle = http_vehicle::Vehicle::new(
                         http.url
                             .parse::<Uri>()
@@ -581,7 +583,7 @@ impl OutboundManager {
 
                     provider_registry.insert(name, Arc::new(RwLock::new(provider)));
                 }
-                OutboundProxyProvider::File(file) => {
+                OutboundProxyProviderDef::File(file) => {
                     let vehicle = file_vehicle::Vehicle::new(&file.path);
                     let hc = HealthCheck::new(
                         vec![],
@@ -607,7 +609,16 @@ impl OutboundManager {
 
         for p in provider_registry.values() {
             info!("initializing provider {}", p.read().await.name());
-            p.write().await.initialize().await?;
+            match p.write().await.initialize().await {
+                Ok(_) => {}
+                Err(err) => {
+                    error!(
+                        "failed to initialize proxy provider {}: {}",
+                        p.read().await.name(),
+                        err
+                    );
+                }
+            }
         }
 
         Ok(())

clash_lib/src/app/proxy_manager/mod.rs → clash_lib/src/app/remote_content_manager/mod.rs

@@ -10,7 +10,7 @@ use std::{
 use boring::ssl::{SslConnector, SslMethod};
 
 use chrono::{DateTime, Utc};
-use futures::StreamExt;
+
 use http::{Request, Version};
 use hyper_boring::HttpsConnector;
 use serde::Serialize;
@@ -246,11 +246,10 @@ impl ProxyManager {
 mod tests {
     use std::{net::Ipv4Addr, sync::Arc, time::Duration};
 
-    use anyhow::Chain;
     use futures::TryFutureExt;
 
     use crate::{
-        app::{dispatcher::ChainedStreamWrapper, dns::MockClashResolver},
+        app::{dispatcher::ChainedStreamWrapper, dns::MockClashResolver, remote_content_manager},
         config::internal::proxy::PROXY_DIRECT,
         proxy::mocks::MockDummyOutboundHandler,
     };
@@ -262,7 +261,7 @@ mod tests {
             .expect_resolve()
             .returning(|_, _| Ok(Some(std::net::IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)))));
 
-        let manager = super::ProxyManager::new(Arc::new(mock_resolver));
+        let manager = remote_content_manager::ProxyManager::new(Arc::new(mock_resolver));
 
         let mut mock_handler = MockDummyOutboundHandler::new();
         mock_handler
@@ -317,7 +316,7 @@ mod tests {
             .expect_resolve()
             .returning(|_, _| Ok(Some(std::net::IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)))));
 
-        let manager = super::ProxyManager::new(Arc::new(mock_resolver));
+        let manager = remote_content_manager::ProxyManager::new(Arc::new(mock_resolver));
 
         let mut mock_handler = MockDummyOutboundHandler::new();
         mock_handler

clash_lib/src/app/proxy_manager/providers/fetcher.rs → clash_lib/src/app/remote_content_manager/providers/fetcher.rs

@@ -7,10 +7,8 @@ use std::{
 
 use chrono::{DateTime, Utc};
 use futures::future::BoxFuture;
-use tokio::{
-    sync::{Mutex, RwLock},
-    time::Instant,
-};
+use serde::de;
+use tokio::sync::{Mutex, RwLock};
 use tracing::{debug, info, trace, warn};
 
 use crate::common::utils;
@@ -20,14 +18,15 @@ use super::{ProviderVehicleType, ThreadSafeProviderVehicle};
 struct Inner {
     updated_at: SystemTime,
     hash: [u8; 16],
+
+    thread_handle: Option<tokio::task::JoinHandle<()>>,
 }
 
 pub struct Fetcher<U, P> {
     name: String,
     interval: Duration,
     vehicle: ThreadSafeProviderVehicle,
-    thread_handle: Option<tokio::task::JoinHandle<()>>,
-    ticker: Option<tokio::time::Interval>,
+    ticker_interval: Duration,
     inner: std::sync::Arc<tokio::sync::RwLock<Inner>>,
     parser: Arc<Mutex<P>>,
     pub on_update: Option<Arc<Mutex<U>>>,
@@ -50,17 +49,11 @@ where
             name,
             interval,
             vehicle,
-            thread_handle: None,
-            ticker: match interval.as_secs() {
-                0 => None,
-                _ => Some(tokio::time::interval_at(
-                    Instant::now() + interval,
-                    interval,
-                )),
-            },
+            ticker_interval: interval,
             inner: Arc::new(tokio::sync::RwLock::new(Inner {
                 updated_at: SystemTime::UNIX_EPOCH,
                 hash: [0; 16],
+                thread_handle: None,
             })),
             parser: Arc::new(Mutex::new(parser)),
             on_update: on_update.map(|f| Arc::new(Mutex::new(f))),
@@ -78,7 +71,7 @@ where
         self.inner.read().await.updated_at.into()
     }
 
-    pub async fn initial(&mut self) -> anyhow::Result<T> {
+    pub async fn initial(&self) -> anyhow::Result<T> {
         let mut is_local = false;
         let mut immediately_update = false;
 
@@ -100,14 +93,16 @@ where
             Err(_) => self.vehicle.read().await?,
         };
 
-        let proxies = match (self.parser.lock().await)(&content) {
+        let parser_guard = self.parser.lock().await;
+
+        let proxies = match (parser_guard)(&content) {
             Ok(proxies) => proxies,
             Err(e) => {
                 if !is_local {
                     return Err(e);
                 }
                 let content = self.vehicle.read().await?;
-                (self.parser.lock().await)(&content)?
+                (parser_guard)(&content)?
             }
         };
 
@@ -127,8 +122,12 @@ where
 
         drop(inner);
 
-        if let Some(ticker) = self.ticker.take() {
-            self.pull_loop(immediately_update, ticker);
+        if !self.ticker_interval.is_zero() {
+            self.pull_loop(
+                immediately_update,
+                tokio::time::interval(self.ticker_interval),
+            )
+            .await;
         }
 
         Ok(proxies)
@@ -180,21 +179,21 @@ where
         Ok((proxies, false))
     }
 
-    pub fn destroy(&mut self) {
-        if let Some(handle) = self.thread_handle.take() {
+    pub async fn destroy(&mut self) {
+        if let Some(handle) = self.inner.write().await.thread_handle.take() {
             handle.abort();
         }
     }
 
-    fn pull_loop(&mut self, immediately_update: bool, mut ticker: tokio::time::Interval) {
+    async fn pull_loop(&self, immediately_update: bool, mut ticker: tokio::time::Interval) {
         let inner = self.inner.clone();
         let vehicle = self.vehicle.clone();
         let parser = self.parser.clone();
         let on_update = self.on_update.clone();
         let name = self.name.clone();
         let fire_immediately = immediately_update;
 
-        self.thread_handle = Some(tokio::spawn(async move {
+        let thread_handle = Some(tokio::spawn(async move {
             debug!("fetcher {} started", &name);
             loop {
                 let inner = inner.clone();
@@ -232,6 +231,8 @@ where
                 }
             }
         }));
+
+        self.inner.write().await.thread_handle = thread_handle;
     }
 }
 
@@ -242,7 +243,7 @@ mod tests {
     use futures::future::BoxFuture;
     use tokio::time::sleep;
 
-    use crate::app::proxy_manager::providers::{MockProviderVehicle, ProviderVehicleType};
+    use crate::app::remote_content_manager::providers::{MockProviderVehicle, ProviderVehicleType};
 
     use super::Fetcher;
 

clash_lib/src/app/proxy_manager/providers/http_vehicle.rs → clash_lib/src/app/remote_content_manager/providers/http_vehicle.rs

@@ -51,7 +51,7 @@ impl ProviderVehicle for Vehicle {
     }
 
     fn typ(&self) -> ProviderVehicleType {
-        ProviderVehicleType::HTTP
+        ProviderVehicleType::Http
     }
 }