conmonrs/streaming_server/
web.rs

1//! The HTTP side of the streaming server on top of hyper: short responses
2//! and the upgrade of a request to a WebSocket. The WebSocket handshake
3//! checks the request like RFC 6455 says, a request which does not pass gets
4//! a plain text response which tells what is missing.
5
6mod accept_key;
7
8use crate::streaming_server::idle::{Activity, STALL_TIMEOUT};
9use accept_key::accept_key;
10use bytes::Bytes;
11use futures::{
12    sink::{Sink, SinkExt},
13    stream::{SplitSink, SplitStream, Stream, StreamExt},
14    task::AtomicWaker,
15};
16use hyper::{
17    Method, Request, StatusCode,
18    body::{Body as HttpBody, Frame, SizeHint},
19    header::{self, HeaderMap, HeaderName, HeaderValue},
20    upgrade::{OnUpgrade, Upgraded},
21};
22use hyper_util::rt::TokioIo;
23use std::{
24    convert::Infallible,
25    future::Future,
26    io,
27    pin::Pin,
28    sync::{
29        Arc,
30        atomic::{AtomicBool, Ordering},
31    },
32    task::{Context, Poll, ready},
33};
34use tokio::{
35    io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf},
36    task::{self, JoinHandle},
37    time::{self, Duration, Instant},
38};
39use tokio_tungstenite::{
40    WebSocketStream,
41    tungstenite::{
42        error::{CapacityError, ProtocolError},
43        protocol::{Role, WebSocketConfig},
44    },
45};
46use tracing::{debug, trace};
47
48pub(super) use tokio_tungstenite::tungstenite::{
49    Error, Message,
50    protocol::{CloseFrame, frame::coding::CloseCode},
51};
52
53/// A WebSocket of a client, after the upgrade of its request.
54pub(super) type WebSocket = WebSocketStream<ClientIo<TokioIo<Upgraded>>>;
55
56/// How long a client gets to answer the close of the server, and to end the
57/// connection after a message which was too large, see [`drain`].
58pub(super) const CLOSE_TIMEOUT: Duration = Duration::from_secs(5);
59
60/// The most a client may still send after the close for a message which was
61/// too large, see [`drain`]. More than the rest of a message of the size
62/// Kubernetes takes.
63const DRAIN_LIMIT: usize = 64 * 1024 * 1024;
64
65/// The buffer a WebSocket reads into, which it allocates up front. Clients
66/// mostly send small stdin and resize events, larger messages grow it while
67/// they get read. Tungstenite allocates 128 KiB otherwise, for every session.
68const WEBSOCKET_READ_BUFFER_SIZE: usize = 16 * 1024;
69
70/// The close of the server once it is done, or after the client sent a
71/// message larger than the server takes, see [`is_too_large`], which tells
72/// the client why.
73pub(super) fn close(too_large: bool, reason: &str) -> Message {
74    let (code, reason) = if too_large {
75        (CloseCode::Size, "message too large")
76    } else {
77        (CloseCode::Normal, reason)
78    };
79    Message::Close(Some(CloseFrame {
80        code,
81        reason: reason.into(),
82    }))
83}
84
85/// Send the close of the server without a reason, see [`close`], the sink
86/// once it went out.
87pub(super) async fn send_close<S>(mut sink: S, too_large: bool) -> Option<S>
88where
89    S: Sink<Message> + Unpin,
90{
91    sink.send(close(too_large, ""))
92        .await
93        .is_ok()
94        .then_some(sink)
95}
96
97/// Answer the close of a client, which a read of `messages` just returned.
98/// Tungstenite queues the answer, but writes it only with a later read or
99/// flush, and every send after the close fails without writing anything.
100/// The next read writes it, after what got queued before, and ends the
101/// stream. Without it the client sees the connection end without a close
102/// handshake. A slow client still gets it, one which takes nothing for
103/// [`STALL_TIMEOUT`] does not, see [`ClientIo::written`].
104pub(super) async fn answer_close<S, E>(messages: &mut S, written: &Activity)
105where
106    S: Stream<Item = Result<Message, E>> + Unpin,
107{
108    if written
109        .until_stalled(Instant::now(), STALL_TIMEOUT, messages.next())
110        .await
111        .is_none()
112    {
113        debug!("Client did not take the answer to its close");
114    }
115}
116
117/// What a read of a WebSocket client gave, see [`next_message`].
118#[derive(Debug)]
119pub(super) enum ClientMessage {
120    /// A binary or a text message.
121    Data(Message),
122
123    /// The close of the client, which still needs its answer, see
124    /// [`answer_close`].
125    Closed(Option<CloseFrame>),
126
127    /// A message larger than the server takes, the input of the client gets
128    /// dropped from now on, see [`is_too_large`] and [`Discard`].
129    TooLarge,
130
131    /// The client went away or the read failed.
132    Gone,
133}
134
135/// Read the next message of a WebSocket client which the server handles,
136/// pings and pongs only keep the WebSocket alive. A client which sent a
137/// message larger than the server takes may still be busy writing the rest,
138/// which would hold up what the server sends, so its input gets dropped.
139pub(super) async fn next_message<S>(messages: &mut S, discard: &Discard) -> ClientMessage
140where
141    S: Stream<Item = Result<Message, Error>> + Unpin,
142{
143    loop {
144        match messages.next().await {
145            Some(Ok(Message::Ping(_))) => trace!("Got ping"),
146            Some(Ok(Message::Pong(_))) => trace!("Got pong"),
147            // Raw frames only get sent, reads never return them.
148            Some(Ok(Message::Frame(_))) => {}
149            Some(Ok(Message::Close(frame))) => return ClientMessage::Closed(frame),
150            Some(Ok(message)) => return ClientMessage::Data(message),
151            Some(Err(e)) if is_too_large(&e) => {
152                debug!("Unable to read the client: {e}");
153                discard.discard();
154                return ClientMessage::TooLarge;
155            }
156            Some(Err(_)) | None => return ClientMessage::Gone,
157        }
158    }
159}
160
161/// Write the answer to the close of a client after a send failed for it,
162/// if the read could not, see [`answer_close`].
163pub(super) async fn write_close_answer<S>(sink: &mut S)
164where
165    S: Sink<Message, Error = Error> + Unpin,
166{
167    if let Err(e) = sink.flush().await {
168        debug!("Unable to answer the close of the client: {e}");
169    }
170}
171
172/// Whether a send failed after a close, which is the one of the client
173/// while the server did not close yet, see [`answer_close`].
174pub(super) fn is_closed_by_client(e: &Error) -> bool {
175    matches!(e, Error::Protocol(ProtocolError::SendAfterClosing))
176}
177
178/// Whether a read failed for a message larger than the server takes. The
179/// WebSocket cannot be read anymore then, the rest of the message did not
180/// get read, but the server can still close it, see [`close`], and has to
181/// [`drain`] it then.
182fn is_too_large(e: &Error) -> bool {
183    matches!(e, Error::Capacity(CapacityError::MessageTooLong { .. }))
184}
185
186/// End a WebSocket whose client sent a message larger than the server
187/// takes, once the close went out. The rest of the message cannot be read
188/// as messages. The server stops writing and drops what the client still
189/// sends until it ends the connection, for up to [`CLOSE_TIMEOUT`] and
190/// [`DRAIN_LIMIT`], on top of what got dropped while the close went out, see
191/// [`Discard`]. A socket closed with input it did not read resets the
192/// connection, which can drop the close and the status before it.
193pub(super) async fn drain<T>(
194    sink: SplitSink<WebSocketStream<ClientIo<T>>, Message>,
195    messages: SplitStream<WebSocketStream<ClientIo<T>>>,
196) where
197    T: AsyncRead + AsyncWrite + Unpin,
198{
199    let Ok(mut socket) = messages.reunite(sink) else {
200        return;
201    };
202    let io = socket.get_mut();
203    io.drain();
204    let drained = async {
205        if let Err(e) = io.shutdown().await {
206            debug!("Unable to shut down the WebSocket: {e}");
207        }
208        // Dropped input never reaches the buffer, a read ends with the
209        // connection or the limit.
210        if let Err(e) = io.read(&mut [0; 1]).await {
211            debug!("Stopped draining the WebSocket: {e}");
212        }
213    };
214    if time::timeout(CLOSE_TIMEOUT, drained).await.is_err() {
215        debug!("Client did not end the WebSocket in time");
216    }
217}
218
219/// The connection of a WebSocket client, whose input can get dropped
220/// instead of read, see [`Discard`].
221pub(super) struct ClientIo<T> {
222    io: T,
223    discard: Arc<DiscardState>,
224
225    /// How much input may still get dropped.
226    left: usize,
227
228    /// Whether the client ended its side of the connection.
229    ended: bool,
230
231    /// Touched whenever a part of what the server sends went out.
232    written: Activity,
233}
234
235#[derive(Debug, Default)]
236struct DiscardState {
237    /// Whether the input gets dropped.
238    on: AtomicBool,
239
240    /// The task whose write waits for the client, which has to drop its
241    /// input from now on.
242    writer: AtomicWaker,
243}
244
245/// Drops the input of a [`ClientIo`], also after its WebSocket got split. A
246/// client which sent a message larger than the server takes may still be
247/// busy writing the rest of it and not read meanwhile, which holds up the
248/// status and the close for it, unless its input gets read. The WebSocket
249/// cannot read it anymore, it fails on the message again. Writes drop what
250/// the client sent before they write, up to [`DRAIN_LIMIT`], and the input
251/// of the client wakes a write which waits for it. Reads see nothing but the
252/// end of the connection.
253#[derive(Clone, Debug, Default)]
254pub(super) struct Discard(Arc<DiscardState>);
255
256impl Discard {
257    /// Drop the input of the client from now on.
258    pub(super) fn discard(&self) {
259        self.0.on.store(true, Ordering::Release);
260        // A write which waits for a client busy with the rest of its message
261        // would never get woken otherwise.
262        self.0.writer.wake();
263    }
264}
265
266impl<T> ClientIo<T> {
267    /// How many reads of dropped input one poll does at most, so that a
268    /// client which keeps sending does not hold the thread.
269    const DISCARD_READS: usize = 16;
270
271    /// Wrap the connection of a client.
272    pub(super) fn new(io: T) -> Self {
273        Self {
274            io,
275            discard: Arc::default(),
276            left: DRAIN_LIMIT,
277            ended: false,
278            written: Activity::new(),
279        }
280    }
281
282    /// Touched whenever a part of what the server sends went out, which
283    /// tells that the client still takes it, see [`Activity::until_stalled`].
284    pub(super) fn written(&self) -> Activity {
285        self.written.clone()
286    }
287
288    /// Drops the input of the connection.
289    pub(super) fn discarder(&self) -> Discard {
290        Discard(self.discard.clone())
291    }
292
293    fn discarding(&self) -> bool {
294        self.discard.on.load(Ordering::Acquire)
295    }
296
297    /// Drop the input from now on, with a budget of its own for the end of
298    /// the WebSocket, see [`drain`].
299    fn drain(&mut self) {
300        self.discard.on.store(true, Ordering::Release);
301        self.left = DRAIN_LIMIT;
302    }
303}
304
305impl<T: AsyncRead + Unpin> ClientIo<T> {
306    /// Read and drop the input until the connection ended, which it is
307    /// ready for.
308    fn poll_discard(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
309        if self.ended {
310            return Poll::Ready(Ok(()));
311        }
312        let mut scratch = [0; 8192];
313        for _ in 0..Self::DISCARD_READS {
314            if self.left == 0 {
315                return Poll::Ready(Err(io::Error::other(
316                    "client sent too much after the close",
317                )));
318            }
319            let mut dropped = ReadBuf::new(&mut scratch);
320            ready!(Pin::new(&mut self.io).poll_read(cx, &mut dropped))?;
321            match dropped.filled().len() {
322                0 => {
323                    self.ended = true;
324                    return Poll::Ready(Ok(()));
325                }
326                read => self.left = self.left.saturating_sub(read),
327            }
328        }
329        // More input may wait, which is readable already and wakes nobody.
330        cx.waker().wake_by_ref();
331        Poll::Pending
332    }
333
334    /// Run a write, after dropping the input if it gets dropped. The read
335    /// leaves the waker of the write with the connection, so that input
336    /// wakes a write which waits for the client, also one which started
337    /// before the input got dropped.
338    fn poll_writing<R>(
339        &mut self,
340        cx: &mut Context<'_>,
341        write: impl FnOnce(Pin<&mut T>, &mut Context<'_>) -> Poll<R>,
342    ) -> Poll<R> {
343        let discarding = self.discarding();
344        if discarding {
345            // Its failures end the dropping, the write still goes on.
346            let _ = self.poll_discard(cx);
347        }
348        let poll = write(Pin::new(&mut self.io), cx);
349        if poll.is_pending() && !discarding {
350            self.discard.writer.register(cx.waker());
351            if self.discarding() {
352                cx.waker().wake_by_ref();
353            }
354        }
355        poll
356    }
357}
358
359impl<T: AsyncRead + Unpin> AsyncRead for ClientIo<T> {
360    fn poll_read(
361        self: Pin<&mut Self>,
362        cx: &mut Context<'_>,
363        buf: &mut ReadBuf<'_>,
364    ) -> Poll<io::Result<()>> {
365        let this = self.get_mut();
366        if this.discarding() {
367            return this.poll_discard(cx);
368        }
369        Pin::new(&mut this.io).poll_read(cx, buf)
370    }
371}
372
373impl<T> ClientIo<T> {
374    /// Touch [`ClientIo::written`] for a write which wrote something.
375    fn wrote(&self, poll: Poll<io::Result<usize>>) -> Poll<io::Result<usize>> {
376        if matches!(poll, Poll::Ready(Ok(1..))) {
377            self.written.touch();
378        }
379        poll
380    }
381}
382
383impl<T: AsyncRead + AsyncWrite + Unpin> AsyncWrite for ClientIo<T> {
384    fn poll_write(
385        self: Pin<&mut Self>,
386        cx: &mut Context<'_>,
387        buf: &[u8],
388    ) -> Poll<io::Result<usize>> {
389        let this = self.get_mut();
390        let poll = this.poll_writing(cx, |io, cx| io.poll_write(cx, buf));
391        this.wrote(poll)
392    }
393
394    fn poll_write_vectored(
395        self: Pin<&mut Self>,
396        cx: &mut Context<'_>,
397        bufs: &[io::IoSlice<'_>],
398    ) -> Poll<io::Result<usize>> {
399        let this = self.get_mut();
400        let poll = this.poll_writing(cx, |io, cx| io.poll_write_vectored(cx, bufs));
401        this.wrote(poll)
402    }
403
404    fn is_write_vectored(&self) -> bool {
405        self.io.is_write_vectored()
406    }
407
408    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
409        self.get_mut().poll_writing(cx, AsyncWrite::poll_flush)
410    }
411
412    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
413        self.get_mut().poll_writing(cx, AsyncWrite::poll_shutdown)
414    }
415}
416
417/// Wait for the task which writes to a WebSocket and closes it, abort it
418/// once the client did not take anything for [`STALL_TIMEOUT`], see
419/// [`ClientIo::written`]. A slow client still gets what is queued.
420/// Dropping its handle would leave it running, a writer to a client which
421/// does not read would hold the connection. The output of the writer, `None`
422/// if it did not end.
423async fn join_or_abort<T>(mut writer: JoinHandle<T>, written: &Activity) -> Option<T> {
424    if let Some(output) = written
425        .until_stalled(Instant::now(), STALL_TIMEOUT, &mut writer)
426        .await
427    {
428        output.ok()
429    } else {
430        writer.abort();
431        None
432    }
433}
434
435/// Read the client after the close of the server, until it answered the
436/// close, for up to [`CLOSE_TIMEOUT`]. What it sends meanwhile gets dropped.
437/// A socket closed with input it did not read resets the connection, which
438/// can drop what the client did not read yet, like the close and the output
439/// before it.
440pub(super) async fn await_close_answer<S, E>(messages: &mut S)
441where
442    S: Stream<Item = Result<Message, E>> + Unpin,
443{
444    let answer = async {
445        while let Some(Ok(message)) = messages.next().await {
446            if matches!(message, Message::Close(_)) {
447                break;
448            }
449        }
450    };
451    if time::timeout(CLOSE_TIMEOUT, answer).await.is_err() {
452        debug!("Client did not answer the close in time");
453    }
454}
455
456/// End a WebSocket: wait for the writer, which sends the close of the
457/// server, see [`join_or_abort`], then for the answer of the client, see
458/// [`await_close_answer`]. The writer returns its sink once its close went
459/// out, a client gets no time to answer a close it did not get.
460async fn finish<S, E, T>(writer: JoinHandle<Option<T>>, written: &Activity, messages: &mut S)
461where
462    S: Stream<Item = Result<Message, E>> + Unpin,
463{
464    if let Some(Some(_)) = join_or_abort(writer, written).await {
465        await_close_answer(messages).await;
466    }
467}
468
469/// How a WebSocket session ended, which decides what its end waits for, see
470/// [`end`].
471#[derive(Clone, Copy, Debug, PartialEq)]
472pub(super) enum End {
473    /// The client sent a message larger than the server takes, which it
474    /// cannot answer the close for, see [`drain`].
475    TooLarge,
476
477    /// The client closed the WebSocket or went away.
478    Gone,
479
480    /// The server ended the session, the client answers its close, see
481    /// [`finish`].
482    Done,
483}
484
485/// End a WebSocket session once its `writer`, which sends the close of the
486/// server, got the rest out, see [`join_or_abort`]. What comes after depends
487/// on how it ended, see [`End`].
488pub(super) async fn end<T>(
489    writer: JoinHandle<Option<SplitSink<WebSocketStream<ClientIo<T>>, Message>>>,
490    written: &Activity,
491    mut messages: SplitStream<WebSocketStream<ClientIo<T>>>,
492    end: End,
493) where
494    T: AsyncRead + AsyncWrite + Unpin,
495{
496    match end {
497        End::TooLarge => {
498            if let Some(Some(sink)) = join_or_abort(writer, written).await {
499                drain(sink, messages).await;
500            }
501        }
502        End::Gone => {
503            join_or_abort(writer, written).await;
504        }
505        End::Done => finish(writer, written, &mut messages).await,
506    }
507}
508
509/// A response of the streaming server.
510pub(super) type Response = hyper::Response<Body>;
511
512/// The body of a response. The server sends short texts only, at once.
513#[derive(Debug, Default)]
514pub(super) struct Body(Option<Bytes>);
515
516impl From<Bytes> for Body {
517    fn from(data: Bytes) -> Self {
518        Self((!data.is_empty()).then_some(data))
519    }
520}
521
522impl HttpBody for Body {
523    type Data = Bytes;
524    type Error = Infallible;
525
526    fn poll_frame(
527        self: Pin<&mut Self>,
528        _: &mut Context<'_>,
529    ) -> Poll<Option<Result<Frame<Bytes>, Infallible>>> {
530        Poll::Ready(self.get_mut().0.take().map(|data| Ok(Frame::data(data))))
531    }
532
533    fn is_end_stream(&self) -> bool {
534        self.0.is_none()
535    }
536
537    fn size_hint(&self) -> SizeHint {
538        SizeHint::with_exact(self.0.as_ref().map_or(0, |data| data.len() as u64))
539    }
540}
541
542/// A response with the status only.
543pub(super) fn status(status: StatusCode) -> Response {
544    let mut response = Response::default();
545    *response.status_mut() = status;
546    response
547}
548
549/// A response with a plain text, which tells the client what went wrong.
550pub(super) fn text(status: StatusCode, text: impl Into<String>) -> Response {
551    let mut response = Response::new(Bytes::from(text.into()).into());
552    *response.status_mut() = status;
553    response.headers_mut().insert(
554        header::CONTENT_TYPE,
555        HeaderValue::from_static("text/plain; charset=utf-8"),
556    );
557    response
558}
559
560/// A request which asks for a WebSocket, checked like RFC 6455 says.
561pub(super) struct WebSocketUpgrade {
562    key: HeaderValue,
563    on_upgrade: OnUpgrade,
564    protocol: Option<&'static str>,
565    config: WebSocketConfig,
566}
567
568impl WebSocketUpgrade {
569    /// Check the request, the rejection is the response to send otherwise.
570    /// The server speaks HTTP/1 only, where a WebSocket starts with a GET.
571    pub(super) fn from_request<B>(request: &mut Request<B>) -> Result<Self, Box<Response>> {
572        let reject = |status, message: &'static str| Err(Box::new(text(status, message)));
573        if request.method() != Method::GET {
574            return reject(
575                StatusCode::METHOD_NOT_ALLOWED,
576                "Request method must be `GET`",
577            );
578        }
579        let headers = request.headers();
580        if !header_has_token(headers, header::CONNECTION, "upgrade") {
581            return reject(
582                StatusCode::BAD_REQUEST,
583                "Connection header did not include 'upgrade'",
584            );
585        }
586        if !header_has_token(headers, header::UPGRADE, "websocket") {
587            return reject(
588                StatusCode::BAD_REQUEST,
589                "`Upgrade` header did not include 'websocket'",
590            );
591        }
592        let Some(key) = headers.get(header::SEC_WEBSOCKET_KEY).cloned() else {
593            return reject(
594                StatusCode::BAD_REQUEST,
595                "`Sec-WebSocket-Key` header missing",
596            );
597        };
598        if !header_eq(headers, header::SEC_WEBSOCKET_VERSION, "13") {
599            return reject(
600                StatusCode::BAD_REQUEST,
601                "`Sec-WebSocket-Version` header did not include '13'",
602            );
603        }
604        let Some(on_upgrade) = request.extensions_mut().remove::<OnUpgrade>() else {
605            return reject(
606                StatusCode::UPGRADE_REQUIRED,
607                "WebSocket request couldn't be upgraded since no upgrade state was present",
608            );
609        };
610        Ok(Self {
611            key,
612            on_upgrade,
613            protocol: None,
614            config: WebSocketConfig::default().read_buffer_size(WEBSOCKET_READ_BUFFER_SIZE),
615        })
616    }
617
618    /// The subprotocol the server picked out of the ones of the client.
619    pub(super) fn protocol(mut self, protocol: Option<&'static str>) -> Self {
620        self.protocol = protocol;
621        self
622    }
623
624    /// The largest message and frame the client may send.
625    pub(super) fn max_message_size(mut self, max: usize) -> Self {
626        self.config = self
627            .config
628            .max_message_size(Some(max))
629            .max_frame_size(Some(max));
630        self
631    }
632
633    /// Answer the request, the WebSocket goes to `callback` once the
634    /// connection got upgraded.
635    pub(super) fn on_upgrade<C, F>(self, callback: C) -> Response
636    where
637        C: FnOnce(WebSocket) -> F + Send + 'static,
638        F: Future<Output = ()> + Send + 'static,
639    {
640        let Self {
641            key,
642            on_upgrade,
643            protocol,
644            config,
645        } = self;
646        task::spawn(async move {
647            let upgraded = match on_upgrade.await {
648                Ok(upgraded) => upgraded,
649                Err(e) => {
650                    debug!("Unable to upgrade connection to a WebSocket: {e}");
651                    return;
652                }
653            };
654            let socket = WebSocketStream::from_raw_socket(
655                ClientIo::new(TokioIo::new(upgraded)),
656                Role::Server,
657                Some(config),
658            )
659            .await;
660            callback(socket).await;
661        });
662
663        let mut response = status(StatusCode::SWITCHING_PROTOCOLS);
664        let headers = response.headers_mut();
665        headers.insert(header::CONNECTION, HeaderValue::from_static("upgrade"));
666        headers.insert(header::UPGRADE, HeaderValue::from_static("websocket"));
667        if let Ok(accept) = HeaderValue::from_str(&accept_key(key.as_bytes())) {
668            headers.insert(header::SEC_WEBSOCKET_ACCEPT, accept);
669        }
670        if let Some(protocol) = protocol {
671            headers.insert(
672                header::SEC_WEBSOCKET_PROTOCOL,
673                HeaderValue::from_static(protocol),
674            );
675        }
676        response
677    }
678}
679
680/// Whether one of the values of the header is `value`, ignoring the case.
681fn header_eq(headers: &HeaderMap, name: HeaderName, value: &str) -> bool {
682    headers
683        .get_all(name)
684        .iter()
685        .any(|v| v.as_bytes().eq_ignore_ascii_case(value.as_bytes()))
686}
687
688/// The comma separated tokens of all values of the header, without the
689/// empty ones, like `keep-alive` and `Upgrade` of
690/// `Connection: keep-alive, Upgrade`.
691pub(super) fn header_tokens(headers: &HeaderMap, name: HeaderName) -> impl Iterator<Item = &str> {
692    headers
693        .get_all(name)
694        .into_iter()
695        .filter_map(|v| v.to_str().ok())
696        .flat_map(|v| v.split(','))
697        .map(str::trim)
698        .filter(|t| !t.is_empty())
699}
700
701/// Whether one of the tokens of the header is `token`, ignoring the case,
702/// see [`header_tokens`].
703pub(super) fn header_has_token(headers: &HeaderMap, name: HeaderName, token: &str) -> bool {
704    header_tokens(headers, name).any(|t| t.eq_ignore_ascii_case(token))
705}
706
707#[cfg(test)]
708mod tests;
conmonrs/streaming_server/web.rs

conmonrs/streaming_server/
web.rs
