SecretStore: ClusterSession::wait_session helper

secret_store/src/key_server_cluster/admin_sessions/key_version_negotiation_session.rs

@@ -198,13 +198,7 @@ impl<T> SessionImpl<T> where T: SessionTransport {
 
 	/// Wait for session completion.
 	pub fn wait(&self) -> Result<(H256, NodeId), Error> {
-		let mut data = self.data.lock();
+		Self::wait_session(&self.core.completed, &self.data, None, |data| data.result.clone())
-		if !data.result.is_some() {
-			self.core.completed.wait(&mut data);
-		}
-
-		data.result.clone()
-			.expect("checked above or waited for completed; completed is only signaled when result.is_some(); qed")
 	}
 
 	/// Initialize session.

secret_store/src/key_server_cluster/admin_sessions/servers_set_change_session.rs

@@ -207,13 +207,7 @@ impl SessionImpl {
 
 	/// Wait for session completion.
 	pub fn wait(&self) -> Result<(), Error> {
-		let mut data = self.data.lock();
+		Self::wait_session(&self.core.completed, &self.data, None, |data| data.result.clone())
-		if !data.result.is_some() {
-			self.core.completed.wait(&mut data);
-		}
-
-		data.result.clone()
-			.expect("checked above or waited for completed; completed is only signaled when result.is_some(); qed")
 	}
 
 	/// Initialize servers set change session on master node.

secret_store/src/key_server_cluster/client_sessions/decryption_session.rs

@@ -202,14 +202,7 @@ impl SessionImpl {
 
 	/// Wait for session completion.
 	pub fn wait(&self) -> Result<EncryptedDocumentKeyShadow, Error> {
-		let mut data = self.data.lock();
+		Self::wait_session(&self.core.completed, &self.data, None, |data| data.result.clone())
-		if !data.result.is_some() {
-			self.core.completed.wait(&mut data);
-		}
-
-		data.result.as_ref()
-			.expect("checked above or waited for completed; completed is only signaled when result.is_some(); qed")
-			.clone()
 	}
 
 	/// Delegate session to other node.

secret_store/src/key_server_cluster/client_sessions/encryption_session.rs

@@ -132,17 +132,7 @@ impl SessionImpl {
 
 	/// Wait for session completion.
 	pub fn wait(&self, timeout: Option<time::Duration>) -> Result<(), Error> {
-		let mut data = self.data.lock();
+		Self::wait_session(&self.completed, &self.data, timeout, |data| data.result.clone())
-		if !data.result.is_some() {
-			match timeout {
-				None => self.completed.wait(&mut data),
-				Some(timeout) => { self.completed.wait_for(&mut data, timeout); },
-			}
-		}
-
-		data.result.as_ref()
-			.expect("checked above or waited for completed; completed is only signaled when result.is_some(); qed")
-			.clone()
 	}
 
 

secret_store/src/key_server_cluster/client_sessions/generation_session.rs

@@ -223,17 +223,8 @@ impl SessionImpl {
 
 	/// Wait for session completion.
 	pub fn wait(&self, timeout: Option<time::Duration>) -> Result<Public, Error> {
-		let mut data = self.data.lock();
+		Self::wait_session(&self.completed, &self.data, timeout, |data| data.joint_public_and_secret.clone()
-		if !data.joint_public_and_secret.is_some() {
+			.map(|r| r.map(|r| r.0.clone())))
-			match timeout {
-				None => self.completed.wait(&mut data),
-				Some(timeout) => { self.completed.wait_for(&mut data, timeout); },
-			}
-		}
-
-		data.joint_public_and_secret.clone()
-			.expect("checked above or waited for completed; completed is only signaled when joint_public.is_some(); qed")
-			.map(|p| p.0)
 	}
 
 	/// Get generated public and secret (if any).

secret_store/src/key_server_cluster/client_sessions/signing_session.rs

@@ -207,14 +207,7 @@ impl SessionImpl {
 
 	/// Wait for session completion.
 	pub fn wait(&self) -> Result<(Secret, Secret), Error> {
-		let mut data = self.data.lock();
+		Self::wait_session(&self.core.completed, &self.data, None, |data| data.result.clone())
-		if !data.result.is_some() {
-			self.core.completed.wait(&mut data);
-		}
-
-		data.result.as_ref()
-			.expect("checked above or waited for completed; completed is only signaled when result.is_some(); qed")
-			.clone()
 	}
 
 	/// Delegate session to other node.

secret_store/src/key_server_cluster/cluster_sessions.rs

@@ -18,7 +18,7 @@ use std::time;
 use std::sync::{Arc, Weak};
 use std::sync::atomic::AtomicBool;
 use std::collections::{VecDeque, BTreeMap};
-use parking_lot::{Mutex, RwLock};
+use parking_lot::{Mutex, RwLock, Condvar};
 use bigint::hash::H256;
 use ethkey::{Secret, Signature};
 use key_server_cluster::{Error, NodeId, SessionId};
@@ -79,6 +79,25 @@ pub trait ClusterSession {
 	fn on_session_error(&self, sender: &NodeId, error: Error);
 	/// Process session message.
 	fn on_message(&self, sender: &NodeId, message: &Message) -> Result<(), Error>;
+
+	/// 'Wait for session completion' helper.
+	fn wait_session<T, U, F: Fn(&U) -> Option<Result<T, Error>>>(completion_event: &Condvar, session_data: &Mutex<U>, timeout: Option<time::Duration>, result_reader: F) -> Result<T, Error> {
+		let mut locked_data = session_data.lock();
+		match result_reader(&locked_data) {
+			Some(result) => result,
+			None => {
+				match timeout {
+					None => completion_event.wait(&mut locked_data),
+					Some(timeout) => {
+						completion_event.wait_for(&mut locked_data, timeout);
+					},
+				}
+
+				result_reader(&locked_data)
+					.expect("waited for completion; completion is only signaled when result.is_some(); qed")
+			},
+		}
+	}
 }
 
 /// Administrative session.
@@ -529,7 +548,6 @@ pub fn create_cluster_view(data: &Arc<ClusterData>, requires_all_connections: bo
 	Ok(Arc::new(ClusterView::new(data.clone(), connected_nodes)))
 }
 
-
 #[cfg(test)]
 mod tests {
 	use std::sync::Arc;