mirror of
https://github.com/OMGeeky/tarpc.git
synced 2026-02-15 22:18:20 +01:00
Cut down size of readme_expanded.
This commit is contained in:
Tim Kuehn
@@ -19,256 +19,58 @@ extern crate tokio_core;
|
||||
extern crate tokio_service;
|
||||
|
||||
use bincode::serde::DeserializeError;
|
||||
use futures::Future;
|
||||
use futures::{Future, IntoFuture};
|
||||
use std::io;
|
||||
use std::net::{SocketAddr, ToSocketAddrs};
|
||||
use std::thread;
|
||||
use tarpc::WireError;
|
||||
use std::net::SocketAddr;
|
||||
use tarpc::future::Connect;
|
||||
use tarpc::util::FirstSocketAddr;
|
||||
use tarpc::util::Never;
|
||||
use tokio_core::reactor::{Handle, Remote};
|
||||
use tokio_service::Service;
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
enum Request {
|
||||
Hello(String),
|
||||
}
|
||||
#[derive(Clone, Copy)]
|
||||
struct HelloServer;
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
enum Response {
|
||||
Hello(String),
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
enum Error {
|
||||
Hello(Never),
|
||||
}
|
||||
|
||||
/// Defines the `Future` RPC service. Implementors must be `Clone`, `Send`, and `'static`,
|
||||
/// as required by `tokio_proto::NewService`. This is required so that the service can be used
|
||||
/// to respond to multiple requests concurrently.
|
||||
pub trait FutureService: Send + Clone + 'static {
|
||||
type HelloFut: Future<Item = String, Error = Never>;
|
||||
fn hello(&self, name: String) -> Self::HelloFut;
|
||||
}
|
||||
|
||||
/// Provides a function for starting the service. This is a separate trait from
|
||||
/// `FutureService` to prevent collisions with the names of RPCs.
|
||||
pub trait FutureServiceExt: FutureService {
|
||||
fn listen(self, addr: SocketAddr) -> tarpc::ListenFuture {
|
||||
let (tx, rx) = futures::oneshot();
|
||||
tarpc::REMOTE.spawn(move |handle| Ok(tx.complete(Self::listen_with(self, addr, handle.clone()))));
|
||||
tarpc::ListenFuture::from_oneshot(rx)
|
||||
}
|
||||
|
||||
/// Spawns the service, binding to the given address and running on the default tokio `Loop`.
|
||||
fn listen_with(self, addr: SocketAddr, handle: Handle) -> io::Result<SocketAddr> {
|
||||
return tarpc::listen_with(addr, move || Ok(AsyncServer(self.clone())), handle);
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
struct AsyncServer<S>(S);
|
||||
|
||||
type Fut = futures::Finished<tarpc::Response<Response, Error>, io::Error>;
|
||||
|
||||
enum FutureReply<S: FutureService> {
|
||||
DeserializeError(Fut),
|
||||
Hello(futures::Then<S::HelloFut, Fut, fn(Result<String, Never>) -> Fut>),
|
||||
}
|
||||
|
||||
impl<S: FutureService> Future for FutureReply<S> {
|
||||
type Item = tarpc::Response<Response, Error>;
|
||||
type Error = io::Error;
|
||||
|
||||
fn poll(&mut self) -> futures::Poll<Self::Item, Self::Error> {
|
||||
match *self {
|
||||
FutureReply::DeserializeError(ref mut future) => future.poll(),
|
||||
FutureReply::Hello(ref mut future) => future.poll(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<S> Service for AsyncServer<S>
|
||||
where S: FutureService
|
||||
{
|
||||
type Request = Result<Request, DeserializeError>;
|
||||
type Response = tarpc::Response<Response, Error>;
|
||||
type Error = io::Error;
|
||||
type Future = FutureReply<S>;
|
||||
|
||||
fn call(&self, request: Self::Request) -> Self::Future {
|
||||
let request = match request {
|
||||
Ok(request) => request,
|
||||
Err(deserialize_err) => {
|
||||
let err = Err(WireError::ServerDeserialize(deserialize_err.to_string()));
|
||||
return FutureReply::DeserializeError(futures::finished(err));
|
||||
}
|
||||
};
|
||||
|
||||
match request {
|
||||
Request::Hello(name) => {
|
||||
fn wrap(response: Result<String, Never>) -> Fut {
|
||||
let fut = response.map(Response::Hello)
|
||||
.map_err(|error| WireError::App(Error::Hello(error)));
|
||||
futures::finished(fut)
|
||||
}
|
||||
return FutureReply::Hello(self.0.hello(name).then(wrap));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Defines the blocking RPC service. Must be `Clone`, `Send`, and `'static`,
|
||||
/// as required by `tokio_proto::NewService`. This is required so that the service can be used
|
||||
/// to respond to multiple requests concurrently.
|
||||
pub trait SyncService: Send + Clone + 'static {
|
||||
fn hello(&self, name: String) -> Result<String, Never>;
|
||||
}
|
||||
|
||||
/// Provides a function for starting the service. This is a separate trait from
|
||||
/// `SyncService` to prevent collisions with the names of RPCs.
|
||||
pub trait SyncServiceExt: SyncService {
|
||||
fn listen<L>(self, addr: L) -> io::Result<SocketAddr>
|
||||
where L: ToSocketAddrs
|
||||
{
|
||||
let addr = addr.try_first_socket_addr()?;
|
||||
impl HelloServer {
|
||||
fn listen(addr: SocketAddr) -> impl Future<Item=SocketAddr, Error=io::Error> {
|
||||
let (tx, rx) = futures::oneshot();
|
||||
tarpc::REMOTE.spawn(move |handle| {
|
||||
Ok(tx.complete(Self::listen_with(self, addr, handle.clone())))
|
||||
Ok(tx.complete(tarpc::listen_with(addr, move || Ok(HelloServer), handle.clone())))
|
||||
});
|
||||
tarpc::ListenFuture::from_oneshot(rx).wait()
|
||||
}
|
||||
|
||||
/// Spawns the service, binding to the given address and running on
|
||||
/// the default tokio `Loop`.
|
||||
fn listen_with<L>(self, addr: L, handle: Handle) -> io::Result<SocketAddr>
|
||||
where L: ToSocketAddrs
|
||||
{
|
||||
let service = SyncServer { service: self };
|
||||
return service.listen_with(addr.try_first_socket_addr()?, handle);
|
||||
|
||||
#[derive(Clone)]
|
||||
struct SyncServer<S> {
|
||||
service: S,
|
||||
}
|
||||
|
||||
impl<S> FutureService for SyncServer<S>
|
||||
where S: SyncService
|
||||
{
|
||||
type HelloFut = futures::Flatten<futures::MapErr<futures::Oneshot<futures::Done<String, Never>>, fn(futures::Canceled) -> Never>>;
|
||||
|
||||
fn hello(&self, name: String) -> Self::HelloFut {
|
||||
fn unimplemented(_: futures::Canceled) -> Never {
|
||||
unimplemented!()
|
||||
}
|
||||
|
||||
let (complete, promise) = futures::oneshot();
|
||||
let service = self.clone();
|
||||
const UNIMPLEMENTED: fn(futures::Canceled) -> Never = unimplemented;
|
||||
thread::spawn(move || {
|
||||
let reply = SyncService::hello(&service.service, name);
|
||||
complete.complete(futures::IntoFuture::into_future(reply));
|
||||
});
|
||||
promise.map_err(UNIMPLEMENTED).flatten()
|
||||
}
|
||||
}
|
||||
rx.map_err(|e| panic!(e)).and_then(|result| result)
|
||||
}
|
||||
}
|
||||
impl<A> FutureServiceExt for A where A: FutureService {}
|
||||
impl<S> SyncServiceExt for S where S: SyncService {}
|
||||
|
||||
type Client = tarpc::Client<Request, Response, Error>;
|
||||
|
||||
/// Implementation detail: Pending connection.
|
||||
pub struct ConnectFuture<T> {
|
||||
inner: futures::Map<tarpc::ConnectFuture<Request, Response, Error>, fn(Client) -> T>,
|
||||
}
|
||||
|
||||
impl<T> Future for ConnectFuture<T> {
|
||||
type Item = T;
|
||||
impl Service for HelloServer {
|
||||
type Request = Result<String, DeserializeError>;
|
||||
type Response = tarpc::Response<String, Never>;
|
||||
type Error = io::Error;
|
||||
type Future = Box<Future<Item = tarpc::Response<String, Never>, Error = io::Error>>;
|
||||
|
||||
fn poll(&mut self) -> futures::Poll<Self::Item, Self::Error> {
|
||||
self.inner.poll()
|
||||
}
|
||||
}
|
||||
|
||||
/// Implementation detail: Pending connection.
|
||||
pub struct ConnectWithFuture<'a, T> {
|
||||
inner: futures::Map<tarpc::ConnectWithFuture<'a, Request, Response, Error>, fn(Client) -> T>,
|
||||
}
|
||||
|
||||
impl<'a, T> Future for ConnectWithFuture<'a, T> {
|
||||
type Item = T;
|
||||
type Error = io::Error;
|
||||
fn poll(&mut self) -> futures::Poll<Self::Item, Self::Error> {
|
||||
self.inner.poll()
|
||||
fn call(&self, request: Self::Request) -> Self::Future {
|
||||
Ok(Ok(format!("Hello, {}!", request.unwrap()))).into_future().boxed()
|
||||
}
|
||||
}
|
||||
|
||||
/// The client stub that makes RPC calls to the server. Exposes a Future interface.
|
||||
#[derive(Debug)]
|
||||
pub struct FutureClient(Client);
|
||||
|
||||
impl<'a> tarpc::future::Connect<'a> for FutureClient {
|
||||
type ConnectFut = ConnectFuture<Self>;
|
||||
type ConnectWithFut = ConnectWithFuture<'a, Self>;
|
||||
|
||||
fn connect_remotely(addr: &SocketAddr, remote: &Remote) -> Self::ConnectFut {
|
||||
let client = Client::connect_remotely(addr, remote);
|
||||
ConnectFuture { inner: client.map(FutureClient) }
|
||||
}
|
||||
|
||||
fn connect_with(addr: &SocketAddr, handle: &'a Handle) -> Self::ConnectWithFut {
|
||||
let client = Client::connect_with(addr, handle);
|
||||
ConnectWithFuture { inner: client.map(FutureClient) }
|
||||
}
|
||||
}
|
||||
pub struct FutureClient(tarpc::Client<String, String, Never>);
|
||||
|
||||
impl FutureClient {
|
||||
fn connect(addr: &SocketAddr) -> impl Future<Item = FutureClient, Error = io::Error> {
|
||||
tarpc::Client::connect_remotely(addr, &tarpc::REMOTE).map(FutureClient)
|
||||
}
|
||||
|
||||
pub fn hello(&self, name: String)
|
||||
-> impl Future<Item = String, Error = tarpc::Error<Never>> + 'static
|
||||
{
|
||||
let request = Request::Hello(name);
|
||||
|
||||
self.0.call(request).then(move |msg| {
|
||||
match msg? {
|
||||
Ok(Response::Hello(msg)) => Ok(msg),
|
||||
Err(err) => {
|
||||
Err(match err {
|
||||
tarpc::Error::App(Error::Hello(err)) => tarpc::Error::App(err),
|
||||
tarpc::Error::ServerDeserialize(err) => {
|
||||
tarpc::Error::ServerDeserialize(err)
|
||||
}
|
||||
tarpc::Error::ServerSerialize(err) => tarpc::Error::ServerSerialize(err),
|
||||
tarpc::Error::ClientDeserialize(err) => {
|
||||
tarpc::Error::ClientDeserialize(err)
|
||||
}
|
||||
tarpc::Error::ClientSerialize(err) => tarpc::Error::ClientSerialize(err),
|
||||
tarpc::Error::Io(error) => tarpc::Error::Io(error),
|
||||
})
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct HelloServer;
|
||||
|
||||
impl SyncService for HelloServer {
|
||||
fn hello(&self, name: String) -> Result<String, Never> {
|
||||
info!("Got request: {}", name);
|
||||
Ok(format!("Hello, {}!", name))
|
||||
self.0.call(name).then(|msg| msg.unwrap())
|
||||
}
|
||||
}
|
||||
|
||||
fn main() {
|
||||
let _ = env_logger::init();
|
||||
let mut core = tokio_core::reactor::Core::new().unwrap();
|
||||
let addr = HelloServer.listen("localhost:10000").unwrap();
|
||||
let addr = HelloServer::listen("localhost:10000".first_socket_addr()).wait().unwrap();
|
||||
let f = FutureClient::connect(&addr)
|
||||
.map_err(tarpc::Error::from)
|
||||
.and_then(|client| {
|
||||
|
||||
Reference in New Issue
Block a user