Support ByteStreams in bytes starts-with and bytes ends-with (#12887)

# Description
Restores `bytes starts-with` so that it is able to work with byte
streams once again. For parity/consistency, this PR also adds byte
stream support to `bytes ends-with`.

# User-Facing Changes
- `bytes ends-with` now supports byte streams.

# Tests + Formatting
Re-enabled tests for `bytes starts-with` and added tests for `bytes
ends-with`.

crates/nu-command/src/bytes/ends_with.rs

@@ -1,5 +1,9 @@
 use nu_cmd_base::input_handler::{operate, CmdArgument};
 use nu_engine::command_prelude::*;
+use std::{
+    collections::VecDeque,
+    io::{self, BufRead},
+};
 
 struct Arguments {
     pattern: Vec<u8>,
@@ -52,14 +56,54 @@ impl Command for BytesEndsWith {
         call: &Call,
         input: PipelineData,
     ) -> Result<PipelineData, ShellError> {
+        let head = call.head;
         let pattern: Vec<u8> = call.req(engine_state, stack, 0)?;
         let cell_paths: Vec<CellPath> = call.rest(engine_state, stack, 1)?;
         let cell_paths = (!cell_paths.is_empty()).then_some(cell_paths);
-        let arg = Arguments {
-            pattern,
-            cell_paths,
-        };
-        operate(ends_with, arg, input, call.head, engine_state.ctrlc.clone())
+
+        if let PipelineData::ByteStream(stream, ..) = input {
+            let span = stream.span();
+            if pattern.is_empty() {
+                return Ok(Value::bool(true, head).into_pipeline_data());
+            }
+            let Some(mut reader) = stream.reader() else {
+                return Ok(Value::bool(false, head).into_pipeline_data());
+            };
+            let cap = pattern.len();
+            let mut end = VecDeque::<u8>::with_capacity(cap);
+            loop {
+                let buf = match reader.fill_buf() {
+                    Ok(&[]) => break,
+                    Ok(buf) => buf,
+                    Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
+                    Err(e) => return Err(e.into_spanned(span).into()),
+                };
+                let len = buf.len();
+                if len >= cap {
+                    end.clear();
+                    end.extend(&buf[(len - cap)..])
+                } else {
+                    let new_len = len + end.len();
+                    if new_len > cap {
+                        // The `drain` below will panic if `(new_len - cap) > end.len()`.
+                        // But this cannot happen since we know `len < cap` (as checked above):
+                        //   (len + end.len() - cap) > end.len()
+                        //   => (len - cap) > 0
+                        //   => len > cap
+                        end.drain(..(new_len - cap));
+                    }
+                    end.extend(buf);
+                }
+                reader.consume(len);
+            }
+            Ok(Value::bool(end == pattern, head).into_pipeline_data())
+        } else {
+            let arg = Arguments {
+                pattern,
+                cell_paths,
+            };
+            operate(ends_with, arg, input, head, engine_state.ctrlc.clone())
+        }
     }
 
     fn examples(&self) -> Vec<Example> {

crates/nu-command/src/bytes/starts_with.rs

@@ -1,5 +1,6 @@
 use nu_cmd_base::input_handler::{operate, CmdArgument};
 use nu_engine::command_prelude::*;
+use std::io::Read;
 
 struct Arguments {
     pattern: Vec<u8>,
@@ -53,20 +54,33 @@ impl Command for BytesStartsWith {
         call: &Call,
         input: PipelineData,
     ) -> Result<PipelineData, ShellError> {
+        let head = call.head;
         let pattern: Vec<u8> = call.req(engine_state, stack, 0)?;
         let cell_paths: Vec<CellPath> = call.rest(engine_state, stack, 1)?;
         let cell_paths = (!cell_paths.is_empty()).then_some(cell_paths);
-        let arg = Arguments {
-            pattern,
-            cell_paths,
-        };
-        operate(
-            starts_with,
-            arg,
-            input,
-            call.head,
-            engine_state.ctrlc.clone(),
-        )
+
+        if let PipelineData::ByteStream(stream, ..) = input {
+            let span = stream.span();
+            if pattern.is_empty() {
+                return Ok(Value::bool(true, head).into_pipeline_data());
+            }
+            let Some(reader) = stream.reader() else {
+                return Ok(Value::bool(false, head).into_pipeline_data());
+            };
+            let mut start = Vec::with_capacity(pattern.len());
+            reader
+                .take(pattern.len() as u64)
+                .read_to_end(&mut start)
+                .err_span(span)?;
+
+            Ok(Value::bool(start == pattern, head).into_pipeline_data())
+        } else {
+            let arg = Arguments {
+                pattern,
+                cell_paths,
+            };
+            operate(starts_with, arg, input, head, engine_state.ctrlc.clone())
+        }
     }
 
     fn examples(&self) -> Vec<Example> {