hyper/client/conn/
http2.rs

1//! HTTP/2 client connections.
2
3use std::error::Error;
4use std::fmt;
5use std::future::Future;
6use std::marker::PhantomData;
7use std::pin::Pin;
8use std::sync::Arc;
9use std::task::{Context, Poll};
10use std::time::Duration;
11
12use crate::rt::{Read, Write};
13use futures_core::ready;
14use http::{Request, Response};
15
16use super::super::dispatch::{self, TrySendError};
17use crate::body::{Body, Incoming as IncomingBody};
18use crate::common::time::Time;
19use crate::proto;
20use crate::rt::bounds::Http2ClientConnExec;
21use crate::rt::Timer;
22
23/// The sender side of an established connection.
24pub struct SendRequest<B> {
25    dispatch: dispatch::UnboundedSender<Request<B>, Response<IncomingBody>>,
26}
27
28impl<B> Clone for SendRequest<B> {
29    fn clone(&self) -> SendRequest<B> {
30        SendRequest {
31            dispatch: self.dispatch.clone(),
32        }
33    }
34}
35
36/// A future that processes all HTTP state for the IO object.
37///
38/// In most cases, this should just be spawned into an executor, so that it
39/// can process incoming and outgoing messages, notice hangups, and the like.
40///
41/// Instances of this type are typically created via the [`handshake`] function.
42///
43/// # Drop behavior
44///
45/// Dropping the `Connection` will close the underlying IO resource.
46/// Any in-flight requests that have not received a response will be
47/// interrupted. If graceful shutdown is desired, poll the connection
48/// until it completes instead of dropping.
49#[must_use = "futures do nothing unless polled"]
50pub struct Connection<T, B, E>
51where
52    T: Read + Write + Unpin,
53    B: Body + 'static,
54    E: Http2ClientConnExec<B, T> + Unpin,
55    B::Error: Into<Box<dyn Error + Send + Sync>>,
56{
57    inner: (PhantomData<T>, proto::h2::ClientTask<B, E, T>),
58}
59
60/// A builder to configure an HTTP connection.
61///
62/// After setting options, the builder is used to create a handshake future.
63///
64/// **Note**: The default values of options are *not considered stable*. They
65/// are subject to change at any time.
66#[derive(Clone, Debug)]
67pub struct Builder<Ex> {
68    pub(super) exec: Ex,
69    pub(super) timer: Time,
70    h2_builder: proto::h2::client::Config,
71}
72
73/// Returns a handshake future over some IO.
74///
75/// This is a shortcut for `Builder::new(exec).handshake(io)`.
76/// See [`client::conn`](crate::client::conn) for more.
77pub async fn handshake<E, T, B>(
78    exec: E,
79    io: T,
80) -> crate::Result<(SendRequest<B>, Connection<T, B, E>)>
81where
82    T: Read + Write + Unpin,
83    B: Body + 'static,
84    B::Data: Send,
85    B::Error: Into<Box<dyn Error + Send + Sync>>,
86    E: Http2ClientConnExec<B, T> + Unpin + Clone,
87{
88    Builder::new(exec).handshake(io).await
89}
90
91// ===== impl SendRequest
92
93impl<B> SendRequest<B> {
94    /// Polls to determine whether this sender can be used yet for a request.
95    ///
96    /// If the associated connection is closed, this returns an Error.
97    pub fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<crate::Result<()>> {
98        if self.is_closed() {
99            Poll::Ready(Err(crate::Error::new_closed()))
100        } else {
101            Poll::Ready(Ok(()))
102        }
103    }
104
105    /// Waits until the dispatcher is ready.
106    ///
107    /// If the associated connection is closed, this returns an Error.
108    pub async fn ready(&mut self) -> crate::Result<()> {
109        crate::common::future::poll_fn(|cx| self.poll_ready(cx)).await
110    }
111
112    /// Checks if the connection is currently ready to send a request.
113    ///
114    /// # Note
115    ///
116    /// This is mostly a hint. Due to inherent latency of networks, it is
117    /// possible that even after checking this is ready, sending a request
118    /// may still fail because the connection was closed in the meantime.
119    pub fn is_ready(&self) -> bool {
120        self.dispatch.is_ready()
121    }
122
123    /// Checks if the connection side has been closed.
124    pub fn is_closed(&self) -> bool {
125        self.dispatch.is_closed()
126    }
127}
128
129impl<B> SendRequest<B>
130where
131    B: Body + 'static,
132{
133    /// Sends a `Request` on the associated connection.
134    ///
135    /// Returns a future that if successful, yields the `Response`.
136    ///
137    /// `req` must have a `Host` header.
138    ///
139    /// Absolute-form `Uri`s are not required. If received, they will be serialized
140    /// as-is.
141    ///
142    /// # Cancel safety
143    ///
144    /// Dropping the returned future is the supported way to cancel an
145    /// in-flight HTTP/2 request. The stream is reset with `RST_STREAM`
146    /// (`CANCEL` error code); the shared connection remains usable for
147    /// other in-flight and future requests. The peer is notified
148    /// immediately rather than continuing to send a response body that
149    /// would be discarded.
150    pub fn send_request(
151        &mut self,
152        req: Request<B>,
153    ) -> impl Future<Output = crate::Result<Response<IncomingBody>>> {
154        let sent = self.dispatch.send(req);
155
156        async move {
157            match sent {
158                Ok(rx) => match rx.await {
159                    Ok(Ok(resp)) => Ok(resp),
160                    Ok(Err(err)) => Err(err),
161                    // this is definite bug if it happens, but it shouldn't happen!
162                    Err(_canceled) => panic!("dispatch dropped without returning error"),
163                },
164                Err(_req) => {
165                    debug!("connection was not ready");
166
167                    Err(crate::Error::new_canceled().with("connection was not ready"))
168                }
169            }
170        }
171    }
172
173    /// Sends a `Request` on the associated connection.
174    ///
175    /// Returns a future that if successful, yields the `Response`.
176    ///
177    /// # Error
178    ///
179    /// If there was an error before trying to serialize the request to the
180    /// connection, the message will be returned as part of this error.
181    pub fn try_send_request(
182        &mut self,
183        req: Request<B>,
184    ) -> impl Future<Output = Result<Response<IncomingBody>, TrySendError<Request<B>>>> {
185        let sent = self.dispatch.try_send(req);
186        async move {
187            match sent {
188                Ok(rx) => match rx.await {
189                    Ok(Ok(res)) => Ok(res),
190                    Ok(Err(err)) => Err(err),
191                    // this is definite bug if it happens, but it shouldn't happen!
192                    Err(_) => panic!("dispatch dropped without returning error"),
193                },
194                Err(req) => {
195                    debug!("connection was not ready");
196                    let error = crate::Error::new_canceled().with("connection was not ready");
197                    Err(TrySendError {
198                        error,
199                        message: Some(req),
200                    })
201                }
202            }
203        }
204    }
205}
206
207impl<B> fmt::Debug for SendRequest<B> {
208    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
209        f.debug_struct("SendRequest").finish()
210    }
211}
212
213// ===== impl Connection
214
215impl<T, B, E> Connection<T, B, E>
216where
217    T: Read + Write + Unpin + 'static,
218    B: Body + Unpin + 'static,
219    B::Data: Send,
220    B::Error: Into<Box<dyn Error + Send + Sync>>,
221    E: Http2ClientConnExec<B, T> + Unpin,
222{
223    /// Returns whether the [extended CONNECT protocol][1] is enabled or not.
224    ///
225    /// This setting is configured by the server peer by sending the
226    /// [`SETTINGS_ENABLE_CONNECT_PROTOCOL` parameter][2] in a `SETTINGS` frame.
227    /// This method returns the currently acknowledged value received from the
228    /// remote.
229    ///
230    /// [1]: https://datatracker.ietf.org/doc/html/rfc8441#section-4
231    /// [2]: https://datatracker.ietf.org/doc/html/rfc8441#section-3
232    pub fn is_extended_connect_protocol_enabled(&self) -> bool {
233        self.inner.1.is_extended_connect_protocol_enabled()
234    }
235
236    /// Returns the current maximum send stream count.
237    ///
238    /// This setting is configured in a [`SETTINGS_MAX_CONCURRENT_STREAMS` parameter][1] in a `SETTINGS` frame,
239    /// and may change throughout the connection lifetime.
240    ///
241    /// [1]: https://datatracker.ietf.org/doc/html/rfc7540#section-5.1.2
242    pub fn current_max_send_streams(&self) -> usize {
243        self.inner.1.current_max_send_streams()
244    }
245
246    /// Returns the current maximum receive stream count.
247    ///
248    /// This setting is configured in a [`SETTINGS_MAX_CONCURRENT_STREAMS` parameter][1] in a `SETTINGS` frame,
249    /// and may change throughout the connection lifetime.
250    ///
251    /// [1]: https://datatracker.ietf.org/doc/html/rfc7540#section-5.1.2
252    pub fn current_max_recv_streams(&self) -> usize {
253        self.inner.1.current_max_recv_streams()
254    }
255}
256
257impl<T, B, E> fmt::Debug for Connection<T, B, E>
258where
259    T: Read + Write + fmt::Debug + 'static + Unpin,
260    B: Body + 'static,
261    E: Http2ClientConnExec<B, T> + Unpin,
262    B::Error: Into<Box<dyn Error + Send + Sync>>,
263{
264    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
265        f.debug_struct("Connection").finish()
266    }
267}
268
269impl<T, B, E> Future for Connection<T, B, E>
270where
271    T: Read + Write + Unpin + 'static,
272    B: Body + 'static + Unpin,
273    B::Data: Send,
274    E: Unpin,
275    B::Error: Into<Box<dyn Error + Send + Sync>>,
276    E: Http2ClientConnExec<B, T> + Unpin,
277{
278    type Output = crate::Result<()>;
279
280    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
281        match ready!(Pin::new(&mut self.inner.1).poll(cx))? {
282            proto::Dispatched::Shutdown => Poll::Ready(Ok(())),
283            #[cfg(feature = "http1")]
284            proto::Dispatched::Upgrade(_pending) => unreachable!("http2 cannot upgrade"),
285        }
286    }
287}
288
289// ===== impl Builder
290
291impl<Ex> Builder<Ex>
292where
293    Ex: Clone,
294{
295    /// Creates a new connection builder.
296    #[inline]
297    pub fn new(exec: Ex) -> Builder<Ex> {
298        Builder {
299            exec,
300            timer: Time::Empty,
301            h2_builder: Default::default(),
302        }
303    }
304
305    /// Provide a timer to execute background HTTP2 tasks.
306    pub fn timer<M>(&mut self, timer: M) -> &mut Builder<Ex>
307    where
308        M: Timer + Send + Sync + 'static,
309    {
310        self.timer = Time::Timer(Arc::new(timer));
311        self
312    }
313
314    /// Sets the [`SETTINGS_INITIAL_WINDOW_SIZE`][spec] option for HTTP2
315    /// stream-level flow control.
316    ///
317    /// Passing `None` will do nothing.
318    ///
319    /// If not set, hyper will use a default.
320    ///
321    /// [spec]: https://httpwg.org/specs/rfc9113.html#SETTINGS_INITIAL_WINDOW_SIZE
322    pub fn initial_stream_window_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
323        if let Some(sz) = sz.into() {
324            self.h2_builder.adaptive_window = false;
325            self.h2_builder.initial_stream_window_size = sz;
326        }
327        self
328    }
329
330    /// Sets the max connection-level flow control for HTTP2.
331    ///
332    /// Passing `None` will do nothing.
333    ///
334    /// If not set, hyper will use a default.
335    pub fn initial_connection_window_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
336        if let Some(sz) = sz.into() {
337            self.h2_builder.adaptive_window = false;
338            self.h2_builder.initial_conn_window_size = sz;
339        }
340        self
341    }
342
343    /// Sets the initial maximum of locally initiated (send) streams.
344    ///
345    /// This value will be overwritten by the value included in the initial
346    /// SETTINGS frame received from the peer as part of a [connection preface].
347    ///
348    /// Passing `None` will do nothing.
349    ///
350    /// If not set, hyper will use a default.
351    ///
352    /// [connection preface]: https://httpwg.org/specs/rfc9113.html#preface
353    pub fn initial_max_send_streams(&mut self, initial: impl Into<Option<usize>>) -> &mut Self {
354        if let Some(initial) = initial.into() {
355            self.h2_builder.initial_max_send_streams = initial;
356        }
357        self
358    }
359
360    /// Sets whether to use an adaptive flow control.
361    ///
362    /// Enabling this will override the limits set in
363    /// `initial_stream_window_size` and
364    /// `initial_connection_window_size`.
365    pub fn adaptive_window(&mut self, enabled: bool) -> &mut Self {
366        use proto::h2::SPEC_WINDOW_SIZE;
367
368        self.h2_builder.adaptive_window = enabled;
369        if enabled {
370            self.h2_builder.initial_conn_window_size = SPEC_WINDOW_SIZE;
371            self.h2_builder.initial_stream_window_size = SPEC_WINDOW_SIZE;
372        }
373        self
374    }
375
376    /// Sets the maximum frame size to use for HTTP2.
377    ///
378    /// Default is currently 16KB, but can change.
379    pub fn max_frame_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
380        self.h2_builder.max_frame_size = sz.into();
381        self
382    }
383
384    /// Sets the max size of received header frames.
385    ///
386    /// Default is currently 16KB, but can change.
387    pub fn max_header_list_size(&mut self, max: u32) -> &mut Self {
388        self.h2_builder.max_header_list_size = max;
389        self
390    }
391
392    /// Sets the header table size.
393    ///
394    /// This setting informs the peer of the maximum size of the header compression
395    /// table used to encode header blocks, in octets. The encoder may select any value
396    /// equal to or less than the header table size specified by the sender.
397    ///
398    /// The default value of crate `h2` is 4,096.
399    pub fn header_table_size(&mut self, size: impl Into<Option<u32>>) -> &mut Self {
400        self.h2_builder.header_table_size = size.into();
401        self
402    }
403
404    /// Sets the maximum number of concurrent streams.
405    ///
406    /// The maximum concurrent streams setting only controls the maximum number
407    /// of streams that can be initiated by the remote peer. In other words,
408    /// when this setting is set to 100, this does not limit the number of
409    /// concurrent streams that can be created by the caller.
410    ///
411    /// It is recommended that this value be no smaller than 100, so as to not
412    /// unnecessarily limit parallelism. However, any value is legal, including
413    /// 0. If `max` is set to 0, then the remote will not be permitted to
414    /// initiate streams.
415    ///
416    /// Note that streams in the reserved state, i.e., push promises that have
417    /// been reserved but the stream has not started, do not count against this
418    /// setting.
419    ///
420    /// Also note that if the remote *does* exceed the value set here, it is not
421    /// a protocol level error. Instead, the `h2` library will immediately reset
422    /// the stream.
423    ///
424    /// See [Section 5.1.2] in the HTTP/2 spec for more details.
425    ///
426    /// [Section 5.1.2]: https://httpwg.org/specs/rfc7540.html#rfc.section.5.1.2
427    pub fn max_concurrent_streams(&mut self, max: impl Into<Option<u32>>) -> &mut Self {
428        self.h2_builder.max_concurrent_streams = max.into();
429        self
430    }
431
432    /// Sets an interval for HTTP2 Ping frames should be sent to keep a
433    /// connection alive.
434    ///
435    /// Pass `None` to disable HTTP2 keep-alive.
436    ///
437    /// Default is currently disabled.
438    pub fn keep_alive_interval(&mut self, interval: impl Into<Option<Duration>>) -> &mut Self {
439        self.h2_builder.keep_alive_interval = interval.into();
440        self
441    }
442
443    /// Sets a timeout for receiving an acknowledgement of the keep-alive ping.
444    ///
445    /// If the ping is not acknowledged within the timeout, the connection will
446    /// be closed. Does nothing if `keep_alive_interval` is disabled.
447    ///
448    /// Default is 20 seconds.
449    pub fn keep_alive_timeout(&mut self, timeout: Duration) -> &mut Self {
450        self.h2_builder.keep_alive_timeout = timeout;
451        self
452    }
453
454    /// Sets whether HTTP2 keep-alive should apply while the connection is idle.
455    ///
456    /// If disabled, keep-alive pings are only sent while there are open
457    /// request/responses streams. If enabled, pings are also sent when no
458    /// streams are active. Does nothing if `keep_alive_interval` is
459    /// disabled.
460    ///
461    /// Default is `false`.
462    pub fn keep_alive_while_idle(&mut self, enabled: bool) -> &mut Self {
463        self.h2_builder.keep_alive_while_idle = enabled;
464        self
465    }
466
467    /// Sets the maximum number of HTTP2 concurrent locally reset streams.
468    ///
469    /// See the documentation of [`h2::client::Builder::max_concurrent_reset_streams`] for more
470    /// details.
471    ///
472    /// The default value is determined by the `h2` crate.
473    ///
474    /// [`h2::client::Builder::max_concurrent_reset_streams`]: https://docs.rs/h2/client/struct.Builder.html#method.max_concurrent_reset_streams
475    pub fn max_concurrent_reset_streams(&mut self, max: usize) -> &mut Self {
476        self.h2_builder.max_concurrent_reset_streams = Some(max);
477        self
478    }
479
480    /// Set the maximum write buffer size for each HTTP/2 stream.
481    ///
482    /// Default is currently 1MB, but may change.
483    ///
484    /// # Panics
485    ///
486    /// The value must be no larger than `u32::MAX`.
487    pub fn max_send_buf_size(&mut self, max: usize) -> &mut Self {
488        assert!(max <= u32::MAX as usize);
489        self.h2_builder.max_send_buffer_size = max;
490        self
491    }
492
493    /// Configures the maximum number of pending reset streams allowed before a GOAWAY will be sent.
494    ///
495    /// This will default to the default value set by the [`h2` crate](https://crates.io/crates/h2).
496    /// As of v0.4.0, it is 20.
497    ///
498    /// See <https://github.com/hyperium/hyper/issues/2877> for more information.
499    pub fn max_pending_accept_reset_streams(&mut self, max: impl Into<Option<usize>>) -> &mut Self {
500        self.h2_builder.max_pending_accept_reset_streams = max.into();
501        self
502    }
503
504    /// Configures the maximum number of local resets due to protocol errors made by the remote end.
505    ///
506    /// See the documentation of [`h2::client::Builder::max_local_error_reset_streams`] for more
507    /// details.
508    ///
509    /// The default value is 1024.
510    pub fn max_local_error_reset_streams(&mut self, max: impl Into<Option<usize>>) -> &mut Self {
511        self.h2_builder.max_local_error_reset_streams = max.into();
512        self
513    }
514
515    /// Sets the duration to remember locally reset streams.
516    ///
517    /// When a stream is explicitly reset by either the client or the server,
518    /// the HTTP/2 specification requires that any further frames received for
519    /// that stream must be ignored for "some time".
520    ///
521    /// In order to satisfy the specification, internal state must be maintained
522    /// to implement the behavior. This state grows linearly with the number of
523    /// streams that are locally reset.
524    ///
525    /// The `reset_stream_duration` setting configures the max amount of time
526    /// this state will be maintained in memory. Once the duration elapses, the
527    /// stream state is purged from memory.
528    ///
529    /// Once the stream has been fully purged from memory, any additional frames
530    /// received for that stream will result in a connection level protocol
531    /// error, forcing the connection to terminate.
532    ///
533    /// The default value is determined by the `h2` crate, and is currently
534    /// 1 second.
535    ///
536    /// See the documentation of [`h2::client::Builder::reset_stream_duration`] for more
537    /// details.
538    ///
539    /// [`h2::client::Builder::reset_stream_duration`]: https://docs.rs/h2/client/struct.Builder.html#method.reset_stream_duration
540    pub fn reset_stream_duration(&mut self, dur: Duration) -> &mut Self {
541        self.h2_builder.reset_stream_duration = Some(dur);
542        self
543    }
544
545    /// Constructs a connection with the configured options and IO.
546    /// See [`client::conn`](crate::client::conn) for more.
547    ///
548    /// Note, if [`Connection`] is not `await`-ed, [`SendRequest`] will
549    /// do nothing.
550    pub fn handshake<T, B>(
551        &self,
552        io: T,
553    ) -> impl Future<Output = crate::Result<(SendRequest<B>, Connection<T, B, Ex>)>>
554    where
555        T: Read + Write + Unpin,
556        B: Body + 'static,
557        B::Data: Send,
558        B::Error: Into<Box<dyn Error + Send + Sync>>,
559        Ex: Http2ClientConnExec<B, T> + Unpin,
560    {
561        let opts = self.clone();
562
563        async move {
564            trace!("client handshake HTTP/2");
565
566            let (tx, rx) = dispatch::channel();
567            let h2 = proto::h2::client::handshake(io, rx, &opts.h2_builder, opts.exec, opts.timer)
568                .await?;
569            Ok((
570                SendRequest {
571                    dispatch: tx.unbound(),
572                },
573                Connection {
574                    inner: (PhantomData, h2),
575                },
576            ))
577        }
578    }
579}
580
581#[cfg(test)]
582mod tests {
583
584    #[tokio::test]
585    #[ignore] // only compilation is checked
586    async fn send_sync_executor_of_non_send_futures() {
587        #[derive(Clone)]
588        struct LocalTokioExecutor;
589
590        impl<F> crate::rt::Executor<F> for LocalTokioExecutor
591        where
592            F: std::future::Future + 'static, // not requiring `Send`
593        {
594            fn execute(&self, fut: F) {
595                // This will spawn into the currently running `LocalSet`.
596                tokio::task::spawn_local(fut);
597            }
598        }
599
600        #[allow(unused)]
601        async fn run(io: impl crate::rt::Read + crate::rt::Write + Unpin + 'static) {
602            let (_sender, conn) = crate::client::conn::http2::handshake::<
603                _,
604                _,
605                http_body_util::Empty<bytes::Bytes>,
606            >(LocalTokioExecutor, io)
607            .await
608            .unwrap();
609
610            tokio::task::spawn_local(async move {
611                conn.await.unwrap();
612            });
613        }
614    }
615
616    #[tokio::test]
617    #[ignore] // only compilation is checked
618    async fn not_send_not_sync_executor_of_not_send_futures() {
619        #[derive(Clone)]
620        struct LocalTokioExecutor {
621            _x: std::marker::PhantomData<std::rc::Rc<()>>,
622        }
623
624        impl<F> crate::rt::Executor<F> for LocalTokioExecutor
625        where
626            F: std::future::Future + 'static, // not requiring `Send`
627        {
628            fn execute(&self, fut: F) {
629                // This will spawn into the currently running `LocalSet`.
630                tokio::task::spawn_local(fut);
631            }
632        }
633
634        #[allow(unused)]
635        async fn run(io: impl crate::rt::Read + crate::rt::Write + Unpin + 'static) {
636            let (_sender, conn) =
637                crate::client::conn::http2::handshake::<_, _, http_body_util::Empty<bytes::Bytes>>(
638                    LocalTokioExecutor {
639                        _x: Default::default(),
640                    },
641                    io,
642                )
643                .await
644                .unwrap();
645
646            tokio::task::spawn_local(async move {
647                conn.await.unwrap();
648            });
649        }
650    }
651
652    #[tokio::test]
653    #[ignore] // only compilation is checked
654    async fn send_not_sync_executor_of_not_send_futures() {
655        #[derive(Clone)]
656        struct LocalTokioExecutor {
657            _x: std::marker::PhantomData<std::cell::Cell<()>>,
658        }
659
660        impl<F> crate::rt::Executor<F> for LocalTokioExecutor
661        where
662            F: std::future::Future + 'static, // not requiring `Send`
663        {
664            fn execute(&self, fut: F) {
665                // This will spawn into the currently running `LocalSet`.
666                tokio::task::spawn_local(fut);
667            }
668        }
669
670        #[allow(unused)]
671        async fn run(io: impl crate::rt::Read + crate::rt::Write + Unpin + 'static) {
672            let (_sender, conn) =
673                crate::client::conn::http2::handshake::<_, _, http_body_util::Empty<bytes::Bytes>>(
674                    LocalTokioExecutor {
675                        _x: Default::default(),
676                    },
677                    io,
678                )
679                .await
680                .unwrap();
681
682            tokio::task::spawn_local(async move {
683                conn.await.unwrap();
684            });
685        }
686    }
687
688    #[tokio::test]
689    #[ignore] // only compilation is checked
690    async fn send_sync_executor_of_send_futures() {
691        #[derive(Clone)]
692        struct TokioExecutor;
693
694        impl<F> crate::rt::Executor<F> for TokioExecutor
695        where
696            F: std::future::Future + 'static + Send,
697            F::Output: Send + 'static,
698        {
699            fn execute(&self, fut: F) {
700                tokio::task::spawn(fut);
701            }
702        }
703
704        #[allow(unused)]
705        async fn run(io: impl crate::rt::Read + crate::rt::Write + Send + Unpin + 'static) {
706            let (_sender, conn) = crate::client::conn::http2::handshake::<
707                _,
708                _,
709                http_body_util::Empty<bytes::Bytes>,
710            >(TokioExecutor, io)
711            .await
712            .unwrap();
713
714            tokio::task::spawn(async move {
715                conn.await.unwrap();
716            });
717        }
718    }
719
720    #[tokio::test]
721    #[ignore] // only compilation is checked
722    async fn send_not_sync_executor_of_send_futures() {
723        #[derive(Clone)]
724        struct TokioExecutor {
725            // !Sync
726            _x: std::marker::PhantomData<std::cell::Cell<()>>,
727        }
728
729        impl<F> crate::rt::Executor<F> for TokioExecutor
730        where
731            F: std::future::Future + 'static + Send,
732            F::Output: Send + 'static,
733        {
734            fn execute(&self, fut: F) {
735                tokio::task::spawn(fut);
736            }
737        }
738
739        #[allow(unused)]
740        async fn run(io: impl crate::rt::Read + crate::rt::Write + Send + Unpin + 'static) {
741            let (_sender, conn) =
742                crate::client::conn::http2::handshake::<_, _, http_body_util::Empty<bytes::Bytes>>(
743                    TokioExecutor {
744                        _x: Default::default(),
745                    },
746                    io,
747                )
748                .await
749                .unwrap();
750
751            tokio::task::spawn_local(async move {
752                // can't use spawn here because when executor is !Send
753                conn.await.unwrap();
754            });
755        }
756    }
757}
hyper/client/conn/http2.rs

hyper/client/conn/
http2.rs
