td_client/
runtime.rs

1//! Process-wide native execution, receive tuning, and diagnostics.
2//!
3//! These settings affect every client in this process. Configure them at
4//! application startup rather than treating them as per-client preferences.
5//! One lazily started receiver thread owns the `TDLib` receive stream and parks
6//! when no clients remain. Do not start a competing receiver or invoke raw
7//! receive functions outside this coordination.
8//!
9//! [`execute`] is distinct from asynchronous [`Client::send`](crate::client::Client::send):
10//! only `TDLib` functions documented as synchronously executable belong here.
11//!
12//! [`on_error`] receives malformed/unroutable unsolicited output. It is not a
13//! subscription to `TDLib`'s internal logging stream; [`set_log_level`] only
14//! controls that stream's verbosity. Request errors still return to their caller.
15
16use std::collections::HashMap;
17use std::ffi::CStr;
18use std::sync::atomic::{AtomicU64, Ordering};
19use std::sync::{LazyLock, Mutex, OnceLock, Weak};
20use std::thread;
21use std::time::Duration;
22
23use serde::Deserialize;
24use tokio::sync::watch;
25
26use td_types::traits::Function;
27
28use crate::connection::Connection;
29use crate::error::{Error, Result};
30
31#[derive(Deserialize)]
32struct Response<'a> {
33  #[serde(rename = "@type")]
34  kind: &'a str,
35}
36
37#[derive(Deserialize)]
38struct Incoming<'a> {
39  #[serde(rename = "@client_id")]
40  client_id: i32,
41  #[serde(rename = "@extra")]
42  extra: Option<u64>,
43  #[serde(rename = "@type")]
44  kind: &'a str,
45}
46
47struct Receiver {
48  clients: Mutex<HashMap<i32, Weak<Connection>>>,
49  thread: OnceLock<thread::Thread>,
50  transition: watch::Sender<()>,
51  timeout: AtomicU64,
52}
53
54static RECEIVER: LazyLock<Receiver> = LazyLock::new(|| {
55  let (transition, _) = watch::channel(());
56  let (clients, thread) = Default::default();
57  let timeout = 1f64.to_bits().into();
58  Receiver { clients, thread, transition, timeout }
59});
60
61impl Receiver {
62  fn run(&self) {
63    loop {
64      if self.clients.lock().unwrap().is_empty() {
65        // Publish idle before parking. A concurrent registration leaves an unpark
66        // permit and publishes a transition, so neither side waits for lost work.
67        self.transition.send_replace(());
68        thread::park();
69      } else {
70        self.receive();
71      }
72    }
73  }
74
75  fn receive(&self) {
76    let timeout = f64::from_bits(self.timeout.load(Ordering::Relaxed));
77    // SAFETY: This is the sole td_receive caller.
78    let raw = unsafe { td_sys::td_receive(timeout) };
79    if !raw.is_null() {
80      // SAFETY: TDLib returned a NUL-terminated buffer valid until the next
81      // receive call on this receiver thread.
82      self.route(unsafe { CStr::from_ptr(raw) }.to_bytes());
83    }
84  }
85
86  fn route(&self, raw: &[u8]) {
87    let Incoming { client_id, extra, kind } = match serde_json::from_slice(raw) {
88      Ok(incoming) => incoming,
89      Err(error) => {
90        // Unsolicited native errors can omit @client_id. Keep the common path
91        // single-pass, but preserve the TDLib error instead of a missing-field error.
92        return match serde_json::from_slice(raw) {
93          Ok(Response { kind: "error" }) => report(parse_error(raw)),
94          _ => report(error),
95        };
96      }
97    };
98    if let (None, "error") = (extra, kind) {
99      return report(parse_error(raw));
100    }
101    let connection = self.clients.lock().unwrap().get(&client_id).and_then(Weak::upgrade);
102    let Some(connection) = connection else { return };
103    match extra {
104      Some(extra) => connection.complete_request(extra, kind, raw),
105      None => connection.update(raw),
106    }
107  }
108}
109
110pub(crate) fn register(id: i32, connection: Weak<Connection>) {
111  RECEIVER.clients.lock().unwrap().insert(id, connection);
112  RECEIVER.thread.get_or_init(|| thread::spawn(|| RECEIVER.run()).thread().clone()).unpark();
113  RECEIVER.transition.send_replace(());
114}
115
116pub(crate) async fn unregister(id: i32) {
117  let mut transition = RECEIVER.transition.subscribe();
118  {
119    let mut clients = RECEIVER.clients.lock().unwrap();
120    clients.remove(&id);
121    if !clients.is_empty() {
122      return;
123    }
124    // Subscribe and consume under the client lock: the next observed transition
125    // follows removal or a new registration, never an earlier idle observation.
126    transition.borrow_and_update();
127  }
128  let _ = transition.changed().await;
129}
130
131pub(crate) fn remove(id: i32) {
132  if let Some(receiver) = LazyLock::get(&RECEIVER) {
133    receiver.clients.lock().unwrap().remove(&id);
134  }
135}
136
137/// Executes a supported synchronous `TDLib` function and decodes its response.
138///
139/// Only generated functions documented by `TDLib` as synchronously executable
140/// are supported, such as `getFileMimeType`. This does not require a [`Session`](crate::session::Session).
141///
142/// This is a synchronous call directly to `td_execute`. Because `TDLib` stores
143/// output in thread-local storage, it executes immediately without acquiring locks
144/// or waiting for an in-progress [`set_receive_timeout`]. Do not call it from [`on_error`].
145///
146/// # Errors
147///
148/// Returns [`Error::Td`] for native errors, [`Error::Json`] for encoding or
149/// decoding failures, and [`Error::UnexpectedResponse`] for a null response.
150///
151/// # Examples
152///
153/// ```
154/// use td_client::execute;
155/// use td_types::{enums::Text, fns};
156///
157/// let Text::text(mime) = execute(
158///   &fns::getFileMimeType { file_name: "picture.png".into() }
159/// )?;
160/// assert_eq!(mime.text, "image/png");
161/// # Ok::<(), td_client::Error>(())
162/// ```
163pub fn execute<F: Function>(request: &F) -> Result<F::Return> {
164  let mut bytes = serde_json::to_vec(request)?;
165  bytes.push(0);
166  // SAFETY: bytes is live and NUL-terminated; td_execute writes to thread-local
167  // storage and is valid on this thread until the next native call.
168  let raw = unsafe { td_sys::td_execute(bytes.as_ptr().cast()) };
169  if raw.is_null() {
170    return Err(Error::UnexpectedResponse("synchronous request returned null"));
171  }
172  // SAFETY: TDLib returned a non-null NUL-terminated buffer valid on this thread.
173  let raw = unsafe { CStr::from_ptr(raw) }.to_bytes();
174  match serde_json::from_slice(raw)? {
175    Response { kind: "error" } => Err(parse_error(raw)),
176    _ => serde_json::from_slice(raw).map_err(Into::into),
177  }
178}
179
180/// Sets the maximum wait used by the next native receive call.
181///
182/// The process-wide default is one second. Changing it does not interrupt a
183/// receive already in progress and does not set a request or operation deadline.
184/// Long waits can increase last-client shutdown latency; very short waits
185/// increase idle polling. Zero is accepted but can busy-poll while clients are
186/// registered.
187///
188/// # Examples
189///
190/// ```no_run
191/// use std::time::Duration;
192/// use td_client::set_receive_timeout;
193///
194/// set_receive_timeout(Duration::from_millis(100));
195/// ```
196pub fn set_receive_timeout(timeout: Duration) {
197  RECEIVER.timeout.store(timeout.as_secs_f64().to_bits(), Ordering::Relaxed);
198}
199
200/// Sets `TDLib`'s process-wide native log verbosity.
201///
202/// `TDLib` documents levels 0 through 5 for progressively more verbose output,
203/// with higher levels up to 1024 enabling additional diagnostics. Its default
204/// is 5; this crate does not silently change it. Supply a level supported by
205/// the native library.
206///
207/// This controls native logging, not the optional [`on_error`] callback.
208/// Logs may contain application-sensitive information; choose verbosity and
209/// log destinations as application policy.
210pub fn set_log_level(level: i32) {
211  // SAFETY: No pointers or borrowed storage are passed.
212  unsafe { td_sys::td_set_log_verbosity_level(level) };
213}
214
215type ErrorCallback = Box<dyn Fn(Error) + Send + Sync>;
216static ERROR_CALLBACK: Mutex<Option<ErrorCallback>> = Mutex::new(None);
217
218/// Installs or replaces the process-wide unsolicited-error callback.
219///
220/// Reports malformed native output and unsolicited `TDLib` errors without a
221/// waiting request recipient, including errors with unknown client IDs.
222/// Correlated request failures are returned to their caller instead. Without
223/// a callback these diagnostics are unobserved; they do not enter unrelated
224/// clients' update queues.
225///
226/// The callback is synchronous on the native receiver thread while the
227/// callback lock is held. It must not block, panic, call `on_error` again,
228/// invoke [`execute`], or wait for work that needs the receiver. Copy/forward
229/// the error to application-owned processing when more work is needed.
230/// A panic can terminate the receiver; there is no automatic restart or poison
231/// recovery. Replacing the callback also drops the old callback under its lock.
232///
233/// This is not a native log subscription. To stop observing, replace the
234/// callback with a no-op; there is no separate unsubscribe handle.
235///
236/// # Examples
237///
238/// ```no_run
239/// use std::sync::mpsc;
240/// use td_client::on_error;
241///
242/// let (errors, receiver) = mpsc::channel();
243/// on_error(move |error| {
244///   // Unbounded send does not wait for application processing.
245///   let _ = errors.send(error);
246/// });
247/// // An application-owned worker can now consume receiver.
248/// # let _ = receiver;
249/// ```
250pub fn on_error(callback: impl Fn(Error) + Send + Sync + 'static) {
251  *ERROR_CALLBACK.lock().unwrap() = Some(Box::new(callback));
252}
253
254pub(crate) fn report(error: impl Into<Error>) {
255  if let Some(callback) = ERROR_CALLBACK.lock().unwrap().as_ref() {
256    callback(error.into());
257  }
258}
259
260pub(crate) fn parse_error(raw: &[u8]) -> Error {
261  match serde_json::from_slice(raw) {
262    Ok(error) => Error::Td(error),
263    Err(error) => error.into(),
264  }
265}
266
267#[cfg(test)]
268mod tests {
269  use std::assert_matches;
270  use std::sync::Arc;
271  use tokio::sync::mpsc;
272
273  use super::*;
274
275  #[test]
276  fn unroutable_output_reports_original_errors_without_poisoning_clients() {
277    let (connection, mut updates) = Connection::fixture();
278    let (transition, _) = watch::channel(());
279    let clients = Mutex::new(HashMap::from([(7, Arc::downgrade(&connection))]));
280    let receiver = Receiver { clients, thread: OnceLock::new(), transition, timeout: AtomicU64::new(0) };
281    let errors = Arc::new(Mutex::new(Vec::new()));
282    let observed = Arc::clone(&errors);
283    on_error(move |error| observed.lock().unwrap().push(error));
284    receiver.route(br#"{"@type":"error","code":429,"message":"limited"}"#);
285    receiver.route(br#"{"@client_id":99,"@type":"error","code":500,"message":"unroutable"}"#);
286    receiver.route(br#"{"@client_id":7,"@type":"error","code":400,"message":"unsolicited"}"#);
287    receiver.route(br#"{"@type":"ok"}"#);
288    receiver.route(br#"{"@client_id":7,"@type":"updateMessageSendSucceeded","message":null}"#);
289    on_error(drop);
290
291    let errors = errors.lock().unwrap().drain(..).collect::<Vec<_>>();
292    let [global, unknown_client, known_client, missing_client, malformed]: [Error; 5] = errors.try_into().unwrap();
293    assert_matches!(global, Error::Td(error) if error.code == 429);
294    assert_matches!(unknown_client, Error::Td(error) if error.code == 500);
295    assert_matches!(known_client, Error::Td(error) if error.code == 400);
296    assert_matches!(missing_client, Error::Json(_));
297    assert_matches!(malformed, Error::Json(_));
298    let application = updates.try_recv();
299    assert_matches!(application, Err(mpsc::error::TryRecvError::Empty));
300  }
301}
td_client/runtime.rs

td_client/
runtime.rs
