poll repeatedly when state changes

substrate/candidate-agreement/src/bft/mod.rs

@@ -214,7 +214,7 @@ impl<D, S> Locked<D, S> {
 // the state of the local node during the current state of consensus.
 //
 // behavior is different when locked on a proposal.
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, PartialEq, Eq)]
 enum LocalState {
 	Start,
 	Proposed,
@@ -327,6 +327,39 @@ impl<C: Context> Strategy<C> {
 		where
 			C::RoundTimeout: Future<Error=E>,
 			C::Proposal: Future<Error=E>,
+	{
+		let mut last_watermark = (
+			self.current_accumulator.round_number(),
+			self.local_state
+		);
+
+		// poll until either completion or state doesn't change.
+		loop {
+			match self.poll_once(context, sending)? {
+				Async::Ready(x) => return Ok(Async::Ready(x)),
+				Async::NotReady => {
+					let new_watermark = (
+						self.current_accumulator.round_number(),
+						self.local_state
+					);
+
+					if new_watermark == last_watermark {
+						return Ok(Async::NotReady)
+					} else {
+						last_watermark = new_watermark;
+					}
+				}
+			}
+		}
+	}
+
+	// perform one round of polling: attempt to broadcast messages and change the state.
+	// if the round or internal round-state changes, this should be called again.
+	fn poll_once<E>(&mut self, context: &C, sending: &mut Sending<ContextCommunication<C>>)
+		-> Poll<Committed<C::Candidate, C::Digest, C::Signature>, E>
+		where
+			C::RoundTimeout: Future<Error=E>,
+			C::Proposal: Future<Error=E>,
 	{
 		self.propose(context, sending)?;
 		self.prepare(context, sending);