fuchsia_pkg_testing/serve/
responder.rs

1// Copyright 2020 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! HttpResponder implementations
6
7use crate::serve::HttpResponder;
8use fuchsia_repo::body::Body;
9use fuchsia_sync::Mutex;
10use futures::channel::{mpsc, oneshot};
11use futures::future::{BoxFuture, Shared, pending, ready};
12use futures::prelude::*;
13use http_body_util::BodyExt;
14use hyper::{Request, Response, StatusCode};
15use std::collections::HashSet;
16use std::path::{Path, PathBuf};
17use std::sync::Arc;
18use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU32, Ordering};
19
20/// hyper::Request extension trait that makes writing `HttpResponder`s more convienent.
21pub trait RequestExt {
22    /// The URI path of the Request.
23    fn path(&self) -> &Path;
24}
25
26impl RequestExt for Request<Body> {
27    /// The URI path of the Request.
28    fn path(&self) -> &Path {
29        Path::new(self.uri().path())
30    }
31}
32
33/// Responder that always responds with the given status code
34pub struct StaticResponseCode(StatusCode);
35
36impl HttpResponder for StaticResponseCode {
37    fn respond(&self, _: &Request<Body>, _: Response<Body>) -> BoxFuture<'_, Response<Body>> {
38        ready(Response::builder().status(self.0).body(Body::empty()).unwrap()).boxed()
39    }
40}
41
42impl StaticResponseCode {
43    /// Creates responder that always responds with the given status code
44    pub fn new(status: StatusCode) -> Self {
45        Self(status)
46    }
47
48    /// Creates responder that always responds with 200 OK
49    pub fn ok() -> Self {
50        Self(StatusCode::OK)
51    }
52
53    /// Creates responder that always responds with 404 Not Found
54    pub fn not_found() -> Self {
55        Self(StatusCode::NOT_FOUND)
56    }
57
58    /// Creates responder that always responds with 500 Internal Server Error
59    pub fn server_error() -> Self {
60        Self(StatusCode::INTERNAL_SERVER_ERROR)
61    }
62
63    /// Creates responder that always responds with 429 Too Many Requests
64    pub fn too_many_requests() -> Self {
65        Self(StatusCode::TOO_MANY_REQUESTS)
66    }
67}
68
69/// An atomic HTTP status code carrier.
70#[derive(Debug, Default)]
71pub struct DynamicResponseSetter(Arc<AtomicU16>);
72
73impl DynamicResponseSetter {
74    /// Atomically sets this toggle to the supplied code.
75    pub fn set(&self, code: u16) {
76        self.0.store(code, Ordering::SeqCst);
77    }
78}
79
80/// Responder that replies with an externally-settable HTTP status.
81pub struct DynamicResponseCode {
82    code: Arc<AtomicU16>,
83}
84
85impl HttpResponder for DynamicResponseCode {
86    fn respond<'a>(
87        &'a self,
88        _: &'a Request<Body>,
89        _: Response<Body>,
90    ) -> BoxFuture<'_, Response<Body>> {
91        ready(
92            Response::builder()
93                .status(self.code.load(Ordering::SeqCst))
94                .body(Body::empty())
95                .unwrap(),
96        )
97        .boxed()
98    }
99}
100
101impl DynamicResponseCode {
102    /// Creates a new responder with a (re)settable status code.
103    pub fn new(initial: u16) -> (Self, DynamicResponseSetter) {
104        let setter = DynamicResponseSetter(Arc::new(initial.into()));
105        (Self { code: Arc::clone(&setter.0) }, setter)
106    }
107}
108
109/// An atomic toggle switch.
110#[derive(Debug, Default)]
111pub struct AtomicToggle(Arc<AtomicBool>);
112
113impl AtomicToggle {
114    /// Creates a new AtomicToggle initialized to `initial`.
115    pub fn new(initial: bool) -> Self {
116        Self(Arc::new(initial.into()))
117    }
118
119    /// Atomically sets this toggle to true.
120    pub fn set(&self) {
121        self.0.store(true, Ordering::SeqCst);
122    }
123
124    /// Atomically sets this toggle to false.
125    pub fn unset(&self) {
126        self.0.store(false, Ordering::SeqCst);
127    }
128}
129
130/// Responder that overrides requests with the given responder only when enabled
131pub struct Toggleable<H: HttpResponder> {
132    enabled: Arc<AtomicBool>,
133    responder: H,
134}
135
136impl<H: HttpResponder> HttpResponder for Toggleable<H> {
137    fn respond<'a>(
138        &'a self,
139        request: &'a Request<Body>,
140        response: Response<Body>,
141    ) -> BoxFuture<'_, Response<Body>> {
142        if self.enabled.load(Ordering::SeqCst) {
143            self.responder.respond(request, response)
144        } else {
145            ready(response).boxed()
146        }
147    }
148}
149
150impl<H: HttpResponder> Toggleable<H> {
151    /// Creates responder that overrides requests when should_override is set.
152    pub fn new(should_override: &AtomicToggle, responder: H) -> Self {
153        Self { enabled: Arc::clone(&should_override.0), responder }
154    }
155}
156
157/// Responder that overrides the given request path for the given number of requests.
158pub struct ForRequestCount<H: HttpResponder> {
159    remaining: Mutex<u32>,
160    responder: H,
161}
162
163impl<H: HttpResponder> HttpResponder for ForRequestCount<H> {
164    fn respond<'a>(
165        &'a self,
166        request: &'a Request<Body>,
167        response: Response<Body>,
168    ) -> BoxFuture<'_, Response<Body>> {
169        let mut remaining = self.remaining.lock();
170        if *remaining > 0 {
171            *remaining -= 1;
172            drop(remaining);
173            self.responder.respond(request, response)
174        } else {
175            ready(response).boxed()
176        }
177    }
178}
179
180impl<H: HttpResponder> ForRequestCount<H> {
181    /// Creates responder that overrides the given request path for the given number of requests.
182    pub fn new(count: u32, responder: H) -> Self {
183        Self { remaining: Mutex::new(count), responder }
184    }
185}
186
187/// Responder that overrides the given request path using the given responder.
188pub struct ForPath<H: HttpResponder> {
189    path: PathBuf,
190    responder: H,
191}
192
193impl<H: HttpResponder> HttpResponder for ForPath<H> {
194    fn respond<'a>(
195        &'a self,
196        request: &'a Request<Body>,
197        response: Response<Body>,
198    ) -> BoxFuture<'_, Response<Body>> {
199        if self.path == request.path() {
200            self.responder.respond(request, response)
201        } else {
202            ready(response).boxed()
203        }
204    }
205}
206
207impl<H: HttpResponder> ForPath<H> {
208    /// Creates responder that overrides the given request path using the given responder.
209    pub fn new(path: impl Into<PathBuf>, responder: H) -> Self {
210        Self { path: path.into(), responder }
211    }
212}
213
214/// Responder that overrides the given request paths using the given responder.
215pub struct ForPaths<H: HttpResponder> {
216    paths: HashSet<PathBuf>,
217    responder: H,
218}
219
220impl<H: HttpResponder> HttpResponder for ForPaths<H> {
221    fn respond<'a>(
222        &'a self,
223        request: &'a Request<Body>,
224        response: Response<Body>,
225    ) -> BoxFuture<'_, Response<Body>> {
226        if self.paths.contains(request.path()) {
227            self.responder.respond(request, response)
228        } else {
229            ready(response).boxed()
230        }
231    }
232}
233
234impl<H: HttpResponder> ForPaths<H> {
235    /// Creates responder that overrides the given request paths using the given responder.
236    pub fn new(paths: HashSet<PathBuf>, responder: H) -> Self {
237        Self { paths, responder }
238    }
239}
240
241/// Responder that overrides all the requests that start with the given request path using the
242/// given responder.
243pub struct ForPathPrefix<H: HttpResponder> {
244    prefix: PathBuf,
245    responder: H,
246}
247
248impl<H: HttpResponder> HttpResponder for ForPathPrefix<H> {
249    fn respond<'a>(
250        &'a self,
251        request: &'a Request<Body>,
252        response: Response<Body>,
253    ) -> BoxFuture<'_, Response<Body>> {
254        if request.path().starts_with(&self.prefix) {
255            self.responder.respond(request, response)
256        } else {
257            ready(response).boxed()
258        }
259    }
260}
261
262impl<H: HttpResponder> ForPathPrefix<H> {
263    /// Creates responder that overrides all the requests that start with the given request path
264    /// using the given responder.
265    pub fn new(prefix: impl Into<PathBuf>, responder: H) -> Self {
266        Self { prefix: prefix.into(), responder }
267    }
268}
269
270/// Responder that overrides all the requests that end with the given request path using the given responder.
271///
272/// Useful for hitting all versions of versioned TUF metadata (e.g. X.targets.json).
273/// TODO(ampearce): change ForPathSuffix and ForPathPrefix to use string matches rather than path.
274pub struct ForPathSuffix<H: HttpResponder> {
275    suffix: PathBuf,
276    responder: H,
277}
278
279impl<H: HttpResponder> HttpResponder for ForPathSuffix<H> {
280    fn respond<'a>(
281        &'a self,
282        request: &'a Request<Body>,
283        response: Response<Body>,
284    ) -> BoxFuture<'_, Response<Body>> {
285        if request.path().ends_with(&self.suffix) {
286            self.responder.respond(request, response)
287        } else {
288            ready(response).boxed()
289        }
290    }
291}
292
293impl<H: HttpResponder> ForPathSuffix<H> {
294    /// Creates responder that overrides all the requests that start with the given request path
295    /// using the given responder.
296    pub fn new(suffix: impl Into<PathBuf>, responder: H) -> Self {
297        Self { suffix: suffix.into(), responder }
298    }
299}
300/// Responder that passes responses through the given responder once per unique path.
301pub struct OncePerPath<H: HttpResponder> {
302    responder: H,
303    failed_paths: Mutex<HashSet<PathBuf>>,
304}
305
306impl<H: HttpResponder> HttpResponder for OncePerPath<H> {
307    fn respond<'a>(
308        &'a self,
309        request: &'a Request<Body>,
310        response: Response<Body>,
311    ) -> BoxFuture<'_, Response<Body>> {
312        if self.failed_paths.lock().insert(request.path().to_owned()) {
313            self.responder.respond(request, response)
314        } else {
315            ready(response).boxed()
316        }
317    }
318}
319
320impl<H: HttpResponder> OncePerPath<H> {
321    /// Creates responder that passes responses through the given responder once per unique path.
322    pub fn new(responder: H) -> Self {
323        Self { responder, failed_paths: Mutex::new(HashSet::new()) }
324    }
325}
326
327/// Transform a `serde_json::Value`. Implements `HttpResponder` by assuming the `Response<Body>` is
328/// json-formatted.
329pub trait JsonTransformer: Send + Sync + Clone + 'static {
330    /// Transform a `serde_json::Value`
331    fn transform(&self, v: serde_json::Value) -> serde_json::Value;
332}
333
334impl<F> JsonTransformer for F
335where
336    F: Fn(serde_json::Value) -> serde_json::Value + Send + Sync + Clone + 'static,
337{
338    fn transform(&self, v: serde_json::Value) -> serde_json::Value {
339        (self)(v)
340    }
341}
342
343/// Responder that manipulates requests with json-formatted bodies.
344impl<T: JsonTransformer> HttpResponder for T {
345    fn respond(
346        &self,
347        _: &Request<Body>,
348        response: Response<Body>,
349    ) -> BoxFuture<'_, Response<Body>> {
350        async move {
351            let (mut parts, body) = response.into_parts();
352            parts.headers.remove(http::header::CONTENT_LENGTH);
353            let bytes = body_to_bytes(body).await;
354            let value = self.transform(serde_json::from_reader(bytes.as_slice()).unwrap());
355            let bytes = serde_json::to_vec(&value).unwrap();
356            Response::from_parts(parts, Body::from(bytes))
357        }
358        .boxed()
359    }
360}
361
362/// Responder that notifies a channel when it receives a request.
363pub struct NotifyWhenRequested {
364    notify: mpsc::UnboundedSender<()>,
365}
366
367impl NotifyWhenRequested {
368    /// Creates a new responder and the receiver it notifies on request receipt.
369    pub fn new() -> (Self, mpsc::UnboundedReceiver<()>) {
370        let (tx, rx) = mpsc::unbounded();
371        (Self { notify: tx }, rx)
372    }
373}
374
375impl HttpResponder for NotifyWhenRequested {
376    fn respond(
377        &self,
378        _: &Request<Body>,
379        response: Response<Body>,
380    ) -> BoxFuture<'_, Response<Body>> {
381        self.notify.unbounded_send(()).unwrap();
382        ready(response).boxed()
383    }
384}
385
386/// A response that is waiting to be sent.
387pub struct BlockedResponse {
388    path: PathBuf,
389    unblocker: oneshot::Sender<()>,
390}
391
392impl BlockedResponse {
393    /// The path of the request.
394    pub fn path(&self) -> &Path {
395        &self.path
396    }
397
398    /// Send the response.
399    pub fn unblock(self) {
400        self.unblocker.send(()).expect("request to still be pending")
401    }
402}
403
404/// Responder that blocks sending response headers and bodies until unblocked by a test.
405pub struct BlockResponseHeaders {
406    blocked_responses: mpsc::UnboundedSender<BlockedResponse>,
407}
408
409impl BlockResponseHeaders {
410    /// Creates a new responder and the receiver it notifies on request receipt.
411    pub fn new() -> (Self, mpsc::UnboundedReceiver<BlockedResponse>) {
412        let (sender, receiver) = mpsc::unbounded();
413
414        (Self { blocked_responses: sender }, receiver)
415    }
416}
417
418impl HttpResponder for BlockResponseHeaders {
419    fn respond(
420        &self,
421        request: &Request<Body>,
422        response: Response<Body>,
423    ) -> BoxFuture<'_, Response<Body>> {
424        // Return a future that notifies the test that the request was blocked and wait for it to
425        // unblock the response.
426        let path = request.path().to_owned();
427        let mut blocked_responses = self.blocked_responses.clone();
428        async move {
429            let (unblocker, waiter) = oneshot::channel();
430            blocked_responses
431                .send(BlockedResponse { path, unblocker })
432                .await
433                .expect("receiver to still exist");
434            waiter.await.expect("request to be unblocked");
435            response
436        }
437        .boxed()
438    }
439}
440
441/// A response Body that is waiting to be sent.
442pub struct BlockedBody {
443    path: PathBuf,
444    unblocker: Box<dyn FnOnce() + Send>,
445}
446
447impl BlockedBody {
448    /// The path of the request.
449    pub fn path(&self) -> &Path {
450        &self.path
451    }
452
453    /// Send the Body.
454    pub fn unblock(self) {
455        (self.unblocker)()
456    }
457}
458
459/// Responder that blocks sending response bodies until unblocked by a test.
460pub struct BlockResponseBodies {
461    blocked_responses: mpsc::UnboundedSender<BlockedBody>,
462}
463
464impl BlockResponseBodies {
465    /// Creates a new responder and the receiver it notifies on request receipt.
466    pub fn new() -> (Self, mpsc::UnboundedReceiver<BlockedBody>) {
467        let (sender, receiver) = mpsc::unbounded();
468        (Self { blocked_responses: sender }, receiver)
469    }
470}
471
472impl HttpResponder for BlockResponseBodies {
473    fn respond(
474        &self,
475        request: &Request<Body>,
476        mut response: Response<Body>,
477    ) -> BoxFuture<'_, Response<Body>> {
478        let path = request.path().to_owned();
479        let mut blocked_responses = self.blocked_responses.clone();
480        async move {
481            // Replace the response's body with a stream that will yield data when the test
482            // unblocks the response body.
483            let (mut sender, new_body) = Body::channel();
484            let old_body = std::mem::replace(response.body_mut(), new_body);
485            let contents = body_to_bytes(old_body).await;
486
487            // Notify the test.
488            let unblocker =
489                Box::new(move || sender.try_send_data(contents.into()).expect("sending body"));
490            let () = blocked_responses
491                .send(BlockedBody { path, unblocker })
492                .await
493                .expect("receiver to still exist");
494
495            // Yield the modified response so hyper will send the headers and wait for the body to
496            // be unblocked.
497            response
498        }
499        .boxed()
500    }
501}
502
503/// Responder that blocks sending response body until unblocked by a test.
504/// Panics if requested more than once.
505pub struct BlockResponseBodyOnce {
506    #[allow(clippy::type_complexity)]
507    notify: Mutex<Option<oneshot::Sender<Box<dyn FnOnce() + Send>>>>,
508}
509
510impl BlockResponseBodyOnce {
511    /// Creates a new responder and the receiver it notifies after sending the response headers.
512    pub fn new() -> (Self, oneshot::Receiver<Box<dyn FnOnce() + Send>>) {
513        let (sender, receiver) = oneshot::channel();
514
515        (Self { notify: Mutex::new(Some(sender)) }, receiver)
516    }
517}
518
519impl HttpResponder for BlockResponseBodyOnce {
520    fn respond(
521        &self,
522        _: &Request<Body>,
523        mut response: Response<Body>,
524    ) -> BoxFuture<'_, Response<Body>> {
525        let notify = self.notify.lock().take().expect("a single request for this path");
526
527        async move {
528            // Replace the response's body with a stream that will yield data when the test
529            // unblocks the response body.
530            let (mut sender, new_body) = Body::channel();
531            let old_body = std::mem::replace(response.body_mut(), new_body);
532            let contents = body_to_bytes(old_body).await;
533
534            // Notify the test.
535            notify
536                .send(Box::new(move || {
537                    sender.try_send_data(contents.into()).expect("sending body")
538                }))
539                .map_err(|_| ())
540                .expect("receiver to still exist");
541
542            // Yield the modified response so hyper will send the headers and wait for the body to
543            // be unblocked.
544            response
545        }
546        .boxed()
547    }
548}
549
550async fn body_to_bytes(body: Body) -> Vec<u8> {
551    body.collect().await.expect("body to bytes").to_bytes().to_vec()
552}
553
554/// Responder that yields the response up to the final byte, then produces an error.  Panics if the
555/// response contains an empty body.
556pub struct OneByteShortThenError;
557
558impl HttpResponder for OneByteShortThenError {
559    fn respond(
560        &self,
561        _: &Request<Body>,
562        response: Response<Body>,
563    ) -> BoxFuture<'_, Response<Body>> {
564        async {
565            let (parts, body) = response.into_parts();
566            let mut bytes = body_to_bytes(body).await;
567            if bytes.pop().is_none() {
568                panic!("can't short 0 bytes");
569            }
570            Response::from_parts(
571                parts,
572                Body::wrap_stream(futures::stream::iter(vec![
573                    Ok(bytes),
574                    Err("all_but_one_byte_then_eror has sent all but one bytes".to_string()),
575                ])),
576            )
577        }
578        .boxed()
579    }
580}
581
582/// Responder that yields the response up to the Nth byte, then produces an error.  Panics if the
583/// response does not contain more than N bytes.
584pub struct NBytesThenError {
585    n: usize,
586}
587
588impl NBytesThenError {
589    /// Make a responder that returns N bytes then errors.
590    pub fn new(n: usize) -> Self {
591        Self { n }
592    }
593}
594impl HttpResponder for NBytesThenError {
595    fn respond(
596        &self,
597        _: &Request<Body>,
598        response: Response<Body>,
599    ) -> BoxFuture<'_, Response<Body>> {
600        let n = self.n;
601        async move {
602            let (parts, body) = response.into_parts();
603            let mut bytes = body_to_bytes(body).await;
604            let initial_len = bytes.len();
605            if initial_len <= n {
606                panic!("not enough bytes to shorten, {initial_len} {n}");
607            }
608            bytes.truncate(n);
609            Response::from_parts(
610                parts,
611                Body::wrap_stream(futures::stream::iter(vec![
612                    Ok(bytes),
613                    Err("all_but_one_byte_then_eror has sent all but one bytes".to_string()),
614                ])),
615            )
616        }
617        .boxed()
618    }
619}
620
621/// Responder that yields the response up to the final byte, then disconnects.  Panics if the
622/// response contains an empty body.
623pub struct OneByteShortThenDisconnect;
624
625impl HttpResponder for OneByteShortThenDisconnect {
626    fn respond(
627        &self,
628        _: &Request<Body>,
629        response: Response<Body>,
630    ) -> BoxFuture<'_, Response<Body>> {
631        async {
632            let (parts, body) = response.into_parts();
633            let mut bytes = body_to_bytes(body).await;
634            if bytes.pop().is_none() {
635                panic!("can't short 0 bytes");
636            }
637            Response::from_parts(
638                parts,
639                Body::wrap_stream(futures::stream::iter(vec![Result::<Vec<u8>, String>::Ok(
640                    bytes,
641                )])),
642            )
643        }
644        .boxed()
645    }
646}
647
648/// Responder that flips the first byte of the response.  Panics if the response contains an empty
649/// body.
650pub struct OneByteFlipped;
651
652impl HttpResponder for OneByteFlipped {
653    fn respond(
654        &self,
655        _: &Request<Body>,
656        response: Response<Body>,
657    ) -> BoxFuture<'_, Response<Body>> {
658        async {
659            let (parts, body) = response.into_parts();
660            let mut bytes = body_to_bytes(body).await;
661            if bytes.is_empty() {
662                panic!("can't flip 0 bytes");
663            }
664            bytes[0] = !bytes[0];
665            Response::from_parts(parts, Body::from(bytes))
666        }
667        .boxed()
668    }
669}
670
671/// Responder that never sends bytes.
672pub struct Hang;
673
674impl HttpResponder for Hang {
675    fn respond(&self, _: &Request<Body>, _: Response<Body>) -> BoxFuture<'_, Response<Body>> {
676        pending().boxed()
677    }
678}
679
680/// Responder that sends the header but then never sends body bytes.
681pub struct HangBody;
682
683impl HttpResponder for HangBody {
684    fn respond(
685        &self,
686        _: &Request<Body>,
687        response: Response<Body>,
688    ) -> BoxFuture<'_, Response<Body>> {
689        async {
690            let (parts, _) = response.into_parts();
691            Response::from_parts(
692                parts,
693                Body::wrap_stream(futures::stream::pending::<Result<Vec<u8>, String>>()),
694            )
695        }
696        .boxed()
697    }
698}
699
700/// Responder that forwards to its wrapped responder once.
701pub struct Once<H: HttpResponder> {
702    already_forwarded: AtomicBool,
703    responder: H,
704}
705
706impl<H: HttpResponder> HttpResponder for Once<H> {
707    fn respond<'a>(
708        &'a self,
709        request: &'a Request<Body>,
710        response: Response<Body>,
711    ) -> BoxFuture<'_, Response<Body>> {
712        if self.already_forwarded.fetch_or(true, Ordering::SeqCst) {
713            ready(response).boxed()
714        } else {
715            self.responder.respond(request, response)
716        }
717    }
718}
719
720impl<H: HttpResponder> Once<H> {
721    /// Creates a responder that forwards to `responder` once.
722    pub fn new(responder: H) -> Self {
723        Self { already_forwarded: AtomicBool::new(false), responder }
724    }
725}
726
727/// Responder that forwards to its wrapped responder the nth time it is called.
728pub struct OverrideNth<H: HttpResponder> {
729    n: u32,
730    call_count: AtomicU32,
731    responder: H,
732}
733
734impl<H: HttpResponder> HttpResponder for OverrideNth<H> {
735    fn respond<'a>(
736        &'a self,
737        request: &'a Request<Body>,
738        response: Response<Body>,
739    ) -> BoxFuture<'_, Response<Body>> {
740        if self.call_count.fetch_add(1, Ordering::SeqCst) + 1 == self.n {
741            self.responder.respond(request, response)
742        } else {
743            ready(response).boxed()
744        }
745    }
746}
747
748impl<H: HttpResponder> OverrideNth<H> {
749    /// Creates a responder that forwards to `responder` on the nth call.
750    pub fn new(n: u32, responder: H) -> Self {
751        Self { n, call_count: AtomicU32::new(0), responder }
752    }
753}
754
755/// Information saved by Record for each request it handles.
756#[derive(Debug)]
757pub struct HistoryEntry {
758    uri_path: PathBuf,
759    headers: hyper::HeaderMap<hyper::header::HeaderValue>,
760}
761
762impl HistoryEntry {
763    /// The uri_path of the request.
764    pub fn uri_path(&self) -> &Path {
765        &self.uri_path
766    }
767
768    /// The request headers.
769    pub fn headers(&self) -> &http::HeaderMap<hyper::header::HeaderValue> {
770        &self.headers
771    }
772}
773
774/// The request history recorded by Record.
775pub struct History(Arc<Mutex<Vec<HistoryEntry>>>);
776
777impl History {
778    /// Take the recorded history, clearing it from the Record.
779    pub fn take(&self) -> Vec<HistoryEntry> {
780        std::mem::take(&mut self.0.lock())
781    }
782}
783
784/// Responder that records the requests.
785pub struct Record {
786    history: History,
787}
788
789impl Record {
790    /// Creates a responder that records all the requests.
791    pub fn new() -> (Self, History) {
792        let history = Arc::new(Mutex::new(vec![]));
793        (Self { history: History(Arc::clone(&history)) }, History(history))
794    }
795}
796
797impl HttpResponder for Record {
798    fn respond<'a>(
799        &'a self,
800        request: &'a Request<Body>,
801        response: Response<Body>,
802    ) -> BoxFuture<'_, Response<Body>> {
803        self.history.0.lock().push(HistoryEntry {
804            uri_path: request.path().to_owned(),
805            headers: request.headers().clone(),
806        });
807        ready(response).boxed()
808    }
809}
810
811/// Responder that forwards requests to its wrapped Responder if filter returns true.
812pub struct Filter<F: FilterFn, T: HttpResponder> {
813    filter: F,
814    handler: T,
815}
816
817/// Used by the Filter HttpResponder to decide which requests to forward and which to ignore.
818pub trait FilterFn: Send + Sync + 'static {
819    /// Return true iff Filter should forward the request to its wrapped Responder.
820    fn filter(&self, request: &Request<Body>) -> bool;
821}
822
823impl<F> FilterFn for F
824where
825    F: Fn(&Request<Body>) -> bool + Send + Sync + 'static,
826{
827    fn filter(&self, request: &Request<Body>) -> bool {
828        (self)(request)
829    }
830}
831
832/// Returns true iff the request has a Content-Range header.
833pub fn is_range_request(request: &Request<Body>) -> bool {
834    request.headers().get(http::header::RANGE).is_some()
835}
836
837impl<F: FilterFn, T: HttpResponder> Filter<F, T> {
838    /// Creates a responder that forwards requests that satisfy a filter.
839    pub fn new(filter: F, handler: T) -> Self {
840        Self { filter, handler }
841    }
842}
843
844impl<F: FilterFn, T: HttpResponder> HttpResponder for Filter<F, T> {
845    fn respond<'a>(
846        &'a self,
847        request: &'a Request<Body>,
848        response: Response<Body>,
849    ) -> BoxFuture<'_, Response<Body>> {
850        if self.filter.filter(request) {
851            self.handler.respond(request, response)
852        } else {
853            ready(response).boxed()
854        }
855    }
856}
857
858/// Responder that changes the status code to a given value.
859pub struct OverwriteStatusCode {
860    code: http::StatusCode,
861}
862
863impl OverwriteStatusCode {
864    /// Creates a responder that changes the status code to a given value.
865    pub fn new(code: http::StatusCode) -> Self {
866        Self { code }
867    }
868}
869
870impl HttpResponder for OverwriteStatusCode {
871    fn respond(
872        &self,
873        _: &Request<Body>,
874        mut response: Response<Body>,
875    ) -> BoxFuture<'_, Response<Body>> {
876        *response.status_mut() = self.code;
877        futures::future::ready(response).boxed()
878    }
879}
880
881/// Responder that calls each wrapped responder in order.
882pub struct Chain {
883    responders: Vec<Box<dyn HttpResponder>>,
884}
885
886impl Chain {
887    /// Creates a responder that calls each wrapped responder in order.
888    pub fn new(responders: Vec<Box<dyn HttpResponder>>) -> Self {
889        Self { responders }
890    }
891}
892
893impl HttpResponder for Chain {
894    fn respond<'a>(
895        &'a self,
896        request: &'a Request<Body>,
897        mut response: Response<Body>,
898    ) -> BoxFuture<'a, Response<Body>> {
899        async move {
900            for responder in self.responders.iter() {
901                response = responder.respond(request, response).await;
902            }
903            response
904        }
905        .boxed()
906    }
907}
908
909/// Fails all requests with NOT_FOUND.
910///
911/// All requests made to the first requested path are failed immediately.
912/// All requests to subsequent paths are blocked until the `unblocker` returned by
913///   new() is used, at which point all requests (pending and future) fail immediately.
914pub struct FailOneThenTemporarilyBlock {
915    path_to_fail: Arc<Mutex<Option<PathBuf>>>,
916    block_until: Shared<oneshot::Receiver<()>>,
917}
918
919impl FailOneThenTemporarilyBlock {
920    /// Create a FailOneThenTemporarilyBlock and its paired unblocker.
921    pub fn new() -> (Self, oneshot::Sender<()>) {
922        let (send, recv) = oneshot::channel();
923        (Self { path_to_fail: Arc::new(Mutex::new(None)), block_until: recv.shared() }, send)
924    }
925}
926
927impl HttpResponder for FailOneThenTemporarilyBlock {
928    fn respond(&self, request: &Request<Body>, _: Response<Body>) -> BoxFuture<'_, Response<Body>> {
929        let response =
930            Response::builder().status(StatusCode::NOT_FOUND).body(Body::empty()).unwrap();
931        match &mut *self.path_to_fail.lock() {
932            o @ None => {
933                *o = Some(request.path().to_owned());
934                ready(response).boxed()
935            }
936            Some(path_to_fail) if path_to_fail == request.path() => ready(response).boxed(),
937            _ => {
938                let block_until = self.block_until.clone();
939                async move {
940                    block_until.await.unwrap();
941                    response
942                }
943            }
944            .boxed(),
945        }
946    }
947}
fuchsia_pkg_testing/serve/responder.rs

fuchsia_pkg_testing/serve/
responder.rs
