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    pub fn reserve_transport(&self) -> Result<TransportLease> {
24        let core = self.core.upgrade().ok_or_else(closed)?;
25        let gate = lock(&core.gate);
26        if gate.failed || gate.close_started {
27            return Err(closed());
28        }
29        let permit = core.transports.clone().try_acquire_owned().map_err(|_| {
30            DiscoveryError::new(
31                ErrorCode::CapacityExceeded,
32                "owner transport task capacity reached",
33            )
34        })?;
35        Ok(TransportLease {
36            core: Arc::downgrade(&core),
37            stop: core.stop.child_token(),
38            permit,
39        })
40    }
41}
42
43impl TransportLease {
44    /// 业务作用：让连接、排队和流消费等待响应同一 owner 的关闭。
45    /// 参数说明：无。
46    /// 返回：owner 最终关闭后完成，不授予调用方关闭 owner 的权限。
47    pub async fn cancelled(&self) {
48        self.stop.cancelled().await;
49    }
50
51    /// 业务作用：将有限传输任务交给 owner 接管，等待者丢弃不造成后台任务失去归属。
52    /// 参数说明：`deadline` 是绝对期限，`future` 是不超过该期限的连接或正文处理。
53    /// 返回：只控制本任务的终止句柄；关闭时禁止启动，任务结束自动归还容量。
54    pub fn spawn(
55        self,
56        deadline: Instant,
57        future: impl Future<Output = ()> + Send + 'static,
58    ) -> Result<AbortHandle> {
59        let core = self.core.upgrade().ok_or_else(closed)?;
60        let gate = lock(&core.gate);
61        if gate.failed || gate.close_started {
62            return Err(closed());
63        }
64        if deadline <= Instant::now() {
65            return Err(crate::timeout());
66        }
67        Ok(spawn_owned(&core, async move {
68            let _permit = self.permit;
69            tokio::select! {
70                biased;
71                _ = self.stop.cancelled() => {},
72                _ = tokio::time::sleep_until(deadline) => {},
73                _ = future => {},
74            }
75        }))
76    }
77}
nadisc_runtime/transport.rs

nadisc_runtime/
transport.rs
