Make client able to use non-Send executor #3184
There are no files selected for viewing
This file contains hidden or 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 hidden or 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,132 @@ | ||
| #![deny(warnings)] | ||
| #![warn(rust_2018_idioms)] | ||
| use std::env; | ||
| use std::marker::PhantomData; | ||
| use std::pin::Pin; | ||
| use std::task::{Context, Poll}; | ||
|
|
||
| use bytes::Bytes; | ||
| use http_body_util::BodyExt; | ||
| use hyper::body::{Body as HttpBody, Frame}; | ||
| use hyper::Error; | ||
| use hyper::Request; | ||
| use tokio::io::{self, AsyncWriteExt as _}; | ||
| use tokio::net::TcpStream; | ||
|
|
||
| struct Body { | ||
| // Our Body type is !Send and !Sync: | ||
| _marker: PhantomData<*const ()>, | ||
| data: Option<Bytes>, | ||
| } | ||
|
|
||
| impl From<String> for Body { | ||
| fn from(a: String) -> Self { | ||
| Body { | ||
| _marker: PhantomData, | ||
| data: Some(a.into()), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl HttpBody for Body { | ||
| type Data = Bytes; | ||
| type Error = Error; | ||
|
|
||
| fn poll_frame( | ||
| self: Pin<&mut Self>, | ||
| _: &mut Context<'_>, | ||
| ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> { | ||
| Poll::Ready(self.get_mut().data.take().map(|d| Ok(Frame::data(d)))) | ||
| } | ||
| } | ||
|
|
||
| fn main() -> Result<(), Box<dyn std::error::Error>> { | ||
| pretty_env_logger::init(); | ||
|
|
||
| // Configure a runtime that runs everything on the current thread | ||
| let rt = tokio::runtime::Builder::new_current_thread() | ||
| .enable_all() | ||
| .build() | ||
| .expect("build runtime"); | ||
|
|
||
| // Combine it with a `LocalSet, which means it can spawn !Send futures... | ||
| let local = tokio::task::LocalSet::new(); | ||
| local.block_on(&rt, init()) | ||
| } | ||
|
|
||
| async fn init() -> Result<(), Box<dyn std::error::Error>> { | ||
| // Some simple CLI args requirements... | ||
| let url = match env::args().nth(1) { | ||
| Some(url) => url, | ||
| None => { | ||
| println!("Usage: client <url>"); | ||
| return Ok(()); | ||
| } | ||
| }; | ||
|
|
||
| // HTTPS requires picking a TLS implementation, so give a better | ||
| // warning if the user tries to request an 'https' URL. | ||
| let url = url.parse::<hyper::Uri>().unwrap(); | ||
| if url.scheme_str() != Some("http") { | ||
| println!("This example only works with 'http' URLs."); | ||
| return Ok(()); | ||
| } | ||
|
|
||
| fetch_url(url).await | ||
| } | ||
|
|
||
| async fn fetch_url(url: hyper::Uri) -> Result<(), Box<dyn std::error::Error>> { | ||
| let host = url.host().expect("uri has no host"); | ||
| let port = url.port_u16().unwrap_or(80); | ||
| let addr = format!("{}:{}", host, port); | ||
| let stream = TcpStream::connect(addr).await?; | ||
|
|
||
| let (mut sender, conn) = hyper::client::conn::http2::handshake(LocalExec, stream).await?; | ||
| tokio::task::spawn_local(async move { | ||
| if let Err(err) = conn.await { | ||
| println!("Connection failed: {:?}", err); | ||
| } | ||
| }); | ||
|
|
||
| let authority = url.authority().unwrap().clone(); | ||
|
|
||
| let req = Request::builder() | ||
| .uri(url) | ||
| .header(hyper::header::HOST, authority.as_str()) | ||
| .body(Body::from("test".to_string()))?; | ||
|
|
||
| let mut res = sender.send_request(req).await?; | ||
|
|
||
| println!("Response: {}", res.status()); | ||
| println!("Headers: {:#?}\n", res.headers()); | ||
|
|
||
| // Stream the body, writing each chunk to stdout as we get it | ||
| // (instead of buffering and printing at the end). | ||
| while let Some(next) = res.frame().await { | ||
| let frame = next?; | ||
| if let Some(chunk) = frame.data_ref() { | ||
| io::stdout().write_all(&chunk).await?; | ||
| } | ||
| } | ||
|
|
||
| println!("\n\nDone!"); | ||
|
|
||
| Ok(()) | ||
| } | ||
|
|
||
| // NOTE: This part is only needed for HTTP/2. HTTP/1 doesn't need an executor. | ||
| // | ||
| // Since the Server needs to spawn some background tasks, we needed | ||
| // to configure an Executor that can spawn !Send futures... | ||
| #[derive(Clone, Copy, Debug)] | ||
| struct LocalExec; | ||
|
|
||
| impl<F> hyper::rt::Executor<F> for LocalExec | ||
| where | ||
| F: std::future::Future + 'static, // not requiring `Send` | ||
| { | ||
| fn execute(&self, fut: F) { | ||
| // This will spawn into the currently running `LocalSet`. | ||
| tokio::task::spawn_local(fut); | ||
| } | ||
| } |
This file contains hidden or 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
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.