@@ -134,7 +134,7 @@ impl<T> Streaming<T> {
|
||||
}
|
||||
|
||||
// To fetch the trailers we must clear the body and drop it.
|
||||
while let Some(_) = self.message().await? {}
|
||||
while self.message().await?.is_some() {}
|
||||
|
||||
// Since we call poll_trailers internally on poll_next we need to
|
||||
// check if it got cached again.
|
||||
@@ -219,9 +219,8 @@ impl<T> Stream for Streaming<T> {
|
||||
// FIXME: implement the ability to poll trailers when we _know_ that
|
||||
// the consumer of this stream will only poll for the first message.
|
||||
// This means we skip the poll_trailers step.
|
||||
match self.decode_chunk()? {
|
||||
Some(item) => return Poll::Ready(Some(Ok(item))),
|
||||
None => (),
|
||||
if let Some(item) = self.decode_chunk()? {
|
||||
return Poll::Ready(Some(Ok(item)));
|
||||
}
|
||||
|
||||
let chunk = match ready!(Pin::new(&mut self.body).poll_data(cx)) {
|
||||
@@ -230,8 +229,7 @@ impl<T> Stream for Streaming<T> {
|
||||
let err: crate::Error = e.into();
|
||||
debug!("decoder inner stream error: {:?}", err);
|
||||
let status = Status::from_error(&*err);
|
||||
Err(status)?;
|
||||
break;
|
||||
return Poll::Ready(Some(Err(status)));
|
||||
}
|
||||
None => None,
|
||||
};
|
||||
@@ -252,10 +250,10 @@ impl<T> Stream for Streaming<T> {
|
||||
// FIXME: improve buf usage.
|
||||
if self.buf.has_remaining() {
|
||||
trace!("unexpected EOF decoding stream");
|
||||
Err(Status::new(
|
||||
return Poll::Ready(Some(Err(Status::new(
|
||||
Code::Internal,
|
||||
"Unexpected EOF decoding stream.".to_string(),
|
||||
))?;
|
||||
))));
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
|
||||
@@ -35,7 +35,7 @@ where
|
||||
T::Item: Send + Sync,
|
||||
U: Stream<Item = T::Item> + Send + Sync + 'static,
|
||||
{
|
||||
let stream = encode(encoder, source.map(|x| Ok(x))).into_stream();
|
||||
let stream = encode(encoder, source.map(Ok)).into_stream();
|
||||
EncodeBody::new_client(stream)
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
use crate::{Request, Status};
|
||||
use std::{fmt, sync::Arc};
|
||||
|
||||
type InterceptorFn =
|
||||
Arc<dyn Fn(Request<()>) -> Result<Request<()>, Status> + Send + Sync + 'static>;
|
||||
|
||||
/// Represents a gRPC interceptor.
|
||||
///
|
||||
/// gRPC interceptors are similar to middleware but have much less
|
||||
@@ -16,7 +19,7 @@ use std::{fmt, sync::Arc};
|
||||
/// features to the body of the request, going through the `tower` abstraction is recommended.
|
||||
#[derive(Clone)]
|
||||
pub struct Interceptor {
|
||||
f: Arc<dyn Fn(Request<()>) -> Result<Request<()>, Status> + Send + Sync + 'static>,
|
||||
f: InterceptorFn,
|
||||
}
|
||||
|
||||
impl Interceptor {
|
||||
|
||||
+3
-2
@@ -505,11 +505,12 @@ impl Status {
|
||||
Status {
|
||||
code,
|
||||
message: message.into(),
|
||||
details: details,
|
||||
metadata: metadata,
|
||||
details,
|
||||
metadata,
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::wrong_self_convention)]
|
||||
/// Build an `http::Response` from the given `Status`.
|
||||
pub fn to_http(self) -> http::Response<BoxBody> {
|
||||
let (mut parts, _body) = http::Response::new(()).into_parts();
|
||||
|
||||
@@ -192,7 +192,7 @@ impl Endpoint {
|
||||
tls: Some(
|
||||
tls_config
|
||||
.tls_connector(self.uri.clone())
|
||||
.map_err(|e| Error::from_source(e))?,
|
||||
.map_err(Error::from_source)?,
|
||||
),
|
||||
..self
|
||||
})
|
||||
|
||||
@@ -183,7 +183,7 @@ impl GrpcService<BoxBody> for Channel {
|
||||
type Future = ResponseFuture;
|
||||
|
||||
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
GrpcService::poll_ready(&mut self.svc, cx).map_err(|e| super::Error::from_source(e))
|
||||
GrpcService::poll_ready(&mut self.svc, cx).map_err(super::Error::from_source)
|
||||
}
|
||||
|
||||
fn call(&mut self, request: Request<BoxBody>) -> Self::Future {
|
||||
@@ -197,7 +197,7 @@ impl Future for ResponseFuture {
|
||||
|
||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let val = futures_util::ready!(Pin::new(&mut self.inner).poll(cx))
|
||||
.map_err(|e| super::Error::from_source(e))?;
|
||||
.map_err(super::Error::from_source)?;
|
||||
Ok(val).into()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,7 +9,7 @@ use std::fmt;
|
||||
/// Configures TLS settings for endpoints.
|
||||
#[cfg(feature = "tls")]
|
||||
#[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
|
||||
#[derive(Clone)]
|
||||
#[derive(Clone, Default)]
|
||||
pub struct ClientTlsConfig {
|
||||
domain: Option<String>,
|
||||
cert: Option<Certificate>,
|
||||
@@ -80,7 +80,7 @@ impl ClientTlsConfig {
|
||||
|
||||
pub(crate) fn tls_connector(&self, uri: Uri) -> Result<TlsConnector, crate::Error> {
|
||||
let domain = match &self.domain {
|
||||
None => uri.host().ok_or(Error::new_invalid_uri())?.to_string(),
|
||||
None => uri.host().ok_or_else(Error::new_invalid_uri)?.to_string(),
|
||||
Some(domain) => domain.clone(),
|
||||
};
|
||||
match &self.rustls_raw {
|
||||
|
||||
@@ -107,3 +107,6 @@ pub use self::channel::ClientTlsConfig;
|
||||
#[cfg(feature = "tls")]
|
||||
#[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
|
||||
pub use self::server::ServerTlsConfig;
|
||||
|
||||
type BoxFuture<T, E> =
|
||||
std::pin::Pin<Box<dyn std::future::Future<Output = Result<T, E>> + Send + 'static>>;
|
||||
|
||||
@@ -21,7 +21,10 @@ pub(crate) use incoming::TlsStream;
|
||||
#[cfg(feature = "tls")]
|
||||
use crate::transport::Error;
|
||||
|
||||
use super::service::{Or, Routes, ServerIo, ServiceBuilderExt};
|
||||
use super::{
|
||||
service::{Or, Routes, ServerIo, ServiceBuilderExt},
|
||||
BoxFuture,
|
||||
};
|
||||
use crate::{body::BoxBody, request::ConnectionInfo};
|
||||
use futures_core::Stream;
|
||||
use futures_util::{
|
||||
@@ -34,7 +37,6 @@ use std::{
|
||||
fmt,
|
||||
future::Future,
|
||||
net::SocketAddr,
|
||||
pin::Pin,
|
||||
sync::Arc,
|
||||
task::{Context, Poll},
|
||||
time::Duration,
|
||||
@@ -98,11 +100,13 @@ where
|
||||
B::Error: Into<crate::Error> + Send,
|
||||
{
|
||||
type Response = Response<BoxBody>;
|
||||
type Error = crate::Error;
|
||||
|
||||
#[allow(clippy::type_complexity)]
|
||||
type Future = FutureEither<
|
||||
MapErr<A::Future, fn(A::Error) -> crate::Error>,
|
||||
MapErr<B::Future, fn(B::Error) -> crate::Error>,
|
||||
>;
|
||||
type Error = crate::Error;
|
||||
|
||||
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
Poll::Ready(Ok(()))
|
||||
@@ -143,11 +147,7 @@ impl Server {
|
||||
#[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
|
||||
pub fn tls_config(self, tls_config: ServerTlsConfig) -> Result<Self, Error> {
|
||||
Ok(Server {
|
||||
tls: Some(
|
||||
tls_config
|
||||
.tls_acceptor()
|
||||
.map_err(|e| Error::from_source(e))?,
|
||||
),
|
||||
tls: Some(tls_config.tls_acceptor().map_err(Error::from_source)?),
|
||||
..self
|
||||
})
|
||||
}
|
||||
@@ -320,7 +320,7 @@ impl Server {
|
||||
let init_connection_window_size = self.init_connection_window_size;
|
||||
let init_stream_window_size = self.init_stream_window_size;
|
||||
let max_concurrent_streams = self.max_concurrent_streams;
|
||||
let timeout = self.timeout.clone();
|
||||
let timeout = self.timeout;
|
||||
|
||||
let tcp = incoming::tcp_incoming(incoming, self);
|
||||
let incoming = accept::from_stream::<_, _, crate::Error>(tcp);
|
||||
@@ -538,6 +538,8 @@ where
|
||||
{
|
||||
type Response = Response<BoxBody>;
|
||||
type Error = crate::Error;
|
||||
|
||||
#[allow(clippy::type_complexity)]
|
||||
type Future = MapErr<Instrumented<S::Future>, fn(S::Error) -> crate::Error>;
|
||||
|
||||
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
@@ -578,8 +580,7 @@ where
|
||||
{
|
||||
type Response = BoxService;
|
||||
type Error = crate::Error;
|
||||
type Future =
|
||||
Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
|
||||
type Future = BoxFuture<Self::Response, Self::Error>;
|
||||
|
||||
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
Ok(()).into()
|
||||
@@ -593,7 +594,7 @@ where
|
||||
|
||||
let svc = self.inner.clone();
|
||||
let concurrency_limit = self.concurrency_limit;
|
||||
let timeout = self.timeout.clone();
|
||||
let timeout = self.timeout;
|
||||
let span = self.span.clone();
|
||||
|
||||
Box::pin(async move {
|
||||
|
||||
@@ -7,7 +7,7 @@ use std::fmt;
|
||||
/// Configures TLS settings for servers.
|
||||
#[cfg(feature = "tls")]
|
||||
#[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
|
||||
#[derive(Clone)]
|
||||
#[derive(Clone, Default)]
|
||||
pub struct ServerTlsConfig {
|
||||
identity: Option<Identity>,
|
||||
client_ca_root: Option<Certificate>,
|
||||
|
||||
@@ -35,8 +35,8 @@ where
|
||||
let set_uri = self.origin.clone().into_parts();
|
||||
|
||||
// Update the URI parts, setting hte scheme and authority
|
||||
uri.scheme = Some(set_uri.scheme.expect("expected scheme").clone());
|
||||
uri.authority = Some(set_uri.authority.expect("expected authority").clone());
|
||||
uri.scheme = Some(set_uri.scheme.expect("expected scheme"));
|
||||
uri.authority = Some(set_uri.authority.expect("expected authority"));
|
||||
|
||||
// Update the the request URI
|
||||
head.uri = http::Uri::from_parts(uri).expect("valid uri");
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
use super::super::BoxFuture;
|
||||
use super::{layer::ServiceBuilderExt, reconnect::Reconnect, AddOrigin, UserAgent};
|
||||
use crate::{body::BoxBody, transport::Endpoint};
|
||||
use http::Uri;
|
||||
@@ -6,8 +7,6 @@ use hyper::client::connect::Connection as HyperConnection;
|
||||
use hyper::client::service::Connect as HyperConnect;
|
||||
use std::{
|
||||
fmt,
|
||||
future::Future,
|
||||
pin::Pin,
|
||||
task::{Context, Poll},
|
||||
};
|
||||
use tokio::io::{AsyncRead, AsyncWrite};
|
||||
@@ -51,8 +50,6 @@ impl Connection {
|
||||
settings.http2_keep_alive_while_idle(val);
|
||||
}
|
||||
|
||||
let settings = settings.clone();
|
||||
|
||||
let stack = ServiceBuilder::new()
|
||||
.layer_fn(|s| AddOrigin::new(s, endpoint.uri.clone()))
|
||||
.layer_fn(|s| UserAgent::new(s, endpoint.user_agent.clone()))
|
||||
@@ -95,9 +92,7 @@ impl Connection {
|
||||
impl Service<Request> for Connection {
|
||||
type Response = Response;
|
||||
type Error = crate::Error;
|
||||
|
||||
type Future =
|
||||
Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
|
||||
type Future = BoxFuture<Self::Response, Self::Error>;
|
||||
|
||||
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
Service::poll_ready(&mut self.inner, cx).map_err(Into::into)
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
use super::super::BoxFuture;
|
||||
use super::io::BoxedIo;
|
||||
#[cfg(feature = "tls")]
|
||||
use super::tls::TlsConnector;
|
||||
use http::Uri;
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::task::{Context, Poll};
|
||||
use tower_make::MakeConnection;
|
||||
use tower_service::Service;
|
||||
@@ -48,9 +47,7 @@ where
|
||||
{
|
||||
type Response = BoxedIo;
|
||||
type Error = crate::Error;
|
||||
|
||||
type Future =
|
||||
Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
|
||||
type Future = BoxFuture<Self::Response, Self::Error>;
|
||||
|
||||
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
MakeConnection::poll_ready(&mut self.inner, cx).map_err(Into::into)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
use super::super::service;
|
||||
use super::super::{service, BoxFuture};
|
||||
use super::connection::Connection;
|
||||
use crate::transport::Endpoint;
|
||||
|
||||
@@ -12,12 +12,11 @@ use tokio::{stream::Stream, sync::mpsc::Receiver};
|
||||
|
||||
use tower::discover::{Change, Discover};
|
||||
|
||||
type DiscoverResult<K, S, E> = Result<Change<K, S>, E>;
|
||||
|
||||
pub(crate) struct DynamicServiceStream<K: Hash + Eq + Clone> {
|
||||
changes: Receiver<Change<K, Endpoint>>,
|
||||
connecting: Option<(
|
||||
K,
|
||||
Pin<Box<dyn Future<Output = Result<Connection, crate::Error>> + Send + 'static>>,
|
||||
)>,
|
||||
connecting: Option<(K, BoxFuture<Connection, crate::Error>)>,
|
||||
}
|
||||
|
||||
impl<K: Hash + Eq + Clone> DynamicServiceStream<K> {
|
||||
@@ -37,7 +36,7 @@ impl<K: Hash + Eq + Clone> Discover for DynamicServiceStream<K> {
|
||||
fn poll_discover(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
) -> Poll<Result<Change<Self::Key, Self::Service>, Self::Error>> {
|
||||
) -> Poll<DiscoverResult<Self::Key, Self::Service, Self::Error>> {
|
||||
loop {
|
||||
if let Some((key, connecting)) = &mut self.connecting {
|
||||
let svc = futures_core::ready!(Pin::new(connecting).poll(cx))?;
|
||||
|
||||
@@ -94,7 +94,7 @@ where
|
||||
if !(self.has_been_connected || self.is_lazy) {
|
||||
return Poll::Ready(Err(e.into()));
|
||||
} else {
|
||||
self.error = Some(e.into());
|
||||
self.error = Some(e);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -94,6 +94,8 @@ where
|
||||
{
|
||||
type Response = A::Response;
|
||||
type Error = crate::Error;
|
||||
|
||||
#[allow(clippy::type_complexity)]
|
||||
type Future = Either<
|
||||
MapErr<A::Future, fn(A::Error) -> crate::Error>,
|
||||
MapErr<B::Future, fn(B::Error) -> crate::Error>,
|
||||
|
||||
@@ -133,10 +133,9 @@ impl TlsAcceptor {
|
||||
let mut cert = std::io::Cursor::new(&cert.pem[..]);
|
||||
|
||||
let mut client_root_cert_store = tokio_rustls::rustls::RootCertStore::empty();
|
||||
match client_root_cert_store.add_pem_file(&mut cert) {
|
||||
Err(_) => return Err(Box::new(TlsError::CertificateParseError)),
|
||||
_ => (),
|
||||
};
|
||||
if client_root_cert_store.add_pem_file(&mut cert).is_err() {
|
||||
return Err(Box::new(TlsError::CertificateParseError));
|
||||
}
|
||||
|
||||
let client_auth =
|
||||
tokio_rustls::rustls::AllowAnyAuthenticatedClient::new(client_root_cert_store);
|
||||
@@ -204,7 +203,7 @@ mod rustls_keys {
|
||||
) -> Result<PrivateKey, crate::Error> {
|
||||
// First attempt to load the private key assuming it is PKCS8-encoded
|
||||
if let Ok(mut keys) = pemfile::pkcs8_private_keys(&mut cursor) {
|
||||
if keys.len() > 0 {
|
||||
if !keys.is_empty() {
|
||||
return Ok(keys.remove(0));
|
||||
}
|
||||
}
|
||||
@@ -212,7 +211,7 @@ mod rustls_keys {
|
||||
// If it not, try loading the private key as an RSA key
|
||||
cursor.set_position(0);
|
||||
if let Ok(mut keys) = pemfile::rsa_private_keys(&mut cursor) {
|
||||
if keys.len() > 0 {
|
||||
if !keys.is_empty() {
|
||||
return Ok(keys.remove(0));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,7 +20,7 @@ impl<T> UserAgent<T> {
|
||||
buf.extend(TONIC_USER_AGENT.as_bytes());
|
||||
HeaderValue::from_bytes(&buf).expect("user-agent should be valid")
|
||||
})
|
||||
.unwrap_or(HeaderValue::from_static(TONIC_USER_AGENT));
|
||||
.unwrap_or_else(|| HeaderValue::from_static(TONIC_USER_AGENT));
|
||||
|
||||
Self { inner, user_agent }
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user