pub fn new<A, P, C, N>(
		client: Arc<C>,
		api: Arc<A>,
		network: N,
		transaction_pool: Arc<TransactionPool<P>>,
		thread_pool: ThreadPoolHandle,
		key: ed25519::Pair,
		block_delay: u64,
	) -> Service
		where
			error::Error: From<<A as AuthoringApi>::Error>,
			A: AuthoringApi + BlockNumberToHash + 'static,
			P: PoolChainApi<Block = <A as AuthoringApi>::Block> + 'static,
			C: BlockchainEvents<<A as AuthoringApi>::Block>
				+ ChainHead<<A as AuthoringApi>::Block>
				+ BlockBody<<A as AuthoringApi>::Block>,
			C: consensus::BlockImport<<A as AuthoringApi>::Block>
				+ consensus::Authorities<<A as AuthoringApi>::Block> + Send + Sync + 'static,
			primitives::H256: From<<<A as AuthoringApi>::Block as BlockT>::Hash>,
			<<A as AuthoringApi>::Block as BlockT>::Hash: PartialEq<primitives::H256> + PartialEq,
			N: Network<Block = <A as AuthoringApi>::Block> + Send + 'static,
	{

		let (signal, exit) = ::exit_future::signal();
		let thread = thread::spawn(move || {
			let mut runtime = LocalRuntime::new().expect("Could not create local runtime");
			let key = Arc::new(key);

			let factory = ProposerFactory {
				client: api.clone(),
				transaction_pool: transaction_pool.clone(),
				network,
				handle: thread_pool.clone(),
				offline: Arc::new(RwLock::new(OfflineTracker::new())),
				force_delay: block_delay,
			};
			let bft_service = Arc::new(BftService::new(client.clone(), key, factory));

			let notifications = {
				let client = client.clone();
				let bft_service = bft_service.clone();

				client.import_notification_stream().for_each(move |notification| {
					if notification.is_new_best {
						start_bft(notification.header, bft_service.clone());
					}
					Ok(())
				})
			};

			let interval = Interval::new(
				Instant::now() + Duration::from_millis(TIMER_DELAY_MS),
				Duration::from_millis(TIMER_INTERVAL_MS),
			);

			let mut prev_best = match client.best_block_header() {
				Ok(header) => header.hash(),
				Err(e) => {
					warn!("Cant's start consensus service. Error reading best block header: {:?}", e);
					return;
				}
			};

			let timed = {
				let c = client.clone();
				let s = bft_service.clone();

				interval.map_err(|e| debug!(target: "bft", "Timer error: {:?}", e)).for_each(move |_| {
					if let Ok(best_block) = c.best_block_header() {
						let hash = best_block.hash();

						if hash == prev_best {
							debug!(target: "bft", "Starting consensus round after a timeout");
							start_bft(best_block, s.clone());
						}
						prev_best = hash;
					}
					Ok(())
				})
			};

			runtime.spawn(notifications);
			runtime.spawn(timed);

			if let Err(e) = runtime.block_on(exit) {
				debug!("BFT event loop error {:?}", e);
			}
		});
		Service {
			thread: Some(thread),
			exit_signal: Some(signal),
		}
	}
/// Create and start a new instance.
Create and start a new instance.
Create and start a new instance .
pub fn new<A, P, C, N>(
		client: Arc<C>,
		api: Arc<A>,
		network: N,
		transaction_pool: Arc<TransactionPool<P>>,
		thread_pool: ThreadPoolHandle,
		key: ed25519::Pair,
		block_delay: u64,
	) -> Service
		where
			error::Error: From<<A as AuthoringApi>::Error>,
			A: AuthoringApi + BlockNumberToHash + 'static,
			P: PoolChainApi<Block = <A as AuthoringApi>::Block> + 'static,
			C: BlockchainEvents<<A as AuthoringApi>::Block>
				+ ChainHead<<A as AuthoringApi>::Block>
				+ BlockBody<<A as AuthoringApi>::Block>,
			C: consensus::BlockImport<<A as AuthoringApi>::Block>
				+ consensus::Authorities<<A as AuthoringApi>::Block> + Send + Sync + 'static,
			primitives::H256: From<<<A as AuthoringApi>::Block as BlockT>::Hash>,
			<<A as AuthoringApi>::Block as BlockT>::Hash: PartialEq<primitives::H256> + PartialEq,
			N: Network<Block = <A as AuthoringApi>::Block> + Send + 'static,
	{
		let (signal, exit) = ::exit_future::signal();
		let thread = thread::spawn(move || {
			let mut runtime = LocalRuntime::new().expect("Could not create local runtime");
			let key = Arc::new(key);
			let factory = ProposerFactory {
				client: api.clone(),
				transaction_pool: transaction_pool.clone(),
				network,
				handle: thread_pool.clone(),
				offline: Arc::new(RwLock::new(OfflineTracker::new())),
				force_delay: block_delay,
			};
			let bft_service = Arc::new(BftService::new(client.clone(), key, factory));
			let notifications = {
				let client = client.clone();
				let bft_service = bft_service.clone();
				client.import_notification_stream().for_each(move |notification| {
					if notification.is_new_best {
						start_bft(notification.header, bft_service.clone());
					}
					Ok(())
				})
			};
			let interval = Interval::new(
				Instant::now() + Duration::from_millis(TIMER_DELAY_MS),
				Duration::from_millis(TIMER_INTERVAL_MS),
			);
			let mut prev_best = match client.best_block_header() {
				Ok(header) => header.hash(),
				Err(e) => {
					warn!("Cant's start consensus service. Error reading best block header: {:?}", e);
					return;
				}
			};
			let timed = {
				let c = client.clone();
				let s = bft_service.clone();
				interval.map_err(|e| debug!(target: "bft", "Timer error: {:?}", e)).for_each(move |_| {
					if let Ok(best_block) = c.best_block_header() {
						let hash = best_block.hash();
						if hash == prev_best {
							debug!(target: "bft", "Starting consensus round after a timeout");
							start_bft(best_block, s.clone());
						}
						prev_best = hash;
					}
					Ok(())
				})
			};
			runtime.spawn(notifications);
			runtime.spawn(timed);
			if let Err(e) = runtime.block_on(exit) {
				debug!("BFT event loop error {:?}", e);
			}
		});
		Service {
			thread: Some(thread),
			exit_signal: Some(signal),
		}
	}
