pub fn build_upon<In, Out>(&self, header: &B::Header, input: In, output: Out)
		-> Result<Option<
			BftFuture<
				B,
				<P as Environment<B>>::Proposer,
				I,
				In,
				Out,
			>>, P::Error>
		where
			In: Stream<Item=Communication<B>, Error=Error>,
			Out: Sink<SinkItem=Communication<B>, SinkError=Error>,
	{
		let hash = header.hash();

		let mut live_agreement = self.live_agreement.lock();
		let can_build = live_agreement.as_ref()
			.map_or(true, |x| self.can_build_on_inner(header, x));

		if !can_build {
			return Ok(None)
		}

		let authorities = self.client.authorities(&BlockId::Hash(hash.clone()))
			.map_err(|e| CommonErrorKind::Other(Box::new(e)).into())?;

		let n = authorities.len();
		let max_faulty = max_faulty_of(n);
		trace!(target: "rhd", "Initiating agreement on top of #{}, {:?}", header.number(), hash);
		trace!(target: "rhd", "max_faulty_of({})={}", n, max_faulty);

		let local_id = self.local_id();

		if !authorities.contains(&local_id) {
			// cancel current agreement
			live_agreement.take();
			Err(CommonErrorKind::InvalidAuthority(local_id).into())?;
		}

		let proposer = self.factory.init(header, &authorities, self.key.clone())?;

		let bft_instance = BftInstance {
			proposer,
			parent_hash: hash.clone(),
			cache: self.round_cache.clone(),
			round_timeout_multiplier: self.round_timeout_multiplier,
			key: self.key.clone(),
			authorities: authorities,
		};

		let mut agreement = rhododendron::agree(
			bft_instance,
			n,
			max_faulty,
			input,
			output,
		);

		// fast forward round number if necessary.
		{
			let mut cache = self.round_cache.lock();
			trace!(target: "rhd", "Round cache: {:?}", &*cache);
			if cache.hash.as_ref() == Some(&hash) {
				trace!(target: "rhd", "Fast-forwarding to round {}", cache.start_round);
				let start_round = cache.start_round;
				cache.start_round += 1;

				drop(cache);
				agreement.fast_forward(start_round);
			} else {
				*cache = RoundCache {
					hash: Some(hash.clone()),
					start_round: 1,
				};
			}
		}

		let status = Arc::new(AtomicUsize::new(status::LIVE));
		let (tx, rx) = oneshot::channel();

		// cancel current agreement.
		*live_agreement = Some((header.clone(), AgreementHandle {
			send_cancel: Some(tx),
			status: status.clone(),
		}));

		Ok(Some(BftFuture {
			inner: agreement,
			status: status,
			cancel: rx,
			import: self.client.clone(),
		}))
	}
/// Signal that a valid block with the given header has been imported.
/// Provide communication streams that are localized to this block.
/// It's recommended to use the communication primitives provided by this
/// module for signature checking and decoding. See `CheckedStream` and
/// `SigningSink` for more details.
///
/// Messages received on the stream that don't match the expected format
/// will be dropped.
///
/// If the local signing key is an authority, this will begin the consensus process to build a
/// block on top of it. If the executor fails to run the future, an error will be returned.
/// Returns `None` if the agreement on the block with given parent is already in progress.
Signal that a valid block with the given header has been imported.
Provide communication streams that are localized to this block.
It's recommended to use the communication primitives provided by this
module for signature checking and decoding. See `CheckedStream` and
`SigningSink` for more details.

Messages received on the stream that don't match the expected format
will be dropped.

