Make into bits produce bitstring stream (#13310)

# Description

Fix `into bits` to have consistent behavior when passed a byte stream.

# User-Facing Changes

Previously, it was returning a binary on stream, even though its
input/output types don't describe this possibility. We don't need this
since we have `into binary` anyway.

# Tests + Formatting
Tests added

crates/nu-cmd-extra/src/extra/bits/into.rs

@@ -1,3 +1,5 @@
+use std::io::{self, Read, Write};
+
 use nu_cmd_base::input_handler::{operate, CmdArgument};
 use nu_engine::command_prelude::*;
 
@@ -118,15 +120,41 @@ fn into_bits(
     let cell_paths = call.rest(engine_state, stack, 0)?;
     let cell_paths = (!cell_paths.is_empty()).then_some(cell_paths);
 
-    if let PipelineData::ByteStream(stream, ..) = input {
+    if let PipelineData::ByteStream(stream, metadata) = input {
-        // TODO: in the future, we may want this to stream out, converting each to bytes
+        Ok(PipelineData::ByteStream(
-        Ok(Value::binary(stream.into_bytes()?, head).into_pipeline_data())
+            byte_stream_to_bits(stream, head),
+            metadata,
+        ))
     } else {
         let args = Arguments { cell_paths };
         operate(action, args, input, call.head, engine_state.ctrlc.clone())
     }
 }
 
+fn byte_stream_to_bits(stream: ByteStream, head: Span) -> ByteStream {
+    if let Some(mut reader) = stream.reader() {
+        let mut is_first = true;
+        ByteStream::from_fn(head, None, ByteStreamType::String, move |buffer| {
+            let mut byte = [0];
+            if reader.read(&mut byte[..]).err_span(head)? > 0 {
+                // Format the byte as bits
+                if is_first {
+                    is_first = false;
+                } else {
+                    buffer.push(b' ');
+                }
+                write!(buffer, "{:08b}", byte[0]).expect("format failed");
+                Ok(true)
+            } else {
+                // EOF
+                Ok(false)
+            }
+        })
+    } else {
+        ByteStream::read(io::empty(), head, None, ByteStreamType::String)
+    }
+}
+
 fn convert_to_smallest_number_type(num: i64, span: Span) -> Value {
     if let Some(v) = num.to_i8() {
         let bytes = v.to_ne_bytes();

crates/nu-cmd-extra/tests/commands/bits/into.rs

@@ -0,0 +1,13 @@
+use nu_test_support::nu;
+
+#[test]
+fn byte_stream_into_bits() {
+    let result = nu!("[0x[01] 0x[02 03]] | bytes collect | into bits");
+    assert_eq!("00000001 00000010 00000011", result.out);
+}
+
+#[test]
+fn byte_stream_into_bits_is_stream() {
+    let result = nu!("[0x[01] 0x[02 03]] | bytes collect | into bits | describe");
+    assert_eq!("string (stream)", result.out);
+}

crates/nu-cmd-extra/tests/commands/bits/mod.rs

@@ -0,0 +1 @@
+mod into;

crates/nu-cmd-extra/tests/commands/mod.rs

@@ -1 +1,2 @@
+mod bits;
 mod bytes;