use crate::types::{ExtendedPeerInfo, SyncEvent, SyncEventStream, SyncStatus, SyncStatusProvider};
use futures::{channel::oneshot, Stream};
use sc_network_types::PeerId;
use sc_consensus::{BlockImportError, BlockImportStatus, JustificationSyncLink, Link};
use sc_network::{NetworkBlock, NetworkSyncForkRequest};
use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedSender};
use sp_runtime::traits::{Block as BlockT, NumberFor};
use std::{
pin::Pin,
sync::{
atomic::{AtomicBool, AtomicUsize, Ordering},
Arc,
},
};
pub enum ToServiceCommand<B: BlockT> {
SetSyncForkRequest(Vec<PeerId>, B::Hash, NumberFor<B>),
RequestJustification(B::Hash, NumberFor<B>),
ClearJustificationRequests,
BlocksProcessed(
usize,
usize,
Vec<(Result<BlockImportStatus<NumberFor<B>>, BlockImportError>, B::Hash)>,
),
JustificationImported(PeerId, B::Hash, NumberFor<B>, bool),
AnnounceBlock(B::Hash, Option<Vec<u8>>),
NewBestBlockImported(B::Hash, NumberFor<B>),
EventStream(TracingUnboundedSender<SyncEvent>),
Status(oneshot::Sender<SyncStatus<B>>),
NumActivePeers(oneshot::Sender<usize>),
NumDownloadedBlocks(oneshot::Sender<usize>),
NumSyncRequests(oneshot::Sender<usize>),
PeersInfo(oneshot::Sender<Vec<(PeerId, ExtendedPeerInfo<B>)>>),
OnBlockFinalized(B::Hash, B::Header),
}
#[derive(Clone)]
pub struct SyncingService<B: BlockT> {
tx: TracingUnboundedSender<ToServiceCommand<B>>,
num_connected: Arc<AtomicUsize>,
is_major_syncing: Arc<AtomicBool>,
}
impl<B: BlockT> SyncingService<B> {
pub fn new(
tx: TracingUnboundedSender<ToServiceCommand<B>>,
num_connected: Arc<AtomicUsize>,
is_major_syncing: Arc<AtomicBool>,
) -> Self {
Self { tx, num_connected, is_major_syncing }
}
pub fn num_connected_peers(&self) -> usize {
self.num_connected.load(Ordering::Relaxed)
}
pub async fn num_active_peers(&self) -> Result<usize, oneshot::Canceled> {
let (tx, rx) = oneshot::channel();
let _ = self.tx.unbounded_send(ToServiceCommand::NumActivePeers(tx));
rx.await
}
pub async fn num_downloaded_blocks(&self) -> Result<usize, oneshot::Canceled> {
let (tx, rx) = oneshot::channel();
let _ = self.tx.unbounded_send(ToServiceCommand::NumDownloadedBlocks(tx));
rx.await
}
pub async fn num_sync_requests(&self) -> Result<usize, oneshot::Canceled> {
let (tx, rx) = oneshot::channel();
let _ = self.tx.unbounded_send(ToServiceCommand::NumSyncRequests(tx));
rx.await
}
pub async fn peers_info(
&self,
) -> Result<Vec<(PeerId, ExtendedPeerInfo<B>)>, oneshot::Canceled> {
let (tx, rx) = oneshot::channel();
let _ = self.tx.unbounded_send(ToServiceCommand::PeersInfo(tx));
rx.await
}
pub fn on_block_finalized(&self, hash: B::Hash, header: B::Header) {
let _ = self.tx.unbounded_send(ToServiceCommand::OnBlockFinalized(hash, header));
}
pub async fn status(&self) -> Result<SyncStatus<B>, oneshot::Canceled> {
let (tx, rx) = oneshot::channel();
let _ = self.tx.unbounded_send(ToServiceCommand::Status(tx));
rx.await
}
}
impl<B: BlockT + 'static> NetworkSyncForkRequest<B::Hash, NumberFor<B>> for SyncingService<B> {
fn set_sync_fork_request(&self, peers: Vec<PeerId>, hash: B::Hash, number: NumberFor<B>) {
let _ = self
.tx
.unbounded_send(ToServiceCommand::SetSyncForkRequest(peers, hash, number));
}
}
impl<B: BlockT> JustificationSyncLink<B> for SyncingService<B> {
fn request_justification(&self, hash: &B::Hash, number: NumberFor<B>) {
let _ = self.tx.unbounded_send(ToServiceCommand::RequestJustification(*hash, number));
}
fn clear_justification_requests(&self) {
let _ = self.tx.unbounded_send(ToServiceCommand::ClearJustificationRequests);
}
}
#[async_trait::async_trait]
impl<B: BlockT> SyncStatusProvider<B> for SyncingService<B> {
async fn status(&self) -> Result<SyncStatus<B>, ()> {
let (rtx, rrx) = oneshot::channel();
let _ = self.tx.unbounded_send(ToServiceCommand::Status(rtx));
rrx.await.map_err(|_| ())
}
}
impl<B: BlockT> Link<B> for SyncingService<B> {
fn blocks_processed(
&self,
imported: usize,
count: usize,
results: Vec<(Result<BlockImportStatus<NumberFor<B>>, BlockImportError>, B::Hash)>,
) {
let _ = self
.tx
.unbounded_send(ToServiceCommand::BlocksProcessed(imported, count, results));
}
fn justification_imported(
&self,
who: PeerId,
hash: &B::Hash,
number: NumberFor<B>,
success: bool,
) {
let _ = self
.tx
.unbounded_send(ToServiceCommand::JustificationImported(who, *hash, number, success));
}
fn request_justification(&self, hash: &B::Hash, number: NumberFor<B>) {
let _ = self.tx.unbounded_send(ToServiceCommand::RequestJustification(*hash, number));
}
}
impl<B: BlockT> SyncEventStream for SyncingService<B> {
fn event_stream(&self, name: &'static str) -> Pin<Box<dyn Stream<Item = SyncEvent> + Send>> {
let (tx, rx) = tracing_unbounded(name, 100_000);
let _ = self.tx.unbounded_send(ToServiceCommand::EventStream(tx));
Box::pin(rx)
}
}
impl<B: BlockT> NetworkBlock<B::Hash, NumberFor<B>> for SyncingService<B> {
fn announce_block(&self, hash: B::Hash, data: Option<Vec<u8>>) {
let _ = self.tx.unbounded_send(ToServiceCommand::AnnounceBlock(hash, data));
}
fn new_best_block_imported(&self, hash: B::Hash, number: NumberFor<B>) {
let _ = self.tx.unbounded_send(ToServiceCommand::NewBestBlockImported(hash, number));
}
}
impl<B: BlockT> sp_consensus::SyncOracle for SyncingService<B> {
fn is_major_syncing(&self) -> bool {
self.is_major_syncing.load(Ordering::Relaxed)
}
fn is_offline(&self) -> bool {
self.num_connected.load(Ordering::Relaxed) == 0
}
}