If the local signing key is an authority, this will begin the consensus process to build a
block on top of it. If the executor fails to run the future, an error will be returned.
Returns `None` if the agreement on the block with given parent is already in progress.
Signal that a valid block with the given header has been imported . Provide communication streams that are localized to this block . It ' s recommended to use the communication primitives provided by this module for signature checking and decoding . See ` CheckedStream ` and ` SigningSink ` for more details . Messages received on the stream that don ' t match the expected format will be dropped . If the local signing key is an authority this will begin the consensus process to build a block on top of it . If the executor fails to run the future an error will be returned . Returns ` None ` if the agreement on the block with given parent is already in progress .
pub fn build_upon<In, Out>(&self, header: &B::Header, input: In, output: Out)
		-> Result<Option<
			BftFuture<
				B,
				<P as Environment<B>>::Proposer,
				I,
				In,
				Out,
			>>, P::Error>
		where
			In: Stream<Item=Communication<B>, Error=Error>,
			Out: Sink<SinkItem=Communication<B>, SinkError=Error>,
	{
		let hash = header.hash();
		let mut live_agreement = self.live_agreement.lock();
		let can_build = live_agreement.as_ref()
			.map_or(true, |x| self.can_build_on_inner(header, x));
		if !can_build {
			return Ok(None)
		}
		let authorities = self.client.authorities(&BlockId::Hash(hash.clone()))
			.map_err(|e| CommonErrorKind::Other(Box::new(e)).into())?;
		let n = authorities.len();
		let max_faulty = max_faulty_of(n);
		trace!(target: "rhd", "Initiating agreement on top of #{}, {:?}", header.number(), hash);
		trace!(target: "rhd", "max_faulty_of({})={}", n, max_faulty);
		let local_id = self.local_id();
		if !authorities.contains(&local_id) {
			// cancel current agreement
			live_agreement.take();
			Err(CommonErrorKind::InvalidAuthority(local_id).into())?;
		}
		let proposer = self.factory.init(header, &authorities, self.key.clone())?;
		let bft_instance = BftInstance {
			proposer,
			parent_hash: hash.clone(),
			cache: self.round_cache.clone(),
			round_timeout_multiplier: self.round_timeout_multiplier,
			key: self.key.clone(),
			authorities: authorities,
		};
		let mut agreement = rhododendron::agree(
			bft_instance,
			n,
			max_faulty,
			input,
			output,
		);
		// fast forward round number if necessary.
		{
			let mut cache = self.round_cache.lock();
			trace!(target: "rhd", "Round cache: {:?}", &*cache);
			if cache.hash.as_ref() == Some(&hash) {
				trace!(target: "rhd", "Fast-forwarding to round {}", cache.start_round);
				let start_round = cache.start_round;
				cache.start_round += 1;
				drop(cache);
				agreement.fast_forward(start_round);
			} else {
				*cache = RoundCache {
					hash: Some(hash.clone()),
					start_round: 1,
				};
			}
		}
		let status = Arc::new(AtomicUsize::new(status::LIVE));
		let (tx, rx) = oneshot::channel();
		// cancel current agreement.
		*live_agreement = Some((header.clone(), AgreementHandle {
			send_cancel: Some(tx),
			status: status.clone(),
		}));
		Ok(Some(BftFuture {
			inner: agreement,
			status: status,
			cancel: rx,
			import: self.client.clone(),
		}))
	}
pub fn build_upon < In , Out > ( & self , header : & B :: Header , input : In , output : Out ) -> Result < Option < BftFuture < B , < P as Environment < B > > :: Proposer , I , In , Out , > > , P :: Error > where In : Stream < Item = Communication < B > , Error = Error > , Out : Sink < SinkItem = Communication < B > , SinkError = Error > , { let hash = header . hash ( ) ; let mut live_agreement = self . live_agreement . lock ( ) ; let can_build = live_agreement . as_ref ( ) . map_or ( true , | x | self . can_build_on_inner ( header , x ) ) ; if ! can_build { return Ok ( None ) } let authorities = self . client . authorities ( & BlockId :: Hash ( hash . clone ( ) ) ) . map_err ( | e | CommonErrorKind :: Other ( Box :: new ( e ) ) . into ( ) ) ? ; let n = authorities . len ( ) ; let max_faulty = max_faulty_of ( n ) ; trace ! ( target " " " " header number ( ) hash ) ; trace ! ( target " " " " n max_faulty ) ; let local_id = self . local_id ( ) ; if ! authorities . contains ( & local_id ) { // cancel current agreement live_agreement . take ( ) ; Err ( CommonErrorKind :: InvalidAuthority ( local_id ) . into ( ) ) ? ; } let proposer = self . factory . init ( header , & authorities , self . key . clone ( ) ) ? ; let bft_instance = BftInstance { proposer , parent_hash : hash . clone ( ) , cache : self . round_cache . clone ( ) , round_timeout_multiplier : self . round_timeout_multiplier , key : self . key . clone ( ) , authorities : authorities , } ; let mut agreement = rhododendron :: agree ( bft_instance , n , max_faulty , input , output , ) ; // fast forward round number if necessary. { let mut cache = self . round_cache . lock ( ) ; trace ! ( target " " " " cache ) ; if cache . hash . as_ref ( ) == Some ( & hash ) { trace ! ( target " " " " cache start_round ) ; let start_round = cache . start_round ; cache . start_round += 1 ; drop ( cache ) ; agreement . fast_forward ( start_round ) ; } else { * cache = RoundCache { hash : Some ( hash . clone ( ) ) , start_round : 1 , } ; } } let status = Arc :: new ( AtomicUsize :: new ( status :: LIVE ) ) ; let ( tx , rx ) = oneshot :: channel ( ) ; // cancel current agreement. * live_agreement = Some ( ( header . clone ( ) , AgreementHandle { send_cancel : Some ( tx ) , status : status . clone ( ) , } ) ) ; Ok ( Some ( BftFuture { inner : agreement , status : status , cancel : rx , import : self . client . clone ( ) , } ) ) }
Signal that a valid block with the given header has been imported.
Signal that a valid block with the given header has been imported .
&self
header
&B::Header
input
In
output
Out
&self
header
&B::Header
input
In
output
Out