Struct aws_smithy_http::byte_stream::ByteStream

pub struct ByteStream(_);

Stream of binary data

ByteStream wraps a stream of binary data for ease of use.

Getting data out of a ByteStream

ByteStream provides two primary mechanisms for accessing the data:

1. With .collect():

   .collect() reads the complete ByteStream into memory and stores it in AggregatedBytes, a non-contiguous ByteBuffer.

   use aws_smithy_http::byte_stream::{ByteStream, AggregatedBytes};
   use aws_smithy_http::body::SdkBody;
   use bytes::Buf;
   async fn example() {
      let stream = ByteStream::new(SdkBody::from("hello! This is some data"));
      // Load data from the stream into memory:
      let data = stream.collect().await.expect("error reading data");
      // collect returns a `bytes::Buf`:
      println!("first chunk: {:?}", data.chunk());
   }

2. Via impl Stream:

   Note: An import of StreamExt is required to use .try_next().

   For use-cases where holding the entire ByteStream in memory is unnecessary, use the Stream implementation:

   use aws_smithy_http::byte_stream::{ByteStream, AggregatedBytes, Error};
   use aws_smithy_http::body::SdkBody;
   use tokio_stream::StreamExt;
   
   async fn example() -> Result<(), Error> {
      let mut stream = ByteStream::from(vec![1, 2, 3, 4, 5, 99]);
      let mut digest = crc32::Digest::new();
      while let Some(bytes) = stream.try_next().await? {
          digest.write(&bytes);
      }
      println!("digest: {}", digest.finish());
      Ok(())
   }

3. Via .into_async_read():

   Note: The rt-tokio feature must be active to use .into_async_read().

   It’s possible to convert a ByteStream into a struct that implements tokio::io::AsyncRead. Then, you can use pre-existing tools like tokio::io::BufReader:

   use aws_smithy_http::byte_stream::ByteStream;
   use aws_smithy_http::body::SdkBody;
   use tokio::io::{AsyncBufReadExt, BufReader};
   #[cfg(feature = "rt-tokio")]
   async fn example() -> std::io::Result<()> {
      let stream = ByteStream::new(SdkBody::from("hello!\nThis is some data"));
      // Wrap the stream in a BufReader
      let buf_reader = BufReader::new(stream.into_async_read());
      let mut lines = buf_reader.lines();
      assert_eq!(lines.next_line().await?, Some("hello!".to_owned()));
      assert_eq!(lines.next_line().await?, Some("This is some data".to_owned()));
      assert_eq!(lines.next_line().await?, None);
      Ok(())
   }

Getting data into a ByteStream

ByteStreams can be created in one of three ways:

1. From in-memory binary data: ByteStreams created from in-memory data are always retryable. Data will be converted into Bytes enabling a cheap clone during retries.

   use bytes::Bytes;
   use aws_smithy_http::byte_stream::ByteStream;
   let stream = ByteStream::from(vec![1,2,3]);
   let stream = ByteStream::from(Bytes::from_static(b"hello!"));

2. From a file: ByteStreams created from a path can be retried. A new file descriptor will be opened if a retry occurs.

   #[cfg(feature = "tokio-rt")]
   use aws_smithy_http::byte_stream::ByteStream;
   let stream = ByteStream::from_path("big_file.csv");

3. From an SdkBody directly: For more advanced / custom use cases, a ByteStream can be created directly from an SdkBody. When created from an SdkBody, care must be taken to ensure retriability. An SdkBody is retryable when constructed from in-memory data or when using SdkBody::retryable.

   use aws_smithy_http::byte_stream::ByteStream;
   use aws_smithy_http::body::SdkBody;
   use bytes::Bytes;
   let (mut tx, channel_body) = hyper::Body::channel();
   // this will not be retryable because the SDK has no way to replay this stream
   let stream = ByteStream::new(SdkBody::from(channel_body));
   tx.send_data(Bytes::from_static(b"hello world!"));
   tx.send_data(Bytes::from_static(b"hello again!"));
   // NOTE! You must ensure that `tx` is dropped to ensure that EOF is sent

Implementations

impl ByteStream

pub fn new(body: SdkBody) -> Self

