zebra_network/peer/connection/
peer_tx.rsuse futures::{FutureExt, Sink, SinkExt};
use zebra_chain::serialization::SerializationError;
use crate::{constants::REQUEST_TIMEOUT, protocol::external::Message, PeerError};
#[derive(Clone, Debug)]
pub struct PeerTx<Tx>
where
Tx: Sink<Message, Error = SerializationError> + Unpin,
{
inner: Tx,
}
impl<Tx> PeerTx<Tx>
where
Tx: Sink<Message, Error = SerializationError> + Unpin,
{
pub async fn send(&mut self, msg: Message) -> Result<(), PeerError> {
tokio::time::timeout(REQUEST_TIMEOUT, self.inner.send(msg))
.await
.map_err(|_| PeerError::ConnectionSendTimeout)?
.map_err(Into::into)
}
pub async fn close(&mut self) -> Result<(), SerializationError> {
self.inner.close().await
}
}
impl<Tx> From<Tx> for PeerTx<Tx>
where
Tx: Sink<Message, Error = SerializationError> + Unpin,
{
fn from(tx: Tx) -> Self {
PeerTx { inner: tx }
}
}
impl<Tx> Drop for PeerTx<Tx>
where
Tx: Sink<Message, Error = SerializationError> + Unpin,
{
fn drop(&mut self) {
self.close().now_or_never();
}
}