Expose stream reader input completion state

Author: eldryoth

CHANGELOG.md

@@ -37,6 +37,8 @@
   state inspection.
 - Added stream adapter `engine()` and `is_padded()` helpers for dependency-free
   policy inspection in diagnostics and audit logs.
+- Added reader-side stream `has_finished_input()` helpers so callers can
+  distinguish EOF or terminal padding from fully drained buffered output.
 
 ## 0.8.0 - 2026-05-16
 

README.md

@@ -525,11 +525,12 @@ returned to the caller. Decoders additionally expose `has_terminal_padding()`
 so framed protocols can tell when a padded payload has ended and leave adjacent
 bytes for the next protocol layer. Reader adapters also expose `is_finished()`
 once EOF or terminal padding has been reached and all buffered output has been
-drained. Writer adapters expose `try_finish()` to finalize pending input and
-flush the wrapped writer without consuming the adapter, plus `is_finalized()`
-for explicit state inspection; after successful finalization, later writes are
-rejected. Their `Debug` output reports adapter state without formatting the
-wrapped reader or writer.
+drained, and `has_finished_input()` when the wrapped reader has reached EOF or
+terminal padding but buffered output may still remain. Writer adapters expose
+`try_finish()` to finalize pending input and flush the wrapped writer without
+consuming the adapter, plus `is_finalized()` for explicit state inspection;
+after successful finalization, later writes are rejected. Their `Debug` output
+reports adapter state without formatting the wrapped reader or writer.
 
 URL-safe, no-padding encoding:
 

docs/MIGRATION.md

@@ -188,8 +188,8 @@ Base64 input and flush the wrapped writer without immediately consuming the
 adapter. After successful finalization, later non-empty writes return
 `InvalidInput`. Stream adapters also expose non-sensitive state helpers such as
 `engine()`, `is_padded()`, `pending_len()`, `has_pending_input()`, reader-side
-`buffered_output_len()`, and decoder-side `has_terminal_padding()` for framed
-protocols and audit logging.
+`buffered_output_len()` and `has_finished_input()`, and decoder-side
+`has_terminal_padding()` for framed protocols and audit logging.
 
 The `tokio` feature is reserved for future async wrappers. It is currently
 inert and dependency-free; use the explicit `stream` feature for `std::io`

src/lib.rs

@@ -1205,6 +1205,16 @@ pub mod stream {
             self.terminal_seen
         }
 
+        /// Returns whether this decoder reader has reached EOF or terminal
+        /// padding in the wrapped reader.
+        ///
+        /// This may become `true` before [`Self::is_finished`] when decoded
+        /// output is still buffered for the caller.
+        #[must_use]
+        pub const fn has_finished_input(&self) -> bool {
+            self.finished
+        }
+
         /// Returns whether this reader has reached EOF or terminal padding
         /// and has no decoded output buffered for the caller.
         #[must_use]
@@ -1453,6 +1463,16 @@ pub mod stream {
             !self.output.is_empty()
         }
 
+        /// Returns whether this encoder reader has reached EOF in the wrapped
+        /// reader.
+        ///
+        /// This may become `true` before [`Self::is_finished`] when encoded
+        /// output is still buffered for the caller.
+        #[must_use]
+        pub const fn has_finished_input(&self) -> bool {
+            self.finished
+        }
+
         /// Returns whether this reader has reached EOF and has no encoded
         /// output buffered for the caller.
         #[must_use]

tests/rfc4648.rs

