Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(client): provide tower::Service support for clients (#1915)
- Loading branch information
1 parent
eef407d
commit eee2a72
Showing
6 changed files
with
184 additions
and
9 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
|
|
||
| use hyper::client::service::{Connect, Service, MakeService}; | ||
| use hyper::client::conn::Builder; | ||
| use hyper::client::connect::HttpConnector; | ||
| use hyper::{Body, Request}; | ||
|
|
||
| #[tokio::main] | ||
| async fn main() -> Result<(), Box<dyn std::error::Error>> { | ||
| pretty_env_logger::init(); | ||
|
|
||
| let mut mk_svc = Connect::new(HttpConnector::new(), Builder::new()); | ||
|
|
||
| let uri = "http://127.0.0.1:8080".parse::<http::Uri>()?; | ||
|
|
||
|
|
||
| let mut svc = mk_svc.make_service(uri.clone()).await?; | ||
|
|
||
| let body = Body::empty(); | ||
|
|
||
| let req = Request::get(uri).body(body)?; | ||
| let res = svc.call(req).await?; | ||
|
|
||
| println!("RESPONSE={:?}", res); | ||
|
|
||
| Ok(()) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| //! Utilities used to interact with the Tower ecosystem. | ||
| //! | ||
| //! This module provides exports of `Service`, `MakeService` and `Connect` which | ||
| //! all provide hook-ins into the Tower ecosystem. | ||
|
|
||
| use super::conn::{SendRequest, Builder}; | ||
| use std::marker::PhantomData; | ||
| use crate::{common::{Poll, task, Pin}, body::Payload}; | ||
| use std::future::Future; | ||
| use std::error::Error as StdError; | ||
| use tower_make::MakeConnection; | ||
|
|
||
| pub use tower_service::Service; | ||
| pub use tower_make::MakeService; | ||
|
|
||
| /// Creates a connection via `SendRequest`. | ||
| /// | ||
| /// This accepts a `hyper::client::conn::Builder` and provides | ||
| /// a `MakeService` implementation to create connections from some | ||
| /// target `T`. | ||
| #[derive(Debug)] | ||
| pub struct Connect<C, B, T> { | ||
| inner: C, | ||
| builder: Builder, | ||
| _pd: PhantomData<fn(T, B)> | ||
| } | ||
|
|
||
| impl<C, B, T> Connect<C, B, T> { | ||
| /// Create a new `Connect` with some inner connector `C` and a connection | ||
| /// builder. | ||
| pub fn new(inner: C, builder: Builder) -> Self { | ||
| Self { | ||
| inner, | ||
| builder, | ||
| _pd: PhantomData | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl<C, B, T> Service<T> for Connect<C, B, T> | ||
| where | ||
| C: MakeConnection<T>, | ||
| C::Connection: Unpin + Send + 'static, | ||
| C::Future: Send + 'static, | ||
| C::Error: Into<Box<dyn StdError + Send + Sync>> + Send, | ||
| B: Payload + Unpin + 'static, | ||
| B::Data: Unpin, | ||
| { | ||
| type Response = SendRequest<B>; | ||
| type Error = crate::Error; | ||
| type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>; | ||
|
|
||
| fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> { | ||
| self.inner.poll_ready(cx).map_err(|e| crate::Error::new(crate::error::Kind::Connect).with(e.into())) | ||
| } | ||
|
|
||
| fn call(&mut self, req: T) -> Self::Future { | ||
| let builder = self.builder.clone(); | ||
| let io = self.inner.make_connection(req); | ||
|
|
||
| let fut = async move { | ||
| match io.await { | ||
| Ok(io) => { | ||
| match builder.handshake(io).await { | ||
| Ok((sr, conn)) => { | ||
| builder.exec.execute(async move { | ||
| if let Err(e) = conn.await { | ||
| debug!("connection error: {:?}", e); | ||
| } | ||
| })?; | ||
| Ok(sr) | ||
| }, | ||
| Err(e) => Err(e) | ||
| } | ||
| }, | ||
| Err(e) => { | ||
| let err = crate::Error::new(crate::error::Kind::Connect).with(e.into()); | ||
| Err(err) | ||
| } | ||
| } | ||
| }; | ||
|
|
||
| Box::pin(fut) | ||
| } | ||
| } |