[api] use shared communicator for Rust client, add session close

src/client/client.rs

@@ -1,62 +1,30 @@
-use capnp::capability::Promise;
-use capnp_rpc::rpc_twoparty_capnp;
-use common::rpc::new_rpc_system;
-use futures::Future;
 use std::error::Error;
 use std::net::SocketAddr;
-use tokio_core::net::TcpStream;
-use tokio_core::reactor::Core;
 
 use CLIENT_PROTOCOL_VERSION;
+use common::wrapped::WrappedRcRefCell;
 use super::session::Session;
+use super::communicator::Communicator;
 
 pub struct Client {
- core: Core,
- service: ::client_capnp::client_service::Client,
+ comm: WrappedRcRefCell<Communicator>,
 }
 
 impl Client {
- pub fn new(scheduler: &SocketAddr) -> Result<Self, Box<Error>> {
- let mut core = Core::new()?;
- let handle = core.handle();
- let stream = core.run(TcpStream::connect(&scheduler, &handle))?;
- stream.set_nodelay(true)?;
+ pub fn new(scheduler: SocketAddr) -> Result<Self, Box<Error>> {
+ let comm = WrappedRcRefCell::wrap(Communicator::new(scheduler, CLIENT_PROTOCOL_VERSION)?);
 
- debug!("Connection to server {} established", scheduler);
-
- let mut rpc = Box::new(new_rpc_system(stream, None));
- let bootstrap: ::server_capnp::server_bootstrap::Client =
- rpc.bootstrap(rpc_twoparty_capnp::Side::Server);
- handle.spawn(rpc.map_err(|err| panic!("RPC error: {}", err)));
-
- let mut request = bootstrap.register_as_client_request();
- request.get().set_version(CLIENT_PROTOCOL_VERSION);
-
- let service = core.run(
- request
- .send()
- .promise
- .and_then(|response| Promise::ok(pry!(response.get()).get_service())),
- )??;
-
- Ok(Client { core, service })
+ Ok(Client {
+ comm
+ })
  }
 
- pub fn new_session(&mut self) -> Result<Session, Box<Error>> {
- let id: i32 = self.core.run(
- self.service
- .new_session_request()
- .send()
- .promise
- .and_then(|response| Promise::ok(pry!(response.get()).get_session_id())),
- )?;
-
- Ok(Session { id })
+ pub fn new_session(&self) -> Result<Session, Box<Error>> {
+ let session_id = self.comm.get_mut().new_session()?;
+ Ok(Session::new(session_id, self.comm.clone()))
  }
 
- pub fn terminate_server(&mut self) -> Result<(), Box<Error>> {
- self.core
- .run(self.service.terminate_server_request().send().promise)?;
- Ok(())
+ pub fn terminate_server(&self) -> Result<(), Box<Error>> {
+ self.comm.get_mut().terminate_server()
  }
 }

src/client/communicator.rs

@@ -0,0 +1,69 @@
+use tokio_core::reactor::Core;
+use std::net::SocketAddr;
+use tokio_core::net::TcpStream;
+use std::error::Error;
+use common::rpc::new_rpc_system;
+use capnp::capability::Promise;
+use capnp_rpc::rpc_twoparty_capnp;
+use futures::Future;
+
+pub struct Communicator {
+ core: Core,
+ service: ::client_capnp::client_service::Client,
+}
+
+impl Communicator {
+ pub fn new(scheduler: SocketAddr, version: i32) -> Result<Self, Box<Error>> {
+ let mut core = Core::new()?;
+ let handle = core.handle();
+ let stream = core.run(TcpStream::connect(&scheduler, &handle))?;
+ stream.set_nodelay(true)?;
+
+ debug!("Connection to server {} established", scheduler);
+
+ let mut rpc = Box::new(new_rpc_system(stream, None));
+ let bootstrap: ::server_capnp::server_bootstrap::Client =
+ rpc.bootstrap(rpc_twoparty_capnp::Side::Server);
+ handle.spawn(rpc.map_err(|err| panic!("RPC error: {}", err)));
+
+ let mut request = bootstrap.register_as_client_request();
+ request.get().set_version(version);
+
+ let service = core.run(
+ request
+ .send()
+ .promise
+ .and_then(|response| Promise::ok(pry!(response.get()).get_service())),
+ )??;
+
+ Ok(Self { core, service })
+ }
+
+ pub fn new_session(&mut self) -> Result<i32, Box<Error>> {
+ let id: i32 = self.core.run(
+ self.service
+ .new_session_request()
+ .send()
+ .promise
+ .and_then(|response| Promise::ok(pry!(response.get()).get_session_id())),
+ )?;
+
+ Ok(id)
+ }
+
+ pub fn close_session(&mut self, id: i32) -> Result<bool, Box<Error>> {
+ self.core.run({
+ let mut req = self.service.close_session_request();
+ req.get().set_session_id(id);
+ req.send().promise
+ })?;
+
+ Ok(true)
+ }
+
+ pub fn terminate_server(&mut self) -> Result<(), Box<Error>> {
+ self.core
+ .run(self.service.terminate_server_request().send().promise)?;
+ Ok(())
+ }
+}

src/client/mod.rs

@@ -1,2 +1,4 @@
 pub mod client;
 pub mod session;
+
+mod communicator;

src/client/session.rs

@@ -1,3 +1,25 @@
+use common::wrapped::WrappedRcRefCell;
+
+use super::communicator::Communicator;
+
 pub struct Session {
- pub id: i32,
+ id: i32,
+ comm: WrappedRcRefCell<Communicator>
+}
+
+impl Session {
+ pub fn new(id: i32, comm: WrappedRcRefCell<Communicator>) -> Self {
+ debug!("Session {} created", id);
+
+ Session {
+ id, comm
+ }
+ }
+}
+
+impl Drop for Session {
+ fn drop(&mut self) {
+ self.comm.get_mut().close_session(self.id).unwrap();
+ debug!("Session {} destroyed", self.id);
+ }
 }

src/server/rpc/client.rs

@@ -521,10 +521,11 @@ impl client_service::Server for ClientServiceImpl {
  Promise::ok(())
  }
 
+ #[allow(unreachable_code)]
  fn terminate_server(
  &mut self,
- params: client_service::TerminateServerParams,
- results: client_service::TerminateServerResults,
+ _params: client_service::TerminateServerParams,
+ _results: client_service::TerminateServerResults,
  ) -> Promise<(), ::capnp::Error> {
  exit(0);
  Promise::ok(())