src/testing/sl4f/src/fidl/sl4f.rs - fuchsia - Git at Google

 // Copyright 2020 The Fuchsia Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.

 use crate::common_utils::common::{read_json_from_vmo, write_json_to_vmo};
 use crate::server::constants::CONCURRENT_REQ_LIMIT;
 use crate::server::Facade;
 use anyhow::Error;
 use async_trait::async_trait;
 use fidl_fuchsia_testing_sl4f::{
     FacadeIteratorRequest, FacadeProviderRequest, FacadeProviderRequestStream,
 };
 use futures::stream::{StreamExt, TryStreamExt};
 use serde_json::Value;
 use std::sync::Arc;
 use tracing::{error, info, warn};

 /// Trait for the implementation of the FacadeProvider protocol.
 #[async_trait(?Send)]
 pub trait FacadeProvider {
     /// Returns the object implementing Facade for the given facade name.
     /// # Arguments:
     /// * 'name' - A string representing the name of the facade.
     fn get_facade(&self, name: &str) -> Option<Arc<dyn Facade>>;

     /// Returns an iterator over all of the Facade implementations.
     fn get_facades(&self) -> Box<dyn Iterator<Item = &Arc<dyn Facade>> + '_>;

     /// Returns an iterator over all of the facade names.
     fn get_facade_names(&self) -> Vec<String>;

     /// Invoked on FacadeProvider.GetFacades(). Sends the list of hosted facades back to the client on
     /// subsequent calls to FacadeIterator.GetNext() over the channel given to GetFacades().
     /// # Arguments
     /// * 'request' - A request from the FacadeProvider client. Must be a GetFacades() request.
     async fn get_facades_impl(&self, request: FacadeProviderRequest) {
         let iterator = match request {
             FacadeProviderRequest::GetFacades { iterator, control_handle: _ } => iterator,
             _ => panic!(
                 "get_facade_impl() must only be called with a FacadeProviderRequest::GetFacades."
             ),
         };

         // Wrap operation in an async block in order to capture any error.
         let get_facades_fut = async {
             let mut iterator = iterator.into_stream()?;
             if let Some(FacadeIteratorRequest::GetNext { responder }) = iterator.try_next().await? {
                 // NOTE: if the list of facade names would exceed the channel buffer size,
                 // they should be split over back-to-back responses to GetNext().
                 responder.send(&self.get_facade_names())?;
                 if let Some(FacadeIteratorRequest::GetNext { responder }) =
                     iterator.try_next().await?
                 {
                     responder.send(&[])?; // Indicates completion.
                 }
             }
             Ok::<(), Error>(())
         };
         if let Err(error) = get_facades_fut.await {
             error!(%error, "Failed to handle GetFacades()");
         }
     }

     /// Invoked on FacadeProvider.Execute(). Executes the given command on a hosted facade.
     /// # Arguments
     /// * 'request' - A request from a FacadeProvider client. Must be an Execute() request.
     async fn execute_impl(&self, request: FacadeProviderRequest) {
         let (facade, command, params_blob, responder) = match request {
             FacadeProviderRequest::Execute { facade, command, params_blob, responder } => {
                 (facade, command, params_blob, responder)
             }
             _ => {
                 panic!("execute_impl() must only be called with a FacadeProviderRequest::Execute.")
             }
         };

         // Look-up the facade.
         let facade = if let Some(f) = self.get_facade(&facade) {
             f
         } else {
             let err_str = format!("Could not find facade: {}", facade);
             error!("{}", err_str);
             if let Err(send_error) = responder.send(None, Some(&err_str)) {
                 error!("Failed to send response with: {}", send_error);
             }
             return;
         };

         // Construct a JSON Value out of the params_blob.
         let params = match read_json_from_vmo(&params_blob) {
             Ok(value) => value,
             Err(error) => {
                 if let Err(send_error) =
                     responder.send(None, Some(&format!("Failed to extract params: {}", error)))
                 {
                     error!(error = %send_error, "Failed to send response");
                 }
                 return;
             }
         };

         // Execute the command on the facade. On error or empty result, send the response.
         let result = match facade.handle_request(command, params).await {
             Ok(Value::Null) => {
                 if let Err(error) = responder.send(None, None) {
                     error!(%error, "Failed to send response");
                 }
                 return;
             }
             Ok(result) => result,
             Err(error) => {
                 if let Err(error) = responder.send(None, Some(&error.to_string())) {
                     error!(%error, "Failed to send response");
                 }
                 return;
             }
         };

         // Write the result blob into a VMO and send the response.
         if let Err(send_error) = match write_json_to_vmo(&params_blob, &result) {
             Ok(()) => responder.send(Some(params_blob), None),
             Err(error) => responder.send(None, Some(&format!("Failed to write result: {}", error))),
         } {
             error!(error = %send_error, "Failed to send response");
         }
     }

     /// Cleans up transient state on all hosted facades. Invoked on FacadeProvider.Cleanup().
     fn cleanup_impl(&self) {
         for facade in self.get_facades() {
             facade.cleanup();
         }
     }

     /// Prints state for all hosted facades. Invoked on FacadeProvider.Print().
     fn print_impl(&self) {
         for facade in self.get_facades() {
             facade.print();
         }
     }

     /// Invoked on each incoming request. Invokes the appropriate handler code.
     /// # Arguments
     /// * 'request' - Incoming request on the FacadeProvider connection.
     async fn handle_request(&self, request: FacadeProviderRequest) {
         match request {
             FacadeProviderRequest::GetFacades { iterator, control_handle } => {
                 self.get_facades_impl(FacadeProviderRequest::GetFacades {
                     iterator,
                     control_handle,
                 })
                 .await;
             }
             FacadeProviderRequest::Execute { facade, command, params_blob, responder } => {
                 info!("Received command {}.{}", facade, command);
                 self.execute_impl(FacadeProviderRequest::Execute {
                     facade,
                     command,
                     params_blob,
                     responder,
                 })
                 .await;
             }
             FacadeProviderRequest::Cleanup { responder } => {
                 self.cleanup_impl();
                 if let Err(error) = responder.send() {
                     warn!(%error, "Failed to notify completion of Cleanup()");
                 }
             }
             FacadeProviderRequest::Print { responder } => {
                 self.print_impl();
                 if let Err(error) = responder.send() {
                     error!(%error, "Failed to notify completion of Print()");
                 }
             }
         }
     }

     /// Invoked on an incoming FacadeProvider connection request. Requests arriving on the stream
     /// will be handled concurrently.
     /// NOTE: The main SL4F server doesn't appear to wait before any outstanding requests are
     /// completed before allowing a cleanup operation to move forward, and this code similarly
     /// makes no such effort. As such, it is up to the test harness to ensure that cleanup and
     /// command requests do not interleave in order to ensure that the facades and device remain in
     /// a consistent state.
     /// # Arguments
     /// * 'stream' - Incoming FacadeProvider request stream.
     async fn run_facade_provider(&self, stream: FacadeProviderRequestStream) {
         stream
             .for_each_concurrent(CONCURRENT_REQ_LIMIT, |request| {
                 self.handle_request(request.unwrap())
             })
             .await;
     }
 }

 #[cfg(test)]
 mod tests {
     use super::*;
     use anyhow::format_err;
     use fidl_fuchsia_testing_sl4f::FacadeProviderMarker;
     use fuchsia_async as fasync;
     use fuchsia_zircon as zx;
     use serde_json::json;
     use std::cell::RefCell;
     use std::collections::HashMap;
     use std::pin::pin;
     use std::task::Poll;

     /// TestFacade provides a trivial Facade implementation which supports commands to interact
     /// with the state.
     #[derive(Debug)]
     struct TestFacade {
         // Trivial state accessed through TestFacade's implementation of Facade.
         state: RefCell<bool>,
     }

     impl TestFacade {
         pub fn new() -> TestFacade {
             TestFacade { state: RefCell::new(false) }
         }
     }

     #[async_trait(?Send)]
     impl Facade for TestFacade {
         async fn handle_request(&self, method: String, args: Value) -> Result<Value, Error> {
             match method.as_str() {
                 "set" => {
                     if let Value::Bool(new_state) = args {
                         *self.state.borrow_mut() = new_state;
                         return Ok(json!(null));
                     }
                     panic!("Received invalid args");
                 }
                 "get" => {
                     if let Value::Null = args {
                         return Ok(json!(*self.state.borrow()));
                     }
                     panic!("Received invalid args");
                 }
                 _ => return Err(format_err!("Invalid TestFacade method {}", method)),
             }
         }

         fn cleanup(&self) {
             *self.state.borrow_mut() = false;
         }

         fn print(&self) {}
     }

     /// TestFacadeProvider provides a simple implementation of the FacadeProvider trait and hosts
     /// the TestFacade.
     struct TestFacadeProvider {
         facades: HashMap<String, Arc<dyn Facade>>,
     }

     impl TestFacadeProvider {
         pub fn new() -> TestFacadeProvider {
             let mut facades: HashMap<String, Arc<dyn Facade>> = HashMap::new();
             facades
                 .insert("test_facade".to_string(), Arc::new(TestFacade::new()) as Arc<dyn Facade>);
             TestFacadeProvider { facades }
         }
     }

     #[async_trait(?Send)]
     impl FacadeProvider for TestFacadeProvider {
         fn get_facade(&self, name: &str) -> Option<Arc<dyn Facade>> {
             self.facades.get(name).map(Arc::clone)
         }

         fn get_facades(&self) -> Box<dyn Iterator<Item = &Arc<dyn Facade>> + '_> {
             Box::new(self.facades.values())
         }

         fn get_facade_names(&self) -> Vec<String> {
             self.facades.keys().cloned().collect()
         }
     }

     /// This test exercises the default FacadeProvider server implementation provided in the
     /// FacadeProvider trait using TestFacadeProvider and TestFacade. It creates server and client
     /// endpoints to a channel, passing each respectively to async blocks for the server and client
     /// functionality. These async blocks are joined and passed to a single-threaded executor. The
     /// server simply waits on its endpoint. The client does the following:
     /// 1. Gets and verifies initial state.
     /// 2. Sets a new state.
     /// 3. Gets and verifies a new state.
     /// 4. Cleans up state.
     /// 5. Gets and verifies the clean state.
     /// 6. Releases the client end so that the server end can exit.
     #[test]
     fn test_facade_provider() -> Result<(), Error> {
         let mut executor = fasync::TestExecutor::new();

         let (proxy, stream) = fidl::endpoints::create_proxy_and_stream::<FacadeProviderMarker>()?;
         let server_fut = async {
             // Run the FacadeProvider server.
             let sl4f = TestFacadeProvider::new();
             sl4f.run_facade_provider(stream).await;
         };
         let client_fut = async {
             // Get the initial state.
             let vmo = zx::Vmo::create_with_opts(zx::VmoOptions::RESIZABLE, 0)?;
             write_json_to_vmo(&vmo, &json!(null))?;
             if let (Some(vmo), None) = proxy.execute("test_facade", "get", vmo).await? {
                 assert_eq!(false, read_json_from_vmo(&vmo)?.as_bool().unwrap());
             } else {
                 panic!("Failed to get initial state.");
             }

             // Set the new state.
             let vmo = zx::Vmo::create_with_opts(zx::VmoOptions::RESIZABLE, 0)?;
             write_json_to_vmo(&vmo, &json!(true))?;
             if let (None, None) = proxy.execute("test_facade", "set", vmo).await? {
             } else {
                 panic!("Failed to set new state.");
             }

             // Get the new state.
             let vmo = zx::Vmo::create_with_opts(zx::VmoOptions::RESIZABLE, 0)?;
             write_json_to_vmo(&vmo, &json!(null))?;
             if let (Some(vmo), None) = proxy.execute("test_facade", "get", vmo).await? {
                 assert_eq!(true, read_json_from_vmo(&vmo)?.as_bool().unwrap());
             } else {
                 panic!("Failed to get new state.");
             }

             // Clean up the state.
             proxy.cleanup().await?;

             // Get the final state.
             let vmo = zx::Vmo::create_with_opts(zx::VmoOptions::RESIZABLE, 0)?;
             write_json_to_vmo(&vmo, &json!(null))?;
             if let (Some(vmo), None) = proxy.execute("test_facade", "get", vmo).await? {
                 assert_eq!(false, read_json_from_vmo(&vmo)?.as_bool().unwrap());
             } else {
                 panic!("Failed to get initial state.");
             }

             // Close the proxy.
             drop(proxy);
             Ok::<(), Error>(())
         };
         let mut combined_fut = pin!(async {
             let (_, res) = futures::join!(server_fut, client_fut);
             res.unwrap();
         });

         assert_eq!(Poll::Ready(()), executor.run_until_stalled(&mut combined_fut));

         Ok(())
     }
 }
	// Copyright 2020 The Fuchsia Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style license that can be
	// found in the LICENSE file.

	use crate::common_utils::common::{read_json_from_vmo, write_json_to_vmo};
	use crate::server::constants::CONCURRENT_REQ_LIMIT;
	use crate::server::Facade;
	use anyhow::Error;
	use async_trait::async_trait;
	use fidl_fuchsia_testing_sl4f::{
	FacadeIteratorRequest, FacadeProviderRequest, FacadeProviderRequestStream,
	};
	use futures::stream::{StreamExt, TryStreamExt};
	use serde_json::Value;
	use std::sync::Arc;
	use tracing::{error, info, warn};

	/// Trait for the implementation of the FacadeProvider protocol.
	#[async_trait(?Send)]
	pub trait FacadeProvider {
	/// Returns the object implementing Facade for the given facade name.
	/// # Arguments:
	/// * 'name' - A string representing the name of the facade.
	fn get_facade(&self, name: &str) -> Option<Arc<dyn Facade>>;

	/// Returns an iterator over all of the Facade implementations.
	fn get_facades(&self) -> Box<dyn Iterator<Item = &Arc<dyn Facade>> + '_>;

	/// Returns an iterator over all of the facade names.
	fn get_facade_names(&self) -> Vec<String>;

	/// Invoked on FacadeProvider.GetFacades(). Sends the list of hosted facades back to the client on
	/// subsequent calls to FacadeIterator.GetNext() over the channel given to GetFacades().
	/// # Arguments
	/// * 'request' - A request from the FacadeProvider client. Must be a GetFacades() request.
	async fn get_facades_impl(&self, request: FacadeProviderRequest) {
	let iterator = match request {
	FacadeProviderRequest::GetFacades { iterator, control_handle: _ } => iterator,
	_ => panic!(
	"get_facade_impl() must only be called with a FacadeProviderRequest::GetFacades."
	),
	};

	// Wrap operation in an async block in order to capture any error.
	let get_facades_fut = async {
	let mut iterator = iterator.into_stream()?;
	if let Some(FacadeIteratorRequest::GetNext { responder }) = iterator.try_next().await? {
	// NOTE: if the list of facade names would exceed the channel buffer size,
	// they should be split over back-to-back responses to GetNext().
	responder.send(&self.get_facade_names())?;
	if let Some(FacadeIteratorRequest::GetNext { responder }) =
	iterator.try_next().await?
	{
	responder.send(&[])?; // Indicates completion.
	}
	}
	Ok::<(), Error>(())
	};
	if let Err(error) = get_facades_fut.await {
	error!(%error, "Failed to handle GetFacades()");
	}
	}

	/// Invoked on FacadeProvider.Execute(). Executes the given command on a hosted facade.
	/// # Arguments
	/// * 'request' - A request from a FacadeProvider client. Must be an Execute() request.
	async fn execute_impl(&self, request: FacadeProviderRequest) {
	let (facade, command, params_blob, responder) = match request {
	FacadeProviderRequest::Execute { facade, command, params_blob, responder } => {
	(facade, command, params_blob, responder)
	}
	_ => {
	panic!("execute_impl() must only be called with a FacadeProviderRequest::Execute.")
	}
	};

	// Look-up the facade.
	let facade = if let Some(f) = self.get_facade(&facade) {
	f
	} else {
	let err_str = format!("Could not find facade: {}", facade);
	error!("{}", err_str);
	if let Err(send_error) = responder.send(None, Some(&err_str)) {
	error!("Failed to send response with: {}", send_error);
	}
	return;
	};

	// Construct a JSON Value out of the params_blob.
	let params = match read_json_from_vmo(&params_blob) {
	Ok(value) => value,
	Err(error) => {
	if let Err(send_error) =
	responder.send(None, Some(&format!("Failed to extract params: {}", error)))
	{
	error!(error = %send_error, "Failed to send response");
	}
	return;
	}
	};

	// Execute the command on the facade. On error or empty result, send the response.
	let result = match facade.handle_request(command, params).await {
	Ok(Value::Null) => {
	if let Err(error) = responder.send(None, None) {
	error!(%error, "Failed to send response");
	}
	return;
	}
	Ok(result) => result,
	Err(error) => {
	if let Err(error) = responder.send(None, Some(&error.to_string())) {
	error!(%error, "Failed to send response");
	}
	return;
	}
	};

	// Write the result blob into a VMO and send the response.
	if let Err(send_error) = match write_json_to_vmo(&params_blob, &result) {
	Ok(()) => responder.send(Some(params_blob), None),
	Err(error) => responder.send(None, Some(&format!("Failed to write result: {}", error))),
	} {
	error!(error = %send_error, "Failed to send response");
	}
	}

	/// Cleans up transient state on all hosted facades. Invoked on FacadeProvider.Cleanup().
	fn cleanup_impl(&self) {
	for facade in self.get_facades() {
	facade.cleanup();
	}
	}

	/// Prints state for all hosted facades. Invoked on FacadeProvider.Print().
	fn print_impl(&self) {
	for facade in self.get_facades() {
	facade.print();
	}
	}

	/// Invoked on each incoming request. Invokes the appropriate handler code.
	/// # Arguments
	/// * 'request' - Incoming request on the FacadeProvider connection.
	async fn handle_request(&self, request: FacadeProviderRequest) {
	match request {
	FacadeProviderRequest::GetFacades { iterator, control_handle } => {
	self.get_facades_impl(FacadeProviderRequest::GetFacades {
	iterator,
	control_handle,
	})
	.await;
	}
	FacadeProviderRequest::Execute { facade, command, params_blob, responder } => {
	info!("Received command {}.{}", facade, command);
	self.execute_impl(FacadeProviderRequest::Execute {
	facade,
	command,
	params_blob,
	responder,
	})
	.await;
	}
	FacadeProviderRequest::Cleanup { responder } => {
	self.cleanup_impl();
	if let Err(error) = responder.send() {
	warn!(%error, "Failed to notify completion of Cleanup()");
	}
	}
	FacadeProviderRequest::Print { responder } => {
	self.print_impl();
	if let Err(error) = responder.send() {
	error!(%error, "Failed to notify completion of Print()");
	}
	}
	}
	}

	/// Invoked on an incoming FacadeProvider connection request. Requests arriving on the stream
	/// will be handled concurrently.
	/// NOTE: The main SL4F server doesn't appear to wait before any outstanding requests are
	/// completed before allowing a cleanup operation to move forward, and this code similarly
	/// makes no such effort. As such, it is up to the test harness to ensure that cleanup and
	/// command requests do not interleave in order to ensure that the facades and device remain in
	/// a consistent state.
	/// # Arguments
	/// * 'stream' - Incoming FacadeProvider request stream.
	async fn run_facade_provider(&self, stream: FacadeProviderRequestStream) {
	stream
	.for_each_concurrent(CONCURRENT_REQ_LIMIT, \|request\| {
	self.handle_request(request.unwrap())
	})
	.await;
	}
	}

	#[cfg(test)]
	mod tests {
	use super::*;
	use anyhow::format_err;
	use fidl_fuchsia_testing_sl4f::FacadeProviderMarker;
	use fuchsia_async as fasync;
	use fuchsia_zircon as zx;
	use serde_json::json;
	use std::cell::RefCell;
	use std::collections::HashMap;
	use std::pin::pin;
	use std::task::Poll;

	/// TestFacade provides a trivial Facade implementation which supports commands to interact
	/// with the state.
	#[derive(Debug)]
	struct TestFacade {
	// Trivial state accessed through TestFacade's implementation of Facade.
	state: RefCell<bool>,
	}

	impl TestFacade {
	pub fn new() -> TestFacade {
	TestFacade { state: RefCell::new(false) }
	}
	}

	#[async_trait(?Send)]
	impl Facade for TestFacade {
	async fn handle_request(&self, method: String, args: Value) -> Result<Value, Error> {
	match method.as_str() {
	"set" => {
	if let Value::Bool(new_state) = args {
	*self.state.borrow_mut() = new_state;
	return Ok(json!(null));
	}
	panic!("Received invalid args");
	}
	"get" => {
	if let Value::Null = args {
	return Ok(json!(*self.state.borrow()));
	}
	panic!("Received invalid args");
	}
	_ => return Err(format_err!("Invalid TestFacade method {}", method)),
	}
	}

	fn cleanup(&self) {
	*self.state.borrow_mut() = false;
	}

	fn print(&self) {}
	}

	/// TestFacadeProvider provides a simple implementation of the FacadeProvider trait and hosts
	/// the TestFacade.
	struct TestFacadeProvider {
	facades: HashMap<String, Arc<dyn Facade>>,
	}

	impl TestFacadeProvider {
	pub fn new() -> TestFacadeProvider {
	let mut facades: HashMap<String, Arc<dyn Facade>> = HashMap::new();
	facades
	.insert("test_facade".to_string(), Arc::new(TestFacade::new()) as Arc<dyn Facade>);
	TestFacadeProvider { facades }
	}
	}

	#[async_trait(?Send)]
	impl FacadeProvider for TestFacadeProvider {
	fn get_facade(&self, name: &str) -> Option<Arc<dyn Facade>> {
	self.facades.get(name).map(Arc::clone)
	}

	fn get_facades(&self) -> Box<dyn Iterator<Item = &Arc<dyn Facade>> + '_> {
	Box::new(self.facades.values())
	}

	fn get_facade_names(&self) -> Vec<String> {
	self.facades.keys().cloned().collect()
	}
	}

	/// This test exercises the default FacadeProvider server implementation provided in the
	/// FacadeProvider trait using TestFacadeProvider and TestFacade. It creates server and client
	/// endpoints to a channel, passing each respectively to async blocks for the server and client
	/// functionality. These async blocks are joined and passed to a single-threaded executor. The
	/// server simply waits on its endpoint. The client does the following:
	/// 1. Gets and verifies initial state.
	/// 2. Sets a new state.
	/// 3. Gets and verifies a new state.
	/// 4. Cleans up state.
	/// 5. Gets and verifies the clean state.
	/// 6. Releases the client end so that the server end can exit.
	#[test]
	fn test_facade_provider() -> Result<(), Error> {
	let mut executor = fasync::TestExecutor::new();

	let (proxy, stream) = fidl::endpoints::create_proxy_and_stream::<FacadeProviderMarker>()?;
	let server_fut = async {
	// Run the FacadeProvider server.
	let sl4f = TestFacadeProvider::new();
	sl4f.run_facade_provider(stream).await;
	};
	let client_fut = async {
	// Get the initial state.
	let vmo = zx::Vmo::create_with_opts(zx::VmoOptions::RESIZABLE, 0)?;
	write_json_to_vmo(&vmo, &json!(null))?;
	if let (Some(vmo), None) = proxy.execute("test_facade", "get", vmo).await? {
	assert_eq!(false, read_json_from_vmo(&vmo)?.as_bool().unwrap());
	} else {
	panic!("Failed to get initial state.");
	}

	// Set the new state.
	let vmo = zx::Vmo::create_with_opts(zx::VmoOptions::RESIZABLE, 0)?;
	write_json_to_vmo(&vmo, &json!(true))?;
	if let (None, None) = proxy.execute("test_facade", "set", vmo).await? {
	} else {
	panic!("Failed to set new state.");
	}

	// Get the new state.
	let vmo = zx::Vmo::create_with_opts(zx::VmoOptions::RESIZABLE, 0)?;
	write_json_to_vmo(&vmo, &json!(null))?;
	if let (Some(vmo), None) = proxy.execute("test_facade", "get", vmo).await? {
	assert_eq!(true, read_json_from_vmo(&vmo)?.as_bool().unwrap());
	} else {
	panic!("Failed to get new state.");
	}

	// Clean up the state.
	proxy.cleanup().await?;

	// Get the final state.
	let vmo = zx::Vmo::create_with_opts(zx::VmoOptions::RESIZABLE, 0)?;
	write_json_to_vmo(&vmo, &json!(null))?;
	if let (Some(vmo), None) = proxy.execute("test_facade", "get", vmo).await? {
	assert_eq!(false, read_json_from_vmo(&vmo)?.as_bool().unwrap());
	} else {
	panic!("Failed to get initial state.");
	}

	// Close the proxy.
	drop(proxy);
	Ok::<(), Error>(())
	};
	let mut combined_fut = pin!(async {
	let (_, res) = futures::join!(server_fut, client_fut);
	res.unwrap();
	});

	assert_eq!(Poll::Ready(()), executor.run_until_stalled(&mut combined_fut));

	Ok(())
	}
	}