reala_subscription/
lib.rs

1#![doc = include_str!("../README.md")]
2
3mod model;
4pub mod util;
5
6use std::sync::{Arc, LazyLock};
7use std::time::Duration;
8
9use tokio::sync::mpsc;
10use tokio::task::JoinHandle;
11use ureq::Agent;
12
13pub use self::model::Proxy;
14
15#[non_exhaustive]
16#[derive(Debug)]
17/// Errors that may occur when fetching or parsing subscription content.
18pub enum Error {
19    /// Failed to fetch the subscription content.
20    Fetch(ureq::Error),
21
22    /// Failed to parse the subscription content.
23    Parse,
24}
25
26impl std::error::Error for Error {
27    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
28        match self {
29            Error::Fetch(e) => Some(e),
30            Error::Parse => None,
31        }
32    }
33}
34
35impl std::fmt::Display for Error {
36    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37        match self {
38            Error::Fetch(e) => write!(f, "Failed to fetch subscription content: {}", e),
39            Error::Parse => write!(f, "Failed to parse subscription content"),
40        }
41    }
42}
43
44impl From<ureq::Error> for Error {
45    fn from(e: ureq::Error) -> Self {
46        Error::Fetch(e)
47    }
48}
49
50#[non_exhaustive]
51#[derive(Debug, Clone, Copy)]
52#[derive(serde::Serialize, serde::Deserialize)]
53/// The type of subscription.
54///
55/// Currently, only Clash-compatible subscriptions are supported.
56pub enum SubscriptionKind {
57    #[serde(rename = "clash")]
58    /// A Clash-compatible subscription, i.e., the content is in Clash config
59    /// format.
60    ClashCompatible,
61}
62
63#[derive(Debug, Clone)]
64#[derive(serde::Serialize, serde::Deserialize)]
65/// Options for fetching subscription content.
66pub struct Subscription {
67    /// The type of subscription.
68    pub kind: SubscriptionKind,
69
70    /// The target URL.
71    pub url: util::Uri,
72
73    #[serde(skip_serializing_if = "Option::is_none")]
74    /// The User-Agent.
75    ///
76    /// By default, `mihomo.party/v1.8.4 (clash.meta)` is used.
77    pub user_agent: Option<Arc<str>>,
78
79    #[serde(skip_serializing_if = "Option::is_none")]
80    /// The HTTP / Socks5 proxy you may use to fetch the subscription.
81    ///
82    /// If `None`, no proxy is used.
83    pub proxy: Option<util::Proxy>,
84}
85
86impl Subscription {
87    /// Fetches the subscription content, and returns the parsed proxies.
88    ///
89    /// Notes that the proxies may be empty if no compatible proxies are found.
90    pub fn fetch(&self) -> Result<Vec<Proxy>, Error> {
91        static AGENT: LazyLock<Agent> = LazyLock::new(|| {
92            const DEFAULT_USER_AGENT: &str = "mihomo.party/v1.8.4 (clash.meta)";
93
94            Agent::config_builder()
95                .timeout_global(Some(Duration::from_secs(5)))
96                .user_agent(DEFAULT_USER_AGENT)
97                .build()
98                .into()
99        });
100
101        let mut request = AGENT
102            .get(&*self.url)
103            .config()
104            .proxy(self.proxy.as_deref().cloned());
105
106        if let Some(user_agent) = &self.user_agent {
107            request = request.user_agent(user_agent);
108        }
109
110        let response = request.build().call()?.into_body().read_to_vec()?;
111
112        let response = String::from_utf8(response).map_err(|_e| Error::Parse)?;
113
114        match self.kind {
115            SubscriptionKind::ClashCompatible => Proxy::parse_clash_compatible(&response),
116        }
117    }
118}
119
120wrapper_lite::wrapper!(
121    #[wrapper_impl(DebugName)]
122    /// A background subscription update task.
123    pub struct Task {
124        /// The subscription info.
125        pending: Subscription,
126
127        /// The update interval.
128        ///
129        /// The default is 30 minutes.
130        interval: Option<Duration>,
131
132        /// The channel TX we use to send back the parsed proxies.
133        proxies_tx: mpsc::Sender<Vec<Proxy>>,
134    }
135);
136
137impl Task {
138    /// Creates a new subscription update task.
139    pub fn new(
140        pending: Subscription,
141        interval: Option<Duration>,
142    ) -> (Self, mpsc::Receiver<Vec<Proxy>>) {
143        let (proxies_tx, proxies_rx) = mpsc::channel(1);
144
145        (
146            Self {
147                pending,
148                interval,
149                proxies_tx,
150            },
151            proxies_rx,
152        )
153    }
154
155    /// Spawns the subscription update task.
156    pub async fn spawn(self) -> JoinHandle<()> {
157        tokio::spawn(async move {
158            loop {
159                let pending = self.pending.clone();
160
161                let proxies = tokio::task::spawn_blocking(move || pending.fetch()).await;
162
163                match proxies {
164                    Ok(Ok(proxies)) => {
165                        let Ok(()) = self.proxies_tx.send(proxies).await else {
166                            tracing::info!("Subscription task channel closed, exiting task.");
167
168                            return;
169                        };
170                    }
171                    Ok(Err(e)) => {
172                        tracing::error!("Failed to fetch subscription: {:?}", e);
173                    }
174                    Err(e) => {
175                        tracing::error!("Failed to join subscription task: {:?}", e);
176                    }
177                }
178
179                tokio::time::sleep(self.interval.unwrap_or(Duration::from_mins(30))).await;
180            }
181        })
182    }
183}
reala_subscription/lib.rs

reala_subscription/
lib.rs
