OMGeeky/tarpc
1
0
Fork 0
mirror of https://github.com/OMGeeky/tarpc.git synced 2026-01-19 18:00:48 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #27 from tikue/listener

Browse Source 
[Breaking] Add support for arbitrary transports.
This commit is contained in:
shaladdle
2016-04-24 18:39:27 -07:00
parent 3693c95a67 4d636d2882
commit 6ce3a3d943
11 changed files with 414 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
README.md
View File

@@ -46,8 +46,9 @@ impl HelloService for HelloServer {
}
fn main() {
let server_handle = HelloServer.spawn("0.0.0.0:0").unwrap();
let client = hello_service::Client::new(server_handle.local_addr()).unwrap();
let addr = "localhost:10000";
let server_handle = HelloServer.spawn(addr).unwrap();
let client = hello_service::Client::new(addr).unwrap();
assert_eq!("Hello, Mom!", client.hello("Mom".into()).unwrap());
drop(client);
server_handle.shutdown();
@@ -56,17 +57,18 @@ fn main() {
The `service!` macro expands to a collection of items that collectively form an rpc service. In the
above example, the macro is called within the `hello_service` module. This module will contain a
`Client` (and `AsyncClient`) type, and a `Service` trait. The trait provides `default fn`s for
starting the service: `spawn` and `spawn_with_config`, which start the service listening on a tcp
port. A `Client` (or `AsyncClient`) can connect to such a service. These generated types make it
easy and ergonomic to write servers without dealing with sockets or serialization directly. See the
tarpc_examples package for more sophisticated examples.
`Client` (and `AsyncClient`) type, and a `Service` trait. The trait provides default `fn`s for
starting the service: `spawn` and `spawn_with_config`, which start the service listening over an
arbitrary transport. A `Client` (or `AsyncClient`) can connect to such a service. These generated
types make it easy and ergonomic to write servers without dealing with sockets or serialization
directly. See the tarpc_examples package for more sophisticated examples.
## Documentation
Use `cargo doc` as you normally would to see the documentation created for all
items expanded by a `service!` invocation.
## Additional Features
- Connect over any transport that `impl`s the `Transport` trait.
- Concurrent requests from a single client.
- Any type that `impl`s `serde`'s `Serialize` and `Deserialize` can be used in the rpc signatures.
- Attributes can be specified on rpc methods. These will be included on both the `Service` trait

14
tarpc/Cargo.toml
View File

@@ -11,11 +11,13 @@ readme = "../README.md"
description = "An RPC framework for Rust with a focus on ease of use."
[dependencies]
bincode = "^0.5"
log = "^0.3"
scoped-pool = "^0.1"
serde = "^0.7"
bincode = "0.5"
log = "0.3"
scoped-pool = "0.1"
serde = "0.7"
unix_socket = "0.5"
[dev-dependencies]
lazy_static = "^0.1"
env_logger = "^0.3"
lazy_static = "0.1"
env_logger = "0.3"
tempdir = "0.3"

11
tarpc/src/lib.rs
View File

@@ -30,14 +30,13 @@
//! }
//!
//! fn main() {
//! let addr = "127.0.0.1:9000";
//! let shutdown = Server.spawn(addr).unwrap();
//! let client = Client::new(addr).unwrap();
//! let serve_handle = Server.spawn("localhost:0").unwrap();
//! let client = Client::new(serve_handle.dialer()).unwrap();
//! assert_eq!(3, client.add(1, 2).unwrap());
//! assert_eq!("Hello, Mom!".to_string(),
//! client.hello("Mom".to_string()).unwrap());
//! drop(client);
//! shutdown.shutdown();
//! serve_handle.shutdown();
//! }
//! ```
@@ -48,6 +47,7 @@ extern crate bincode;
#[macro_use]
extern crate log;
extern crate scoped_pool;
extern crate unix_socket;
macro_rules! pos {
() => (concat!(file!(), ":", line!()))
@@ -60,4 +60,7 @@ pub mod protocol;
/// Provides the macro used for constructing rpc services and client stubs.
pub mod macros;
/// Provides transport traits and implementations.
pub mod transport;
pub use protocol::{Config, Error, Result, ServeHandle};

102
tarpc/src/macros.rs
View File

@@ -262,7 +262,7 @@ macro_rules! service {
#[doc(hidden)]
#[macro_export]
macro_rules! service_inner {
// Pattern for when the next rpc has an implicit unit return type
// Pattern for when the next rpc has an implicit unit return type
(
{
$(#[$attr:meta])*
@@ -281,7 +281,7 @@ macro_rules! service_inner {
rpc $fn_name( $( $arg : $in_ ),* ) -> ();
}
};
// Pattern for when the next rpc has an explicit return type
// Pattern for when the next rpc has an explicit return type
(
{
$(#[$attr:meta])*
@@ -300,7 +300,7 @@ macro_rules! service_inner {
rpc $fn_name( $( $arg : $in_ ),* ) -> $out;
}
};
// Pattern when all return types have been expanded
// Pattern when all return types have been expanded
(
{ } // none left to expand
$(
@@ -316,21 +316,30 @@ macro_rules! service_inner {
)*
#[doc="Spawn a running service."]
fn spawn<A>(self, addr: A) -> $crate::Result<$crate::protocol::ServeHandle>
where A: ::std::net::ToSocketAddrs,
fn spawn<T>(self,
transport: T)
-> $crate::Result<
$crate::protocol::ServeHandle<
<T::Listener as $crate::transport::Listener>::Dialer>>
where T: $crate::transport::Transport,
Self: 'static,
{
self.spawn_with_config(addr, $crate::Config::default())
self.spawn_with_config(transport, $crate::Config::default())
}
#[doc="Spawn a running service."]
fn spawn_with_config<A>(self, addr: A, config: $crate::Config)
-> $crate::Result<$crate::protocol::ServeHandle>
where A: ::std::net::ToSocketAddrs,
fn spawn_with_config<T>(self,
transport: T,
config: $crate::Config)
-> $crate::Result<
$crate::protocol::ServeHandle<
<T::Listener as $crate::transport::Listener>::Dialer>>
where T: $crate::transport::Transport,
Self: 'static,
{
let server = ::std::sync::Arc::new(__Server(self));
let handle = try!($crate::protocol::Serve::spawn_with_config(server, addr, config));
let server = __Server(self);
let result = $crate::protocol::Serve::spawn_with_config(server, transport, config);
let handle = try!(result);
::std::result::Result::Ok(handle)
}
}
@@ -386,25 +395,29 @@ macro_rules! service_inner {
#[allow(unused)]
#[doc="The client stub that makes RPC calls to the server."]
pub struct Client($crate::protocol::Client<__Request, __Reply>);
pub struct Client<S = ::std::net::TcpStream>(
$crate::protocol::Client<__Request, __Reply, S>
) where S: $crate::transport::Stream;
impl Client {
impl<S> Client<S>
where S: $crate::transport::Stream
{
#[allow(unused)]
#[doc="Create a new client with default configuration that connects to the given \
address."]
pub fn new<A>(addr: A) -> $crate::Result<Self>
where A: ::std::net::ToSocketAddrs,
pub fn new<D>(dialer: D) -> $crate::Result<Self>
where D: $crate::transport::Dialer<Stream=S>,
{
Self::with_config(addr, $crate::Config::default())
Self::with_config(dialer, $crate::Config::default())
}
#[allow(unused)]
#[doc="Create a new client with the specified configuration that connects to the \
given address."]
pub fn with_config<A>(addr: A, config: $crate::Config) -> $crate::Result<Self>
where A: ::std::net::ToSocketAddrs,
pub fn with_config<D>(dialer: D, config: $crate::Config) -> $crate::Result<Self>
where D: $crate::transport::Dialer<Stream=S>,
{
let inner = try!($crate::protocol::Client::with_config(addr, config));
let inner = try!($crate::protocol::Client::with_config(dialer, config));
::std::result::Result::Ok(Client(inner))
}
@@ -425,25 +438,27 @@ macro_rules! service_inner {
#[allow(unused)]
#[doc="The client stub that makes asynchronous RPC calls to the server."]
pub struct AsyncClient($crate::protocol::Client<__Request, __Reply>);
pub struct AsyncClient<S = ::std::net::TcpStream>(
$crate::protocol::Client<__Request, __Reply, S>
) where S: $crate::transport::Stream;
impl AsyncClient {
impl<S> AsyncClient<S>
where S: $crate::transport::Stream {
#[allow(unused)]
#[doc="Create a new asynchronous client with default configuration that connects to \
the given address."]
pub fn new<A>(addr: A) -> $crate::Result<Self>
where A: ::std::net::ToSocketAddrs,
pub fn new<D>(dialer: D) -> $crate::Result<Self>
where D: $crate::transport::Dialer<Stream=S>,
{
Self::with_config(addr, $crate::Config::default())
Self::with_config(dialer, $crate::Config::default())
}
#[allow(unused)]
#[doc="Create a new asynchronous client that connects to the given address."]
pub fn with_config<A>(addr: A, config: $crate::Config)
-> $crate::Result<Self>
where A: ::std::net::ToSocketAddrs,
pub fn with_config<D>(dialer: D, config: $crate::Config) -> $crate::Result<Self>
where D: $crate::transport::Dialer<Stream=S>,
{
let inner = try!($crate::protocol::Client::with_config(addr, config));
let inner = try!($crate::protocol::Client::with_config(dialer, config));
::std::result::Result::Ok(AsyncClient(inner))
}
@@ -463,7 +478,8 @@ macro_rules! service_inner {
}
#[allow(unused)]
struct __Server<S: 'static + Service>(S);
struct __Server<S>(S)
where S: 'static + Service;
impl<S> $crate::protocol::Serve for __Server<S>
where S: 'static + Service
@@ -513,6 +529,8 @@ mod syntax_test {
#[cfg(test)]
mod functional_test {
extern crate env_logger;
extern crate tempdir;
use transport::unix::UnixTransport;
service! {
rpc add(x: i32, y: i32) -> i32;
@@ -534,7 +552,7 @@ mod functional_test {
fn simple() {
let _ = env_logger::init();
let handle = Server.spawn("localhost:0").unwrap();
let client = Client::new(handle.local_addr()).unwrap();
let client = Client::new(handle.dialer()).unwrap();
assert_eq!(3, client.add(1, 2).unwrap());
assert_eq!("Hey, Tim.", client.hey("Tim".into()).unwrap());
drop(client);
@@ -545,7 +563,7 @@ mod functional_test {
fn simple_async() {
let _ = env_logger::init();
let handle = Server.spawn("localhost:0").unwrap();
let client = AsyncClient::new(handle.local_addr()).unwrap();
let client = AsyncClient::new(handle.dialer()).unwrap();
assert_eq!(3, client.add(1, 2).get().unwrap());
assert_eq!("Hey, Adam.", client.hey("Adam".into()).get().unwrap());
drop(client);
@@ -555,7 +573,7 @@ mod functional_test {
#[test]
fn try_clone() {
let handle = Server.spawn("localhost:0").unwrap();
let client1 = Client::new(handle.local_addr()).unwrap();
let client1 = Client::new(handle.dialer()).unwrap();
let client2 = client1.try_clone().unwrap();
assert_eq!(3, client1.add(1, 2).unwrap());
assert_eq!(3, client2.add(1, 2).unwrap());
@@ -564,7 +582,19 @@ mod functional_test {
#[test]
fn async_try_clone() {
let handle = Server.spawn("localhost:0").unwrap();
let client1 = AsyncClient::new(handle.local_addr()).unwrap();
let client1 = AsyncClient::new(handle.dialer()).unwrap();
let client2 = client1.try_clone().unwrap();
assert_eq!(3, client1.add(1, 2).get().unwrap());
assert_eq!(3, client2.add(1, 2).get().unwrap());
}
#[test]
fn async_try_clone_unix() {
let temp_dir = tempdir::TempDir::new("tarpc").unwrap();
let temp_file = temp_dir.path()
.join("async_try_clone_unix.tmp");
let handle = Server.spawn(UnixTransport(temp_file)).unwrap();
let client1 = AsyncClient::new(handle.dialer()).unwrap();
let client2 = client1.try_clone().unwrap();
assert_eq!(3, client1.add(1, 2).get().unwrap());
assert_eq!(3, client2.add(1, 2).get().unwrap());
@@ -576,6 +606,12 @@ mod functional_test {
let _ = ::std::sync::Arc::new(Server).spawn("localhost:0");
}
// Tests that a tcp client can be created from &str
#[allow(dead_code)]
fn test_client_str() {
let _ = Client::new("localhost:0");
}
#[test]
fn serde() {
use bincode;

49
tarpc/src/protocol/client.rs
View File

@@ -8,38 +8,44 @@ use std::fmt;
use std::io::{self, BufReader, BufWriter, Read};
use std::collections::HashMap;
use std::mem;
use std::net::{TcpStream, ToSocketAddrs};
use std::sync::{Arc, Mutex};
use std::sync::mpsc::{Receiver, Sender, channel};
use std::thread;
use super::{Config, Deserialize, Error, Packet, Result, Serialize};
use transport::{Dialer, Stream};
/// A client stub that connects to a server to run rpcs.
pub struct Client<Request, Reply>
where Request: serde::ser::Serialize
pub struct Client<Request, Reply, S>
where Request: serde::ser::Serialize,
S: Stream
{
// The guard is in an option so it can be joined in the drop fn
reader_guard: Arc<Option<thread::JoinHandle<()>>>,
outbound: Sender<(Request, Sender<Result<Reply>>)>,
requests: Arc<Mutex<RpcFutures<Reply>>>,
shutdown: TcpStream,
shutdown: S,
}
impl<Request, Reply> Client<Request, Reply>
impl<Request, Reply, S> Client<Request, Reply, S>
where Request: serde::ser::Serialize + Send + 'static,
Reply: serde::de::Deserialize + Send + 'static
Reply: serde::de::Deserialize + Send + 'static,
S: Stream
{
/// Create a new client that connects to `addr`. The client uses the given timeout
/// for both reads and writes.
pub fn new<A: ToSocketAddrs>(addr: A) -> io::Result<Self> {
Self::with_config(addr, Config::default())
pub fn new<D>(dialer: D) -> io::Result<Self>
where D: Dialer<Stream = S>
{
Self::with_config(dialer, Config::default())
}
/// Create a new client that connects to `addr`. The client uses the given timeout
/// for both reads and writes.
pub fn with_config<A: ToSocketAddrs>(addr: A, config: Config) -> io::Result<Self> {
let stream = try!(TcpStream::connect(addr));
pub fn with_config<D>(dialer: D, config: Config) -> io::Result<Self>
where D: Dialer<Stream = S>
{
let stream = try!(dialer.dial());
try!(stream.set_read_timeout(config.timeout));
try!(stream.set_write_timeout(config.timeout));
let reader_stream = try!(stream.try_clone());
@@ -59,7 +65,7 @@ impl<Request, Reply> Client<Request, Reply>
}
/// Clones the Client so that it can be shared across threads.
pub fn try_clone(&self) -> io::Result<Client<Request, Reply>> {
pub fn try_clone(&self) -> io::Result<Self> {
Ok(Client {
reader_guard: self.reader_guard.clone(),
outbound: self.outbound.clone(),
@@ -97,14 +103,15 @@ impl<Request, Reply> Client<Request, Reply>
}
}
impl<Request, Reply> Drop for Client<Request, Reply>
where Request: serde::ser::Serialize
impl<Request, Reply, S> Drop for Client<Request, Reply, S>
where Request: serde::ser::Serialize,
S: Stream
{
fn drop(&mut self) {
debug!("Dropping Client.");
if let Some(reader_guard) = Arc::get_mut(&mut self.reader_guard) {
debug!("Attempting to shut down writer and reader threads.");
if let Err(e) = self.shutdown.shutdown(::std::net::Shutdown::Both) {
if let Err(e) = self.shutdown.shutdown() {
warn!("Client: couldn't shutdown writer and reader threads: {:?}",
e);
} else {
@@ -185,11 +192,12 @@ impl<Reply> RpcFutures<Reply> {
}
}
fn write<Request, Reply>(outbound: Receiver<(Request, Sender<Result<Reply>>)>,
requests: Arc<Mutex<RpcFutures<Reply>>>,
stream: TcpStream)
fn write<Request, Reply, S>(outbound: Receiver<(Request, Sender<Result<Reply>>)>,
requests: Arc<Mutex<RpcFutures<Reply>>>,
stream: S)
where Request: serde::Serialize,
Reply: serde::Deserialize
Reply: serde::Deserialize,
S: Stream
{
let mut next_id = 0;
let mut stream = BufWriter::new(stream);
@@ -238,8 +246,9 @@ fn write<Request, Reply>(outbound: Receiver<(Request, Sender<Result<Reply>>)>,
}
fn read<Reply>(requests: Arc<Mutex<RpcFutures<Reply>>>, stream: TcpStream)
where Reply: serde::Deserialize
fn read<Reply, S>(requests: Arc<Mutex<RpcFutures<Reply>>>, stream: S)
where Reply: serde::Deserialize,
S: Stream
{
let mut stream = BufReader::new(stream);
loop {

18
tarpc/src/protocol/mod.rs
View File

@@ -93,6 +93,7 @@ mod test {
extern crate env_logger;
use super::{Client, Config, Serve};
use scoped_pool::Pool;
use std::net::TcpStream;
use std::sync::{Arc, Barrier, Mutex};
use std::thread;
use std::time::Duration;
@@ -132,7 +133,7 @@ mod test {
let _ = env_logger::init();
let server = Arc::new(Server::new());
let serve_handle = server.spawn("localhost:0").unwrap();
let client: Client<(), u64> = Client::new(serve_handle.local_addr()).unwrap();
let client: Client<(), u64, TcpStream> = Client::new(serve_handle.dialer()).unwrap();
drop(client);
serve_handle.shutdown();
}
@@ -142,9 +143,8 @@ mod test {
let _ = env_logger::init();
let server = Arc::new(Server::new());
let serve_handle = server.clone().spawn("localhost:0").unwrap();
let addr = serve_handle.local_addr().clone();
// The explicit type is required so that it doesn't deserialize a u32 instead of u64
let client: Client<(), u64> = Client::new(addr).unwrap();
let client: Client<(), u64, _> = Client::new(serve_handle.dialer()).unwrap();
assert_eq!(0, client.rpc(()).unwrap());
assert_eq!(1, server.count());
assert_eq!(1, client.rpc(()).unwrap());
@@ -187,8 +187,7 @@ mod test {
let serve_handle = server.spawn_with_config("localhost:0",
Config { timeout: Some(Duration::new(0, 10)) })
.unwrap();
let addr = serve_handle.local_addr().clone();
let client: Client<(), u64> = Client::new(addr).unwrap();
let client: Client<(), u64, _> = Client::new(serve_handle.dialer()).unwrap();
let thread = thread::spawn(move || serve_handle.shutdown());
info!("force_shutdown:: rpc1: {:?}", client.rpc(()));
thread.join().unwrap();
@@ -201,8 +200,7 @@ mod test {
let serve_handle = server.spawn_with_config("localhost:0",
Config { timeout: test_timeout() })
.unwrap();
let addr = serve_handle.local_addr().clone();
let client: Arc<Client<(), u64>> = Arc::new(Client::new(addr).unwrap());
let client: Arc<Client<(), u64, _>> = Arc::new(Client::new(serve_handle.dialer()).unwrap());
client.rpc(()).unwrap();
serve_handle.shutdown();
match client.rpc(()) {
@@ -219,8 +217,7 @@ mod test {
let pool = Pool::new(concurrency);
let server = Arc::new(BarrierServer::new(concurrency));
let serve_handle = server.clone().spawn("localhost:0").unwrap();
let addr = serve_handle.local_addr().clone();
let client: Client<(), u64> = Client::new(addr).unwrap();
let client: Client<(), u64, _> = Client::new(serve_handle.dialer()).unwrap();
pool.scoped(|scope| {
for _ in 0..concurrency {
let client = client.try_clone().unwrap();
@@ -239,8 +236,7 @@ mod test {
let _ = env_logger::init();
let server = Arc::new(Server::new());
let serve_handle = server.spawn("localhost:0").unwrap();
let addr = serve_handle.local_addr().clone();
let client: Client<(), u64> = Client::new(addr).unwrap();
let client: Client<(), u64, _> = Client::new(serve_handle.dialer()).unwrap();
// Drop future immediately; does the reader channel panic when sending?
client.rpc_async(());

76
tarpc/src/protocol/server.rs
View File

@@ -7,24 +7,27 @@ use serde;
use scoped_pool::{Pool, Scope};
use std::fmt;
use std::io::{self, BufReader, BufWriter};
use std::net::{SocketAddr, TcpListener, TcpStream, ToSocketAddrs};
use std::sync::mpsc::{Receiver, Sender, TryRecvError, channel};
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use std::thread::{self, JoinHandle};
use super::{Config, Deserialize, Error, Packet, Result, Serialize};
use transport::{Dialer, Listener, Stream, Transport};
use transport::tcp::TcpDialer;
struct ConnectionHandler<'a, S>
where S: Serve
struct ConnectionHandler<'a, S, St>
where S: Serve,
St: Stream
{
read_stream: BufReader<TcpStream>,
write_stream: BufWriter<TcpStream>,
read_stream: BufReader<St>,
write_stream: BufWriter<St>,
server: S,
shutdown: &'a AtomicBool,
}
impl<'a, S> ConnectionHandler<'a, S>
where S: Serve
impl<'a, S, St> ConnectionHandler<'a, S, St>
where S: Serve,
St: Stream
{
fn handle_conn<'b>(&'b mut self, scope: &Scope<'b>) -> Result<()> {
let ConnectionHandler {
@@ -83,7 +86,7 @@ impl<'a, S> ConnectionHandler<'a, S>
}
}
fn write(rx: Receiver<Packet<<S as Serve>::Reply>>, stream: &mut BufWriter<TcpStream>) {
fn write(rx: Receiver<Packet<<S as Serve>::Reply>>, stream: &mut BufWriter<St>) {
loop {
match rx.recv() {
Err(e) => {
@@ -101,21 +104,25 @@ impl<'a, S> ConnectionHandler<'a, S>
}
/// Provides methods for blocking until the server completes,
pub struct ServeHandle {
pub struct ServeHandle<D = TcpDialer>
where D: Dialer
{
tx: Sender<()>,
join_handle: JoinHandle<()>,
addr: SocketAddr,
dialer: D,
}
impl ServeHandle {
impl<D> ServeHandle<D>
where D: Dialer
{
/// Block until the server completes
pub fn wait(self) {
self.join_handle.join().expect(pos!());
}
/// Returns the address the server is bound to
pub fn local_addr(&self) -> &SocketAddr {
&self.addr
/// Returns the dialer to the server.
pub fn dialer(&self) -> &D {
&self.dialer
}
/// Shutdown the server. Gracefully shuts down the serve thread but currently does not
@@ -123,7 +130,7 @@ impl ServeHandle {
pub fn shutdown(self) {
info!("ServeHandle: attempting to shut down the server.");
self.tx.send(()).expect(pos!());
if let Ok(_) = TcpStream::connect(self.addr) {
if let Ok(_) = self.dialer.dial() {
self.join_handle.join().expect(pos!());
} else {
warn!("ServeHandle: best effort shutdown of serve thread failed");
@@ -131,16 +138,19 @@ impl ServeHandle {
}
}
struct Server<'a, S: 'a> {
struct Server<'a, S: 'a, L>
where L: Listener
{
server: &'a S,
listener: TcpListener,
listener: L,
read_timeout: Option<Duration>,
die_rx: Receiver<()>,
shutdown: &'a AtomicBool,
}
impl<'a, S: 'a> Server<'a, S>
where S: Serve + 'static
impl<'a, S, L> Server<'a, S, L>
where S: Serve + 'static,
L: Listener
{
fn serve<'b>(self, scope: &Scope<'b>)
where 'a: 'b
@@ -194,7 +204,9 @@ impl<'a, S: 'a> Server<'a, S>
}
}
impl<'a, S> Drop for Server<'a, S> {
impl<'a, S, L> Drop for Server<'a, S, L>
where L: Listener
{
fn drop(&mut self) {
debug!("Shutting down connection handlers.");
self.shutdown.store(true, Ordering::SeqCst);
@@ -212,29 +224,33 @@ pub trait Serve: Send + Sync + Sized {
fn serve(&self, request: Self::Request) -> Self::Reply;
/// spawn
fn spawn<A>(self, addr: A) -> io::Result<ServeHandle>
where A: ToSocketAddrs,
fn spawn<T>(self, transport: T) -> io::Result<ServeHandle<<T::Listener as Listener>::Dialer>>
where T: Transport,
Self: 'static
{
self.spawn_with_config(addr, Config::default())
self.spawn_with_config(transport, Config::default())
}
/// spawn
fn spawn_with_config<A>(self, addr: A, config: Config) -> io::Result<ServeHandle>
where A: ToSocketAddrs,
fn spawn_with_config<T>(self,
transport: T,
config: Config)
-> io::Result<ServeHandle<<T::Listener as Listener>::Dialer>>
where T: Transport,
Self: 'static
{
let listener = try!(TcpListener::bind(&addr));
let addr = try!(listener.local_addr());
info!("spawn_with_config: spinning up server on {:?}", addr);
let listener = try!(transport.bind());
let dialer = try!(listener.dialer());
info!("spawn_with_config: spinning up server.");
let (die_tx, die_rx) = channel();
let timeout = config.timeout;
let join_handle = thread::spawn(move || {
let pool = Pool::new(100); // TODO(tjk): make this configurable, and expire idle threads
let shutdown = AtomicBool::new(false);
let server = Server {
server: &self,
listener: listener,
read_timeout: config.timeout,
read_timeout: timeout,
die_rx: die_rx,
shutdown: &shutdown,
};
@@ -245,7 +261,7 @@ pub trait Serve: Send + Sync + Sized {
Ok(ServeHandle {
tx: die_tx,
join_handle: join_handle,
addr: addr.clone(),
dialer: dialer,
})
}

91
tarpc/src/transport/mod.rs Normal file
View File

@@ -0,0 +1,91 @@
use std::io::{self, Read, Write};
use std::time::Duration;
/// A factory for creating a listener on a given address.
/// For TCP, an address might be an IPv4 address; for Unix sockets, it
/// is just a file name.
pub trait Transport {
/// The type of listener that binds to the given address.
type Listener: Listener;
/// Return a listener on the given address, and a dialer to that address.
fn bind(&self) -> io::Result<Self::Listener>;
}
/// Accepts incoming connections from dialers.
pub trait Listener: Send + 'static {
/// The type of address being listened on.
type Dialer: Dialer;
/// The type of stream this listener accepts.
type Stream: Stream;
/// Accept an incoming stream.
fn accept(&self) -> io::Result<Self::Stream>;
/// Returns the local address being listened on.
fn dialer(&self) -> io::Result<Self::Dialer>;
/// Iterate over incoming connections.
fn incoming(&self) -> Incoming<Self> {
Incoming { listener: self }
}
}
/// A cloneable Reader/Writer.
pub trait Stream: Read + Write + Send + Sized + 'static {
/// Creates a new independently owned handle to the Stream.
///
/// The returned Stream should reference the same stream that this
/// object references. Both handles should read and write the same
/// stream of data, and options set on one stream should be propagated
/// to the other stream.
fn try_clone(&self) -> io::Result<Self>;
/// Sets a read timeout.
///
/// If the value specified is `None`, then read calls will block indefinitely.
/// It is an error to pass the zero `Duration` to this method.
fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()>;
/// Sets a write timeout.
///
/// If the value specified is `None`, then write calls will block indefinitely.
/// It is an error to pass the zero `Duration` to this method.
fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()>;
/// Shuts down both ends of the stream.
///
/// Implementations should cause all pending and future I/O on the specified
/// portions to return immediately with an appropriate value.
fn shutdown(&self) -> io::Result<()>;
}
/// A `Stream` factory.
pub trait Dialer {
/// The type of `Stream` this can create.
type Stream: Stream;
/// Open a stream.
fn dial(&self) -> io::Result<Self::Stream>;
}
impl<P, D: ?Sized> Dialer for P
where P: ::std::ops::Deref<Target = D>,
D: Dialer + 'static
{
type Stream = D::Stream;
fn dial(&self) -> io::Result<Self::Stream> {
(**self).dial()
}
}
/// Iterates over incoming connections.
pub struct Incoming<'a, L: Listener + ?Sized + 'a> {
listener: &'a L,
}
impl<'a, L: Listener> Iterator for Incoming<'a, L> {
type Item = io::Result<L::Stream>;
fn next(&mut self) -> Option<Self::Item> {
Some(self.listener.accept())
}
}
/// Provides a TCP transport.
pub mod tcp;
/// Provides a unix socket transport.
pub mod unix;

75
tarpc/src/transport/tcp.rs Normal file
View File

@@ -0,0 +1,75 @@
use std::io;
use std::net::{SocketAddr, TcpListener, TcpStream, ToSocketAddrs};
use std::time::Duration;
/// A transport for TCP.
pub struct TcpTransport<A: ToSocketAddrs>(pub A);
impl<A: ToSocketAddrs> super::Transport for TcpTransport<A> {
type Listener = TcpListener;
fn bind(&self) -> io::Result<TcpListener> {
TcpListener::bind(&self.0)
}
}
impl<A: ToSocketAddrs> super::Transport for A {
type Listener = TcpListener;
fn bind(&self) -> io::Result<TcpListener> {
TcpListener::bind(self)
}
}
impl super::Listener for TcpListener {
type Dialer = TcpDialer<SocketAddr>;
type Stream = TcpStream;
fn accept(&self) -> io::Result<TcpStream> {
self.accept().map(|(stream, _)| stream)
}
fn dialer(&self) -> io::Result<TcpDialer<SocketAddr>> {
self.local_addr().map(|addr| TcpDialer(addr))
}
}
impl super::Stream for TcpStream {
fn try_clone(&self) -> io::Result<Self> {
self.try_clone()
}
fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
self.set_read_timeout(dur)
}
fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
self.set_write_timeout(dur)
}
fn shutdown(&self) -> io::Result<()> {
self.shutdown(::std::net::Shutdown::Both)
}
}
/// Connects to a socket address.
pub struct TcpDialer<A = SocketAddr>(pub A) where A: ToSocketAddrs;
impl<A> super::Dialer for TcpDialer<A>
where A: ToSocketAddrs
{
type Stream = TcpStream;
fn dial(&self) -> io::Result<TcpStream> {
TcpStream::connect(&self.0)
}
}
impl super::Dialer for str {
type Stream = TcpStream;
fn dial(&self) -> io::Result<TcpStream> {
TcpStream::connect(self)
}
}

70
tarpc/src/transport/unix.rs Normal file
View File

@@ -0,0 +1,70 @@
use std::io;
use std::path::{Path, PathBuf};
use std::time::Duration;
use unix_socket::{UnixListener, UnixStream};
/// A transport for unix sockets.
pub struct UnixTransport<P>(pub P) where P: AsRef<Path>;
impl<P> super::Transport for UnixTransport<P>
where P: AsRef<Path>
{
type Listener = UnixListener;
fn bind(&self) -> io::Result<UnixListener> {
UnixListener::bind(&self.0)
}
}
/// Connects to a unix socket address.
pub struct UnixDialer<P>(pub P) where P: AsRef<Path>;
impl<P> super::Dialer for UnixDialer<P>
where P: AsRef<Path>
{
type Stream = UnixStream;
fn dial(&self) -> io::Result<UnixStream> {
UnixStream::connect(&self.0)
}
}
impl super::Listener for UnixListener {
type Stream = UnixStream;
type Dialer = UnixDialer<PathBuf>;
fn accept(&self) -> io::Result<UnixStream> {
self.accept().map(|(stream, _)| stream)
}
fn dialer(&self) -> io::Result<UnixDialer<PathBuf>> {
self.local_addr().and_then(|addr| {
match addr.as_pathname() {
Some(path) => Ok(UnixDialer(path.to_owned())),
None => {
Err(io::Error::new(io::ErrorKind::AddrNotAvailable,
"Couldn't get a path to bound unix socket"))
}
}
})
}
}
impl super::Stream for UnixStream {
fn try_clone(&self) -> io::Result<Self> {
self.try_clone()
}
fn set_read_timeout(&self, timeout: Option<Duration>) -> io::Result<()> {
self.set_read_timeout(timeout)
}
fn set_write_timeout(&self, timeout: Option<Duration>) -> io::Result<()> {
self.set_write_timeout(timeout)
}
fn shutdown(&self) -> io::Result<()> {
self.shutdown(::std::net::Shutdown::Both)
}
}

5
tarpc_examples/src/lib.rs
View File

@@ -41,8 +41,9 @@ mod benchmark {
Arc::new(Mutex::new(handle))
};
static ref CLIENT: Arc<Mutex<AsyncClient>> = {
let addr = HANDLE.lock().unwrap().local_addr().clone();
let client = AsyncClient::new(addr).unwrap();
let lock = HANDLE.lock().unwrap();
let dialer = lock.dialer();
let client = AsyncClient::new(dialer).unwrap();
Arc::new(Mutex::new(client))
};
}