pub fn from_static(bytes: &'static [u8]) -> Self

pub fn into_inner(self) -> SdkBody

Consumes the ByteStream, returning the wrapped SdkBody

pub async fn collect(self) -> Result<AggregatedBytes, Error>

Read all the data from this ByteStream into memory

If an error in the underlying stream is encountered, ByteStreamError is returned.

Data is read into an AggregatedBytes that stores data non-contiguously as it was received over the network. If a contiguous slice is required, use into_bytes().

use bytes::Bytes;
use aws_smithy_http::body;
use aws_smithy_http::body::SdkBody;
use aws_smithy_http::byte_stream::{ByteStream, Error};
async fn get_data() {
    let stream = ByteStream::new(SdkBody::from("hello!"));
    let data: Result<Bytes, Error> = stream.collect().await.map(|data| data.into_bytes());
}

pub fn read_from() -> FsBuilder

Available on crate feature rt-tokio only.

Returns a FsBuilder, allowing you to build a ByteStream with full control over how the file is read (eg. specifying the length of the file or the size of the buffer used to read the file).

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

async fn bytestream_from_file() -> ByteStream {
    let bytestream = ByteStream::read_from()
        .path("docs/some-large-file.csv")
        // 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)
        .length(Length::Exact(123_456))
        .build()
        .await
        .expect("valid path");
    bytestream
}

pub async fn from_path(path: impl AsRef<Path>) -> Result<Self, Error>

Available on crate feature rt-tokio only.

Create a ByteStream that streams data from the filesystem

This function creates a retryable ByteStream for a given path. The returned ByteStream will provide a size hint when used as an HTTP body. If the request fails, the read will begin again by reloading the file handle.

Warning

The contents of the file MUST not change during retries. The length & checksum of the file will be cached. If the contents of the file change, the operation will almost certainly fail.

Furthermore, a partial write MAY seek in the file and resume from the previous location.

Note: If you want more control, such as specifying the size of the buffer used to read the file or the length of the file, use a FsBuilder as returned from ByteStream::read_from

Examples

use aws_smithy_http::byte_stream::ByteStream;
use std::path::Path;
 async fn make_bytestream() -> ByteStream {
    ByteStream::from_path("docs/rows.csv").await.expect("file should be readable")
}

pub async fn from_file(file: File) -> Result<Self, Error>

👎 Deprecated since 0.40.0:

Prefer the more extensible ByteStream::read_from() API

Available on crate feature rt-tokio only.

Create a ByteStream from a file

NOTE: This will NOT result in a retryable ByteStream. For a ByteStream that can be retried in the case of upstream failures, use ByteStream::from_path

pub fn with_body_callback(
    &mut self,
    body_callback: Box<dyn BodyCallback>
) -> &mut Self

Set a callback on this ByteStream. The callback’s methods will be called at various points throughout this ByteStream’s life cycle. See the BodyCallback trait for more information.

pub fn into_async_read(self) -> impl AsyncRead

Convert this ByteStream into a struct that implements AsyncRead.

Example

use tokio::io::{BufReader, AsyncBufReadExt};
use aws_smithy_http::byte_stream::ByteStream;

let mut lines =  BufReader::new(my_bytestream.into_async_read()).lines();
while let Some(line) = lines.next_line().await? {
  // Do something line by line
}

Trait Implementations

impl Debug for ByteStream

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for ByteStream

fn default() -> Self

Returns the “default value” for a type.

impl From<Body> for ByteStream

fn from(input: Body) -> Self

Converts to this type from the input type.

impl From<Bytes> for ByteStream

Construct a retryable ByteStream from bytes::Bytes

fn from(input: Bytes) -> Self

Converts to this type from the input type.

impl From<SdkBody> for ByteStream

fn from(inp: SdkBody) -> Self

Converts to this type from the input type.

impl From<Vec<u8, Global>> for ByteStream

Construct a retryable ByteStream from a Vec<u8>.

This will convert the Vec<u8> into bytes::Bytes to enable efficient retries.

fn from(input: Vec<u8>) -> Self

Converts to this type from the input type.

impl Stream for ByteStream

type Item = Result<Bytes, Error>

Values yielded by the stream.

fn poll_next(
    self: Pin<&mut Self>,
    cx: &mut Context<'_>
) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.

impl<'pin> Unpin for ByteStream where
    __ByteStream<'pin>: Unpin,