Update `FsBuilder`-based `ByteStreams` to support file offsets (#1361)

references = ["smithy-rs#1352"]

meta = { "breaking" = false, "tada" = false, "bug" = false }

author = "jdisanti"

[[smithy-rs]]

message = """

The `aws_smithy_http::byte_stream::bytestream_util::FsBuilder` has been updated to allow for easier creation of

multi-part requests.

- `FsBuilder::offset` is a new method allowing users to specify an offset to start reading a file from.

- `FsBuilder::file_size` has been reworked into `FsBuilder::length` and is now used to specify the amount of data to read.

With these two methods, it's now simple to create a `ByteStream` that will read a single "chunk" of a file. The example

below demonstrates how you could divide a single `File` into consecutive chunks to create multiple `ByteStream`s.

```rust

let example_file_path = Path::new("/example.txt");

let example_file_size = tokio::fs::metadata(&example_file_path).await.unwrap().len();

let chunks = 6;

let chunk_size = file_size / chunks;

let mut byte_streams = Vec::new();

for i in 0..chunks {

    let length = if i == chunks - 1 {

        // If we're on the last chunk, the length to read might be less than a whole chunk.

        // We substract the size of all previous chunks from the total file size to get the

        // size of the final chunk.

        file_size - (i * chunk_size)

    } else {

        chunk_size

    };

    let byte_stream = ByteStream::read_from()

        .path(&file_path)

        .offset(i * chunk_size)

        .length(length)

        .build()

        .await?;

    byte_streams.push(byte_stream);

}

for chunk in byte_streams {

    // Make requests to a service

}

```

"""

references = ["aws-sdk-rust#494", "aws-sdk-rust#519"]

meta = { "breaking" = true, "tada" = true, "bug" = false }

author = "Velfi"

repository = "https://github.com/awslabs/smithy-rs"

[features]

rt-tokio = ["tokio/rt", "tokio/fs", "tokio-util/io"]

rt-tokio = ["tokio/rt", "tokio/fs", "tokio/io-util", "tokio-util/io"]

event-stream = ["aws-smithy-eventstream"]

[dependencies]

//! ```no_run

//! # #[cfg(feature = "rt-tokio")]

//! # {

//! use aws_smithy_http::byte_stream::ByteStream;

//! use aws_smithy_http::byte_stream::{ByteStream, Length};

//! use std::path::Path;

//! struct GetObjectInput {

//!     body: ByteStream

//! async fn bytestream_from_file() -> GetObjectInput {

//!     let bytestream = ByteStream::read_from().path("docs/some-large-file.csv")

//!         .buffer_size(32_784)

//!         .file_size(123_456)

//!         .length(Length::Exact(123_456))

//!         .build()

//!         .await

//!         .expect("valid path");

#[cfg(feature = "rt-tokio")]

mod bytestream_util;

#[cfg(feature = "rt-tokio")]

pub use bytestream_util::Length;

#[cfg(feature = "rt-tokio")]

pub use self::bytestream_util::FsBuilder;

    /// ```no_run

    /// # #[cfg(feature = "rt-tokio")]

    /// # {

    /// use aws_smithy_http::byte_stream::ByteStream;

    /// use aws_smithy_http::byte_stream::{ByteStream, Length};

    ///

    /// async fn bytestream_from_file() -> ByteStream {

    ///     let bytestream = ByteStream::read_from()

    ///         // Specify the size of the buffer used to read the file (in bytes, default is 4096)

    ///         .buffer_size(32_784)

    ///         // Specify the length of the file used (skips an additional call to retrieve the size)

    ///         .file_size(123_456)

    ///         .length(Length::Exact(123_456))

    ///         .build()

    ///         .await

    ///         .expect("valid path");

        Ok(())

    }

    #[cfg(feature = "rt-tokio")]

    #[tokio::test]

    async fn path_based_bytestreams_with_builder() -> Result<(), Box<dyn std::error::Error>> {

        use super::ByteStream;

        use bytes::Buf;

        use http_body::Body;

        use std::io::Write;

        use tempfile::NamedTempFile;

        let mut file = NamedTempFile::new()?;

        for i in 0..10000 {

            writeln!(file, "Brian was here. Briefly. {}", i)?;

        }

        let body = ByteStream::read_from()

            .path(&file)

            .buffer_size(16384)

            // This isn't the right file length - one shouldn't do this in real code

            .file_size(200)

            .build()

            .await?

            .into_inner();

        // assert that the file length specified size is used as size hint

        assert_eq!(body.size_hint().exact(), Some(200));

        let mut body1 = body.try_clone().expect("retryable bodies are cloneable");

        // read a little bit from one of the clones

        let some_data = body1

            .data()

            .await

            .expect("should have some data")

            .expect("read should not fail");

        // The size of one read should be equal to that of the buffer size

        assert_eq!(some_data.len(), 16384);

        assert_eq!(

            ByteStream::new(body1).collect().await?.remaining(),

            298890 - some_data.len()

        );

        Ok(())

    }

}