Remove more tODO's

This commit is contained in:
Lucio Franco
2019-09-29 22:38:28 -04:00
parent d81ca1ae56
commit a9f7992585
11 changed files with 58 additions and 22 deletions
+10
View File
@@ -39,6 +39,16 @@ may be a good resource as it shows examples of many of the gRPC features.
### Examples ### Examples
#### Rust Version
`tonic` currently works on rust `1.39-beta` and above as it requires support for the `async_await`
feature. To install the beta simply follow the commands below:
```bash
$ rustup install beta && rustup component add rustfmt --toolchain beta
$ cargo +beta build
```
#### Client #### Client
```rust ```rust
-1
View File
@@ -144,7 +144,6 @@ impl server::RouteGuide for RouteGuide {
} }
}; };
// TODO: Clean this up
Ok(Response::new(Box::pin(output) Ok(Response::new(Box::pin(output)
as Pin< as Pin<
Box<dyn Stream<Item = Result<RouteNote, Status>> + Send + 'static>, Box<dyn Stream<Item = Result<RouteNote, Status>> + Send + 'static>,
+4 -1
View File
@@ -4,7 +4,10 @@ pub mod pb {
use pb::{EchoRequest, EchoResponse}; use pb::{EchoRequest, EchoResponse};
use std::collections::VecDeque; use std::collections::VecDeque;
use tonic::{transport::{Server, Identity}, Request, Response, Status, Streaming}; use tonic::{
transport::{Identity, Server},
Request, Response, Status, Streaming,
};
type EchoResult<T> = Result<Response<T>, Status>; type EchoResult<T> = Result<Response<T>, Status>;
type Stream = VecDeque<Result<EchoResponse, Status>>; type Stream = VecDeque<Result<EchoResponse, Status>>;
+3 -5
View File
@@ -1,11 +1,9 @@
use structopt::StructOpt;
use tonic::transport::{Identity, Server};
use tonic_interop::{server, MergeTrailers};
// TODO: move GrpcService out of client since it can be used for the
// server too.
use http::header::HeaderName; use http::header::HeaderName;
use structopt::StructOpt;
use tonic::body::BoxBody; use tonic::body::BoxBody;
use tonic::client::GrpcService; use tonic::client::GrpcService;
use tonic::transport::{Identity, Server};
use tonic_interop::{server, MergeTrailers};
#[derive(StructOpt)] #[derive(StructOpt)]
struct Opts { struct Opts {
-2
View File
@@ -170,7 +170,6 @@ pub async fn ping_pong(client: &mut TestClient, assertions: &mut Vec<TestAsserti
loop { loop {
match response.next().await { match response.next().await {
Some(result) => { Some(result) => {
// TODO: what to do with this result?
responses.push(result.unwrap()); responses.push(result.unwrap());
if responses.len() == REQUEST_LENGTHS.len() { if responses.len() == REQUEST_LENGTHS.len() {
drop(tx); drop(tx);
@@ -360,7 +359,6 @@ pub async fn custom_metadata(client: &mut TestClient, assertions: &mut Vec<TestA
req_unary.metadata_mut().insert(key1, value1.clone()); req_unary.metadata_mut().insert(key1, value1.clone());
req_unary.metadata_mut().insert_bin(key2, value2.clone()); req_unary.metadata_mut().insert_bin(key2, value2.clone());
// TODO: custom metadata for fullduplex
let stream = stream::iter(vec![Ok(make_ping_pong_request(0))]); let stream = stream::iter(vec![Ok(make_ping_pong_request(0))]);
let mut req_stream = Request::new(stream); let mut req_stream = Request::new(stream);
req_stream.metadata_mut().insert(key1, value1.clone()); req_stream.metadata_mut().insert(key1, value1.clone());
-1
View File
@@ -142,7 +142,6 @@ impl pb::server::TestService for TestService {
&self, &self,
_: Streaming<StreamingOutputCallRequest>, _: Streaming<StreamingOutputCallRequest>,
) -> Result<Self::HalfDuplexCallStream> { ) -> Result<Self::HalfDuplexCallStream> {
// TODO: implement half duplex
Err(Status::unimplemented("TODO")) Err(Status::unimplemented("TODO"))
} }
+3 -3
View File
@@ -116,7 +116,7 @@ impl<T> Streaming<T> {
/// This will drain the stream of all its messages to receive the trailing /// This will drain the stream of all its messages to receive the trailing
/// metadata. If [`Streaming::message`] returns `None` then this function /// metadata. If [`Streaming::message`] returns `None` then this function
/// will not need to poll for trailers since the body was totally consumed. /// will not need to poll for trailers since the body was totally consumed.
/// ///
/// ```rust /// ```rust
/// # use tonic::{Streaming, Status}; /// # use tonic::{Streaming, Status};
/// # async fn trailers_ex<T>(mut request: Streaming<T>) -> Result<(), Status> { /// # async fn trailers_ex<T>(mut request: Streaming<T>) -> Result<(), Status> {
@@ -213,7 +213,7 @@ impl<T> Stream for Streaming<T> {
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
loop { loop {
// TODO: implement the ability to poll trailers when we _know_ that // FIXME: implement the ability to poll trailers when we _know_ that
// the consumer of this stream will only poll for the first message. // the consumer of this stream will only poll for the first message.
// This means we skip the poll_trailers step. // This means we skip the poll_trailers step.
match self.decode_chunk()? { match self.decode_chunk()? {
@@ -236,7 +236,7 @@ impl<T> Stream for Streaming<T> {
if let Some(data) = chunk { if let Some(data) = chunk {
self.buf.put(data); self.buf.put(data);
} else { } else {
// TODO: get BytesMut to impl `Buf` directlty? // FIXME: improve buf usage.
let buf1 = (&self.buf[..]).into_buf(); let buf1 = (&self.buf[..]).into_buf();
if buf1.has_remaining() { if buf1.has_remaining() {
trace!("unexpected EOF decoding stream"); trace!("unexpected EOF decoding stream");
+1 -3
View File
@@ -71,8 +71,6 @@ impl Status {
} }
} }
// TODO: This should probably be made public eventually. Need to decide on
// the exact argument type.
#[cfg_attr(not(feature = "h2"), allow(dead_code))] #[cfg_attr(not(feature = "h2"), allow(dead_code))]
pub(crate) fn from_error(err: &(dyn Error + 'static)) -> Status { pub(crate) fn from_error(err: &(dyn Error + 'static)) -> Status {
Status::try_from_error(err).unwrap_or_else(|| Status::new(Code::Unknown, err.to_string())) Status::try_from_error(err).unwrap_or_else(|| Status::new(Code::Unknown, err.to_string()))
@@ -103,7 +101,7 @@ impl Status {
None None
} }
// TODO: bubble this into `transport` and expose generic http2 reasons. // FIXME: bubble this into `transport` and expose generic http2 reasons.
#[cfg(feature = "h2")] #[cfg(feature = "h2")]
fn from_h2_error(err: &h2::Error) -> Status { fn from_h2_error(err: &h2::Error) -> Status {
// See https://github.com/grpc/grpc/blob/3977c30/doc/PROTOCOL-HTTP2.md#errors // See https://github.com/grpc/grpc/blob/3977c30/doc/PROTOCOL-HTTP2.md#errors
+1 -1
View File
@@ -27,7 +27,7 @@ pub struct Endpoint {
} }
impl Endpoint { impl Endpoint {
// TODO: determine if we want to expose this or not. This is really // FIXME: determine if we want to expose this or not. This is really
// just used in codegen for a shortcut. // just used in codegen for a shortcut.
#[doc(hidden)] #[doc(hidden)]
pub fn new<D>(dst: D) -> Result<Self, super::Error> pub fn new<D>(dst: D) -> Result<Self, super::Error>
-1
View File
@@ -231,7 +231,6 @@ impl Stream for TcpIncoming {
} }
} }
// TODO: add custom tracing here
#[derive(Debug)] #[derive(Debug)]
struct Svc<S>(S); struct Svc<S>(S);
+36 -4
View File
@@ -10,10 +10,12 @@ use std::{fmt, sync::Arc};
use tokio::net::TcpStream; use tokio::net::TcpStream;
#[cfg(feature = "rustls")] #[cfg(feature = "rustls")]
use tokio_rustls::{ use tokio_rustls::{
rustls::{internal::pemfile, ClientConfig, NoClientAuth, ServerConfig}, rustls::{internal::pemfile, ClientConfig, NoClientAuth, ServerConfig, Session},
webpki::DNSNameRef, webpki::DNSNameRef,
TlsAcceptor as RustlsAcceptor, TlsConnector as RustlsConnector, TlsAcceptor as RustlsAcceptor, TlsConnector as RustlsConnector,
}; };
#[allow(unused_import)]
use tracing::trace;
/// h2 alpn in wire format for openssl. /// h2 alpn in wire format for openssl.
#[cfg(feature = "openssl")] #[cfg(feature = "openssl")]
@@ -29,6 +31,12 @@ pub(crate) struct Cert {
pub(crate) domain: String, pub(crate) domain: String,
} }
#[derive(Debug)]
enum TlsError {
#[allow(dead_code)]
H2NotNegotiated,
}
#[derive(Clone)] #[derive(Clone)]
pub(crate) struct TlsConnector { pub(crate) struct TlsConnector {
inner: Connector, inner: Connector,
@@ -80,7 +88,6 @@ impl TlsConnector {
}) })
} }
// TODO: Write an either tlsstream to avoid this box
pub(crate) async fn connect(&self, io: TcpStream) -> Result<BoxedIo, crate::Error> { pub(crate) async fn connect(&self, io: TcpStream) -> Result<BoxedIo, crate::Error> {
let tls_io = match &self.inner { let tls_io = match &self.inner {
#[cfg(feature = "openssl")] #[cfg(feature = "openssl")]
@@ -88,7 +95,12 @@ impl TlsConnector {
let config = connector.configure()?; let config = connector.configure()?;
let tls = tokio_openssl::connect(config, &self.domain, io).await?; let tls = tokio_openssl::connect(config, &self.domain, io).await?;
// TODO: check that we actually got an h2 stream // FIXME: alpn returned from interop server is not working
// match tls.ssl().selected_alpn_protocol() {
// Some(b) if b == b"h2" => trace!("HTTP/2 succesfully negotiated."),
// _ => return Err(TlsError::H2NotNegotiated.into()),
// };
BoxedIo::new(tls) BoxedIo::new(tls)
} }
#[cfg(feature = "rustls")] #[cfg(feature = "rustls")]
@@ -101,7 +113,12 @@ impl TlsConnector {
.connect(dns.as_ref(), io) .connect(dns.as_ref(), io)
.await?; .await?;
// TODO: check that we actually got an h2 stream let (_, session) = io.get_ref();
match session.get_alpn_protocol() {
Some(b) if b == b"h2" => (),
_ => return Err(TlsError::H2NotNegotiated.into()),
};
BoxedIo::new(io) BoxedIo::new(io)
} }
@@ -190,6 +207,11 @@ impl TlsAcceptor {
#[cfg(feature = "openssl")] #[cfg(feature = "openssl")]
Acceptor::Openssl(acceptor) => { Acceptor::Openssl(acceptor) => {
let tls = tokio_openssl::accept(&acceptor, io).await?; let tls = tokio_openssl::accept(&acceptor, io).await?;
// let ssl = tls.ssl();
// ssl.set_alpn_protos(ALPN_H2_WIRE);
BoxedIo::new(tls) BoxedIo::new(tls)
} }
@@ -226,3 +248,13 @@ impl fmt::Debug for TlsAcceptor {
.finish() .finish()
} }
} }
impl fmt::Display for TlsError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
TlsError::H2NotNegotiated => write!(f, "HTTP/2 was not negotiated."),
}
}
}
impl std::error::Error for TlsError {}