500 lines
20 KiB
Rust
500 lines
20 KiB
Rust
//! Transport-neutral MTP framing.
|
|
//!
|
|
//! These types are used by non-wtransport backends. The established
|
|
//! [`crate::Sender`] and [`crate::Receiver`] remain source compatible native
|
|
//! wrappers while the framing implementation below is shared by adapters.
|
|
|
|
use crate::{
|
|
Policy, TransportConnection, TransportRecvStream, TransportSendStream, framing::write_frame,
|
|
};
|
|
use mtp_codec::CommunicationValue;
|
|
use mtp_common::CommunicationError;
|
|
use std::sync::Arc;
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use tokio::sync::{Mutex, RwLock, Semaphore, mpsc};
|
|
use tokio::time::timeout;
|
|
|
|
#[cfg(feature = "pipes")]
|
|
use crate::pipe::{PipeReader, PipeWriter};
|
|
|
|
pub struct GenericSender<C: TransportConnection> {
|
|
connection: C,
|
|
policy: Arc<Policy>,
|
|
persistent: Arc<Mutex<Option<C::SendStream>>>,
|
|
send_lock: Arc<Mutex<()>>,
|
|
}
|
|
|
|
impl<C: TransportConnection> Clone for GenericSender<C> {
|
|
fn clone(&self) -> Self {
|
|
Self {
|
|
connection: self.connection.clone(),
|
|
policy: self.policy.clone(),
|
|
persistent: self.persistent.clone(),
|
|
send_lock: self.send_lock.clone(),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<C: TransportConnection> GenericSender<C> {
|
|
pub fn new(connection: C, policy: Arc<Policy>) -> Self {
|
|
Self {
|
|
connection,
|
|
policy,
|
|
persistent: Arc::new(Mutex::new(None)),
|
|
send_lock: Arc::new(Mutex::new(())),
|
|
}
|
|
}
|
|
|
|
async fn open(&self) -> Result<C::SendStream, CommunicationError> {
|
|
timeout(self.policy.open_stream_timeout, self.connection.open_uni())
|
|
.await
|
|
.map_err(|_| CommunicationError::StreamError)?
|
|
}
|
|
|
|
pub async fn send(&self, value: &CommunicationValue) -> Result<(), CommunicationError> {
|
|
let _lock = self.send_lock.lock().await;
|
|
if self.connection.close_reason().is_some() {
|
|
return Err(CommunicationError::StreamClosed);
|
|
}
|
|
match self.policy.send_mode {
|
|
crate::SendMode::SingleStreamPerMessage => {
|
|
let mut stream = self.open().await?;
|
|
timeout(
|
|
self.policy.write_timeout,
|
|
write_frame(&mut stream, value, &self.policy),
|
|
)
|
|
.await
|
|
.map_err(|_| CommunicationError::StreamError)??;
|
|
timeout(self.policy.write_timeout, stream.finish())
|
|
.await
|
|
.map_err(|_| CommunicationError::StreamError)?
|
|
}
|
|
crate::SendMode::PersistentStream => {
|
|
let mut stream = self.persistent.lock().await;
|
|
let mut attempts = 0;
|
|
loop {
|
|
if stream.is_none() {
|
|
*stream = Some(self.open().await?);
|
|
}
|
|
let result = timeout(
|
|
self.policy.write_timeout,
|
|
write_frame(stream.as_mut().unwrap(), value, &self.policy),
|
|
)
|
|
.await
|
|
.map_err(|_| CommunicationError::StreamError)
|
|
.and_then(|r| r);
|
|
if result.is_ok() {
|
|
return result;
|
|
}
|
|
*stream = None;
|
|
attempts += 1;
|
|
if attempts > self.policy.persistent_stream_max_retries {
|
|
return result;
|
|
}
|
|
tokio::time::sleep(
|
|
self.policy.persistent_stream_retry_backoff * attempts as u32,
|
|
)
|
|
.await;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(feature = "pipes")]
|
|
pub async fn open_pipe(
|
|
&self,
|
|
pipe_id: u32,
|
|
description: &str,
|
|
) -> Result<PipeWriter<C::SendStream>, CommunicationError> {
|
|
if self.connection.close_reason().is_some() {
|
|
return Err(CommunicationError::StreamClosed);
|
|
}
|
|
|
|
let mut stream = self.open().await?;
|
|
|
|
let request = CommunicationValue::new(mtp_codec::CommunicationType::PipeRequest)
|
|
.with_id(pipe_id)
|
|
.add_typed_default(
|
|
mtp_codec::DataType::Description,
|
|
mtp_codec::DataValue::Str(description.to_string()),
|
|
);
|
|
|
|
timeout(
|
|
self.policy.write_timeout,
|
|
write_frame(&mut stream, &request, &self.policy),
|
|
)
|
|
.await
|
|
.map_err(|_| CommunicationError::StreamError)??;
|
|
|
|
Ok(PipeWriter { stream })
|
|
}
|
|
|
|
pub fn close(&self) {
|
|
self.connection
|
|
.close(self.policy.application_close_code, b"mtp-close");
|
|
}
|
|
|
|
/// Finish the current persistent stream.
|
|
///
|
|
/// This is used by hosts that put the opening/authentication exchange on
|
|
/// a persistent stream and then transition to application streams.
|
|
pub async fn finish_stream(&self) -> Result<(), CommunicationError> {
|
|
let _lock = self.send_lock.lock().await;
|
|
let mut stream = self.persistent.lock().await;
|
|
let Some(mut stream) = stream.take() else {
|
|
return Ok(());
|
|
};
|
|
timeout(self.policy.write_timeout, stream.finish())
|
|
.await
|
|
.map_err(|_| CommunicationError::StreamError)?
|
|
}
|
|
pub fn is_closed(&self) -> bool {
|
|
self.connection.close_reason().is_some()
|
|
}
|
|
pub fn is_open(&self) -> bool {
|
|
!self.is_closed()
|
|
}
|
|
pub fn close_reason(&self) -> Option<CommunicationError> {
|
|
self.connection.close_reason()
|
|
}
|
|
}
|
|
|
|
pub struct GenericReceiver<C: TransportConnection> {
|
|
incoming: Arc<Mutex<mpsc::Receiver<Result<CommunicationValue, CommunicationError>>>>,
|
|
#[cfg(feature = "pipes")]
|
|
pipes: Arc<Mutex<mpsc::Receiver<PipeReader<C::RecvStream>>>>,
|
|
connection: C,
|
|
ping_sender: Arc<RwLock<Option<GenericSender<C>>>>,
|
|
max_message_size: Arc<AtomicU64>,
|
|
_accept_task: Arc<tokio::task::JoinHandle<()>>,
|
|
}
|
|
|
|
impl<C: TransportConnection> Clone for GenericReceiver<C> {
|
|
fn clone(&self) -> Self {
|
|
Self {
|
|
incoming: self.incoming.clone(),
|
|
#[cfg(feature = "pipes")]
|
|
pipes: self.pipes.clone(),
|
|
connection: self.connection.clone(),
|
|
ping_sender: self.ping_sender.clone(),
|
|
max_message_size: self.max_message_size.clone(),
|
|
_accept_task: self._accept_task.clone(),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<C: TransportConnection> Drop for GenericReceiver<C> {
|
|
fn drop(&mut self) {
|
|
if Arc::strong_count(&self._accept_task) == 1 {
|
|
self._accept_task.abort();
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<C: TransportConnection> GenericReceiver<C> {
|
|
pub fn new(connection: C, policy: Arc<Policy>) -> Self {
|
|
let (tx, rx) = mpsc::channel(policy.receiver_queue_capacity);
|
|
#[cfg(feature = "pipes")]
|
|
let (pipe_tx, pipe_rx) = mpsc::channel(policy.receiver_queue_capacity);
|
|
let ping_sender: Arc<RwLock<Option<GenericSender<C>>>> = Arc::new(RwLock::new(None));
|
|
let max_message_size = Arc::new(AtomicU64::new(
|
|
policy
|
|
.handshake_max_message_size
|
|
.min(policy.max_message_size),
|
|
));
|
|
let task_ping_sender = ping_sender.clone();
|
|
let task_connection = connection.clone();
|
|
let task_policy = policy.clone();
|
|
let task_max_message_size = max_message_size.clone();
|
|
let task_accept_task_tx = tx.clone();
|
|
#[cfg(feature = "pipes")]
|
|
let task_accept_task_pipe_tx = pipe_tx.clone();
|
|
let accept_task = tokio::spawn(async move {
|
|
let limit = Arc::new(Semaphore::new(
|
|
task_policy.max_concurrent_stream_tasks.max(1),
|
|
));
|
|
loop {
|
|
// Backpressure: stop accepting new streams if the output queue is full.
|
|
#[cfg(feature = "pipes")]
|
|
let cap_full =
|
|
task_accept_task_tx.capacity() == 0 || task_accept_task_pipe_tx.capacity() == 0;
|
|
#[cfg(not(feature = "pipes"))]
|
|
let cap_full = task_accept_task_tx.capacity() == 0;
|
|
|
|
if cap_full {
|
|
tokio::time::sleep(std::time::Duration::from_millis(1)).await;
|
|
continue;
|
|
}
|
|
|
|
let stream = match tokio::time::timeout(
|
|
task_policy.accept_stream_timeout,
|
|
task_connection.accept_uni(),
|
|
)
|
|
.await
|
|
{
|
|
Ok(Ok(stream)) => stream,
|
|
Ok(Err(error)) => {
|
|
let _ = task_accept_task_tx.send(Err(error)).await;
|
|
break;
|
|
}
|
|
Err(_) => {
|
|
if task_connection.close_reason().is_some() {
|
|
let _ = task_accept_task_tx
|
|
.send(Err(CommunicationError::StreamClosed))
|
|
.await;
|
|
break;
|
|
} else {
|
|
continue;
|
|
}
|
|
}
|
|
};
|
|
// Acquire semaphore permit BEFORE spawning the task.
|
|
let permit = match limit.clone().acquire_owned().await {
|
|
Ok(permit) => permit,
|
|
Err(_) => break,
|
|
};
|
|
let tx = task_accept_task_tx.clone();
|
|
#[cfg(feature = "pipes")]
|
|
let pipe_tx = task_accept_task_pipe_tx.clone();
|
|
let policy = task_policy.clone();
|
|
let max_message_size = task_max_message_size.clone();
|
|
let ping_sender = task_ping_sender.clone();
|
|
let connection = task_connection.clone();
|
|
tokio::spawn(async move {
|
|
let _permit = permit;
|
|
let mut stream = stream;
|
|
let mut frames = 0usize;
|
|
loop {
|
|
if policy
|
|
.max_frames_per_stream
|
|
.is_some_and(|max| frames >= max)
|
|
{
|
|
let close_error = CommunicationError::StreamError;
|
|
let _ = tx.send(Err(close_error.clone())).await;
|
|
connection.close(policy.application_close_code, b"max frames exceeded");
|
|
break;
|
|
}
|
|
let mut len = [0; 4];
|
|
match tokio::time::timeout(policy.read_timeout, stream.read_exact(&mut len))
|
|
.await
|
|
{
|
|
Ok(Ok(())) => {}
|
|
Ok(Err(CommunicationError::StreamClosed)) => break,
|
|
Ok(Err(error)) => {
|
|
tracing::warn!(%error, "MTP receive stream failed while reading frame header");
|
|
let _ = tx.send(Err(error)).await;
|
|
connection.close(
|
|
policy.application_close_code,
|
|
b"frame header read error",
|
|
);
|
|
break;
|
|
}
|
|
Err(_) => {
|
|
break;
|
|
}
|
|
}
|
|
let len = u32::from_be_bytes(len);
|
|
if len == policy.close_frame_len {
|
|
break;
|
|
}
|
|
let frame_limit = max_message_size.load(Ordering::Relaxed);
|
|
if len as u64 > frame_limit {
|
|
tracing::warn!(len, "MTP receive stream frame is too large");
|
|
let _ = tx.send(Err(CommunicationError::MessageTooLarge)).await;
|
|
connection.close(policy.application_close_code, b"frame too large");
|
|
break;
|
|
}
|
|
let target_len = len as usize;
|
|
let mut body = Vec::new();
|
|
if body.try_reserve(target_len.min(16 * 1024)).is_err() {
|
|
tracing::warn!(
|
|
target_len,
|
|
"MTP receive stream could not reserve frame body"
|
|
);
|
|
let _ = tx.send(Err(CommunicationError::MessageTooLarge)).await;
|
|
connection
|
|
.close(policy.application_close_code, b"frame allocation failed");
|
|
break;
|
|
}
|
|
while body.len() < target_len {
|
|
let chunk_len = (target_len - body.len()).min(16 * 1024);
|
|
let mut chunk = [0u8; 16 * 1024];
|
|
let body_read = tokio::time::timeout(
|
|
policy.read_timeout,
|
|
stream.read_exact(&mut chunk[..chunk_len]),
|
|
)
|
|
.await;
|
|
if !matches!(&body_read, Ok(Ok(())))
|
|
|| body.try_reserve(chunk_len).is_err()
|
|
{
|
|
tracing::warn!(
|
|
pipe_chunk_len = chunk_len,
|
|
?body_read,
|
|
"MTP receive stream failed while reading frame body"
|
|
);
|
|
let _ = tx.send(Err(CommunicationError::StreamError)).await;
|
|
connection
|
|
.close(policy.application_close_code, b"frame body read error");
|
|
break;
|
|
}
|
|
body.extend_from_slice(&chunk[..chunk_len]);
|
|
}
|
|
if body.len() != target_len {
|
|
break;
|
|
}
|
|
frames += 1;
|
|
let message = match CommunicationValue::from_bytes(&body) {
|
|
Ok(message) => message,
|
|
Err(_) => {
|
|
tracing::warn!("MTP receive stream contained an invalid frame");
|
|
let _ = tx
|
|
.send(Err(CommunicationError::ParseCommunicationValue))
|
|
.await;
|
|
connection.close(policy.application_close_code, b"invalid frame");
|
|
break;
|
|
}
|
|
};
|
|
|
|
#[cfg(feature = "pipes")]
|
|
{
|
|
let pipe_request_type = mtp_codec::CommunicationType::PipeRequest
|
|
.try_to_id(&mtp_codec::TypeMap::latest());
|
|
if Some(message.get_type()) == pipe_request_type && frames == 1 {
|
|
let pipe_id = message.get_id();
|
|
let description = message
|
|
.get_str(mtp_codec::DataType::Description)
|
|
.unwrap_or("")
|
|
.to_string();
|
|
|
|
let pipe_reader = PipeReader {
|
|
stream,
|
|
description,
|
|
pipe_id,
|
|
};
|
|
|
|
tracing::debug!(pipe_id, description = %pipe_reader.description, "classified incoming pipe stream");
|
|
|
|
if pipe_tx.send(pipe_reader).await.is_err() {
|
|
break;
|
|
}
|
|
return;
|
|
}
|
|
}
|
|
|
|
if message.is_type(mtp_codec::CommunicationType::Ping) {
|
|
if let Some(sender) = ping_sender.read().await.clone() {
|
|
let mut pong =
|
|
CommunicationValue::new(mtp_codec::CommunicationType::Pong)
|
|
.with_id(message.get_id());
|
|
if let Some(timestamp) =
|
|
message.get_data_opt(mtp_codec::DataType::Timestamp)
|
|
{
|
|
pong = pong.add_typed_default(
|
|
mtp_codec::DataType::Timestamp,
|
|
timestamp.clone(),
|
|
);
|
|
}
|
|
let _ = sender.send(&pong).await;
|
|
}
|
|
continue;
|
|
}
|
|
if tx.send(Ok(message)).await.is_err() {
|
|
break;
|
|
}
|
|
}
|
|
});
|
|
}
|
|
});
|
|
Self {
|
|
incoming: Arc::new(Mutex::new(rx)),
|
|
#[cfg(feature = "pipes")]
|
|
pipes: Arc::new(Mutex::new(pipe_rx)),
|
|
connection,
|
|
ping_sender,
|
|
max_message_size,
|
|
_accept_task: Arc::new(accept_task),
|
|
}
|
|
}
|
|
pub async fn respond_to_pings(&self, sender: GenericSender<C>) {
|
|
*self.ping_sender.write().await = Some(sender);
|
|
}
|
|
|
|
/// Switch from the handshake frame limit to the application frame limit.
|
|
pub fn set_max_message_size(&self, max_message_size: u64) {
|
|
self.max_message_size
|
|
.store(max_message_size, Ordering::Relaxed);
|
|
}
|
|
pub async fn receive(&self) -> Result<CommunicationValue, CommunicationError> {
|
|
self.incoming
|
|
.lock()
|
|
.await
|
|
.recv()
|
|
.await
|
|
.unwrap_or(Err(CommunicationError::StreamClosed))
|
|
}
|
|
|
|
#[cfg(feature = "pipes")]
|
|
pub async fn receive_event(
|
|
&self,
|
|
) -> Result<crate::TransportEvent<C::RecvStream>, CommunicationError> {
|
|
let mut incoming = self.incoming.lock().await;
|
|
let mut pipes = self.pipes.lock().await;
|
|
tokio::select! {
|
|
msg = incoming.recv() => {
|
|
match msg {
|
|
Some(Ok(val)) => Ok(crate::TransportEvent::Message(val)),
|
|
Some(Err(e)) => Err(e),
|
|
None => Err(self
|
|
.connection
|
|
.close_reason()
|
|
.unwrap_or(CommunicationError::StreamClosed)),
|
|
}
|
|
}
|
|
pipe = pipes.recv() => {
|
|
match pipe {
|
|
Some(reader) => Ok(crate::TransportEvent::Pipe(reader)),
|
|
None => Err(self
|
|
.connection
|
|
.close_reason()
|
|
.unwrap_or(CommunicationError::StreamClosed)),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(feature = "pipes")]
|
|
pub async fn receive_pipe(&self) -> Result<PipeReader<C::RecvStream>, CommunicationError> {
|
|
self.pipes
|
|
.lock()
|
|
.await
|
|
.recv()
|
|
.await
|
|
.ok_or(CommunicationError::StreamClosed)
|
|
}
|
|
|
|
#[cfg(feature = "pipes")]
|
|
pub fn try_receive_pipe(
|
|
&self,
|
|
) -> Result<Option<PipeReader<C::RecvStream>>, CommunicationError> {
|
|
match self.pipes.try_lock() {
|
|
Ok(mut rx) => match rx.try_recv() {
|
|
Ok(reader) => Ok(Some(reader)),
|
|
Err(mpsc::error::TryRecvError::Empty) => Ok(None),
|
|
Err(mpsc::error::TryRecvError::Disconnected) => {
|
|
Err(CommunicationError::StreamClosed)
|
|
}
|
|
},
|
|
Err(_) => Ok(None),
|
|
}
|
|
}
|
|
|
|
pub fn is_closed(&self) -> bool {
|
|
self.connection.close_reason().is_some()
|
|
}
|
|
pub fn is_open(&self) -> bool {
|
|
!self.is_closed()
|
|
}
|
|
pub fn close_reason(&self) -> Option<CommunicationError> {
|
|
self.connection.close_reason()
|
|
}
|
|
}
|