alux_http_hyper/connections.rs
1//! Accepting at one address, for interpreters whose framework states a service rather than a loop.
2
3use core::cell::RefCell;
4use core::error::Error;
5use core::pin::pin;
6use core::time::Duration;
7use futures::{FutureExt, StreamExt};
8use hyper::body::{Body, Incoming};
9use hyper::service::Service;
10use hyper::{Request, Response};
11use hyper_util::rt::{TokioExecutor, TokioIo};
12use hyper_util::server::conn::auto::Builder;
13use std::rc::Rc;
14use tokio::net::TcpListener;
15use tokio::sync::watch;
16use tokio::task::JoinHandle;
17use tokio_stream::wrappers::TcpListenerStream;
18
19/// Carries an error returned while serving connections.
20pub type ConnectionsError = Box<dyn Error + Send + Sync>;
21
22/// The drain to serve with where a caller states none of its own.
23///
24/// The same bound the framework-owning interpretations state, so a request outliving a close is
25/// ended after the same wait wherever it was made.
26pub const DRAIN: Duration = Duration::from_secs(5);
27
28/// Accepts at one address and serves each connection with one service.
29///
30/// This is all a framework needs to be served when it exposes a service rather than an accept loop.
31/// The listener stays here, so closing drops it and frees the address immediately, while
32/// connections already accepted are given the drain to finish.
33pub struct HyperConnections {
34 accepting: JoinHandle<Result<(), ConnectionsError>>,
35 /// Tells every connection to finish what it is answering and then stop.
36 closing: watch::Sender<bool>,
37 /// Every connection accepted and not yet finished, which closing drains and then ends.
38 serving: Rc<RefCell<Vec<JoinHandle<()>>>>,
39 /// How long closing waits for those before ending them.
40 drain: Duration,
41}
42
43impl Drop for HyperConnections {
44 fn drop(&mut self) {
45 self.accepting.abort();
46 for connection in self.serving.borrow().iter() {
47 connection.abort();
48 }
49 }
50}
51
52impl HyperConnections {
53 /// Serves every connection accepted at `listener` with `service`, draining for `drain`.
54 ///
55 /// The drain is stated here rather than fixed, because how long a request may take is what the
56 /// surface being served decides, not what accepting it does. [`DRAIN`] is the usual answer.
57 pub fn serve<Answering, ResponseBody>(listener: TcpListener, service: Answering, drain: Duration) -> Self
58 where
59 Answering: Service<Request<Incoming>, Response = Response<ResponseBody>> + Clone + 'static,
60 Answering::Error: Into<ConnectionsError>,
61 Answering::Future: Send,
62 ResponseBody: Body + Send + 'static,
63 ResponseBody::Data: Send,
64 ResponseBody::Error: Into<ConnectionsError>,
65 {
66 let serving = Rc::new(RefCell::new(Vec::<JoinHandle<()>>::new()));
67 let (closing, closed) = watch::channel(false);
68 // An accept error affects that connection only, not the listener, so keep accepting.
69 let connections = TcpListenerStream::new(listener).filter_map(|accepted| async move { accepted.ok() });
70 let accepting = tokio::task::spawn_local(
71 connections
72 .for_each({
73 let serving = Rc::clone(&serving);
74 move |stream| {
75 let service = service.clone();
76 let serving = Rc::clone(&serving);
77 let mut closed = closed.clone();
78 async move {
79 // One task per connection, so releasing the address leaves these
80 // running, and holding the handle is what lets closing end them.
81 let connection = tokio::task::spawn_local(async move {
82 let connection = Builder::new(TokioExecutor::new());
83 let mut connection = pin!(connection.serve_connection(TokioIo::new(stream), service));
84 tokio::select! {
85 _ = connection.as_mut() => {}
86 _ = closed.changed() => {
87 // Finish the answer being produced, then stop rather than
88 // wait for a request that is never coming.
89 connection.as_mut().graceful_shutdown();
90 let _ = connection.await;
91 }
92 }
93 });
94 let mut serving = serving.borrow_mut();
95 serving.retain(|connection| !connection.is_finished());
96 serving.push(connection);
97 }
98 }
99 })
100 .map(Ok),
101 );
102
103 Self { accepting, closing, serving, drain }
104 }
105
106 /// Releases the address, leaving what is already being served to finish on its own.
107 ///
108 /// Aborting the accept task drops the listener, which is what frees the address, and awaiting
109 /// the handle is what makes it free on return rather than shortly after. Each connection is
110 /// then told to shut down gracefully, which is what makes one finish: it answers what it is
111 /// producing and then ends, rather than waiting for a request no caller will send. This does
112 /// not wait for that, which is what [`Self::end`] is for.
113 pub async fn close(&mut self) {
114 self.accepting.abort();
115 // Only where it is still running: closing twice, or closing and then ending, would
116 // otherwise poll a handle that has already answered, which panics.
117 if !self.accepting.is_finished() {
118 let _ = (&mut self.accepting).await;
119 }
120 let _ = self.closing.send(true);
121 }
122
123 /// Releases the address, then gives what is already being served up to the drain to finish.
124 ///
125 /// Whatever is still serving when the drain runs out is ended, so this resolves within the
126 /// drain of [`Self::close`] however slow a request is.
127 pub async fn end(&mut self) {
128 self.close().await;
129
130 let mut serving = self.serving.take();
131 let draining = async {
132 for connection in &mut serving {
133 let _ = connection.await;
134 }
135 };
136 let _ = tokio::time::timeout(self.drain, draining).await;
137
138 for connection in &serving {
139 connection.abort();
140 }
141 }
142}