use std::{collections::BTreeMap, future::Future};
use futures::{sink::SinkExt, stream::StreamExt};
use linera_base::{
crypto::CryptoHash,
data_types::{BlobContent, BlockHeight, NetworkDescription},
identifiers::{BlobId, ChainId, StreamId},
time::{timer, Duration},
};
use linera_chain::{
data_types::BlockProposal,
types::{
ConfirmedBlockCertificate, LiteCertificate, TimeoutCertificate, ValidatedBlockCertificate,
},
};
use linera_core::{
data_types::{ChainInfoQuery, ChainInfoResponse},
node::{BlobStream, CrossChainMessageDelivery, NodeError, NotificationStream, ValidatorNode},
};
use linera_storage::Arc as CacheArc;
use linera_version::VersionInfo;
use super::{codec, transport::TransportProtocol};
use crate::{
config::ValidatorPublicNetworkPreConfig, HandleConfirmedCertificateRequest,
HandleLiteCertRequest, HandleTimeoutCertificateRequest, HandleValidatedCertificateRequest,
RpcMessage,
};
#[derive(Clone)]
pub struct SimpleClient {
network: ValidatorPublicNetworkPreConfig<TransportProtocol>,
send_timeout: Duration,
recv_timeout: Duration,
}
impl SimpleClient {
pub(crate) fn new(
network: ValidatorPublicNetworkPreConfig<TransportProtocol>,
send_timeout: Duration,
recv_timeout: Duration,
) -> Self {
Self {
network,
send_timeout,
recv_timeout,
}
}
async fn send_recv_internal(&self, message: RpcMessage) -> Result<RpcMessage, codec::Error> {
let address = format!("{}:{}", self.network.host, self.network.port);
let mut stream = self.network.protocol.connect(address).await?;
timer::timeout(self.send_timeout, stream.send(message))
.await
.map_err(|timeout| codec::Error::IoError(timeout.into()))??;
timer::timeout(self.recv_timeout, stream.next())
.await
.map_err(|timeout| codec::Error::IoError(timeout.into()))?
.transpose()?
.ok_or_else(|| codec::Error::IoError(std::io::ErrorKind::UnexpectedEof.into()))
}
async fn query<Response>(&self, query: RpcMessage) -> Result<Response, Response::Error>
where
Response: TryFrom<RpcMessage>,
Response::Error: From<codec::Error>,
{
self.send_recv_internal(query).await?.try_into()
}
}
impl ValidatorNode for SimpleClient {
type NotificationStream = NotificationStream;
fn address(&self) -> String {
format!(
"{}://{}:{}",
self.network.protocol, self.network.host, self.network.port
)
}
async fn handle_block_proposal(
&self,
proposal: BlockProposal,
) -> Result<ChainInfoResponse, NodeError> {
let request = RpcMessage::BlockProposal(Box::new(proposal));
self.query(request).await
}
async fn handle_lite_certificate(
&self,
certificate: LiteCertificate<'_>,
delivery: CrossChainMessageDelivery,
) -> Result<ChainInfoResponse, NodeError> {
let wait_for_outgoing_messages = delivery.wait_for_outgoing_messages();
let request = RpcMessage::LiteCertificate(Box::new(HandleLiteCertRequest {
certificate: certificate.cloned(),
wait_for_outgoing_messages,
}));
self.query(request).await
}
async fn handle_validated_certificate(
&self,
certificate: ValidatedBlockCertificate,
) -> Result<ChainInfoResponse, NodeError> {
let request = HandleValidatedCertificateRequest { certificate };
let request = RpcMessage::ValidatedCertificate(Box::new(request));
self.query(request).await
}
async fn handle_confirmed_certificate(
&self,
certificate: CacheArc<ConfirmedBlockCertificate>,
delivery: CrossChainMessageDelivery,
) -> Result<ChainInfoResponse, NodeError> {
let wait_for_outgoing_messages = delivery.wait_for_outgoing_messages();
let request = HandleConfirmedCertificateRequest {
certificate: CacheArc::unwrap_or_clone(certificate),
wait_for_outgoing_messages,
};
let request = RpcMessage::ConfirmedCertificate(Box::new(request));
self.query(request).await
}
async fn handle_timeout_certificate(
&self,
certificate: TimeoutCertificate,
) -> Result<ChainInfoResponse, NodeError> {
let request = HandleTimeoutCertificateRequest { certificate };
let request = RpcMessage::TimeoutCertificate(Box::new(request));
self.query(request).await
}
async fn handle_chain_info_query(
&self,
query: ChainInfoQuery,
) -> Result<ChainInfoResponse, NodeError> {
let request = RpcMessage::ChainInfoQuery(Box::new(query));
self.query(request).await
}
fn subscribe(
&self,
_chains: Vec<ChainId>,
) -> impl Future<Output = Result<NotificationStream, NodeError>> + Send {
let transport = self.network.protocol.to_string();
async { Err(NodeError::SubscriptionError { transport }) }
}
async fn get_version_info(&self) -> Result<VersionInfo, NodeError> {
self.query(RpcMessage::VersionInfoQuery).await
}
async fn get_network_description(&self) -> Result<NetworkDescription, NodeError> {
self.query(RpcMessage::NetworkDescriptionQuery).await
}
async fn upload_blob(&self, content: BlobContent) -> Result<BlobId, NodeError> {
self.query(RpcMessage::UploadBlob(Box::new(content))).await
}
async fn download_blob(&self, blob_id: BlobId) -> Result<BlobContent, NodeError> {
self.query(RpcMessage::DownloadBlob(Box::new(blob_id)))
.await
}
async fn download_blobs(&self, blob_ids: Vec<BlobId>) -> Result<BlobStream, NodeError> {
let mut stream = self
.network
.protocol
.connect((self.network.host.clone(), self.network.port))
.await
.map_err(|e| NodeError::ClientIoError {
error: e.to_string(),
})?;
timer::timeout(
self.send_timeout,
stream.send(RpcMessage::DownloadBlobs(blob_ids)),
)
.await
.map_err(|timeout| NodeError::ClientIoError {
error: timeout.to_string(),
})?
.map_err(|e| NodeError::ClientIoError {
error: e.to_string(),
})?;
let blob_stream = stream.filter_map(|result| async {
match result {
Ok(RpcMessage::DownloadBlobResponse(blob)) => Some(Ok(*blob)),
Ok(RpcMessage::Error(err)) => Some(Err(*err)),
Ok(_) => Some(Err(NodeError::UnexpectedMessage)),
Err(e) => Some(Err(NodeError::ClientIoError {
error: e.to_string(),
})),
}
});
Ok(Box::pin(blob_stream))
}
async fn download_pending_blob(
&self,
chain_id: ChainId,
blob_id: BlobId,
) -> Result<BlobContent, NodeError> {
self.query(RpcMessage::DownloadPendingBlob(Box::new((
chain_id, blob_id,
))))
.await
}
async fn handle_pending_blob(
&self,
chain_id: ChainId,
blob: BlobContent,
) -> Result<ChainInfoResponse, NodeError> {
self.query(RpcMessage::HandlePendingBlob(Box::new((chain_id, blob))))
.await
}
async fn download_certificate(
&self,
hash: CryptoHash,
) -> Result<ConfirmedBlockCertificate, NodeError> {
Ok(self
.download_certificates(vec![hash])
.await?
.into_iter()
.next()
.unwrap()) }
async fn download_certificates(
&self,
hashes: Vec<CryptoHash>,
) -> Result<Vec<ConfirmedBlockCertificate>, NodeError> {
let certificates = self
.query::<Vec<ConfirmedBlockCertificate>>(RpcMessage::DownloadCertificates(
hashes.clone(),
))
.await?;
if certificates.len() != hashes.len() {
let missing_hashes: Vec<CryptoHash> = hashes
.into_iter()
.filter(|hash| !certificates.iter().any(|cert| cert.hash() == *hash))
.collect();
Err(NodeError::MissingCertificates(missing_hashes))
} else {
Ok(certificates)
}
}
async fn download_certificates_by_heights(
&self,
chain_id: ChainId,
heights: Vec<BlockHeight>,
) -> Result<Vec<ConfirmedBlockCertificate>, NodeError> {
let expected_count = heights.len();
let certificates: Vec<ConfirmedBlockCertificate> = self
.query(RpcMessage::DownloadCertificatesByHeights(
chain_id,
heights.clone(),
))
.await?;
if certificates.len() < expected_count {
return Err(NodeError::MissingCertificatesByHeights { chain_id, heights });
}
Ok(certificates)
}
async fn blob_last_used_by(&self, blob_id: BlobId) -> Result<CryptoHash, NodeError> {
self.query(RpcMessage::BlobLastUsedBy(Box::new(blob_id)))
.await
}
async fn missing_blob_ids(&self, blob_ids: Vec<BlobId>) -> Result<Vec<BlobId>, NodeError> {
self.query(RpcMessage::MissingBlobIds(blob_ids)).await
}
async fn blob_last_used_by_certificate(
&self,
blob_id: BlobId,
) -> Result<ConfirmedBlockCertificate, NodeError> {
self.query::<ConfirmedBlockCertificate>(RpcMessage::BlobLastUsedByCertificate(Box::new(
blob_id,
)))
.await
}
async fn previous_event_blocks(
&self,
chain_id: ChainId,
stream_ids: Vec<StreamId>,
) -> Result<BTreeMap<StreamId, (BlockHeight, CryptoHash)>, NodeError> {
self.query(RpcMessage::PreviousEventBlocks(Box::new((
chain_id, stream_ids,
))))
.await
}
}