THRIFT-2945 Add Rust support
Client: Rust
Patch: Allen George <allen.george@gmail.com>
This closes #1147
diff --git a/lib/rs/src/transport/buffered.rs b/lib/rs/src/transport/buffered.rs
new file mode 100644
index 0000000..3f240d8
--- /dev/null
+++ b/lib/rs/src/transport/buffered.rs
@@ -0,0 +1,400 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::cell::RefCell;
+use std::cmp;
+use std::io;
+use std::io::{Read, Write};
+use std::rc::Rc;
+
+use super::{TTransport, TTransportFactory};
+
+/// Default capacity of the read buffer in bytes.
+const DEFAULT_RBUFFER_CAPACITY: usize = 4096;
+
+/// Default capacity of the write buffer in bytes..
+const DEFAULT_WBUFFER_CAPACITY: usize = 4096;
+
+/// Transport that communicates with endpoints using a byte stream.
+///
+/// A `TBufferedTransport` maintains a fixed-size internal write buffer. All
+/// writes are made to this buffer and are sent to the wrapped transport only
+/// when `TTransport::flush()` is called. On a flush a fixed-length header with a
+/// count of the buffered bytes is written, followed by the bytes themselves.
+///
+/// A `TBufferedTransport` also maintains a fixed-size internal read buffer.
+/// On a call to `TTransport::read(...)` one full message - both fixed-length
+/// header and bytes - is read from the wrapped transport and buffered.
+/// Subsequent read calls are serviced from the internal buffer until it is
+/// exhausted, at which point the next full message is read from the wrapped
+/// transport.
+///
+/// # Examples
+///
+/// Create and use a `TBufferedTransport`.
+///
+/// ```no_run
+/// use std::cell::RefCell;
+/// use std::rc::Rc;
+/// use std::io::{Read, Write};
+/// use thrift::transport::{TBufferedTransport, TTcpTransport, TTransport};
+///
+/// let mut t = TTcpTransport::new();
+/// t.open("localhost:9090").unwrap();
+///
+/// let t = Rc::new(RefCell::new(Box::new(t) as Box<TTransport>));
+/// let mut t = TBufferedTransport::new(t);
+///
+/// // read
+/// t.read(&mut vec![0u8; 1]).unwrap();
+///
+/// // write
+/// t.write(&[0x00]).unwrap();
+/// t.flush().unwrap();
+/// ```
+pub struct TBufferedTransport {
+ rbuf: Box<[u8]>,
+ rpos: usize,
+ rcap: usize,
+ wbuf: Vec<u8>,
+ inner: Rc<RefCell<Box<TTransport>>>,
+}
+
+impl TBufferedTransport {
+ /// Create a `TBufferedTransport` with default-sized internal read and
+ /// write buffers that wraps an `inner` `TTransport`.
+ pub fn new(inner: Rc<RefCell<Box<TTransport>>>) -> TBufferedTransport {
+ TBufferedTransport::with_capacity(DEFAULT_RBUFFER_CAPACITY, DEFAULT_WBUFFER_CAPACITY, inner)
+ }
+
+ /// Create a `TBufferedTransport` with an internal read buffer of size
+ /// `read_buffer_capacity` and an internal write buffer of size
+ /// `write_buffer_capacity` that wraps an `inner` `TTransport`.
+ pub fn with_capacity(read_buffer_capacity: usize,
+ write_buffer_capacity: usize,
+ inner: Rc<RefCell<Box<TTransport>>>)
+ -> TBufferedTransport {
+ TBufferedTransport {
+ rbuf: vec![0; read_buffer_capacity].into_boxed_slice(),
+ rpos: 0,
+ rcap: 0,
+ wbuf: Vec::with_capacity(write_buffer_capacity),
+ inner: inner,
+ }
+ }
+
+ fn get_bytes(&mut self) -> io::Result<&[u8]> {
+ if self.rcap - self.rpos == 0 {
+ self.rpos = 0;
+ self.rcap = self.inner.borrow_mut().read(&mut self.rbuf)?;
+ }
+
+ Ok(&self.rbuf[self.rpos..self.rcap])
+ }
+
+ fn consume(&mut self, consumed: usize) {
+ // TODO: was a bug here += <-- test somehow
+ self.rpos = cmp::min(self.rcap, self.rpos + consumed);
+ }
+}
+
+impl Read for TBufferedTransport {
+ fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+ let mut bytes_read = 0;
+
+ loop {
+ let nread = {
+ let avail_bytes = self.get_bytes()?;
+ let avail_space = buf.len() - bytes_read;
+ let nread = cmp::min(avail_space, avail_bytes.len());
+ buf[bytes_read..(bytes_read + nread)].copy_from_slice(&avail_bytes[..nread]);
+ nread
+ };
+
+ self.consume(nread);
+ bytes_read += nread;
+
+ if bytes_read == buf.len() || nread == 0 {
+ break;
+ }
+ }
+
+ Ok(bytes_read)
+ }
+}
+
+impl Write for TBufferedTransport {
+ fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+ let avail_bytes = cmp::min(buf.len(), self.wbuf.capacity() - self.wbuf.len());
+ self.wbuf.extend_from_slice(&buf[..avail_bytes]);
+ assert!(self.wbuf.len() <= self.wbuf.capacity(),
+ "copy overflowed buffer");
+ Ok(avail_bytes)
+ }
+
+ fn flush(&mut self) -> io::Result<()> {
+ self.inner.borrow_mut().write_all(&self.wbuf)?;
+ self.inner.borrow_mut().flush()?;
+ self.wbuf.clear();
+ Ok(())
+ }
+}
+
+/// Factory for creating instances of `TBufferedTransport`
+#[derive(Default)]
+pub struct TBufferedTransportFactory;
+
+impl TBufferedTransportFactory {
+ /// Create a `TBufferedTransportFactory`.
+ pub fn new() -> TBufferedTransportFactory {
+ TBufferedTransportFactory {}
+ }
+}
+
+impl TTransportFactory for TBufferedTransportFactory {
+ fn create(&self, inner: Rc<RefCell<Box<TTransport>>>) -> Box<TTransport> {
+ Box::new(TBufferedTransport::new(inner)) as Box<TTransport>
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::cell::RefCell;
+ use std::io::{Read, Write};
+ use std::rc::Rc;
+
+ use super::*;
+ use ::transport::{TPassThruTransport, TTransport};
+ use ::transport::mem::TBufferTransport;
+
+ macro_rules! new_transports {
+ ($wbc:expr, $rbc:expr) => (
+ {
+ let mem = Rc::new(RefCell::new(Box::new(TBufferTransport::with_capacity($wbc, $rbc))));
+ let thru: Box<TTransport> = Box::new(TPassThruTransport { inner: mem.clone() });
+ let thru = Rc::new(RefCell::new(thru));
+ (mem, thru)
+ }
+ );
+ }
+
+ #[test]
+ fn must_return_zero_if_read_buffer_is_empty() {
+ let (_, thru) = new_transports!(10, 0);
+ let mut t = TBufferedTransport::with_capacity(10, 0, thru);
+
+ let mut b = vec![0; 10];
+ let read_result = t.read(&mut b);
+
+ assert_eq!(read_result.unwrap(), 0);
+ }
+
+ #[test]
+ fn must_return_zero_if_caller_reads_into_zero_capacity_buffer() {
+ let (_, thru) = new_transports!(10, 0);
+ let mut t = TBufferedTransport::with_capacity(10, 0, thru);
+
+ let read_result = t.read(&mut []);
+
+ assert_eq!(read_result.unwrap(), 0);
+ }
+
+ #[test]
+ fn must_return_zero_if_nothing_more_can_be_read() {
+ let (mem, thru) = new_transports!(4, 0);
+ let mut t = TBufferedTransport::with_capacity(4, 0, thru);
+
+ mem.borrow_mut().set_readable_bytes(&[0, 1, 2, 3]);
+
+ // read buffer is exactly the same size as bytes available
+ let mut buf = vec![0u8; 4];
+ let read_result = t.read(&mut buf);
+
+ // we've read exactly 4 bytes
+ assert_eq!(read_result.unwrap(), 4);
+ assert_eq!(&buf, &[0, 1, 2, 3]);
+
+ // try read again
+ let buf_again = vec![0u8; 4];
+ let read_result = t.read(&mut buf);
+
+ // this time, 0 bytes and we haven't changed the buffer
+ assert_eq!(read_result.unwrap(), 0);
+ assert_eq!(&buf_again, &[0, 0, 0, 0])
+ }
+
+ #[test]
+ fn must_fill_user_buffer_with_only_as_many_bytes_as_available() {
+ let (mem, thru) = new_transports!(4, 0);
+ let mut t = TBufferedTransport::with_capacity(4, 0, thru);
+
+ mem.borrow_mut().set_readable_bytes(&[0, 1, 2, 3]);
+
+ // read buffer is much larger than the bytes available
+ let mut buf = vec![0u8; 8];
+ let read_result = t.read(&mut buf);
+
+ // we've read exactly 4 bytes
+ assert_eq!(read_result.unwrap(), 4);
+ assert_eq!(&buf[..4], &[0, 1, 2, 3]);
+
+ // try read again
+ let read_result = t.read(&mut buf[4..]);
+
+ // this time, 0 bytes and we haven't changed the buffer
+ assert_eq!(read_result.unwrap(), 0);
+ assert_eq!(&buf, &[0, 1, 2, 3, 0, 0, 0, 0])
+ }
+
+ #[test]
+ fn must_read_successfully() {
+ // this test involves a few loops within the buffered transport
+ // itself where it has to drain the underlying transport in order
+ // to service a read
+
+ // we have a much smaller buffer than the
+ // underlying transport has bytes available
+ let (mem, thru) = new_transports!(10, 0);
+ let mut t = TBufferedTransport::with_capacity(2, 0, thru);
+
+ // fill the underlying transport's byte buffer
+ let mut readable_bytes = [0u8; 10];
+ for i in 0..10 {
+ readable_bytes[i] = i as u8;
+ }
+ mem.borrow_mut().set_readable_bytes(&readable_bytes);
+
+ // we ask to read into a buffer that's much larger
+ // than the one the buffered transport has; as a result
+ // it's going to have to keep asking the underlying
+ // transport for more bytes
+ let mut buf = [0u8; 8];
+ let read_result = t.read(&mut buf);
+
+ // we should have read 8 bytes
+ assert_eq!(read_result.unwrap(), 8);
+ assert_eq!(&buf, &[0, 1, 2, 3, 4, 5, 6, 7]);
+
+ // let's clear out the buffer and try read again
+ for i in 0..8 {
+ buf[i] = 0;
+ }
+ let read_result = t.read(&mut buf);
+
+ // this time we were only able to read 2 bytes
+ // (all that's remaining from the underlying transport)
+ // let's also check that the remaining bytes are untouched
+ assert_eq!(read_result.unwrap(), 2);
+ assert_eq!(&buf[0..2], &[8, 9]);
+ assert_eq!(&buf[2..], &[0, 0, 0, 0, 0, 0]);
+
+ // try read again (we should get 0)
+ // and all the existing bytes were untouched
+ let read_result = t.read(&mut buf);
+ assert_eq!(read_result.unwrap(), 0);
+ assert_eq!(&buf[0..2], &[8, 9]);
+ assert_eq!(&buf[2..], &[0, 0, 0, 0, 0, 0]);
+ }
+
+ #[test]
+ fn must_return_zero_if_nothing_can_be_written() {
+ let (_, thru) = new_transports!(0, 0);
+ let mut t = TBufferedTransport::with_capacity(0, 0, thru);
+
+ let b = vec![0; 10];
+ let r = t.write(&b);
+
+ assert_eq!(r.unwrap(), 0);
+ }
+
+ #[test]
+ fn must_return_zero_if_caller_calls_write_with_empty_buffer() {
+ let (mem, thru) = new_transports!(0, 10);
+ let mut t = TBufferedTransport::with_capacity(0, 10, thru);
+
+ let r = t.write(&[]);
+
+ assert_eq!(r.unwrap(), 0);
+ assert_eq!(mem.borrow_mut().write_buffer_as_ref(), &[]);
+ }
+
+ #[test]
+ fn must_return_zero_if_write_buffer_full() {
+ let (_, thru) = new_transports!(0, 0);
+ let mut t = TBufferedTransport::with_capacity(0, 4, thru);
+
+ let b = [0x00, 0x01, 0x02, 0x03];
+
+ // we've now filled the write buffer
+ let r = t.write(&b);
+ assert_eq!(r.unwrap(), 4);
+
+ // try write the same bytes again - nothing should be writable
+ let r = t.write(&b);
+ assert_eq!(r.unwrap(), 0);
+ }
+
+ #[test]
+ fn must_only_write_to_inner_transport_on_flush() {
+ let (mem, thru) = new_transports!(10, 10);
+ let mut t = TBufferedTransport::new(thru);
+
+ let b: [u8; 5] = [0, 1, 2, 3, 4];
+ assert_eq!(t.write(&b).unwrap(), 5);
+ assert_eq!(mem.borrow_mut().write_buffer_as_ref().len(), 0);
+
+ assert!(t.flush().is_ok());
+
+ {
+ let inner = mem.borrow_mut();
+ let underlying_buffer = inner.write_buffer_as_ref();
+ assert_eq!(b, underlying_buffer);
+ }
+ }
+
+ #[test]
+ fn must_write_successfully_after_flush() {
+ let (mem, thru) = new_transports!(0, 5);
+ let mut t = TBufferedTransport::with_capacity(0, 5, thru);
+
+ // write and flush
+ let b: [u8; 5] = [0, 1, 2, 3, 4];
+ assert_eq!(t.write(&b).unwrap(), 5);
+ assert!(t.flush().is_ok());
+
+ // check the flushed bytes
+ {
+ let inner = mem.borrow_mut();
+ let underlying_buffer = inner.write_buffer_as_ref();
+ assert_eq!(b, underlying_buffer);
+ }
+
+ // reset our underlying transport
+ mem.borrow_mut().empty_write_buffer();
+
+ // write and flush again
+ assert_eq!(t.write(&b).unwrap(), 5);
+ assert!(t.flush().is_ok());
+
+ // check the flushed bytes
+ {
+ let inner = mem.borrow_mut();
+ let underlying_buffer = inner.write_buffer_as_ref();
+ assert_eq!(b, underlying_buffer);
+ }
+ }
+}
diff --git a/lib/rs/src/transport/framed.rs b/lib/rs/src/transport/framed.rs
new file mode 100644
index 0000000..75c12f4
--- /dev/null
+++ b/lib/rs/src/transport/framed.rs
@@ -0,0 +1,187 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
+use std::cell::RefCell;
+use std::cmp;
+use std::io;
+use std::io::{ErrorKind, Read, Write};
+use std::rc::Rc;
+
+use super::{TTransport, TTransportFactory};
+
+/// Default capacity of the read buffer in bytes.
+const WRITE_BUFFER_CAPACITY: usize = 4096;
+
+/// Default capacity of the write buffer in bytes..
+const DEFAULT_WBUFFER_CAPACITY: usize = 4096;
+
+/// Transport that communicates with endpoints using framed messages.
+///
+/// A `TFramedTransport` maintains a fixed-size internal write buffer. All
+/// writes are made to this buffer and are sent to the wrapped transport only
+/// when `TTransport::flush()` is called. On a flush a fixed-length header with a
+/// count of the buffered bytes is written, followed by the bytes themselves.
+///
+/// A `TFramedTransport` also maintains a fixed-size internal read buffer.
+/// On a call to `TTransport::read(...)` one full message - both fixed-length
+/// header and bytes - is read from the wrapped transport and buffered.
+/// Subsequent read calls are serviced from the internal buffer until it is
+/// exhausted, at which point the next full message is read from the wrapped
+/// transport.
+///
+/// # Examples
+///
+/// Create and use a `TFramedTransport`.
+///
+/// ```no_run
+/// use std::cell::RefCell;
+/// use std::rc::Rc;
+/// use std::io::{Read, Write};
+/// use thrift::transport::{TFramedTransport, TTcpTransport, TTransport};
+///
+/// let mut t = TTcpTransport::new();
+/// t.open("localhost:9090").unwrap();
+///
+/// let t = Rc::new(RefCell::new(Box::new(t) as Box<TTransport>));
+/// let mut t = TFramedTransport::new(t);
+///
+/// // read
+/// t.read(&mut vec![0u8; 1]).unwrap();
+///
+/// // write
+/// t.write(&[0x00]).unwrap();
+/// t.flush().unwrap();
+/// ```
+pub struct TFramedTransport {
+ rbuf: Box<[u8]>,
+ rpos: usize,
+ rcap: usize,
+ wbuf: Box<[u8]>,
+ wpos: usize,
+ inner: Rc<RefCell<Box<TTransport>>>,
+}
+
+impl TFramedTransport {
+ /// Create a `TFramedTransport` with default-sized internal read and
+ /// write buffers that wraps an `inner` `TTransport`.
+ pub fn new(inner: Rc<RefCell<Box<TTransport>>>) -> TFramedTransport {
+ TFramedTransport::with_capacity(WRITE_BUFFER_CAPACITY, DEFAULT_WBUFFER_CAPACITY, inner)
+ }
+
+ /// Create a `TFramedTransport` with an internal read buffer of size
+ /// `read_buffer_capacity` and an internal write buffer of size
+ /// `write_buffer_capacity` that wraps an `inner` `TTransport`.
+ pub fn with_capacity(read_buffer_capacity: usize,
+ write_buffer_capacity: usize,
+ inner: Rc<RefCell<Box<TTransport>>>)
+ -> TFramedTransport {
+ TFramedTransport {
+ rbuf: vec![0; read_buffer_capacity].into_boxed_slice(),
+ rpos: 0,
+ rcap: 0,
+ wbuf: vec![0; write_buffer_capacity].into_boxed_slice(),
+ wpos: 0,
+ inner: inner,
+ }
+ }
+}
+
+impl Read for TFramedTransport {
+ fn read(&mut self, b: &mut [u8]) -> io::Result<usize> {
+ if self.rcap - self.rpos == 0 {
+ let message_size = self.inner.borrow_mut().read_i32::<BigEndian>()? as usize;
+ if message_size > self.rbuf.len() {
+ return Err(io::Error::new(ErrorKind::Other,
+ format!("bytes to be read ({}) exceeds buffer \
+ capacity ({})",
+ message_size,
+ self.rbuf.len())));
+ }
+ self.inner.borrow_mut().read_exact(&mut self.rbuf[..message_size])?;
+ self.rpos = 0;
+ self.rcap = message_size as usize;
+ }
+
+ let nread = cmp::min(b.len(), self.rcap - self.rpos);
+ b[..nread].clone_from_slice(&self.rbuf[self.rpos..self.rpos + nread]);
+ self.rpos += nread;
+
+ Ok(nread)
+ }
+}
+
+impl Write for TFramedTransport {
+ fn write(&mut self, b: &[u8]) -> io::Result<usize> {
+ if b.len() > (self.wbuf.len() - self.wpos) {
+ return Err(io::Error::new(ErrorKind::Other,
+ format!("bytes to be written ({}) exceeds buffer \
+ capacity ({})",
+ b.len(),
+ self.wbuf.len() - self.wpos)));
+ }
+
+ let nwrite = b.len(); // always less than available write buffer capacity
+ self.wbuf[self.wpos..(self.wpos + nwrite)].clone_from_slice(b);
+ self.wpos += nwrite;
+ Ok(nwrite)
+ }
+
+ fn flush(&mut self) -> io::Result<()> {
+ let message_size = self.wpos;
+
+ if let 0 = message_size {
+ return Ok(());
+ } else {
+ self.inner.borrow_mut().write_i32::<BigEndian>(message_size as i32)?;
+ }
+
+ let mut byte_index = 0;
+ while byte_index < self.wpos {
+ let nwrite = self.inner.borrow_mut().write(&self.wbuf[byte_index..self.wpos])?;
+ byte_index = cmp::min(byte_index + nwrite, self.wpos);
+ }
+
+ self.wpos = 0;
+ self.inner.borrow_mut().flush()
+ }
+}
+
+/// Factory for creating instances of `TFramedTransport`.
+#[derive(Default)]
+pub struct TFramedTransportFactory;
+
+impl TFramedTransportFactory {
+ // Create a `TFramedTransportFactory`.
+ pub fn new() -> TFramedTransportFactory {
+ TFramedTransportFactory {}
+ }
+}
+
+impl TTransportFactory for TFramedTransportFactory {
+ fn create(&self, inner: Rc<RefCell<Box<TTransport>>>) -> Box<TTransport> {
+ Box::new(TFramedTransport::new(inner)) as Box<TTransport>
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ // use std::io::{Read, Write};
+ //
+ // use super::*;
+ // use ::transport::mem::TBufferTransport;
+}
diff --git a/lib/rs/src/transport/mem.rs b/lib/rs/src/transport/mem.rs
new file mode 100644
index 0000000..8ec2a98
--- /dev/null
+++ b/lib/rs/src/transport/mem.rs
@@ -0,0 +1,342 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::cmp;
+use std::io;
+
+/// Simple transport that contains both a fixed-length internal read buffer and
+/// a fixed-length internal write buffer.
+///
+/// On a `write` bytes are written to the internal write buffer. Writes are no
+/// longer accepted once this buffer is full. Callers must `empty_write_buffer()`
+/// before subsequent writes are accepted.
+///
+/// You can set readable bytes in the internal read buffer by filling it with
+/// `set_readable_bytes(...)`. Callers can then read until the buffer is
+/// depleted. No further reads are accepted until the internal read buffer is
+/// replenished again.
+pub struct TBufferTransport {
+ rbuf: Box<[u8]>,
+ rpos: usize,
+ ridx: usize,
+ rcap: usize,
+ wbuf: Box<[u8]>,
+ wpos: usize,
+ wcap: usize,
+}
+
+impl TBufferTransport {
+ /// Constructs a new, empty `TBufferTransport` with the given
+ /// read buffer capacity and write buffer capacity.
+ pub fn with_capacity(read_buffer_capacity: usize,
+ write_buffer_capacity: usize)
+ -> TBufferTransport {
+ TBufferTransport {
+ rbuf: vec![0; read_buffer_capacity].into_boxed_slice(),
+ ridx: 0,
+ rpos: 0,
+ rcap: read_buffer_capacity,
+ wbuf: vec![0; write_buffer_capacity].into_boxed_slice(),
+ wpos: 0,
+ wcap: write_buffer_capacity,
+ }
+ }
+
+ /// Return a slice containing the bytes held by the internal read buffer.
+ /// Returns an empty slice if no readable bytes are present.
+ pub fn read_buffer(&self) -> &[u8] {
+ &self.rbuf[..self.ridx]
+ }
+
+ // FIXME: do I really need this API call?
+ // FIXME: should this simply reset to the last set of readable bytes?
+ /// Reset the number of readable bytes to zero.
+ ///
+ /// Subsequent calls to `read` will return nothing.
+ pub fn empty_read_buffer(&mut self) {
+ self.rpos = 0;
+ self.ridx = 0;
+ }
+
+ /// Copy bytes from the source buffer `buf` into the internal read buffer,
+ /// overwriting any existing bytes. Returns the number of bytes copied,
+ /// which is `min(buf.len(), internal_read_buf.len())`.
+ pub fn set_readable_bytes(&mut self, buf: &[u8]) -> usize {
+ self.empty_read_buffer();
+ let max_bytes = cmp::min(self.rcap, buf.len());
+ self.rbuf[..max_bytes].clone_from_slice(&buf[..max_bytes]);
+ self.ridx = max_bytes;
+ max_bytes
+ }
+
+ /// Return a slice containing the bytes held by the internal write buffer.
+ /// Returns an empty slice if no bytes were written.
+ pub fn write_buffer_as_ref(&self) -> &[u8] {
+ &self.wbuf[..self.wpos]
+ }
+
+ /// Return a vector with a copy of the bytes held by the internal write buffer.
+ /// Returns an empty vector if no bytes were written.
+ pub fn write_buffer_to_vec(&self) -> Vec<u8> {
+ let mut buf = vec![0u8; self.wpos];
+ buf.copy_from_slice(&self.wbuf[..self.wpos]);
+ buf
+ }
+
+ /// Resets the internal write buffer, making it seem like no bytes were
+ /// written. Calling `write_buffer` after this returns an empty slice.
+ pub fn empty_write_buffer(&mut self) {
+ self.wpos = 0;
+ }
+
+ /// Overwrites the contents of the read buffer with the contents of the
+ /// write buffer. The write buffer is emptied after this operation.
+ pub fn copy_write_buffer_to_read_buffer(&mut self) {
+ let buf = {
+ let b = self.write_buffer_as_ref();
+ let mut b_ret = vec![0; b.len()];
+ b_ret.copy_from_slice(&b);
+ b_ret
+ };
+
+ let bytes_copied = self.set_readable_bytes(&buf);
+ assert_eq!(bytes_copied, buf.len());
+
+ self.empty_write_buffer();
+ }
+}
+
+impl io::Read for TBufferTransport {
+ fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+ let nread = cmp::min(buf.len(), self.ridx - self.rpos);
+ buf[..nread].clone_from_slice(&self.rbuf[self.rpos..self.rpos + nread]);
+ self.rpos += nread;
+ Ok(nread)
+ }
+}
+
+impl io::Write for TBufferTransport {
+ fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+ let nwrite = cmp::min(buf.len(), self.wcap - self.wpos);
+ self.wbuf[self.wpos..self.wpos + nwrite].clone_from_slice(&buf[..nwrite]);
+ self.wpos += nwrite;
+ Ok(nwrite)
+ }
+
+ fn flush(&mut self) -> io::Result<()> {
+ Ok(()) // nothing to do on flush
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::io::{Read, Write};
+
+ use super::TBufferTransport;
+
+ #[test]
+ fn must_empty_write_buffer() {
+ let mut t = TBufferTransport::with_capacity(0, 1);
+
+ let bytes_to_write: [u8; 1] = [0x01];
+ let result = t.write(&bytes_to_write);
+ assert_eq!(result.unwrap(), 1);
+ assert_eq!(&t.write_buffer_as_ref(), &bytes_to_write);
+
+ t.empty_write_buffer();
+ assert_eq!(t.write_buffer_as_ref().len(), 0);
+ }
+
+ #[test]
+ fn must_accept_writes_after_buffer_emptied() {
+ let mut t = TBufferTransport::with_capacity(0, 2);
+
+ let bytes_to_write: [u8; 2] = [0x01, 0x02];
+
+ // first write (all bytes written)
+ let result = t.write(&bytes_to_write);
+ assert_eq!(result.unwrap(), 2);
+ assert_eq!(&t.write_buffer_as_ref(), &bytes_to_write);
+
+ // try write again (nothing should be written)
+ let result = t.write(&bytes_to_write);
+ assert_eq!(result.unwrap(), 0);
+ assert_eq!(&t.write_buffer_as_ref(), &bytes_to_write); // still the same as before
+
+ // now reset the buffer
+ t.empty_write_buffer();
+ assert_eq!(t.write_buffer_as_ref().len(), 0);
+
+ // now try write again - the write should succeed
+ let result = t.write(&bytes_to_write);
+ assert_eq!(result.unwrap(), 2);
+ assert_eq!(&t.write_buffer_as_ref(), &bytes_to_write);
+ }
+
+ #[test]
+ fn must_accept_multiple_writes_until_buffer_is_full() {
+ let mut t = TBufferTransport::with_capacity(0, 10);
+
+ // first write (all bytes written)
+ let bytes_to_write_0: [u8; 2] = [0x01, 0x41];
+ let write_0_result = t.write(&bytes_to_write_0);
+ assert_eq!(write_0_result.unwrap(), 2);
+ assert_eq!(t.write_buffer_as_ref(), &bytes_to_write_0);
+
+ // second write (all bytes written, starting at index 2)
+ let bytes_to_write_1: [u8; 7] = [0x24, 0x41, 0x32, 0x33, 0x11, 0x98, 0xAF];
+ let write_1_result = t.write(&bytes_to_write_1);
+ assert_eq!(write_1_result.unwrap(), 7);
+ assert_eq!(&t.write_buffer_as_ref()[2..], &bytes_to_write_1);
+
+ // third write (only 1 byte written - that's all we have space for)
+ let bytes_to_write_2: [u8; 3] = [0xBF, 0xDA, 0x98];
+ let write_2_result = t.write(&bytes_to_write_2);
+ assert_eq!(write_2_result.unwrap(), 1);
+ assert_eq!(&t.write_buffer_as_ref()[9..], &bytes_to_write_2[0..1]); // how does this syntax work?!
+
+ // fourth write (no writes are accepted)
+ let bytes_to_write_3: [u8; 3] = [0xBF, 0xAA, 0xFD];
+ let write_3_result = t.write(&bytes_to_write_3);
+ assert_eq!(write_3_result.unwrap(), 0);
+
+ // check the full write buffer
+ let mut expected: Vec<u8> = Vec::with_capacity(10);
+ expected.extend_from_slice(&bytes_to_write_0);
+ expected.extend_from_slice(&bytes_to_write_1);
+ expected.extend_from_slice(&bytes_to_write_2[0..1]);
+ assert_eq!(t.write_buffer_as_ref(), &expected[..]);
+ }
+
+ #[test]
+ fn must_empty_read_buffer() {
+ let mut t = TBufferTransport::with_capacity(1, 0);
+
+ let bytes_to_read: [u8; 1] = [0x01];
+ let result = t.set_readable_bytes(&bytes_to_read);
+ assert_eq!(result, 1);
+ assert_eq!(&t.read_buffer(), &bytes_to_read);
+
+ t.empty_read_buffer();
+ assert_eq!(t.read_buffer().len(), 0);
+ }
+
+ #[test]
+ fn must_allow_readable_bytes_to_be_set_after_read_buffer_emptied() {
+ let mut t = TBufferTransport::with_capacity(1, 0);
+
+ let bytes_to_read_0: [u8; 1] = [0x01];
+ let result = t.set_readable_bytes(&bytes_to_read_0);
+ assert_eq!(result, 1);
+ assert_eq!(&t.read_buffer(), &bytes_to_read_0);
+
+ t.empty_read_buffer();
+ assert_eq!(t.read_buffer().len(), 0);
+
+ let bytes_to_read_1: [u8; 1] = [0x02];
+ let result = t.set_readable_bytes(&bytes_to_read_1);
+ assert_eq!(result, 1);
+ assert_eq!(&t.read_buffer(), &bytes_to_read_1);
+ }
+
+ #[test]
+ fn must_accept_multiple_reads_until_all_bytes_read() {
+ let mut t = TBufferTransport::with_capacity(10, 0);
+
+ let readable_bytes: [u8; 10] = [0xFF, 0xEE, 0xDD, 0xCC, 0xBB, 0x00, 0x1A, 0x2B, 0x3C, 0x4D];
+
+ // check that we're able to set the bytes to be read
+ let result = t.set_readable_bytes(&readable_bytes);
+ assert_eq!(result, 10);
+ assert_eq!(&t.read_buffer(), &readable_bytes);
+
+ // first read
+ let mut read_buf_0 = vec![0; 5];
+ let read_result = t.read(&mut read_buf_0);
+ assert_eq!(read_result.unwrap(), 5);
+ assert_eq!(read_buf_0.as_slice(), &(readable_bytes[0..5]));
+
+ // second read
+ let mut read_buf_1 = vec![0; 4];
+ let read_result = t.read(&mut read_buf_1);
+ assert_eq!(read_result.unwrap(), 4);
+ assert_eq!(read_buf_1.as_slice(), &(readable_bytes[5..9]));
+
+ // third read (only 1 byte remains to be read)
+ let mut read_buf_2 = vec![0; 3];
+ let read_result = t.read(&mut read_buf_2);
+ assert_eq!(read_result.unwrap(), 1);
+ read_buf_2.truncate(1); // FIXME: does the caller have to do this?
+ assert_eq!(read_buf_2.as_slice(), &(readable_bytes[9..]));
+
+ // fourth read (nothing should be readable)
+ let mut read_buf_3 = vec![0; 10];
+ let read_result = t.read(&mut read_buf_3);
+ assert_eq!(read_result.unwrap(), 0);
+ read_buf_3.truncate(0);
+
+ // check that all the bytes we received match the original (again!)
+ let mut bytes_read = Vec::with_capacity(10);
+ bytes_read.extend_from_slice(&read_buf_0);
+ bytes_read.extend_from_slice(&read_buf_1);
+ bytes_read.extend_from_slice(&read_buf_2);
+ bytes_read.extend_from_slice(&read_buf_3);
+ assert_eq!(&bytes_read, &readable_bytes);
+ }
+
+ #[test]
+ fn must_allow_reads_to_succeed_after_read_buffer_replenished() {
+ let mut t = TBufferTransport::with_capacity(3, 0);
+
+ let readable_bytes_0: [u8; 3] = [0x02, 0xAB, 0x33];
+
+ // check that we're able to set the bytes to be read
+ let result = t.set_readable_bytes(&readable_bytes_0);
+ assert_eq!(result, 3);
+ assert_eq!(&t.read_buffer(), &readable_bytes_0);
+
+ let mut read_buf = vec![0; 4];
+
+ // drain the read buffer
+ let read_result = t.read(&mut read_buf);
+ assert_eq!(read_result.unwrap(), 3);
+ assert_eq!(t.read_buffer(), &read_buf[0..3]);
+
+ // check that a subsequent read fails
+ let read_result = t.read(&mut read_buf);
+ assert_eq!(read_result.unwrap(), 0);
+
+ // we don't modify the read buffer on failure
+ let mut expected_bytes = Vec::with_capacity(4);
+ expected_bytes.extend_from_slice(&readable_bytes_0);
+ expected_bytes.push(0x00);
+ assert_eq!(&read_buf, &expected_bytes);
+
+ // replenish the read buffer again
+ let readable_bytes_1: [u8; 2] = [0x91, 0xAA];
+
+ // check that we're able to set the bytes to be read
+ let result = t.set_readable_bytes(&readable_bytes_1);
+ assert_eq!(result, 2);
+ assert_eq!(&t.read_buffer(), &readable_bytes_1);
+
+ // read again
+ let read_result = t.read(&mut read_buf);
+ assert_eq!(read_result.unwrap(), 2);
+ assert_eq!(t.read_buffer(), &read_buf[0..2]);
+ }
+}
diff --git a/lib/rs/src/transport/mod.rs b/lib/rs/src/transport/mod.rs
new file mode 100644
index 0000000..bbabd66
--- /dev/null
+++ b/lib/rs/src/transport/mod.rs
@@ -0,0 +1,51 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Types required to send and receive bytes over an I/O channel.
+//!
+//! The core type is the `TTransport` trait, through which a `TProtocol` can
+//! send and receive primitives over the wire. While `TProtocol` instances deal
+//! with primitive types, `TTransport` instances understand only bytes.
+
+use std::cell::RefCell;
+use std::io;
+use std::rc::Rc;
+
+mod buffered;
+mod framed;
+mod passthru;
+mod socket;
+
+#[cfg(test)]
+pub mod mem;
+
+pub use self::buffered::{TBufferedTransport, TBufferedTransportFactory};
+pub use self::framed::{TFramedTransport, TFramedTransportFactory};
+pub use self::passthru::TPassThruTransport;
+pub use self::socket::TTcpTransport;
+
+/// Identifies an I/O channel that can be used to send and receive bytes.
+pub trait TTransport: io::Read + io::Write {}
+impl<I: io::Read + io::Write> TTransport for I {}
+
+/// Helper type used by servers to create `TTransport` instances for accepted
+/// client connections.
+pub trait TTransportFactory {
+ /// Create a `TTransport` that wraps an `inner` transport, thus creating
+ /// a transport stack.
+ fn create(&self, inner: Rc<RefCell<Box<TTransport>>>) -> Box<TTransport>;
+}
diff --git a/lib/rs/src/transport/passthru.rs b/lib/rs/src/transport/passthru.rs
new file mode 100644
index 0000000..60dc3a6
--- /dev/null
+++ b/lib/rs/src/transport/passthru.rs
@@ -0,0 +1,73 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::cell::RefCell;
+use std::rc::Rc;
+use std::io;
+use std::io::{Read, Write};
+
+use super::TTransport;
+
+/// Proxy that wraps an inner `TTransport` and delegates all calls to it.
+///
+/// Unlike other `TTransport` wrappers, `TPassThruTransport` is generic with
+/// regards to the wrapped transport. This allows callers to use methods
+/// specific to the type being wrapped instead of being constrained to methods
+/// on the `TTransport` trait.
+///
+/// # Examples
+///
+/// Create and use a `TPassThruTransport`.
+///
+/// ```no_run
+/// use std::cell::RefCell;
+/// use std::rc::Rc;
+/// use thrift::transport::{TPassThruTransport, TTcpTransport};
+///
+/// let t = TTcpTransport::new();
+/// let t = TPassThruTransport::new(Rc::new(RefCell::new(Box::new(t))));
+///
+/// // since the type parameter is maintained, we are able
+/// // to use functions specific to `TTcpTransport`
+/// t.inner.borrow_mut().open("localhost:9090").unwrap();
+/// ```
+pub struct TPassThruTransport<I: TTransport> {
+ pub inner: Rc<RefCell<Box<I>>>,
+}
+
+impl<I: TTransport> TPassThruTransport<I> {
+ /// Create a `TPassThruTransport` that wraps an `inner` TTransport.
+ pub fn new(inner: Rc<RefCell<Box<I>>>) -> TPassThruTransport<I> {
+ TPassThruTransport { inner: inner }
+ }
+}
+
+impl<I: TTransport> Read for TPassThruTransport<I> {
+ fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+ self.inner.borrow_mut().read(buf)
+ }
+}
+
+impl<I: TTransport> Write for TPassThruTransport<I> {
+ fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+ self.inner.borrow_mut().write(buf)
+ }
+
+ fn flush(&mut self) -> io::Result<()> {
+ self.inner.borrow_mut().flush()
+ }
+}
diff --git a/lib/rs/src/transport/socket.rs b/lib/rs/src/transport/socket.rs
new file mode 100644
index 0000000..9f2b8ba
--- /dev/null
+++ b/lib/rs/src/transport/socket.rs
@@ -0,0 +1,141 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::convert::From;
+use std::io;
+use std::io::{ErrorKind, Read, Write};
+use std::net::{Shutdown, TcpStream};
+use std::ops::Drop;
+
+use ::{TransportError, TransportErrorKind};
+
+/// Communicate with a Thrift service over a TCP socket.
+///
+/// # Examples
+///
+/// Create a `TTcpTransport`.
+///
+/// ```no_run
+/// use std::io::{Read, Write};
+/// use thrift::transport::TTcpTransport;
+///
+/// let mut t = TTcpTransport::new();
+/// t.open("localhost:9090").unwrap();
+///
+/// let mut buf = vec![0u8; 4];
+/// t.read(&mut buf).unwrap();
+/// t.write(&vec![0, 1, 2]).unwrap();
+/// ```
+///
+/// Create a `TTcpTransport` by wrapping an existing `TcpStream`.
+///
+/// ```no_run
+/// use std::io::{Read, Write};
+/// use std::net::TcpStream;
+/// use thrift::transport::TTcpTransport;
+///
+/// let stream = TcpStream::connect("127.0.0.1:9189").unwrap();
+/// let mut t = TTcpTransport::with_stream(stream);
+///
+/// // no need to call t.open() since we've already connected above
+///
+/// let mut buf = vec![0u8; 4];
+/// t.read(&mut buf).unwrap();
+/// t.write(&vec![0, 1, 2]).unwrap();
+/// ```
+#[derive(Default)]
+pub struct TTcpTransport {
+ stream: Option<TcpStream>,
+}
+
+impl TTcpTransport {
+ /// Create an uninitialized `TTcpTransport`.
+ ///
+ /// The returned instance must be opened using `TTcpTransport::open(...)`
+ /// before it can be used.
+ pub fn new() -> TTcpTransport {
+ TTcpTransport { stream: None }
+ }
+
+ /// Create a `TTcpTransport` that wraps an existing `TcpStream`.
+ ///
+ /// The passed-in stream is assumed to have been opened before being wrapped
+ /// by the created `TTcpTransport` instance.
+ pub fn with_stream(stream: TcpStream) -> TTcpTransport {
+ TTcpTransport { stream: Some(stream) }
+ }
+
+ /// Connect to `remote_address`, which should have the form `host:port`.
+ pub fn open(&mut self, remote_address: &str) -> ::Result<()> {
+ if self.stream.is_some() {
+ Err(::Error::Transport(TransportError::new(TransportErrorKind::AlreadyOpen,
+ "transport previously opened")))
+ } else {
+ match TcpStream::connect(&remote_address) {
+ Ok(s) => {
+ self.stream = Some(s);
+ Ok(())
+ }
+ Err(e) => Err(From::from(e)),
+ }
+ }
+ }
+
+ /// Shutdown this transport.
+ ///
+ /// Both send and receive halves are closed, and this instance can no
+ /// longer be used to communicate with another endpoint.
+ pub fn close(&mut self) -> ::Result<()> {
+ self.if_set(|s| s.shutdown(Shutdown::Both)).map_err(From::from)
+ }
+
+ fn if_set<F, T>(&mut self, mut stream_operation: F) -> io::Result<T>
+ where F: FnMut(&mut TcpStream) -> io::Result<T>
+ {
+
+ if let Some(ref mut s) = self.stream {
+ stream_operation(s)
+ } else {
+ Err(io::Error::new(ErrorKind::NotConnected, "tcp endpoint not connected"))
+ }
+ }
+}
+
+impl Read for TTcpTransport {
+ fn read(&mut self, b: &mut [u8]) -> io::Result<usize> {
+ self.if_set(|s| s.read(b))
+ }
+}
+
+impl Write for TTcpTransport {
+ fn write(&mut self, b: &[u8]) -> io::Result<usize> {
+ self.if_set(|s| s.write(b))
+ }
+
+ fn flush(&mut self) -> io::Result<()> {
+ self.if_set(|s| s.flush())
+ }
+}
+
+// Do I have to implement the Drop trait? TcpStream closes the socket on drop.
+impl Drop for TTcpTransport {
+ fn drop(&mut self) {
+ if let Err(e) = self.close() {
+ warn!("error while closing socket transport: {:?}", e)
+ }
+ }
+}