Add more config options for channels

This commit is contained in:
Lucio Franco
2019-09-18 18:11:06 -04:00
parent 9954b9e10c
commit 2ed69f799b
4 changed files with 20 additions and 8 deletions
+9
View File
@@ -8,6 +8,7 @@ pub struct Endpoint {
pub(super) uri: Uri, pub(super) uri: Uri,
pub(super) timeout: Option<Duration>, pub(super) timeout: Option<Duration>,
pub(super) concurrency_limit: Option<usize>, pub(super) concurrency_limit: Option<usize>,
pub(super) rate_limit: Option<(u64, Duration)>,
pub(super) cert: Option<Cert>, pub(super) cert: Option<Cert>,
} }
@@ -32,6 +33,11 @@ impl Endpoint {
self self
} }
pub fn rate_limit(&mut self, limit: u64, duration: Duration) -> &mut Self {
self.rate_limit = Some((limit, duration));
self
}
pub fn tls_cert(&mut self, ca: Vec<u8>, domain: Option<String>) -> &mut Self { pub fn tls_cert(&mut self, ca: Vec<u8>, domain: Option<String>) -> &mut Self {
self.cert = Some(Cert { self.cert = Some(Cert {
ca, ca,
@@ -41,6 +47,8 @@ impl Endpoint {
self self
} }
// pub fn metadata_interceptor(f: impl Fn(MetadataMap) ->)
pub fn channel(&self) -> Result<Channel, super::Error> { pub fn channel(&self) -> Result<Channel, super::Error> {
Channel::builder().connect(self.clone()) Channel::builder().connect(self.clone())
} }
@@ -51,6 +59,7 @@ impl From<Uri> for Endpoint {
Self { Self {
uri, uri,
concurrency_limit: None, concurrency_limit: None,
rate_limit: None,
timeout: None, timeout: None,
cert: None, cert: None,
} }
@@ -9,18 +9,20 @@ use std::{
task::{Context, Poll}, task::{Context, Poll},
}; };
use tower::{ use tower::{
layer::Layer, limit::concurrency::ConcurrencyLimitLayer, timeout::TimeoutLayer, layer::Layer,
util::BoxService, ServiceBuilder, limit::{concurrency::ConcurrencyLimitLayer, rate::RateLimitLayer},
timeout::TimeoutLayer,
util::BoxService,
ServiceBuilder,
}; };
use tower_load::Load; use tower_load::Load;
use tower_reconnect::Reconnect; use tower_reconnect::Reconnect;
use tower_service::Service; use tower_service::Service;
type Request = http::Request<BoxBody>; pub(crate) type Request = http::Request<BoxBody>;
type Response = http::Response<hyper::Body>; pub(crate) type Response = http::Response<hyper::Body>;
pub struct Connection { pub struct Connection {
// inner: AddOrigin<Reconnect<HyperConnect<Connector, BoxBody, Uri>, Uri>>,
inner: BoxService<Request, Response, crate::Error>, inner: BoxService<Request, Response, crate::Error>,
} }
@@ -38,6 +40,7 @@ impl Connection {
.concurrency_limit .concurrency_limit
.map(|l| ConcurrencyLimitLayer::new(l)), .map(|l| ConcurrencyLimitLayer::new(l)),
) )
.optional_layer(endpoint.rate_limit.map(|(l, d)| RateLimitLayer::new(l, d)))
.into_inner(); .into_inner();
let conn = Reconnect::new(HyperConnect::new(connector, settings), endpoint.uri.clone()); let conn = Reconnect::new(HyperConnect::new(connector, settings), endpoint.uri.clone());
+1 -1
View File
@@ -1,4 +1,4 @@
use super::connect::Connection; use super::connection::Connection;
use crate::transport::Endpoint; use crate::transport::Endpoint;
use std::collections::VecDeque; use std::collections::VecDeque;
use std::pin::Pin; use std::pin::Pin;
+2 -2
View File
@@ -1,6 +1,6 @@
mod add_origin; mod add_origin;
mod boxed; mod boxed;
mod connect; mod connection;
mod connector; mod connector;
mod discover; mod discover;
mod io; mod io;
@@ -8,7 +8,7 @@ mod layer;
pub(crate) use self::add_origin::AddOrigin; pub(crate) use self::add_origin::AddOrigin;
pub(crate) use self::boxed::BoxService; pub(crate) use self::boxed::BoxService;
pub(crate) use self::connect::Connection; pub(crate) use self::connection::Connection;
pub(crate) use self::connector::Connector; pub(crate) use self::connector::Connector;
pub(crate) use self::discover::ServiceList; pub(crate) use self::discover::ServiceList;
pub(crate) use self::io::BoxedIo; pub(crate) use self::io::BoxedIo;