Progress bar Implementation (#7661)

# Description

_(Description of your pull request goes here. **Provide examples and/or
screenshots** if your changes affect the user experience.)_

I implemented the status bar we talk about yesterday. The idea was
inspired by the progress bar of `wget`.
I decided to go for the second suggestion by `@Reilly`
> 2. add an Option<usize> or whatever to RawStream (and ListStream?) for
situations where you do know the length ahead of time

For now only works with the command `save` but after the approve of this
PR we can see how we can implement it on commands like `cp` and `mv`

When using `fetch` nushell will check if there is any `content-length`
attribute in the request header. If so, then `fetch` will send it
through the new `Option` variable in the `RawStream` to the `save`.
If we know the total size we show the progress bar 

![nu_pb01](https://user-images.githubusercontent.com/38369407/210298647-07ee55ea-e751-41b1-a84d-f72ec1f6e9e5.jpg)
but if we don't then we just show the stats like: data already saved,
bytes per second, and time lapse.

![nu_pb02](https://user-images.githubusercontent.com/38369407/210298698-1ef65f51-40cc-4481-83de-309cbd1049cb.jpg)

![nu_pb03](https://user-images.githubusercontent.com/38369407/210298701-eef2ef13-9206-4a98-8202-e4fe5531d79d.jpg)

Please let me know If I need to make any changes and I will be happy to
do it.

# User-Facing Changes

A new flag (`--progress` `-p`) was added to the `save` command 
Examples:
```nu
fetch https://github.com/torvalds/linux/archive/refs/heads/master.zip | save --progress -f main.zip
fetch https://releases.ubuntu.com/22.04.1/ubuntu-22.04.1-desktop-amd64.iso | save --progress -f main.zip
open main.zip --raw | save --progress main.copy
```

# Tests + Formatting

Don't forget to add tests that cover your changes.

Make sure you've run and fixed any issues with these commands:

- `cargo fmt --all -- --check` to check standard code formatting (`cargo
fmt --all` applies these changes)
- `cargo clippy --workspace -- -D warnings -D clippy::unwrap_used -A
clippy::needless_collect` to check that you're using the standard code
style
- `cargo test --workspace` to check that all tests pass
-
I am getting some errors and its weird because the errors are showing up
in files i haven't touch. Is this normal?

# After Submitting

If your PR had any user-facing changes, update [the
documentation](https://github.com/nushell/nushell.github.io) after the
PR is merged, if necessary. This will help us keep the docs up to date.

Co-authored-by: Reilly Wood <reilly.wood@icloud.com>

crates/nu-command/src/filesystem/open.rs

@@ -139,6 +139,7 @@ impl Command for Open {
                     Box::new(BufferedReader { input: buf_reader }),
                     ctrlc,
                     call_span,
+                    None,
                 )),
                 stderr: None,
                 exit_code: None,

crates/nu-command/src/filesystem/save.rs

@@ -9,6 +9,8 @@ use std::fs::File;
 use std::io::{BufWriter, Write};
 use std::path::Path;
 
+use crate::progress_bar;
+
 #[derive(Clone)]
 pub struct Save;
 
@@ -47,6 +49,7 @@ impl Command for Save {
             .switch("raw", "save file as raw binary", Some('r'))
             .switch("append", "append input to the end of the file", Some('a'))
             .switch("force", "overwrite the destination", Some('f'))
+            .switch("progress", "enable progress bar", Some('p'))
             .category(Category::FileSystem)
     }
 
@@ -60,6 +63,7 @@ impl Command for Save {
         let raw = call.has_flag("raw");
         let append = call.has_flag("append");
         let force = call.has_flag("force");
+        let progress = call.has_flag("progress");
 
         let span = call.head;
 
@@ -81,16 +85,16 @@ impl Command for Save {
 
                 // delegate a thread to redirect stderr to result.
                 let handler = stderr.map(|stderr_stream| match stderr_file {
-                    Some(stderr_file) => {
-                        std::thread::spawn(move || stream_to_file(stderr_stream, stderr_file, span))
-                    }
+                    Some(stderr_file) => std::thread::spawn(move || {
+                        stream_to_file(stderr_stream, stderr_file, span, progress)
+                    }),
                     None => std::thread::spawn(move || {
                         let _ = stderr_stream.into_bytes();
                         Ok(PipelineData::empty())
                     }),
                 });
 
-                let res = stream_to_file(stream, file, span);
+                let res = stream_to_file(stream, file, span, progress);
                 if let Some(h) = handler {
                     h.join().map_err(|err| {
                         ShellError::ExternalCommand(
@@ -332,10 +336,29 @@ fn stream_to_file(
     mut stream: RawStream,
     file: File,
     span: Span,
+    progress: bool,
 ) -> Result<PipelineData, ShellError> {
     let mut writer = BufWriter::new(file);
 
-    stream
+    let mut bytes_processed: u64 = 0;
+    let bytes_processed_p = &mut bytes_processed;
+    let file_total_size = stream.known_size;
+    let mut process_failed = false;
+    let process_failed_p = &mut process_failed;
+
+    // Create the progress bar
+    // It looks a bit messy but I am doing it this way to avoid
+    // creating the bar when is not needed
+    let (mut bar_opt, bar_opt_clone) = if progress {
+        let tmp_bar = progress_bar::NuProgressBar::new(file_total_size);
+        let tmp_bar_clone = tmp_bar.clone();
+
+        (Some(tmp_bar), Some(tmp_bar_clone))
+    } else {
+        (None, None)
+    };
+
+    let result = stream
         .try_for_each(move |result| {
             let buf = match result {
                 Ok(v) => match v {
@@ -353,13 +376,39 @@ fn stream_to_file(
                         ));
                     }
                 },
-                Err(err) => return Err(err),
+                Err(err) => {
+                    *process_failed_p = true;
+                    return Err(err);
+                }
             };
 
+            // If the `progress` flag is set then
+            if progress {
+                // Update the total amount of bytes that has been saved and then print the progress bar
+                *bytes_processed_p += buf.len() as u64;
+                if let Some(bar) = &mut bar_opt {
+                    bar.update_bar(*bytes_processed_p);
+                }
+            }
+
             if let Err(err) = writer.write(&buf) {
+                *process_failed_p = true;
                 return Err(ShellError::IOError(err.to_string()));
             }
             Ok(())
         })
-        .map(|_| PipelineData::empty())
+        .map(|_| PipelineData::empty());
+
+    // If the `progress` flag is set then
+    if progress {
+        // If the process failed, stop the progress bar with an error message.
+        if process_failed {
+            if let Some(bar) = bar_opt_clone {
+                bar.abandoned_msg("# Error while saving #".to_owned());
+            }
+        }
+    }
+
+    // And finally return the stream result.
+    result
 }