use std::time::Duration;
use crate::transport::matrix::{
ConnectionMatrixAddError, ConnectionMatrixEnvelope, ConnectionMatrixLifeCycle,
ConnectionMatrixReceiver, ConnectionMatrixRecvError, ConnectionMatrixRecvTimeoutError,
ConnectionMatrixRemoveError, ConnectionMatrixSendError, ConnectionMatrixSender,
};
use crate::transport::Connection;
use super::{Mesh, RecvError, RecvTimeoutError};
#[derive(Clone)]
pub struct MeshLifeCycle {
mesh: Mesh,
}
impl MeshLifeCycle {
pub fn new(mesh: Mesh) -> Self {
MeshLifeCycle { mesh }
}
}
impl ConnectionMatrixLifeCycle for MeshLifeCycle {
fn add(
&self,
connection: Box<dyn Connection>,
id: String,
) -> Result<usize, ConnectionMatrixAddError> {
self.mesh.add(connection, id).map_err(|err| {
ConnectionMatrixAddError::new(
"Unable to add connection to matrix".to_string(),
Some(Box::new(err)),
)
})
}
fn remove(&self, id: &str) -> Result<Box<dyn Connection>, ConnectionMatrixRemoveError> {
self.mesh.remove(id).map_err(|err| {
ConnectionMatrixRemoveError::new(
"Unable to remove connection from matrix".to_string(),
Some(Box::new(err)),
)
})
}
}
#[derive(Clone)]
pub struct MeshMatrixSender {
mesh: Mesh,
}
impl MeshMatrixSender {
pub fn new(mesh: Mesh) -> Self {
MeshMatrixSender { mesh }
}
}
impl ConnectionMatrixSender for MeshMatrixSender {
fn send(&self, id: String, message: Vec<u8>) -> Result<(), ConnectionMatrixSendError> {
let envelope = ConnectionMatrixEnvelope::new(id, message);
self.mesh.send(envelope).map_err(|err| {
ConnectionMatrixSendError::new(
"Unable to send message to connection".to_string(),
Some(Box::new(err)),
)
})
}
}
#[derive(Clone)]
pub struct MeshMatrixReceiver {
mesh: Mesh,
}
impl MeshMatrixReceiver {
pub fn new(mesh: Mesh) -> Self {
MeshMatrixReceiver { mesh }
}
}
impl ConnectionMatrixReceiver for MeshMatrixReceiver {
fn recv(&self) -> Result<ConnectionMatrixEnvelope, ConnectionMatrixRecvError> {
match self.mesh.recv() {
Ok(envelope) => Ok(envelope),
Err(err) => match err {
RecvError::Disconnected => Err(ConnectionMatrixRecvError::Disconnected),
RecvError::PoisonedLock => Err(ConnectionMatrixRecvError::new_internal_error(
"Internal state poisoned".to_string(),
Some(Box::new(err)),
)),
RecvError::Shutdown => Err(ConnectionMatrixRecvError::Shutdown),
},
}
}
fn recv_timeout(
&self,
timeout: Duration,
) -> Result<ConnectionMatrixEnvelope, ConnectionMatrixRecvTimeoutError> {
match self.mesh.recv_timeout(timeout) {
Ok(envelope) => Ok(envelope),
Err(err) => match err {
RecvTimeoutError::Timeout => Err(ConnectionMatrixRecvTimeoutError::Timeout),
RecvTimeoutError::Disconnected => {
Err(ConnectionMatrixRecvTimeoutError::Disconnected)
}
RecvTimeoutError::PoisonedLock => {
Err(ConnectionMatrixRecvTimeoutError::new_internal_error(
"Internal state poisoned".to_string(),
Some(Box::new(err)),
))
}
RecvTimeoutError::Shutdown => Err(ConnectionMatrixRecvTimeoutError::Shutdown),
},
}
}
}