@@ -2657,6 +2657,7 @@ fn stream_encoder_reader_handles_small_reads() {
     let mut reader = EncoderReader::new(&b"hello"[..], STANDARD);
     assert_eq!(reader.engine(), STANDARD);
     assert!(reader.is_padded());
+    assert!(!reader.has_finished_input());
     assert!(!reader.is_finished());
     assert_eq!(reader.pending_len(), 0);
     assert!(!reader.has_pending_input());
@@ -2676,6 +2677,7 @@ fn stream_encoder_reader_handles_small_reads() {
     assert!(!reader.has_pending_input());
     assert_eq!(reader.buffered_output_len(), 0);
     assert!(!reader.has_buffered_output());
+    assert!(reader.has_finished_input());
     assert!(reader.is_finished());
 }
 
@@ -2691,13 +2693,40 @@ fn stream_encoder_reader_reports_buffered_output() {
     assert!(reader.has_pending_input());
     assert_eq!(reader.buffered_output_len(), 3);
     assert!(reader.has_buffered_output());
+    assert!(!reader.has_finished_input());
     assert!(!reader.is_finished());
 
     let mut rest = Vec::new();
     reader.read_to_end(&mut rest).unwrap();
     assert_eq!(rest, b"GVsbG8=");
     assert_eq!(reader.buffered_output_len(), 0);
     assert!(!reader.has_buffered_output());
+    assert!(reader.has_finished_input());
+    assert!(reader.is_finished());
+}
+
+#[cfg(feature = "stream")]
+#[test]
+fn stream_encoder_reader_finished_input_before_buffer_drain() {
+    let mut reader = EncoderReader::new(&b"h"[..], STANDARD);
+    assert!(!reader.has_finished_input());
+    assert!(!reader.is_finished());
+
+    let mut first = [0u8; 1];
+    assert_eq!(reader.read(&mut first).unwrap(), 1);
+    assert_eq!(first, [b'a']);
+    assert_eq!(reader.pending_len(), 0);
+    assert_eq!(reader.buffered_output_len(), 3);
+    assert!(reader.has_buffered_output());
+    assert!(reader.has_finished_input());
+    assert!(!reader.is_finished());
+
+    let mut rest = Vec::new();
+    reader.read_to_end(&mut rest).unwrap();
+    assert_eq!(rest, b"A==");
+    assert_eq!(reader.buffered_output_len(), 0);
+    assert!(!reader.has_buffered_output());
+    assert!(reader.has_finished_input());
     assert!(reader.is_finished());
 }
 
@@ -2885,6 +2914,7 @@ fn stream_decoder_reader_handles_small_reads() {
     let mut reader = DecoderReader::new(&b"aGVsbG8="[..], STANDARD);
     assert_eq!(reader.engine(), STANDARD);
     assert!(reader.is_padded());
+    assert!(!reader.has_finished_input());
     assert!(!reader.is_finished());
     assert_eq!(reader.pending_len(), 0);
     assert!(!reader.has_pending_input());
@@ -2905,6 +2935,7 @@ fn stream_decoder_reader_handles_small_reads() {
     assert_eq!(reader.buffered_output_len(), 0);
     assert!(!reader.has_buffered_output());
     assert!(reader.has_terminal_padding());
+    assert!(reader.has_finished_input());
     assert!(reader.is_finished());
 }
 
@@ -2919,6 +2950,7 @@ fn stream_decoder_reader_reports_buffered_output() {
     assert_eq!(reader.buffered_output_len(), 2);
     assert!(reader.has_buffered_output());
     assert!(!reader.has_terminal_padding());
+    assert!(!reader.has_finished_input());
     assert!(!reader.is_finished());
 
     let mut rest = Vec::new();
@@ -2927,6 +2959,7 @@ fn stream_decoder_reader_reports_buffered_output() {
     assert_eq!(reader.buffered_output_len(), 0);
     assert!(!reader.has_buffered_output());
     assert!(reader.has_terminal_padding());
+    assert!(reader.has_finished_input());
     assert!(reader.is_finished());
 }
 
@@ -2940,6 +2973,7 @@ fn stream_decoder_reader_terminal_padding_finishes_after_buffer_drain() {
     assert_eq!(reader.buffered_output_len(), 1);
     assert!(reader.has_buffered_output());
     assert!(reader.has_terminal_padding());
+    assert!(reader.has_finished_input());
     assert!(!reader.is_finished());
 
     let mut rest = Vec::new();
@@ -2948,6 +2982,7 @@ fn stream_decoder_reader_terminal_padding_finishes_after_buffer_drain() {
     assert_eq!(reader.buffered_output_len(), 0);
     assert!(!reader.has_buffered_output());
     assert!(reader.has_terminal_padding());
+    assert!(reader.has_finished_input());
     assert!(reader.is_finished());
 }