rama_http/layer/traffic_writer/
response.rs

1use super::{
2    PerMessageFileWriter, TrafficWriterId, WriterMode, capture_body_channel,
3    ensure_traffic_writer_id, write_headers_body_flags,
4};
5use crate::io::write_http_response_streaming;
6use crate::{Body, Request, Response, StreamingBody, body::util::BodyExt as _};
7use rama_core::bytes::Bytes;
8use rama_core::error::{BoxError, ErrorContext as _};
9use rama_core::extensions::{Extension, ExtensionsRef};
10use rama_core::rt::Executor;
11use rama_core::telemetry::tracing::{self, Instrument};
12use rama_core::{Layer, Service};
13use rama_http_types::CaptureBody;
14use rama_utils::macros::define_inner_service_accessors;
15use std::{fmt::Debug, io, path::PathBuf, sync::Arc};
16use tokio::io::{AsyncWrite, AsyncWriteExt as _, stderr, stdout};
17use tokio::sync::mpsc::{Sender, UnboundedSender, channel, unbounded_channel};
18
19async fn write_response_entry<W>(
20    writer: &mut W,
21    response: Response,
22    write_headers: bool,
23    write_body: bool,
24) where
25    W: AsyncWrite + Unpin + Send + Sync + 'static,
26{
27    if let Err(err) =
28        write_http_response_streaming(writer, response, write_headers, write_body).await
29    {
30        tracing::error!("failed to write http response to writer: {err:?}")
31    }
32}
33
34macro_rules! drive_response_writer {
35    ($writer:ident, $rx:ident, $write_headers:ident, $write_body:ident) => {{
36        while let Some(response) = $rx.recv().await {
37            write_response_entry(&mut $writer, response, $write_headers, $write_body).await;
38        }
39        if let Err(err) = $writer.flush().await {
40            tracing::error!("failed to flush response writer: {err:?}")
41        }
42    }};
43}
44
45/// Layer that applies [`ResponseWriterService`] which prints the http response in std format.
46///
47/// Response bodies are observed while they stream to the caller. The writer
48/// receives an owned body stream and processes frames concurrently.
49#[derive(Clone)]
50pub struct ResponseWriterLayer<W> {
51    writer: W,
52    executor: Executor,
53}
54
55impl<W> Debug for ResponseWriterLayer<W> {
56    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
57        f.debug_struct("ResponseWriterLayer")
58            .field("writer", &format_args!("{}", std::any::type_name::<W>()))
59            .finish()
60    }
61}
62
63impl<W> ResponseWriterLayer<W> {
64    /// Create a new [`ResponseWriterLayer`] with a custom [`ResponseWriter`].
65    ///
66    /// Use [`Self::new_with_executor`] when streaming writer tasks must participate
67    /// in graceful shutdown.
68    pub const fn new(writer: W) -> Self {
69        Self::new_with_executor(writer, Executor::new())
70    }
71
72    /// Create a new [`ResponseWriterLayer`] with a custom [`ResponseWriter`]
73    /// and executor for streaming writer tasks.
74    pub const fn new_with_executor(writer: W, executor: Executor) -> Self {
75        Self { writer, executor }
76    }
77}
78
79/// A trait for writing http responses.
80pub trait ResponseWriter: Send + Sync + 'static {
81    /// Write the HTTP response while its body is streaming.
82    ///
83    /// Implementations must consume or deliberately drop the body while this
84    /// future runs. Retaining it for later applies backpressure once its
85    /// single-frame observational capture queue is full.
86    fn write_response(&self, res: Response) -> impl Future<Output = ()> + Send + '_;
87}
88
89/// Marker struct to indicate that the response should not be printed.
90#[derive(Debug, Clone, Default, Extension)]
91#[extension(tags(http))]
92#[non_exhaustive]
93pub struct DoNotWriteResponse;
94
95impl DoNotWriteResponse {
96    /// Create a new [`DoNotWriteResponse`] marker.
97    #[must_use]
98    pub const fn new() -> Self {
99        Self
100    }
101}
102
103impl ResponseWriterLayer<UnboundedSender<Response>> {
104    /// Create a new [`ResponseWriterLayer`] that prints responses to an [`AsyncWrite`]r
105    /// over an unbounded channel
106    pub fn writer_unbounded<W>(executor: &Executor, mut writer: W, mode: Option<WriterMode>) -> Self
107    where
108        W: AsyncWrite + Unpin + Send + Sync + 'static,
109    {
110        let (tx, mut rx) = unbounded_channel();
111        let (write_headers, write_body) = write_headers_body_flags(mode);
112
113        let span =
114            tracing::trace_root_span!("TrafficWriter::response::unbounded", otel.kind = "consumer");
115
116        executor.spawn_task(
117            async move {
118                drive_response_writer!(writer, rx, write_headers, write_body);
119            }
120            .instrument(span),
121        );
122
123        Self {
124            writer: tx,
125            executor: executor.clone(),
126        }
127    }
128
129    /// Create a new [`ResponseWriterLayer`] that prints responses to stdout
130    /// over an unbounded channel.
131    #[must_use]
132    pub fn stdout_unbounded(executor: &Executor, mode: Option<WriterMode>) -> Self {
133        Self::writer_unbounded(executor, stdout(), mode)
134    }
135
136    /// Create a new [`ResponseWriterLayer`] that prints responses to stderr
137    /// over an unbounded channel.
138    #[must_use]
139    pub fn stderr_unbounded(executor: &Executor, mode: Option<WriterMode>) -> Self {
140        Self::writer_unbounded(executor, stderr(), mode)
141    }
142}
143
144impl ResponseWriterLayer<PerMessageFileWriter> {
145    /// Create a layer that streams every response into a unique file below
146    /// `directory` using the portable filename `prefix`.
147    pub async fn file_per_response(
148        executor: &Executor,
149        directory: impl Into<PathBuf>,
150        prefix: impl AsRef<str>,
151        mode: Option<WriterMode>,
152    ) -> io::Result<Self> {
153        let writer = PerMessageFileWriter::try_new(directory, prefix)
154            .await?
155            .with_request_mode(None)
156            .with_response_mode(mode);
157        Ok(Self::new_with_executor(writer, executor.clone()))
158    }
159}
160
161impl ResponseWriterLayer<Sender<Response>> {
162    /// Create a new [`ResponseWriterLayer`] that prints responses to an [`AsyncWrite`]r
163    /// over a bounded channel with a fixed buffer size.
164    pub fn writer<W>(
165        executor: &Executor,
166        mut writer: W,
167        buffer_size: usize,
168        mode: Option<WriterMode>,
169    ) -> Self
170    where
171        W: AsyncWrite + Unpin + Send + Sync + 'static,
172    {
173        let (tx, mut rx) = channel(buffer_size);
174        let (write_headers, write_body) = write_headers_body_flags(mode);
175
176        let span =
177            tracing::trace_root_span!("TrafficWriter::response::bounded", otel.kind = "consumer");
178
179        executor.spawn_task(
180            async move {
181                drive_response_writer!(writer, rx, write_headers, write_body);
182            }
183            .instrument(span),
184        );
185        Self {
186            writer: tx,
187            executor: executor.clone(),
188        }
189    }
190
191    /// Create a new [`ResponseWriterLayer`] that prints responses to stdout
192    /// over a bounded channel with a fixed buffer size.
193    #[must_use]
194    pub fn stdout(executor: &Executor, buffer_size: usize, mode: Option<WriterMode>) -> Self {
195        Self::writer(executor, stdout(), buffer_size, mode)
196    }
197
198    /// Create a new [`ResponseWriterLayer`] that prints responses to stderr
199    /// over a bounded channel with a fixed buffer size.
200    #[must_use]
201    pub fn stderr(executor: &Executor, buffer_size: usize, mode: Option<WriterMode>) -> Self {
202        Self::writer(executor, stderr(), buffer_size, mode)
203    }
204}
205
206impl<S, W: Clone> Layer<S> for ResponseWriterLayer<W> {
207    type Service = ResponseWriterService<S, W>;
208
209    fn layer(&self, inner: S) -> Self::Service {
210        ResponseWriterService {
211            inner,
212            writer: Arc::new(self.writer.clone()),
213            executor: self.executor.clone(),
214        }
215    }
216
217    fn into_layer(self, inner: S) -> Self::Service {
218        ResponseWriterService {
219            inner,
220            writer: Arc::new(self.writer),
221            executor: self.executor,
222        }
223    }
224}
225
226/// Middleware to print Http responses in std format.
227///
228/// Response bodies are observed while they stream to the caller. The writer
229/// receives an owned body stream and processes frames concurrently.
230///
231/// See the [module docs](super) for more details.
232#[derive(Clone)]
233pub struct ResponseWriterService<S, W> {
234    inner: S,
235    writer: Arc<W>,
236    executor: Executor,
237}
238
239impl<S, W> ResponseWriterService<S, W> {
240    /// Create a new [`ResponseWriterService`] with a custom [`ResponseWriter`].
241    ///
242    /// Use [`Self::new_with_executor`] when streaming writer tasks must participate
243    /// in graceful shutdown.
244    pub fn new(writer: W, inner: S) -> Self {
245        Self::new_with_executor(writer, inner, Executor::new())
246    }
247
248    /// Create a new [`ResponseWriterService`] with a custom [`ResponseWriter`]
249    /// and executor for streaming writer tasks.
250    pub fn new_with_executor(writer: W, inner: S, executor: Executor) -> Self {
251        Self {
252            inner,
253            writer: Arc::new(writer),
254            executor,
255        }
256    }
257
258    define_inner_service_accessors!();
259}
260
261impl<S: Debug, W> Debug for ResponseWriterService<S, W> {
262    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
263        f.debug_struct("ResponseWriterService")
264            .field("inner", &self.inner)
265            .field("writer", &format_args!("{}", std::any::type_name::<W>()))
266            .finish()
267    }
268}
269
270impl<S> ResponseWriterService<S, UnboundedSender<Response>> {
271    /// Create a new [`ResponseWriterService`] that prints responses to an [`AsyncWrite`]r
272    /// over an unbounded channel
273    pub fn writer_unbounded<W>(
274        executor: &Executor,
275        writer: W,
276        mode: Option<WriterMode>,
277        inner: S,
278    ) -> Self
279    where
280        W: AsyncWrite + Unpin + Send + Sync + 'static,
281    {
282        let layer = ResponseWriterLayer::writer_unbounded(executor, writer, mode);
283        layer.into_layer(inner)
284    }
285
286    /// Create a new [`ResponseWriterService`] that prints responses to stdout
287    /// over an unbounded channel.
288    pub fn stdout_unbounded(executor: &Executor, mode: Option<WriterMode>, inner: S) -> Self {
289        Self::writer_unbounded(executor, stdout(), mode, inner)
290    }
291
292    /// Create a new [`ResponseWriterService`] that prints responses to stderr
293    /// over an unbounded channel.
294    pub fn stderr_unbounded(executor: &Executor, mode: Option<WriterMode>, inner: S) -> Self {
295        Self::writer_unbounded(executor, stderr(), mode, inner)
296    }
297}
298
299impl<S> ResponseWriterService<S, PerMessageFileWriter> {
300    /// Create a service that streams every response into a unique file below
301    /// `directory` using the portable filename `prefix`.
302    pub async fn file_per_response(
303        executor: &Executor,
304        directory: impl Into<PathBuf>,
305        prefix: impl AsRef<str>,
306        mode: Option<WriterMode>,
307        inner: S,
308    ) -> io::Result<Self> {
309        let layer =
310            ResponseWriterLayer::file_per_response(executor, directory, prefix, mode).await?;
311        Ok(layer.into_layer(inner))
312    }
313}
314
315impl<S> ResponseWriterService<S, Sender<Response>> {
316    /// Create a new [`ResponseWriterService`] that prints responses to an [`AsyncWrite`]r
317    /// over a bounded channel with a fixed buffer size.
318    pub fn writer<W>(
319        executor: &Executor,
320        writer: W,
321        buffer_size: usize,
322        mode: Option<WriterMode>,
323        inner: S,
324    ) -> Self
325    where
326        W: AsyncWrite + Unpin + Send + Sync + 'static,
327    {
328        let layer = ResponseWriterLayer::writer(executor, writer, buffer_size, mode);
329        layer.into_layer(inner)
330    }
331
332    /// Create a new [`ResponseWriterService`] that prints responses to stdout
333    /// over a bounded channel with a fixed buffer size.
334    pub fn stdout(
335        executor: &Executor,
336        buffer_size: usize,
337        mode: Option<WriterMode>,
338        inner: S,
339    ) -> Self {
340        Self::writer(executor, stdout(), buffer_size, mode, inner)
341    }
342
343    /// Create a new [`ResponseWriterService`] that prints responses to stderr
344    /// over a bounded channel with a fixed buffer size.
345    pub fn stderr(
346        executor: &Executor,
347        buffer_size: usize,
348        mode: Option<WriterMode>,
349        inner: S,
350    ) -> Self {
351        Self::writer(executor, stderr(), buffer_size, mode, inner)
352    }
353}
354
355impl<S, W> ResponseWriterService<S, W> {}
356
357impl<S, W, ReqBody, ResBody> Service<Request<ReqBody>> for ResponseWriterService<S, W>
358where
359    S: Service<Request<ReqBody>, Output = Response<ResBody>, Error: Into<BoxError>>,
360    W: ResponseWriter,
361    ReqBody: Send + 'static,
362    ResBody: StreamingBody<Data = Bytes, Error: Into<BoxError>> + Send + Sync + 'static,
363{
364    type Output = Response;
365    type Error = BoxError;
366
367    async fn serve(&self, req: Request<ReqBody>) -> Result<Self::Output, Self::Error> {
368        let do_not_print_response: Option<Arc<DoNotWriteResponse>> = req.extensions().get_arc();
369        let traffic_writer_id = do_not_print_response
370            .is_none()
371            .then(|| ensure_traffic_writer_id(req.extensions()));
372        let resp = self.inner.serve(req).await.into_box_error()?;
373        let resp = if let Some(traffic_writer_id) = traffic_writer_id {
374            if resp.extensions().get_ref::<TrafficWriterId>().copied() != Some(traffic_writer_id) {
375                resp.extensions().insert(traffic_writer_id);
376            }
377            let (parts, body) = resp.into_parts();
378            let (capture, captured_body) = capture_body_channel();
379            let captured_parts = parts.clone();
380            let writer = Arc::clone(&self.writer);
381            self.executor.spawn_task(async move {
382                writer
383                    .write_response(Response::from_parts(captured_parts, captured_body))
384                    .await;
385            });
386            Response::from_parts(
387                parts,
388                Body::new(CaptureBody::new(body.map_err(Into::into), capture)),
389            )
390        } else {
391            resp.map(Body::new)
392        };
393        Ok(resp)
394    }
395}
396
397impl ResponseWriter for Sender<Response> {
398    async fn write_response(&self, res: Response) {
399        if let Err(err) = self.send(res).await {
400            tracing::error!("failed to send response to channel: {err:?}")
401        }
402    }
403}
404
405impl ResponseWriter for UnboundedSender<Response> {
406    async fn write_response(&self, res: Response) {
407        if let Err(err) = self.send(res) {
408            tracing::error!("failed to send response to unbounded channel: {err:?}")
409        }
410    }
411}
412
413impl<F, Fut> ResponseWriter for F
414where
415    F: Fn(Response) -> Fut + Send + Sync + 'static,
416    Fut: Future<Output = ()> + Send + 'static,
417{
418    async fn write_response(&self, res: Response) {
419        self(res).await
420    }
421}
422
423#[cfg(test)]
424mod tests {
425    use std::{
426        convert::Infallible,
427        sync::{
428            Arc,
429            atomic::{AtomicUsize, Ordering},
430        },
431        time::Duration,
432    };
433
434    use rama_core::{
435        Layer as _, Service as _, bytes::Bytes, extensions::ExtensionsRef as _, rt::Executor,
436        service::service_fn,
437    };
438    use tokio::io::AsyncReadExt as _;
439
440    use super::{ResponseWriterLayer, ResponseWriterService};
441    use crate::layer::traffic_writer::{TrafficWriterId, WriterMode};
442    use crate::{Body, Request, Response, body::util::BodyExt as _};
443
444    #[tokio::test]
445    async fn shared_response_writer_flushes_when_channel_closes() {
446        let executor = Executor::new();
447        let (output, mut captured) = tokio::io::duplex(128);
448        let layer = ResponseWriterLayer::writer_unbounded(
449            &executor,
450            tokio::io::BufWriter::new(output),
451            Some(WriterMode::All),
452        );
453        let sender = layer.writer.clone();
454        drop(layer);
455
456        sender.send(Response::new(Body::empty())).unwrap();
457        drop(sender);
458
459        let mut bytes = Vec::new();
460        tokio::time::timeout(Duration::from_secs(1), captured.read_to_end(&mut bytes))
461            .await
462            .expect("response writer should flush and close")
463            .unwrap();
464        assert_eq!(bytes, b"HTTP/1.1 200 OK\r\n\r\n");
465    }
466
467    #[tokio::test]
468    async fn file_constructors_only_enable_response_capture() {
469        let temp = tempfile::tempdir().unwrap();
470        let executor = Executor::new();
471        let layer = ResponseWriterLayer::file_per_response(
472            &executor,
473            temp.path(),
474            "responses",
475            Some(WriterMode::Headers),
476        )
477        .await
478        .unwrap();
479        assert_eq!(layer.writer.request_mode(), None);
480        assert_eq!(layer.writer.response_mode(), Some(WriterMode::Headers));
481
482        let service = ResponseWriterService::file_per_response(
483            &executor,
484            temp.path(),
485            "responses",
486            Some(WriterMode::Body),
487            (),
488        )
489        .await
490        .unwrap();
491        assert_eq!(service.writer.request_mode(), None);
492        assert_eq!(service.writer.response_mode(), Some(WriterMode::Body));
493    }
494
495    #[tokio::test]
496    async fn response_writer_propagates_private_request_id_to_both_responses() {
497        let (writer, mut captured_responses) = tokio::sync::mpsc::unbounded_channel();
498        let (seen_id, mut seen_ids) = tokio::sync::mpsc::unbounded_channel();
499        let inner = service_fn(move |request: Request| {
500            let seen_id = seen_id.clone();
501            async move {
502                seen_id
503                    .send(*request.extensions().get_ref::<TrafficWriterId>().unwrap())
504                    .unwrap();
505                Ok::<_, Infallible>(Response::new(Body::empty()))
506            }
507        });
508        let service = ResponseWriterLayer::new(writer).into_layer(inner);
509
510        let response = service.serve(Request::new(Body::empty())).await.unwrap();
511        let captured = tokio::time::timeout(Duration::from_secs(1), captured_responses.recv())
512            .await
513            .unwrap()
514            .unwrap();
515        let request_id = tokio::time::timeout(Duration::from_secs(1), seen_ids.recv())
516            .await
517            .unwrap()
518            .unwrap();
519        assert_eq!(
520            response.extensions().get_ref::<TrafficWriterId>(),
521            Some(&request_id)
522        );
523        assert_eq!(
524            captured.extensions().get_ref::<TrafficWriterId>(),
525            Some(&request_id)
526        );
527    }
528
529    #[tokio::test]
530    async fn writer_observes_response_as_the_caller_streams_it() {
531        let polls = Arc::new(AtomicUsize::new(0));
532        let response_polls = Arc::clone(&polls);
533        let (writer, mut written) = tokio::sync::mpsc::unbounded_channel();
534        let service =
535            ResponseWriterLayer::new(writer).into_layer(service_fn(move |_request: Request| {
536                let polls = Arc::clone(&response_polls);
537                async move {
538                    let body = Body::from_stream(rama_core::futures::stream::once(async move {
539                        polls.fetch_add(1, Ordering::Relaxed);
540                        Ok::<_, Infallible>(Bytes::from_static(b"streamed response"))
541                    }));
542                    Ok::<_, Infallible>(Response::new(body))
543                }
544            }));
545
546        let response = service.serve(Request::new(Body::empty())).await.unwrap();
547        assert_eq!(polls.load(Ordering::Relaxed), 0);
548        written.try_recv().unwrap_err();
549
550        let written_response =
551            tokio::time::timeout(std::time::Duration::from_secs(1), written.recv())
552                .await
553                .expect("response capture should arrive promptly")
554                .unwrap();
555        let (response, captured) = tokio::time::timeout(std::time::Duration::from_secs(1), async {
556            tokio::join!(
557                response.into_body().collect(),
558                written_response.into_body().collect(),
559            )
560        })
561        .await
562        .expect("response streams should finish promptly");
563        assert_eq!(
564            response.unwrap().to_bytes(),
565            Bytes::from_static(b"streamed response")
566        );
567        assert_eq!(polls.load(Ordering::Relaxed), 1);
568        assert_eq!(
569            captured.unwrap().to_bytes(),
570            Bytes::from_static(b"streamed response")
571        );
572    }
573
574    #[tokio::test]
575    async fn shared_writer_backpressures_later_response_without_interleaving() {
576        let (first_sender, first_receiver) = tokio::sync::mpsc::unbounded_channel();
577        let first_receiver = Arc::new(tokio::sync::Mutex::new(Some(first_receiver)));
578        let inner_receiver = Arc::clone(&first_receiver);
579        let inner = service_fn(move |request: Request| {
580            let first_receiver = Arc::clone(&inner_receiver);
581            async move {
582                let body = if request.headers().contains_key("x-long-lived") {
583                    let receiver = first_receiver
584                        .lock()
585                        .await
586                        .take()
587                        .expect("the long-lived response is requested once");
588                    Body::from_stream(rama_core::futures::stream::unfold(
589                        receiver,
590                        |mut receiver| async move { receiver.recv().await.map(|item| (item, receiver)) },
591                    ))
592                } else {
593                    Body::from("second")
594                };
595                Ok::<_, Infallible>(Response::new(body))
596            }
597        });
598        let executor = Executor::new();
599        let (writer, mut output) = tokio::io::duplex(64);
600        let service =
601            ResponseWriterLayer::writer_unbounded(&executor, writer, Some(WriterMode::Body))
602                .into_layer(inner);
603
604        let first_request = Request::builder()
605            .header("x-long-lived", "true")
606            .body(Body::empty())
607            .unwrap();
608        let first_response = service.serve(first_request).await.unwrap();
609        let mut first_body = first_response.into_body();
610        first_sender
611            .send(Ok::<_, Infallible>(Bytes::from_static(b"first")))
612            .unwrap();
613        assert_eq!(
614            first_body
615                .frame()
616                .await
617                .unwrap()
618                .unwrap()
619                .into_data()
620                .unwrap(),
621            "first"
622        );
623
624        let mut first_output = [0; 7];
625        tokio::time::timeout(Duration::from_secs(1), output.read_exact(&mut first_output))
626            .await
627            .expect("the writer should start draining the first response")
628            .unwrap();
629        assert_eq!(&first_output, b"\r\nfirst");
630
631        let second_response = service.serve(Request::new(Body::empty())).await.unwrap();
632        let mut second = Box::pin(second_response.into_body().collect());
633        tokio::time::timeout(Duration::from_millis(50), &mut second)
634            .await
635            .expect_err("the shared writer must backpressure a later body");
636
637        drop(first_body);
638        drop(first_sender);
639        let second = tokio::time::timeout(Duration::from_secs(1), second)
640            .await
641            .expect("the later body should resume when the first capture ends")
642            .unwrap();
643        assert_eq!(second.to_bytes(), Bytes::from_static(b"second"));
644
645        let mut second_output = [0; 8];
646        tokio::time::timeout(
647            Duration::from_secs(1),
648            output.read_exact(&mut second_output),
649        )
650        .await
651        .expect("the second response should be written after the first")
652        .unwrap();
653        assert_eq!(&second_output, b"\r\nsecond");
654    }
655
656    #[tokio::test]
657    async fn headers_only_writer_does_not_consume_or_stall_live_body() {
658        let inner = service_fn(|_request: Request| async {
659            Ok::<_, Infallible>(
660                Response::builder()
661                    .header("x-written", "yes")
662                    .body(Body::from("live payload"))
663                    .unwrap(),
664            )
665        });
666        let executor = Executor::new();
667        let (writer, mut output) = tokio::io::duplex(128);
668        let service =
669            ResponseWriterLayer::writer_unbounded(&executor, writer, Some(WriterMode::Headers))
670                .into_layer(inner);
671
672        let response = service.serve(Request::new(Body::empty())).await.unwrap();
673        assert_eq!(
674            response.into_body().collect().await.unwrap().to_bytes(),
675            Bytes::from_static(b"live payload")
676        );
677        drop(service);
678
679        let mut written = Vec::new();
680        tokio::time::timeout(Duration::from_secs(1), output.read_to_end(&mut written))
681            .await
682            .expect("the headers-only writer should shut down promptly")
683            .unwrap();
684        let written = String::from_utf8(written).unwrap();
685        assert!(written.contains("HTTP/1.1 200 OK\r\n"));
686        assert!(written.contains("x-written: yes\r\n"));
687        assert!(!written.contains("live payload"));
688    }
689}
rama_http/layer/traffic_writer/response.rs

rama_http/layer/traffic_writer/
response.rs
