fn run_to_completion_with<F: FnOnce()>(
	blocks: u64,
	net: Arc<Mutex<GrandpaTestNet>>,
	peers: &[AuthorityKeyring],
	before_waiting: F,
) -> u64 {
	use parking_lot::RwLock;

	let mut finality_notifications = Vec::new();
	let mut runtime = current_thread::Runtime::new().unwrap();

	let highest_finalized = Arc::new(RwLock::new(0));

	for (peer_id, key) in peers.iter().enumerate() {
		let highest_finalized = highest_finalized.clone();
		let (client, link) = {
			let net = net.lock();
			// temporary needed for some reason
			let link = net.peers[peer_id].data.lock().take().expect("link initialized at startup; qed");
			(
				net.peers[peer_id].client().clone(),
				link,
			)
		};
		finality_notifications.push(
			client.finality_notification_stream()
				.take_while(move |n| {
					let mut highest_finalized = highest_finalized.write();
					if *n.header.number() > *highest_finalized {
						*highest_finalized = *n.header.number();
					}
					Ok(n.header.number() < &blocks)
				})
				.for_each(|_| Ok(()))
		);
		fn assert_send<T: Send>(_: &T) { }

		let voter = run_grandpa(
			Config {
				gossip_duration: TEST_GOSSIP_DURATION,
				justification_period: 32,
				local_key: Some(Arc::new(key.clone().into())),
				name: Some(format!("peer#{}", peer_id)),
			},
			link,
			MessageRouting::new(net.clone(), peer_id),
			InherentDataProviders::new(),
			futures::empty(),
		).expect("all in order with client and network");

		assert_send(&voter);

		runtime.spawn(voter);
	}

	// wait for all finalized on each.
	let wait_for = ::futures::future::join_all(finality_notifications)
		.map(|_| ())
		.map_err(|_| ());

	let drive_to_completion = ::tokio::timer::Interval::new_interval(TEST_ROUTING_INTERVAL)
		.for_each(move |_| {
			net.lock().send_import_notifications();
			net.lock().send_finality_notifications();
			net.lock().sync_without_disconnects();
			Ok(())
		})
		.map(|_| ())
		.map_err(|_| ());

	(before_waiting)();

	runtime.block_on(wait_for.select(drive_to_completion).map_err(|_| ())).unwrap();

	let highest_finalized = *highest_finalized.read();

	highest_finalized
}
// run the voters to completion. provide a closure to be invoked after
// the voters are spawned but before blocking on them.
run the voters to completion. provide a closure to be invoked after
the voters are spawned but before blocking on them.
run the voters to completion . provide a closure to be invoked after the voters are spawned but before blocking on them .
fn run_to_completion_with<F: FnOnce()>(
	blocks: u64,
	net: Arc<Mutex<GrandpaTestNet>>,
	peers: &[AuthorityKeyring],
	before_waiting: F,
) -> u64 {
	use parking_lot::RwLock;
	let mut finality_notifications = Vec::new();
	let mut runtime = current_thread::Runtime::new().unwrap();
	let highest_finalized = Arc::new(RwLock::new(0));
	for (peer_id, key) in peers.iter().enumerate() {
		let highest_finalized = highest_finalized.clone();
		let (client, link) = {
			let net = net.lock();
			// temporary needed for some reason
			let link = net.peers[peer_id].data.lock().take().expect("link initialized at startup; qed");
			(
				net.peers[peer_id].client().clone(),
				link,
			)
		};
		finality_notifications.push(
			client.finality_notification_stream()
				.take_while(move |n| {
					let mut highest_finalized = highest_finalized.write();
					if *n.header.number() > *highest_finalized {
						*highest_finalized = *n.header.number();
					}
					Ok(n.header.number() < &blocks)
				})
				.for_each(|_| Ok(()))
		);
		fn assert_send<T: Send>(_: &T) { }
		let voter = run_grandpa(
			Config {
				gossip_duration: TEST_GOSSIP_DURATION,
				justification_period: 32,
				local_key: Some(Arc::new(key.clone().into())),
				name: Some(format!("peer#{}", peer_id)),
			},
			link,
			MessageRouting::new(net.clone(), peer_id),
			InherentDataProviders::new(),
			futures::empty(),
		).expect("all in order with client and network");
		assert_send(&voter);
		runtime.spawn(voter);
	}
	// wait for all finalized on each.
	let wait_for = ::futures::future::join_all(finality_notifications)
		.map(|_| ())
		.map_err(|_| ());
	let drive_to_completion = ::tokio::timer::Interval::new_interval(TEST_ROUTING_INTERVAL)
		.for_each(move |_| {
			net.lock().send_import_notifications();
			net.lock().send_finality_notifications();
			net.lock().sync_without_disconnects();
			Ok(())
		})
		.map(|_| ())
		.map_err(|_| ());
	(before_waiting)();
	runtime.block_on(wait_for.select(drive_to_completion).map_err(|_| ())).unwrap();
	let highest_finalized = *highest_finalized.read();
	highest_finalized
}
fn run_to_completion_with < F : FnOnce ( ) > ( blocks : u64 , net : Arc < Mutex < GrandpaTestNet > > , peers : & [ AuthorityKeyring ] , before_waiting : F , ) -> u64 { use parking_lot :: RwLock ; let mut finality_notifications = Vec :: new ( ) ; let mut runtime = current_thread :: Runtime :: new ( ) . unwrap ( ) ; let highest_finalized = Arc :: new ( RwLock :: new ( 0 ) ) ; for ( peer_id , key ) in peers . iter ( ) . enumerate ( ) { let highest_finalized = highest_finalized . clone ( ) ; let ( client , link ) = { let net = net . lock ( ) ; // temporary needed for some reason let link = net . peers [ peer_id ] . data . lock ( ) . take ( ) . expect ( " " ) ; ( net . peers [ peer_id ] . client ( ) . clone ( ) , link , ) } ; finality_notifications . push ( client . finality_notification_stream ( ) . take_while ( move | n | { let mut highest_finalized = highest_finalized . write ( ) ; if * n . header . number ( ) > * highest_finalized { * highest_finalized = * n . header . number ( ) ; } Ok ( n . header . number ( ) < & blocks ) } ) . for_each ( | _ | Ok ( ( ) ) ) ) ; fn assert_send < T : Send > ( _ : & T ) { } let voter = run_grandpa ( Config { gossip_duration : TEST_GOSSIP_DURATION , justification_period : 32 , local_key : Some ( Arc :: new ( key . clone ( ) . into ( ) ) ) , name : Some ( format ! ( " " peer_id ) ) , } , link , MessageRouting :: new ( net . clone ( ) , peer_id ) , InherentDataProviders :: new ( ) , futures :: empty ( ) , ) . expect ( " " ) ; assert_send ( & voter ) ; runtime . spawn ( voter ) ; } // wait for all finalized on each. let wait_for = :: futures :: future :: join_all ( finality_notifications ) . map ( | _ | ( ) ) . map_err ( | _ | ( ) ) ; let drive_to_completion = :: tokio :: timer :: Interval :: new_interval ( TEST_ROUTING_INTERVAL ) . for_each ( move | _ | { net . lock ( ) . send_import_notifications ( ) ; net . lock ( ) . send_finality_notifications ( ) ; net . lock ( ) . sync_without_disconnects ( ) ; Ok ( ( ) ) } ) . map ( | _ | ( ) ) . map_err ( | _ | ( ) ) ; ( before_waiting ) ( ) ; runtime . block_on ( wait_for . select ( drive_to_completion ) . map_err ( | _ | ( ) ) ) . unwrap ( ) ; let highest_finalized = * highest_finalized . read ( ) ; highest_finalized }
run the voters to completion.
run the voters to completion .
blocks
u64
net
Arc<Mutex<GrandpaTestNet>>
peers
&[AuthorityKeyring]
before_waiting
F
blocks
u64
net
Arc<Mutex<GrandpaTestNet>>
peers
&[AuthorityKeyring]
before_waiting
F