THRIFT-5283: add support for Unix Domain Sockets in lib/rs (#2545)
Client: rs
diff --git a/lib/rs/test/Cargo.toml b/lib/rs/test/Cargo.toml
index 0ba96fd..47b8cbf 100644
--- a/lib/rs/test/Cargo.toml
+++ b/lib/rs/test/Cargo.toml
@@ -9,6 +9,7 @@
[dependencies]
clap = "~2.33"
bitflags = "=1.2"
+log = "0.4"
[dependencies.thrift]
path = "../"
diff --git a/lib/rs/test/src/bin/kitchen_sink_client.rs b/lib/rs/test/src/bin/kitchen_sink_client.rs
index 74197de..b98afb8 100644
--- a/lib/rs/test/src/bin/kitchen_sink_client.rs
+++ b/lib/rs/test/src/bin/kitchen_sink_client.rs
@@ -16,8 +16,16 @@
// under the License.
use clap::{clap_app, value_t};
+use log::*;
use std::convert::Into;
+use std::net::TcpStream;
+use std::net::ToSocketAddrs;
+
+#[cfg(unix)]
+use std::os::unix::net::UnixStream;
+#[cfg(unix)]
+use std::path::Path;
use kitchen_sink::base_two::{TNapkinServiceSyncClient, TRamenServiceSyncClient};
use kitchen_sink::midlayer::{MealServiceSyncClient, TMealServiceSyncClient};
@@ -30,9 +38,9 @@
TBinaryInputProtocol, TBinaryOutputProtocol, TCompactInputProtocol, TCompactOutputProtocol,
TInputProtocol, TOutputProtocol,
};
-use thrift::transport::{
- ReadHalf, TFramedReadTransport, TFramedWriteTransport, TIoChannel, TTcpChannel, WriteHalf,
-};
+use thrift::transport::{TFramedReadTransport, TFramedWriteTransport, TIoChannel, TTcpChannel};
+
+type IoProtocol = (Box<dyn TInputProtocol>, Box<dyn TOutputProtocol>);
fn main() {
match run() {
@@ -51,6 +59,7 @@
(about: "Thrift Rust kitchen sink client")
(@arg host: --host +takes_value "Host on which the Thrift test server is located")
(@arg port: --port +takes_value "Port on which the Thrift test server is listening")
+ (@arg domain_socket: --("domain-socket") + takes_value "Unix Domain Socket on which the Thrift test server is listening")
(@arg protocol: --protocol +takes_value "Thrift protocol implementation to use (\"binary\", \"compact\")")
(@arg service: --service +takes_value "Service type to contact (\"part\", \"full\", \"recursive\")")
)
@@ -58,10 +67,47 @@
let host = matches.value_of("host").unwrap_or("127.0.0.1");
let port = value_t!(matches, "port", u16).unwrap_or(9090);
+ let domain_socket = matches.value_of("domain_socket");
let protocol = matches.value_of("protocol").unwrap_or("compact");
let service = matches.value_of("service").unwrap_or("part");
- let (i_chan, o_chan) = tcp_channel(host, port)?;
+ let (i_prot, o_prot) = match domain_socket {
+ None => {
+ let listen_address = format!("{}:{}", host, port);
+ info!("Client binds to {} with {}", listen_address, protocol);
+ bind(listen_address, protocol)?
+ }
+ Some(domain_socket) => {
+ info!("Client binds to {} (UDS) with {}", domain_socket, protocol);
+ bind_uds(domain_socket, protocol)?
+ }
+ };
+
+ run_client(service, i_prot, o_prot)
+}
+
+fn bind<A: ToSocketAddrs>(listen_address: A, protocol: &str) -> Result<IoProtocol, thrift::Error> {
+ let stream = TcpStream::connect(listen_address)?;
+ let channel = TTcpChannel::with_stream(stream);
+
+ let (i_prot, o_prot) = build(channel, protocol)?;
+ Ok((i_prot, o_prot))
+}
+
+#[cfg(unix)]
+fn bind_uds<P: AsRef<Path>>(domain_socket: P, protocol: &str) -> Result<IoProtocol, thrift::Error> {
+ let stream = UnixStream::connect(domain_socket)?;
+
+ let (i_prot, o_prot) = build(stream, protocol)?;
+ Ok((i_prot, o_prot))
+}
+
+fn build<C: TIoChannel + 'static>(
+ channel: C,
+ protocol: &str,
+) -> thrift::Result<(Box<dyn TInputProtocol>, Box<dyn TOutputProtocol>)> {
+ let (i_chan, o_chan) = channel.split()?;
+
let (i_tran, o_tran) = (
TFramedReadTransport::new(i_chan),
TFramedWriteTransport::new(o_chan),
@@ -79,7 +125,7 @@
unmatched => return Err(format!("unsupported protocol {}", unmatched).into()),
};
- run_client(service, i_prot, o_prot)
+ Ok((i_prot, o_prot))
}
fn run_client(
@@ -98,15 +144,6 @@
}
}
-fn tcp_channel(
- host: &str,
- port: u16,
-) -> thrift::Result<(ReadHalf<TTcpChannel>, WriteHalf<TTcpChannel>)> {
- let mut c = TTcpChannel::new();
- c.open(&format!("{}:{}", host, port))?;
- c.split()
-}
-
fn exec_meal_client(
i_prot: Box<dyn TInputProtocol>,
o_prot: Box<dyn TOutputProtocol>,
diff --git a/lib/rs/test/src/bin/kitchen_sink_server.rs b/lib/rs/test/src/bin/kitchen_sink_server.rs
index 8b910b3..ea571c6 100644
--- a/lib/rs/test/src/bin/kitchen_sink_server.rs
+++ b/lib/rs/test/src/bin/kitchen_sink_server.rs
@@ -16,6 +16,7 @@
// under the License.
use clap::{clap_app, value_t};
+use log::*;
use thrift;
use thrift::protocol::{
@@ -28,6 +29,7 @@
TWriteTransportFactory,
};
+use crate::Socket::{ListenAddress, UnixDomainSocket};
use kitchen_sink::base_one::Noodle;
use kitchen_sink::base_two::{
BrothType, Napkin, NapkinServiceSyncHandler, Ramen, RamenServiceSyncHandler,
@@ -42,6 +44,11 @@
FullMealServiceSyncHandler,
};
+enum Socket {
+ ListenAddress(String),
+ UnixDomainSocket(String),
+}
+
fn main() {
match run() {
Ok(()) => println!("kitchen sink server completed successfully"),
@@ -57,18 +64,29 @@
(version: "0.1.0")
(author: "Apache Thrift Developers <dev@thrift.apache.org>")
(about: "Thrift Rust kitchen sink test server")
- (@arg port: --port +takes_value "port on which the test server listens")
+ (@arg port: --port +takes_value "Port on which the Thrift test server listens")
+ (@arg domain_socket: --("domain-socket") + takes_value "Unix Domain Socket on which the Thrift test server listens")
(@arg protocol: --protocol +takes_value "Thrift protocol implementation to use (\"binary\", \"compact\")")
(@arg service: --service +takes_value "Service type to contact (\"part\", \"full\", \"recursive\")")
)
.get_matches();
let port = value_t!(matches, "port", u16).unwrap_or(9090);
+ let domain_socket = matches.value_of("domain_socket");
let protocol = matches.value_of("protocol").unwrap_or("compact");
let service = matches.value_of("service").unwrap_or("part");
let listen_address = format!("127.0.0.1:{}", port);
- println!("binding to {}", listen_address);
+ let socket = match domain_socket {
+ None => {
+ info!("Server is binding to {}", listen_address);
+ Socket::ListenAddress(listen_address)
+ }
+ Some(domain_socket) => {
+ info!("Server is binding to {} (UDS)", domain_socket);
+ Socket::UnixDomainSocket(domain_socket.to_string())
+ }
+ };
let r_transport_factory = TFramedReadTransportFactory::new();
let w_transport_factory = TFramedWriteTransportFactory::new();
@@ -102,21 +120,21 @@
// Since what I'm doing is uncommon I'm just going to duplicate the code
match &*service {
"part" => run_meal_server(
- &listen_address,
+ socket,
r_transport_factory,
i_protocol_factory,
w_transport_factory,
o_protocol_factory,
),
"full" => run_full_meal_server(
- &listen_address,
+ socket,
r_transport_factory,
i_protocol_factory,
w_transport_factory,
o_protocol_factory,
),
"recursive" => run_recursive_server(
- &listen_address,
+ socket,
r_transport_factory,
i_protocol_factory,
w_transport_factory,
@@ -127,7 +145,7 @@
}
fn run_meal_server<RTF, IPF, WTF, OPF>(
- listen_address: &str,
+ socket: Socket,
r_transport_factory: RTF,
i_protocol_factory: IPF,
w_transport_factory: WTF,
@@ -149,11 +167,14 @@
1,
);
- server.listen(listen_address)
+ match socket {
+ ListenAddress(listen_address) => server.listen(listen_address),
+ UnixDomainSocket(s) => server.listen_uds(s),
+ }
}
fn run_full_meal_server<RTF, IPF, WTF, OPF>(
- listen_address: &str,
+ socket: Socket,
r_transport_factory: RTF,
i_protocol_factory: IPF,
w_transport_factory: WTF,
@@ -175,7 +196,10 @@
1,
);
- server.listen(listen_address)
+ match socket {
+ ListenAddress(listen_address) => server.listen(listen_address),
+ UnixDomainSocket(s) => server.listen_uds(s),
+ }
}
struct PartHandler;
@@ -267,7 +291,7 @@
}
fn run_recursive_server<RTF, IPF, WTF, OPF>(
- listen_address: &str,
+ socket: Socket,
r_transport_factory: RTF,
i_protocol_factory: IPF,
w_transport_factory: WTF,
@@ -289,7 +313,10 @@
1,
);
- server.listen(listen_address)
+ match socket {
+ ListenAddress(listen_address) => server.listen(listen_address),
+ UnixDomainSocket(s) => server.listen_uds(s),
+ }
}
struct RecursiveTestServerHandler;