pub fn new < A , P , C , N > ( client : Arc < C > , api : Arc < A > , network : N , transaction_pool : Arc < TransactionPool < P > > , thread_pool : ThreadPoolHandle , key : ed25519 :: Pair , block_delay : u64 , ) -> Service where error :: Error : From < < A as AuthoringApi > :: Error > , A : AuthoringApi + BlockNumberToHash + ' static , P : PoolChainApi < Block = < A as AuthoringApi > :: Block > + ' static , C : BlockchainEvents < < A as AuthoringApi > :: Block > + ChainHead < < A as AuthoringApi > :: Block > + BlockBody < < A as AuthoringApi > :: Block > , C : consensus :: BlockImport < < A as AuthoringApi > :: Block > + consensus :: Authorities < < A as AuthoringApi > :: Block > + Send + Sync + ' static , primitives :: H256 : From < < < A as AuthoringApi > :: Block as BlockT > :: Hash > , < < A as AuthoringApi > :: Block as BlockT > :: Hash : PartialEq < primitives :: H256 > + PartialEq , N : Network < Block = < A as AuthoringApi > :: Block > + Send + ' static , { let ( signal , exit ) = :: exit_future :: signal ( ) ; let thread = thread :: spawn ( move | | { let mut runtime = LocalRuntime :: new ( ) . expect ( " " ) ; let key = Arc :: new ( key ) ; let factory = ProposerFactory { client : api . clone ( ) , transaction_pool : transaction_pool . clone ( ) , network , handle : thread_pool . clone ( ) , offline : Arc :: new ( RwLock :: new ( OfflineTracker :: new ( ) ) ) , force_delay : block_delay , } ; let bft_service = Arc :: new ( BftService :: new ( client . clone ( ) , key , factory ) ) ; let notifications = { let client = client . clone ( ) ; let bft_service = bft_service . clone ( ) ; client . import_notification_stream ( ) . for_each ( move | notification | { if notification . is_new_best { start_bft ( notification . header , bft_service . clone ( ) ) ; } Ok ( ( ) ) } ) } ; let interval = Interval :: new ( Instant :: now ( ) + Duration :: from_millis ( TIMER_DELAY_MS ) , Duration :: from_millis ( TIMER_INTERVAL_MS ) , ) ; let mut prev_best = match client . best_block_header ( ) { Ok ( header ) => header . hash ( ) , Err ( e ) => { warn ! ( " " e ) ; return ; } } ; let timed = { let c = client . clone ( ) ; let s = bft_service . clone ( ) ; interval . map_err ( | e | debug ! ( target " " " " e ) ) . for_each ( move | _ | { if let Ok ( best_block ) = c . best_block_header ( ) { let hash = best_block . hash ( ) ; if hash == prev_best { debug ! ( target " " " " ) ; start_bft ( best_block , s . clone ( ) ) ; } prev_best = hash ; } Ok ( ( ) ) } ) } ; runtime . spawn ( notifications ) ; runtime . spawn ( timed ) ; if let Err ( e ) = runtime . block_on ( exit ) { debug ! ( " " e ) ; } } ) ; Service { thread : Some ( thread ) , exit_signal : Some ( signal ) , } }
Create and start a new instance.
Create and start a new instance .
client
Arc<C>
api
Arc<A>
network
N
transaction_pool
Arc<TransactionPool<P>>
thread_pool
ThreadPoolHandle
key
ed25519::Pair
block_delay
u64
client
Arc<C>
api
Arc<A>
network
N
transaction_pool
Arc<TransactionPool<P>>
thread_pool
ThreadPoolHandle
key
ed25519::Pair
block_delay
u64