,
table_router: Router,
}
impl MessageProcessTask {
fn process_message(&self, msg: ConsensusMessage) -> Option> {
match msg {
ConsensusMessage::Bft(msg) => {
let local_id = self.table_router.session_key();
match process_bft_message(msg, &local_id, &self.validators[..]) {
Ok(Some(msg)) => {
if let Err(_) = self.bft_messages.unbounded_send(msg) {
// if the BFT receiving stream has ended then
// we should just bail.
trace!(target: "bft", "BFT message stream appears to have closed");
return Some(Async::Ready(()));
}
}
Ok(None) => {} // ignored local message
Err(e) => {
debug!("Message validation failed: {:?}", e);
}
}
}
ConsensusMessage::ChainSpecific(msg, _) => {
if let Some(Message::Statement(parent_hash, statement)) = Decode::decode(&mut msg.as_slice()) {
if ::polkadot_consensus::check_statement(&statement.statement, &statement.signature, statement.sender, &parent_hash) {
self.table_router.import_statement(statement);
}
}
}
}
None
}
}
impl Future for MessageProcessTask {
type Item = ();
type Error = ();
fn poll(&mut self) -> Poll<(), ()> {
loop {
match self.inner_stream.poll() {
Ok(Async::Ready(Some(val))) => if let Some(async) = self.process_message(val) {
return Ok(async);
},
Ok(Async::Ready(None)) => return Ok(Async::Ready(())),
Ok(Async::NotReady) => return Ok(Async::NotReady),
Err(e) => debug!(target: "p_net", "Error getting consensus message: {:?}", e),
}
}
}
}
/// Input stream from the consensus network.
pub struct InputAdapter {
input: mpsc::UnboundedReceiver>,
}
impl Stream for InputAdapter {
type Item = bft::Communication;
type Error = ::polkadot_consensus::Error;
fn poll(&mut self) -> Poll, Self::Error> {
match self.input.poll() {
Err(_) | Ok(Async::Ready(None)) => Err(bft::InputStreamConcluded.into()),
Ok(x) => Ok(x)
}
}
}
/// Wrapper around the network service
pub struct ConsensusNetwork {
network: Arc,
api: Arc,
}
impl
ConsensusNetwork
{
/// Create a new consensus networking object.
pub fn new(network: Arc, api: Arc) -> Self {
ConsensusNetwork { network, api }
}
}
impl
Clone for ConsensusNetwork
{
fn clone(&self) -> Self {
ConsensusNetwork {
network: self.network.clone(),
api: self.api.clone(),
}
}
}
/// A long-lived network which can create parachain statement and BFT message routing processes on demand.
impl Network for ConsensusNetwork {
type TableRouter = Router
;
/// The input stream of BFT messages. Should never logically conclude.
type Input = InputAdapter;
/// The output sink of BFT messages. Messages sent here should eventually pass to all
/// current validators.
type Output = BftSink<::polkadot_consensus::Error>;
/// Instantiate a table router using the given shared table.
fn communication_for(&self, validators: &[SessionKey], table: Arc, task_executor: TaskExecutor) -> (Self::TableRouter, Self::Input, Self::Output) {
let parent_hash = table.consensus_parent_hash().clone();
let sink = BftSink {
network: self.network.clone(),
parent_hash,
_marker: Default::default(),
};
let (bft_send, bft_recv) = mpsc::unbounded();
let knowledge = Arc::new(Mutex::new(Knowledge::new()));
let local_session_key = table.session_key();
let table_router = Router::new(
table,
self.network.clone(),
self.api.clone(),
task_executor.clone(),
parent_hash,
knowledge.clone(),
);
// spin up a task in the background that processes all incoming statements
// TODO: propagate statements on a timer?
let process_task = self.network.with_spec(|spec, ctx| {
spec.new_consensus(ctx, CurrentConsensus {
knowledge,
parent_hash,
local_session_key,
});
MessageProcessTask {
inner_stream: spec.consensus_gossip.messages_for(parent_hash),
bft_messages: bft_send,
validators: validators.to_vec(),
table_router: table_router.clone(),
}
});
match process_task {
Some(task) => task_executor.spawn(task),
None => warn!(target: "p_net", "Cannot process incoming messages: network appears to be down"),
}
(table_router, InputAdapter { input: bft_recv }, sink)
}
}
/// Error when the network appears to be down.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct NetworkDown;
/// A future that resolves when a collation is received.
pub struct AwaitingCollation(Option<::futures::sync::oneshot::Receiver>);
impl Future for AwaitingCollation {
type Item = Collation;
type Error = NetworkDown;
fn poll(&mut self) -> Poll {
match self.0.poll().map_err(|_| NetworkDown)? {
Async::Ready(None) => Err(NetworkDown),
Async::Ready(Some(x)) => Ok(Async::Ready(x)),
Async::NotReady => Ok(Async::NotReady),
}
}
}
impl Collators for ConsensusNetwork {
type Error = NetworkDown;
type Collation = AwaitingCollation;
fn collate(&self, parachain: ParaId, relay_parent: Hash) -> Self::Collation {
AwaitingCollation(
self.network.with_spec(|spec, _| spec.await_collation(relay_parent, parachain))
)
}
fn note_bad_collator(&self, collator: AccountId) {
self.network.with_spec(|spec, ctx| spec.disconnect_bad_collator(ctx, collator));
}
}