nadisc_runtime/
transport.rs

1//! 调用层向共同 owner 登记连接与正文任务，使无人读取的流也服从停机和期限。
2
3use crate::{closed, lock, spawn_owned, Core, DiscoveryHandle};
4use nadisc::managed::{DiscoveryError, ErrorCode, Result};
5use std::{
6    future::Future,
7    sync::{Arc, Weak},
8};
9use tokio::{sync::OwnedSemaphorePermit, task::AbortHandle, time::Instant};
10use tokio_util::sync::CancellationToken;
11
12/// 一次传输任务的容量许可，绑定 owner 的关闭通知与任务归属。
13pub struct TransportLease {
14    core: Weak<Core>,
15    stop: CancellationToken,
16    permit: OwnedSemaphorePermit,
17}
18
19impl DiscoveryHandle {
20    /// 业务作用：在打开传输前预留 owner 级有限容量，关闭和容量不足时不启动连接。
21    ///
22    /// 参数说明：无。
23    ///
24    /// 返回：一次连接或正文任务的额度；释放后可由其他调用复用。
25    pub fn reserve_transport(&self) -> Result<TransportLease> {
26        let core = self.core.upgrade().ok_or_else(closed)?;
27        let gate = lock(&core.gate);
28        if gate.failed || gate.close_started {
29            return Err(closed());
30        }
31        let permit = core.transports.clone().try_acquire_owned().map_err(|_| {
32            DiscoveryError::new(
33                ErrorCode::CapacityExceeded,
34                "owner transport task capacity reached",
35            )
36        })?;
37        Ok(TransportLease {
38            core: Arc::downgrade(&core),
39            stop: core.stop.child_token(),
40            permit,
41        })
42    }
43}
44
45impl TransportLease {
46    /// 业务作用：让连接、排队和流消费等待响应同一 owner 的关闭。
47    ///
48    /// 参数说明：无。
49    ///
50    /// 返回：owner 最终关闭后完成，不授予调用方关闭 owner 的权限。
51    pub async fn cancelled(&self) {
52        self.stop.cancelled().await;
53    }
54
55    /// 业务作用：将有限传输任务交给 owner 接管，等待者丢弃不造成后台任务失去归属。
56    ///
57    /// 参数说明：
58    /// - `deadline`：是绝对期限
59    /// - `future`：是不超过该期限的连接或正文处理。
60    ///
61    /// 返回：只控制本任务的终止句柄；关闭时禁止启动，任务结束自动归还容量。
62    pub fn spawn(
63        self,
64        deadline: Instant,
65        future: impl Future<Output = ()> + Send + 'static,
66    ) -> Result<AbortHandle> {
67        let core = self.core.upgrade().ok_or_else(closed)?;
68        let gate = lock(&core.gate);
69        if gate.failed || gate.close_started {
70            return Err(closed());
71        }
72        if deadline <= Instant::now() {
73            return Err(crate::timeout());
74        }
75        Ok(spawn_owned(&core, async move {
76            let _permit = self.permit;
77            tokio::select! {
78                biased;
79                _ = self.stop.cancelled() => {},
80                _ = tokio::time::sleep_until(deadline) => {},
81                _ = future => {},
82            }
83        }))
84    }
85}
nadisc_runtime/transport.rs

nadisc_runtime/
transport.rs
