logged_stream/
stream.rs

1use crate::ChannelLogger;
2use crate::MemoryStorageLogger;
3use crate::RecordFilter;
4use crate::buffer_formatter::BufferFormatter;
5use crate::logger::Logger;
6use crate::record::Record;
7use crate::record::RecordKind;
8use std::collections;
9use std::fmt;
10use std::io;
11use std::pin::Pin;
12use std::sync::mpsc;
13use std::task::Context;
14use std::task::Poll;
15use tokio::io as tokio_io;
16
17/// Wrapper for an IO object that logs every read, write, error, shutdown and drop that passes
18/// through it.
19///
20/// [`LoggedStream`] wraps an underlying IO object implementing the [`Read`] / [`Write`] traits, or
21/// their asynchronous [`tokio`] analogues [`AsyncRead`] / [`AsyncWrite`], and logs all read and
22/// write operations, errors, shutdowns and drops. It re-implements the same IO trait it wraps, so
23/// it is a drop-in replacement that works transparently in both synchronous and asynchronous code.
24///
25/// # Architecture
26///
27/// [`LoggedStream`] is generic over four independent, pluggable parts. Each logged event flows
28/// through them in order: `event -> Formatter -> Filter -> Logger`.
29///
30/// -   **The inner IO object (`S`).** The stream you are wrapping. [`LoggedStream`] implements the
31///     same IO trait `S` does, so it slots in wherever `S` was used.
32/// -   **Formatter ([`BufferFormatter`]).** Turns the read and written byte buffers into the display
33///     strings you see in the log.
34/// -   **Filter ([`RecordFilter`]).** Decides which records are logged. It runs on every record kind,
35///     including shutdown and drop.
36/// -   **Logger ([`Logger`]).** The sink that consumes accepted records.
37///
38/// All three of [`BufferFormatter`], [`RecordFilter`] and [`Logger`] are public, `Send + 'static`
39/// and object-safe, with blanket implementations for `Box<...>` (and `Arc<T>` where `T: Sync` for
40/// [`BufferFormatter`]). You are free to supply your own implementation of any part.
41///
42/// # Provided implementations
43///
44/// ## Formatters ([`BufferFormatter`])
45///
46/// Control how byte buffers are rendered. Each formatter stores a separator (default `:`) and
47/// exposes parallel constructors: `new`, `new_static`, `new_owned` and `new_default`.
48///
49/// | Formatter | Renders each byte as |
50/// | --- | --- |
51/// | [`LowercaseHexadecimalFormatter`] | lowercase hexadecimal — `0a:ff` |
52/// | [`UppercaseHexadecimalFormatter`] | uppercase hexadecimal — `0A:FF` |
53/// | [`DecimalFormatter`] | decimal — `10:255` |
54/// | [`OctalFormatter`] | octal — `012:377` |
55/// | [`BinaryFormatter`] | binary — `00001010:11111111` |
56///
57/// ## Filters ([`RecordFilter`])
58///
59/// Decide which records reach the logger.
60///
61/// | Filter | Behavior |
62/// | --- | --- |
63/// | [`DefaultFilter`] | Accepts every record. |
64/// | [`RecordKindFilter`] | Accepts only the record kinds in an allow-list given at construction. |
65/// | [`AllFilter`] | AND — a record passes only if every child filter accepts it (an empty list accepts everything). |
66/// | [`AnyFilter`] | OR — a record passes if any child filter accepts it (an empty list rejects everything). |
67///
68/// ## Loggers ([`Logger`])
69///
70/// Consume each accepted record.
71///
72/// | Logger | Destination |
73/// | --- | --- |
74/// | [`ConsoleLogger`] | Emits records through the `log` facade. |
75/// | [`FileLogger`] | Writes records to a file. |
76/// | [`MemoryStorageLogger`] | Retains recent records in a bounded in-memory buffer. |
77/// | [`ChannelLogger`] | Sends records over an `mpsc` channel for handling elsewhere. |
78///
79/// [`Read`]: io::Read
80/// [`Write`]: io::Write
81/// [`AsyncRead`]: tokio::io::AsyncRead
82/// [`AsyncWrite`]: tokio::io::AsyncWrite
83/// [`LowercaseHexadecimalFormatter`]: crate::LowercaseHexadecimalFormatter
84/// [`UppercaseHexadecimalFormatter`]: crate::UppercaseHexadecimalFormatter
85/// [`DecimalFormatter`]: crate::DecimalFormatter
86/// [`BinaryFormatter`]: crate::BinaryFormatter
87/// [`OctalFormatter`]: crate::OctalFormatter
88/// [`DefaultFilter`]: crate::DefaultFilter
89/// [`RecordKindFilter`]: crate::RecordKindFilter
90/// [`AllFilter`]: crate::AllFilter
91/// [`AnyFilter`]: crate::AnyFilter
92/// [`ConsoleLogger`]: crate::ConsoleLogger
93/// [`FileLogger`]: crate::FileLogger
94pub struct LoggedStream<
95    S: 'static,
96    Formatter: 'static,
97    Filter: RecordFilter + 'static,
98    L: Logger + 'static,
99> {
100    inner_stream: S,
101    formatter: Formatter,
102    filter: Filter,
103    logger: L,
104}
105
106impl<S: 'static, Formatter: 'static, Filter: RecordFilter + 'static, L: Logger + 'static>
107    LoggedStream<S, Formatter, Filter, L>
108{
109    /// Construct a new instance of [`LoggedStream`] using provided arguments.
110    pub fn new(stream: S, formatter: Formatter, filter: Filter, logger: L) -> Self {
111        Self {
112            inner_stream: stream,
113            formatter,
114            filter,
115            logger,
116        }
117    }
118
119    /// Emit a custom [`RecordKind::Open`] record carrying `message`.
120    ///
121    /// [`RecordKind::Open`] is never produced automatically by the read, write, shutdown and drop
122    /// machinery — this method is the way to emit one. Use it to annotate the start of a stream,
123    /// for example to record the peer of a freshly established connection
124    /// (`"Established connection with 127.0.0.1:8080"`) or other per-stream metadata.
125    ///
126    /// Like every other record, the `Open` record is passed through the filter before it reaches
127    /// the logger, so a `RecordKindFilter` that does not allow `Open` will suppress it. The message
128    /// is logged verbatim; it is not run through the formatter, which only applies to byte buffers.
129    ///
130    /// For asynchronous streams, call this before splitting the wrapper with `tokio::io::split`,
131    /// since the resulting halves do not expose it.
132    pub fn log_open(&mut self, message: impl Into<String>) {
133        self.emit(Record::new(RecordKind::Open, message.into()));
134    }
135
136    /// Route a record through the filter, logging it only if the filter accepts it.
137    ///
138    /// Every record — reads, writes, errors, shutdowns, drops and manual `Open` markers — is
139    /// emitted through this single method, so the filter is applied consistently to all of them.
140    #[inline]
141    fn emit(&mut self, record: Record) {
142        if self.filter.check(&record) {
143            self.logger.log(record);
144        }
145    }
146}
147
148impl<S: 'static, Formatter: 'static, Filter: RecordFilter + 'static>
149    LoggedStream<S, Formatter, Filter, MemoryStorageLogger>
150{
151    #[inline]
152    pub fn get_log_records(&self) -> collections::VecDeque<Record> {
153        self.logger.get_log_records()
154    }
155
156    #[inline]
157    pub fn clear_log_records(&mut self) {
158        self.logger.clear_log_records()
159    }
160}
161
162impl<S: 'static, Formatter: 'static, Filter: RecordFilter + 'static>
163    LoggedStream<S, Formatter, Filter, ChannelLogger>
164{
165    #[inline]
166    pub fn take_receiver(&mut self) -> Option<mpsc::Receiver<Record>> {
167        self.logger.take_receiver()
168    }
169
170    #[inline]
171    pub fn take_receiver_unchecked(&mut self) -> mpsc::Receiver<Record> {
172        self.logger.take_receiver_unchecked()
173    }
174}
175
176impl<
177    S: fmt::Debug + 'static,
178    Formatter: fmt::Debug + 'static,
179    Filter: RecordFilter + fmt::Debug + 'static,
180    L: Logger + fmt::Debug + 'static,
181> fmt::Debug for LoggedStream<S, Formatter, Filter, L>
182{
183    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
184        f.debug_struct("LoggedStream")
185            .field("inner_stream", &self.inner_stream)
186            .field("formatter", &self.formatter)
187            .field("filter", &self.filter)
188            .field("logger", &self.logger)
189            .finish()
190    }
191}
192
193impl<
194    S: io::Read + 'static,
195    Formatter: BufferFormatter + 'static,
196    Filter: RecordFilter + 'static,
197    L: Logger + 'static,
198> io::Read for LoggedStream<S, Formatter, Filter, L>
199{
200    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
201        let result = self.inner_stream.read(buf);
202
203        match &result {
204            Ok(length) => {
205                let record = Record::new(
206                    RecordKind::Read,
207                    self.formatter.format_buffer(&buf[0..*length]),
208                );
209                self.emit(record);
210            }
211            Err(e) if matches!(e.kind(), io::ErrorKind::WouldBlock) => {}
212            Err(e) => {
213                self.emit(Record::new(
214                    RecordKind::Error,
215                    format!("Error during read: {e}"),
216                ));
217            }
218        };
219
220        result
221    }
222}
223
224impl<
225    S: tokio_io::AsyncRead + Unpin + 'static,
226    Formatter: BufferFormatter + Unpin + 'static,
227    Filter: RecordFilter + Unpin + 'static,
228    L: Logger + Unpin + 'static,
229> tokio_io::AsyncRead for LoggedStream<S, Formatter, Filter, L>
230{
231    fn poll_read(
232        self: Pin<&mut Self>,
233        cx: &mut Context<'_>,
234        buf: &mut tokio_io::ReadBuf<'_>,
235    ) -> Poll<io::Result<()>> {
236        let mut_self = self.get_mut();
237        let length_before_read = buf.filled().len();
238        let result = Pin::new(&mut mut_self.inner_stream).poll_read(cx, buf);
239        let length_after_read = buf.filled().len();
240        let diff = length_after_read - length_before_read;
241
242        match &result {
243            Poll::Ready(Ok(())) if diff == 0 => {}
244            Poll::Ready(Ok(())) => {
245                let record = Record::new(
246                    RecordKind::Read,
247                    mut_self
248                        .formatter
249                        .format_buffer(&(buf.filled())[length_before_read..length_after_read]),
250                );
251                mut_self.emit(record);
252            }
253            Poll::Ready(Err(e)) => {
254                mut_self.emit(Record::new(
255                    RecordKind::Error,
256                    format!("Error during async read: {e}"),
257                ));
258            }
259            Poll::Pending => {}
260        }
261
262        result
263    }
264}
265
266impl<
267    S: io::Write + 'static,
268    Formatter: BufferFormatter + 'static,
269    Filter: RecordFilter + 'static,
270    L: Logger + 'static,
271> io::Write for LoggedStream<S, Formatter, Filter, L>
272{
273    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
274        let result = self.inner_stream.write(buf);
275
276        match &result {
277            Ok(length) => {
278                let record = Record::new(
279                    RecordKind::Write,
280                    self.formatter.format_buffer(&buf[0..*length]),
281                );
282                self.emit(record);
283            }
284            Err(e)
285                if matches!(
286                    e.kind(),
287                    io::ErrorKind::WriteZero | io::ErrorKind::WouldBlock
288                ) => {}
289            Err(e) => {
290                self.emit(Record::new(
291                    RecordKind::Error,
292                    format!("Error during write: {e}"),
293                ));
294            }
295        };
296
297        result
298    }
299
300    fn flush(&mut self) -> io::Result<()> {
301        self.inner_stream.flush()
302    }
303}
304
305impl<
306    S: tokio_io::AsyncWrite + Unpin + 'static,
307    Formatter: BufferFormatter + Unpin + 'static,
308    Filter: RecordFilter + Unpin + 'static,
309    L: Logger + Unpin + 'static,
310> tokio_io::AsyncWrite for LoggedStream<S, Formatter, Filter, L>
311{
312    fn poll_write(
313        self: Pin<&mut Self>,
314        cx: &mut Context<'_>,
315        buf: &[u8],
316    ) -> Poll<Result<usize, io::Error>> {
317        let mut_self = self.get_mut();
318        let result = Pin::new(&mut mut_self.inner_stream).poll_write(cx, buf);
319
320        match &result {
321            Poll::Ready(Ok(length)) => {
322                let record = Record::new(
323                    RecordKind::Write,
324                    mut_self.formatter.format_buffer(&buf[0..*length]),
325                );
326                mut_self.emit(record);
327            }
328            Poll::Ready(Err(e)) => {
329                mut_self.emit(Record::new(
330                    RecordKind::Error,
331                    format!("Error during async write: {e}"),
332                ));
333            }
334            Poll::Pending => {}
335        }
336
337        result
338    }
339
340    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
341        Pin::new(&mut self.get_mut().inner_stream).poll_flush(cx)
342    }
343
344    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
345        let mut_self = self.get_mut();
346        let result = Pin::new(&mut mut_self.inner_stream).poll_shutdown(cx);
347
348        mut_self.emit(Record::new(
349            RecordKind::Shutdown,
350            String::from("Writer shutdown request."),
351        ));
352
353        result
354    }
355}
356
357impl<S: 'static, Formatter: 'static, Filter: RecordFilter + 'static, L: Logger + 'static> Drop
358    for LoggedStream<S, Formatter, Filter, L>
359{
360    fn drop(&mut self) {
361        self.emit(Record::new(RecordKind::Drop, String::from("Deallocated.")));
362    }
363}
364
365#[cfg(test)]
366mod tests {
367    use crate::ChannelLogger;
368    use crate::DecimalFormatter;
369    use crate::DefaultFilter;
370    use crate::LoggedStream;
371    use crate::LowercaseHexadecimalFormatter;
372    use crate::MemoryStorageLogger;
373    use crate::RecordKind;
374    use crate::RecordKindFilter;
375    use crate::UppercaseHexadecimalFormatter;
376    use std::cell::RefCell;
377    use std::io::Cursor;
378    use std::io::ErrorKind;
379    use std::io::Read;
380    use std::io::Write;
381    use std::pin::Pin;
382    use std::rc::Rc;
383    use std::task::Context;
384    use std::task::Poll;
385
386    //////////////////////////////////////////////////////////////////////////////////////////////////////////
387    // Test doubles
388    //////////////////////////////////////////////////////////////////////////////////////////////////////////
389
390    /// A synchronous reader whose every `read` fails with the given [`ErrorKind`].
391    struct ErrReader(ErrorKind);
392
393    impl Read for ErrReader {
394        fn read(&mut self, _buf: &mut [u8]) -> std::io::Result<usize> {
395            Err(std::io::Error::new(self.0, "boom"))
396        }
397    }
398
399    /// A synchronous writer whose every `write` fails with the given [`ErrorKind`].
400    struct ErrWriter(ErrorKind);
401
402    impl Write for ErrWriter {
403        fn write(&mut self, _buf: &[u8]) -> std::io::Result<usize> {
404            Err(std::io::Error::new(self.0, "boom"))
405        }
406
407        fn flush(&mut self) -> std::io::Result<()> {
408            Ok(())
409        }
410    }
411
412    /// An asynchronous reader whose every `poll_read` fails with the given [`ErrorKind`].
413    struct ErrAsyncReader(ErrorKind);
414
415    impl tokio::io::AsyncRead for ErrAsyncReader {
416        fn poll_read(
417            self: Pin<&mut Self>,
418            _cx: &mut Context<'_>,
419            _buf: &mut tokio::io::ReadBuf<'_>,
420        ) -> Poll<std::io::Result<()>> {
421            Poll::Ready(Err(std::io::Error::new(self.0, "boom")))
422        }
423    }
424
425    /// An asynchronous writer whose every `poll_write` fails with the given [`ErrorKind`].
426    struct ErrAsyncWriter(ErrorKind);
427
428    impl tokio::io::AsyncWrite for ErrAsyncWriter {
429        fn poll_write(
430            self: Pin<&mut Self>,
431            _cx: &mut Context<'_>,
432            _buf: &[u8],
433        ) -> Poll<std::io::Result<usize>> {
434            Poll::Ready(Err(std::io::Error::new(self.0, "boom")))
435        }
436
437        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
438            Poll::Ready(Ok(()))
439        }
440
441        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
442            Poll::Ready(Ok(()))
443        }
444    }
445
446    /// A synchronous writer that appends everything written to it into a shared buffer, so a test
447    /// can assert that bytes actually reach the wrapped stream.
448    struct RecordingWriter(Rc<RefCell<Vec<u8>>>);
449
450    impl Write for RecordingWriter {
451        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
452            self.0.borrow_mut().extend_from_slice(buf);
453            Ok(buf.len())
454        }
455
456        fn flush(&mut self) -> std::io::Result<()> {
457            Ok(())
458        }
459    }
460
461    //////////////////////////////////////////////////////////////////////////////////////////////////////////
462    // Construction & transparency
463    //////////////////////////////////////////////////////////////////////////////////////////////////////////
464
465    #[test]
466    fn test_read_returns_inner_bytes_unchanged() {
467        let data = vec![0x01, 0x02, 0x03, 0x04];
468        let mut stream = LoggedStream::new(
469            Cursor::new(data.clone()),
470            LowercaseHexadecimalFormatter::new_default(),
471            DefaultFilter,
472            MemoryStorageLogger::new(16),
473        );
474
475        let mut buf = [0u8; 4];
476        stream.read_exact(&mut buf).unwrap();
477
478        assert_eq!(buf.to_vec(), data);
479    }
480
481    #[test]
482    fn test_write_reaches_inner_stream() {
483        let sink = Rc::new(RefCell::new(Vec::new()));
484        let mut stream = LoggedStream::new(
485            RecordingWriter(Rc::clone(&sink)),
486            LowercaseHexadecimalFormatter::new_default(),
487            DefaultFilter,
488            MemoryStorageLogger::new(16),
489        );
490
491        stream.write_all(&[0xde, 0xad, 0xbe, 0xef]).unwrap();
492
493        assert_eq!(*sink.borrow(), vec![0xde, 0xad, 0xbe, 0xef]);
494    }
495
496    //////////////////////////////////////////////////////////////////////////////////////////////////////////
497    // Read logging
498    //////////////////////////////////////////////////////////////////////////////////////////////////////////
499
500    #[test]
501    fn test_read_logs_read_record_with_formatted_content() {
502        let mut stream = LoggedStream::new(
503            Cursor::new(vec![0x0a, 0xff]),
504            LowercaseHexadecimalFormatter::new_default(),
505            DefaultFilter,
506            MemoryStorageLogger::new(16),
507        );
508
509        let mut buf = [0u8; 2];
510        stream.read_exact(&mut buf).unwrap();
511
512        let records = stream.get_log_records();
513        assert_eq!(records.len(), 1);
514        assert_eq!(records[0].kind, RecordKind::Read);
515        assert_eq!(records[0].message, "0a:ff");
516    }
517
518    #[test]
519    fn test_read_uses_configured_formatter() {
520        let mut stream = LoggedStream::new(
521            Cursor::new(vec![10, 255]),
522            DecimalFormatter::new_default(),
523            DefaultFilter,
524            MemoryStorageLogger::new(16),
525        );
526
527        let mut buf = [0u8; 2];
528        stream.read_exact(&mut buf).unwrap();
529
530        assert_eq!(stream.get_log_records()[0].message, "10:255");
531    }
532
533    #[test]
534    fn test_multiple_reads_log_in_order() {
535        let mut stream = LoggedStream::new(
536            Cursor::new(vec![0x01, 0x02, 0x03, 0x04]),
537            LowercaseHexadecimalFormatter::new_default(),
538            DefaultFilter,
539            MemoryStorageLogger::new(16),
540        );
541
542        let mut buf = [0u8; 2];
543        stream.read_exact(&mut buf).unwrap();
544        stream.read_exact(&mut buf).unwrap();
545
546        let records = stream.get_log_records();
547        assert_eq!(records.len(), 2);
548        assert_eq!(records[0].message, "01:02");
549        assert_eq!(records[1].message, "03:04");
550    }
551
552    //////////////////////////////////////////////////////////////////////////////////////////////////////////
553    // Write logging
554    //////////////////////////////////////////////////////////////////////////////////////////////////////////
555
556    #[test]
557    fn test_write_logs_write_record_with_formatted_content() {
558        let mut stream = LoggedStream::new(
559            Cursor::new(Vec::<u8>::new()),
560            UppercaseHexadecimalFormatter::new_default(),
561            DefaultFilter,
562            MemoryStorageLogger::new(16),
563        );
564
565        stream.write_all(&[0x0a, 0xff]).unwrap();
566
567        let records = stream.get_log_records();
568        assert_eq!(records.len(), 1);
569        assert_eq!(records[0].kind, RecordKind::Write);
570        assert_eq!(records[0].message, "0A:FF");
571    }
572
573    #[test]
574    fn test_flush_does_not_log() {
575        let mut stream = LoggedStream::new(
576            Cursor::new(Vec::<u8>::new()),
577            LowercaseHexadecimalFormatter::new_default(),
578            DefaultFilter,
579            MemoryStorageLogger::new(16),
580        );
581
582        stream.flush().unwrap();
583
584        assert!(stream.get_log_records().is_empty());
585    }
586
587    //////////////////////////////////////////////////////////////////////////////////////////////////////////
588    // Record filtering
589    //////////////////////////////////////////////////////////////////////////////////////////////////////////
590
591    #[test]
592    fn test_read_record_suppressed_by_filter() {
593        // The filter allows only Write, so the Read record is dropped.
594        let mut stream = LoggedStream::new(
595            Cursor::new(vec![0x01, 0x02]),
596            LowercaseHexadecimalFormatter::new_default(),
597            RecordKindFilter::new(&[RecordKind::Write]),
598            MemoryStorageLogger::new(16),
599        );
600
601        let mut buf = [0u8; 2];
602        stream.read_exact(&mut buf).unwrap();
603
604        assert!(stream.get_log_records().is_empty());
605    }
606
607    #[test]
608    fn test_write_record_suppressed_by_filter() {
609        // The filter allows only Read, so the Write record is dropped.
610        let mut stream = LoggedStream::new(
611            Cursor::new(Vec::<u8>::new()),
612            LowercaseHexadecimalFormatter::new_default(),
613            RecordKindFilter::new(&[RecordKind::Read]),
614            MemoryStorageLogger::new(16),
615        );
616
617        stream.write_all(&[0x01, 0x02]).unwrap();
618
619        assert!(stream.get_log_records().is_empty());
620    }
621
622    //////////////////////////////////////////////////////////////////////////////////////////////////////////
623    // Error handling (message content, filtering and swallowed error kinds)
624    //////////////////////////////////////////////////////////////////////////////////////////////////////////
625
626    #[test]
627    fn test_read_error_logs_error_record() {
628        let mut stream = LoggedStream::new(
629            ErrReader(ErrorKind::Other),
630            LowercaseHexadecimalFormatter::new_default(),
631            DefaultFilter,
632            MemoryStorageLogger::new(16),
633        );
634
635        let mut buf = [0u8; 4];
636        let _ = stream.read(&mut buf);
637
638        let records = stream.get_log_records();
639        assert_eq!(records.len(), 1);
640        assert_eq!(records[0].kind, RecordKind::Error);
641        assert!(records[0].message.starts_with("Error during read:"));
642    }
643
644    #[test]
645    fn test_read_error_suppressed_by_filter_without_error() {
646        let mut stream = LoggedStream::new(
647            ErrReader(ErrorKind::Other),
648            LowercaseHexadecimalFormatter::new_default(),
649            RecordKindFilter::new(&[RecordKind::Read, RecordKind::Write]),
650            MemoryStorageLogger::new(16),
651        );
652
653        let mut buf = [0u8; 4];
654        let _ = stream.read(&mut buf);
655
656        assert!(stream.get_log_records().is_empty());
657    }
658
659    #[test]
660    fn test_read_would_block_is_not_logged() {
661        // WouldBlock is a transient non-event and must not produce a record.
662        let mut stream = LoggedStream::new(
663            ErrReader(ErrorKind::WouldBlock),
664            LowercaseHexadecimalFormatter::new_default(),
665            DefaultFilter,
666            MemoryStorageLogger::new(16),
667        );
668
669        let mut buf = [0u8; 4];
670        let _ = stream.read(&mut buf);
671
672        assert!(stream.get_log_records().is_empty());
673    }
674
675    #[test]
676    fn test_write_error_logs_error_record() {
677        let mut stream = LoggedStream::new(
678            ErrWriter(ErrorKind::Other),
679            LowercaseHexadecimalFormatter::new_default(),
680            DefaultFilter,
681            MemoryStorageLogger::new(16),
682        );
683
684        let _ = stream.write(&[0x01, 0x02]);
685
686        let records = stream.get_log_records();
687        assert_eq!(records.len(), 1);
688        assert_eq!(records[0].kind, RecordKind::Error);
689        assert!(records[0].message.starts_with("Error during write:"));
690    }
691
692    #[test]
693    fn test_write_error_suppressed_by_filter_without_error() {
694        let mut stream = LoggedStream::new(
695            ErrWriter(ErrorKind::Other),
696            LowercaseHexadecimalFormatter::new_default(),
697            RecordKindFilter::new(&[RecordKind::Read, RecordKind::Write]),
698            MemoryStorageLogger::new(16),
699        );
700
701        let _ = stream.write(&[0x01, 0x02]);
702
703        assert!(stream.get_log_records().is_empty());
704    }
705
706    #[test]
707    fn test_write_would_block_is_not_logged() {
708        let mut stream = LoggedStream::new(
709            ErrWriter(ErrorKind::WouldBlock),
710            LowercaseHexadecimalFormatter::new_default(),
711            DefaultFilter,
712            MemoryStorageLogger::new(16),
713        );
714
715        let _ = stream.write(&[0x01, 0x02]);
716
717        assert!(stream.get_log_records().is_empty());
718    }
719
720    #[test]
721    fn test_write_write_zero_is_not_logged() {
722        let mut stream = LoggedStream::new(
723            ErrWriter(ErrorKind::WriteZero),
724            LowercaseHexadecimalFormatter::new_default(),
725            DefaultFilter,
726            MemoryStorageLogger::new(16),
727        );
728
729        let _ = stream.write(&[0x01, 0x02]);
730
731        assert!(stream.get_log_records().is_empty());
732    }
733
734    //////////////////////////////////////////////////////////////////////////////////////////////////////////
735    // Manual Open marker (log_open)
736    //////////////////////////////////////////////////////////////////////////////////////////////////////////
737
738    #[test]
739    fn test_log_open_emits_open_record() {
740        let mut stream = LoggedStream::new(
741            Cursor::new(Vec::<u8>::new()),
742            LowercaseHexadecimalFormatter::new_default(),
743            DefaultFilter,
744            MemoryStorageLogger::new(16),
745        );
746        stream.log_open("Established connection with 127.0.0.1:8080");
747
748        let records = stream.get_log_records();
749        assert_eq!(records.len(), 1);
750        assert_eq!(records[0].kind, RecordKind::Open);
751        assert_eq!(
752            records[0].message,
753            "Established connection with 127.0.0.1:8080"
754        );
755    }
756
757    #[test]
758    fn test_log_open_passes_filter_allowing_open() {
759        let mut stream = LoggedStream::new(
760            Cursor::new(Vec::<u8>::new()),
761            LowercaseHexadecimalFormatter::new_default(),
762            RecordKindFilter::new(&[RecordKind::Open]),
763            MemoryStorageLogger::new(16),
764        );
765        stream.log_open("kept");
766
767        let records = stream.get_log_records();
768        assert_eq!(records.len(), 1);
769        assert_eq!(records[0].kind, RecordKind::Open);
770        assert_eq!(records[0].message, "kept");
771    }
772
773    #[test]
774    fn test_log_open_suppressed_by_filter_without_open() {
775        let mut stream = LoggedStream::new(
776            Cursor::new(Vec::<u8>::new()),
777            LowercaseHexadecimalFormatter::new_default(),
778            RecordKindFilter::new(&[RecordKind::Read, RecordKind::Write]),
779            MemoryStorageLogger::new(16),
780        );
781        stream.log_open("should be filtered out");
782        assert!(stream.get_log_records().is_empty());
783    }
784
785    //////////////////////////////////////////////////////////////////////////////////////////////////////////
786    // Lifecycle: drop & shutdown
787    //////////////////////////////////////////////////////////////////////////////////////////////////////////
788
789    #[test]
790    fn test_drop_logs_drop_record() {
791        // The ChannelLogger's receiver outlives the stream, so we can observe the Drop record.
792        let mut stream = LoggedStream::new(
793            Cursor::new(Vec::<u8>::new()),
794            LowercaseHexadecimalFormatter::new_default(),
795            DefaultFilter,
796            ChannelLogger::new(),
797        );
798        let receiver = stream.take_receiver_unchecked();
799
800        drop(stream);
801
802        let record = receiver
803            .recv()
804            .expect("dropping the stream should emit a record");
805        assert_eq!(record.kind, RecordKind::Drop);
806        assert_eq!(record.message, "Deallocated.");
807    }
808
809    #[tokio::test]
810    async fn test_async_shutdown_logs_shutdown_record() {
811        use tokio::io::AsyncWriteExt;
812
813        let (client, _server) = tokio::io::duplex(64);
814        let mut stream = LoggedStream::new(
815            client,
816            LowercaseHexadecimalFormatter::new_default(),
817            DefaultFilter,
818            MemoryStorageLogger::new(16),
819        );
820
821        stream.shutdown().await.unwrap();
822
823        let records = stream.get_log_records();
824        assert_eq!(records.len(), 1);
825        assert_eq!(records[0].kind, RecordKind::Shutdown);
826        assert_eq!(records[0].message, "Writer shutdown request.");
827    }
828
829    //////////////////////////////////////////////////////////////////////////////////////////////////////////
830    // Asynchronous IO
831    //////////////////////////////////////////////////////////////////////////////////////////////////////////
832
833    #[tokio::test]
834    async fn test_async_read_logs_read_record() {
835        use tokio::io::AsyncReadExt;
836        use tokio::io::AsyncWriteExt;
837
838        let (client, mut server) = tokio::io::duplex(64);
839        server.write_all(&[0xaa, 0xbb]).await.unwrap();
840
841        let mut stream = LoggedStream::new(
842            client,
843            LowercaseHexadecimalFormatter::new_default(),
844            DefaultFilter,
845            MemoryStorageLogger::new(16),
846        );
847
848        let mut buf = [0u8; 2];
849        stream.read_exact(&mut buf).await.unwrap();
850
851        assert_eq!(buf, [0xaa, 0xbb]);
852        let records = stream.get_log_records();
853        assert_eq!(records.len(), 1);
854        assert_eq!(records[0].kind, RecordKind::Read);
855        assert_eq!(records[0].message, "aa:bb");
856    }
857
858    #[tokio::test]
859    async fn test_async_write_logs_write_record() {
860        use tokio::io::AsyncWriteExt;
861
862        let (client, _server) = tokio::io::duplex(64);
863        let mut stream = LoggedStream::new(
864            client,
865            LowercaseHexadecimalFormatter::new_default(),
866            DefaultFilter,
867            MemoryStorageLogger::new(16),
868        );
869
870        stream.write_all(&[0x01, 0x02, 0x03, 0x04]).await.unwrap();
871
872        let records = stream.get_log_records();
873        assert_eq!(records.len(), 1);
874        assert_eq!(records[0].kind, RecordKind::Write);
875        assert_eq!(records[0].message, "01:02:03:04");
876    }
877
878    #[tokio::test]
879    async fn test_async_read_error_suppressed_by_filter_without_error() {
880        use tokio::io::AsyncReadExt;
881
882        let mut stream = LoggedStream::new(
883            ErrAsyncReader(ErrorKind::Other),
884            LowercaseHexadecimalFormatter::new_default(),
885            RecordKindFilter::new(&[RecordKind::Read, RecordKind::Write]),
886            MemoryStorageLogger::new(16),
887        );
888
889        let mut buf = [0u8; 4];
890        let _ = stream.read(&mut buf).await;
891
892        assert!(stream.get_log_records().is_empty());
893    }
894
895    #[tokio::test]
896    async fn test_async_write_error_suppressed_by_filter_without_error() {
897        use tokio::io::AsyncWriteExt;
898
899        let mut stream = LoggedStream::new(
900            ErrAsyncWriter(ErrorKind::Other),
901            LowercaseHexadecimalFormatter::new_default(),
902            RecordKindFilter::new(&[RecordKind::Read, RecordKind::Write]),
903            MemoryStorageLogger::new(16),
904        );
905
906        let _ = stream.write(&[0x01, 0x02]).await;
907
908        assert!(stream.get_log_records().is_empty());
909    }
910
911    #[tokio::test]
912    async fn test_async_read_error_logs_error_record() {
913        use tokio::io::AsyncReadExt;
914
915        let mut stream = LoggedStream::new(
916            ErrAsyncReader(ErrorKind::Other),
917            LowercaseHexadecimalFormatter::new_default(),
918            DefaultFilter,
919            MemoryStorageLogger::new(16),
920        );
921
922        let mut buf = [0u8; 4];
923        let _ = stream.read(&mut buf).await;
924
925        let records = stream.get_log_records();
926        assert_eq!(records.len(), 1);
927        assert_eq!(records[0].kind, RecordKind::Error);
928        assert!(records[0].message.starts_with("Error during async read:"));
929    }
930
931    #[tokio::test]
932    async fn test_async_write_error_logs_error_record() {
933        use tokio::io::AsyncWriteExt;
934
935        let mut stream = LoggedStream::new(
936            ErrAsyncWriter(ErrorKind::Other),
937            LowercaseHexadecimalFormatter::new_default(),
938            DefaultFilter,
939            MemoryStorageLogger::new(16),
940        );
941
942        let _ = stream.write(&[0x01, 0x02]).await;
943
944        let records = stream.get_log_records();
945        assert_eq!(records.len(), 1);
946        assert_eq!(records[0].kind, RecordKind::Error);
947        assert!(records[0].message.starts_with("Error during async write:"));
948    }
949
950    //////////////////////////////////////////////////////////////////////////////////////////////////////////
951    // Logger accessors
952    //////////////////////////////////////////////////////////////////////////////////////////////////////////
953
954    #[test]
955    fn test_memory_storage_get_and_clear() {
956        let mut stream = LoggedStream::new(
957            Cursor::new(vec![0x01, 0x02]),
958            LowercaseHexadecimalFormatter::new_default(),
959            DefaultFilter,
960            MemoryStorageLogger::new(16),
961        );
962
963        let mut buf = [0u8; 2];
964        stream.read_exact(&mut buf).unwrap();
965        assert_eq!(stream.get_log_records().len(), 1);
966
967        stream.clear_log_records();
968        assert!(stream.get_log_records().is_empty());
969    }
970
971    #[test]
972    fn test_channel_logger_delivers_records() {
973        let mut stream = LoggedStream::new(
974            Cursor::new(vec![0x01, 0x02]),
975            LowercaseHexadecimalFormatter::new_default(),
976            DefaultFilter,
977            ChannelLogger::new(),
978        );
979        let receiver = stream.take_receiver_unchecked();
980
981        let mut buf = [0u8; 2];
982        stream.read_exact(&mut buf).unwrap();
983
984        let record = receiver.recv().unwrap();
985        assert_eq!(record.kind, RecordKind::Read);
986        assert_eq!(record.message, "01:02");
987    }
988
989    //////////////////////////////////////////////////////////////////////////////////////////////////////////
990    // Trait assertions
991    //////////////////////////////////////////////////////////////////////////////////////////////////////////
992
993    fn assert_send<T: Send>() {}
994
995    fn assert_unpin<T: Unpin>() {}
996
997    #[test]
998    fn test_send() {
999        assert_send::<
1000            LoggedStream<
1001                Cursor<Vec<u8>>,
1002                LowercaseHexadecimalFormatter,
1003                DefaultFilter,
1004                MemoryStorageLogger,
1005            >,
1006        >();
1007    }
1008
1009    #[test]
1010    fn test_unpin() {
1011        assert_unpin::<
1012            LoggedStream<
1013                Cursor<Vec<u8>>,
1014                LowercaseHexadecimalFormatter,
1015                DefaultFilter,
1016                MemoryStorageLogger,
1017            >,
1018        >();
1019    }
1020
1021    #[test]
1022    fn test_debug() {
1023        let stream = LoggedStream::new(
1024            Cursor::new(Vec::<u8>::new()),
1025            LowercaseHexadecimalFormatter::new_default(),
1026            DefaultFilter,
1027            MemoryStorageLogger::new(4),
1028        );
1029
1030        let debug = format!("{stream:?}");
1031        assert!(debug.contains("LoggedStream"));
1032        assert!(debug.contains("formatter"));
1033    }
1034}
logged_stream/stream.rs

logged_stream/
stream.rs
