Struct aws_sdk_s3::primitives::ByteStream

source ·
pub struct ByteStream { /* private fields */ }
Expand description

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_types::byte_stream::{ByteStream, AggregatedBytes};
    use aws_smithy_types::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 .next() or .try_next():

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

    use aws_smithy_types::byte_stream::{ByteStream, AggregatedBytes, error::Error};
    use aws_smithy_types::body::SdkBody;
    
    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::AsyncBufRead.

    use aws_smithy_types::byte_stream::ByteStream;
    use aws_smithy_types::body::SdkBody;
    use tokio::io::AsyncBufReadExt;
    #[cfg(feature = "rt-tokio")]
    async fn example() -> std::io::Result<()> {
       let stream = ByteStream::new(SdkBody::from("hello!\nThis is some data"));
       // Convert the stream to a BufReader
       let buf_reader = 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_types::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_types::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_types::byte_stream::ByteStream;
    use aws_smithy_types::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_body_0_4(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§

source§

impl ByteStream

source

pub fn from_body_0_4<T, E>(body: T) -> ByteStream
where T: Body<Data = Bytes, Error = E> + Send + Sync + 'static, E: Into<Box<dyn Error + Sync + Send>> + 'static,

Construct a ByteStream from a type that implements http_body_0_4::Body<Data = Bytes>.

Note: This is only available with http-body-0-4-x enabled.

source§

impl ByteStream

source

pub fn from_body_1_x<T, E>(body: T) -> ByteStream
where T: Body<Data = Bytes, Error = E> + Send + Sync + 'static, E: Into<Box<dyn Error + Sync + Send>> + 'static,

Construct a ByteStream from a type that implements http_body_1_0::Body<Data = Bytes>.

source§

impl ByteStream

source

pub fn new(body: SdkBody) -> ByteStream

Create a new ByteStream from an SdkBody.

source

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

Create a new ByteStream from a static byte slice.

source

pub fn into_inner(self) -> SdkBody

Consume the ByteStream, returning the wrapped SdkBody.

source

pub async fn next(&mut self) -> Option<Result<Bytes, Error>>

Return the next item in the ByteStream.

There is also a sibling method try_next, which returns a Result<Option<Bytes>, Error> instead of an Option<Result<Bytes, Error>>.

source

pub fn poll_next( self: Pin<&mut ByteStream>, cx: &mut Context<'_>, ) -> Poll<Option<Result<Bytes, Error>>>

Attempt to pull out the next value of this stream, returning None if the stream is exhausted.

source

pub async fn try_next(&mut self) -> Result<Option<Bytes>, Error>

Consume and return the next item in the ByteStream or return an error if an error is encountered.

Similar to the next method, but this returns a Result<Option<Bytes>, Error> rather than an Option<Result<Bytes, Error>>, making for easy use with the ? operator.

source

pub fn size_hint(&self) -> (u64, Option<u64>)

Return the bounds on the remaining length of the ByteStream.

source

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_types::body;
use aws_smithy_types::body::SdkBody;
use aws_smithy_types::byte_stream::{ByteStream, error::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());
}
source

pub fn read_from() -> FsBuilder

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_types::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
}
source

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

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_types::byte_stream::ByteStream;
use std::path::Path;
 async fn make_bytestream() -> ByteStream {
    ByteStream::from_path("docs/rows.csv").await.expect("file should be readable")
}
source

pub fn into_async_read(self) -> impl AsyncBufRead

Convert this ByteStream into a struct that implements AsyncBufRead.

§Example
use tokio::io::AsyncBufReadExt;
use aws_smithy_types::byte_stream::ByteStream;

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

pub fn map( self, f: impl Fn(SdkBody) -> SdkBody + Send + Sync + 'static, ) -> ByteStream

Given a function to modify an SdkBody, run it on the SdkBody inside this Bytestream. returning a new Bytestream.

Trait Implementations§

source§

impl Debug for ByteStream

source§

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

Formats the value using the given formatter. Read more
source§

impl Default for ByteStream

source§

fn default() -> ByteStream

Returns the “default value” for a type. Read more
source§

impl From<Bytes> for ByteStream

Construct a retryable ByteStream from bytes::Bytes.

source§

fn from(input: Bytes) -> ByteStream

Converts to this type from the input type.
source§

impl From<SdkBody> for ByteStream

source§

fn from(inp: SdkBody) -> ByteStream

Converts to this type from the input type.
source§

impl From<Vec<u8>> for ByteStream

Construct a retryable ByteStream from a Vec<u8>.

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

source§

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

Converts to this type from the input type.
source§

impl<'__pin> Unpin for ByteStream
where __Origin<'__pin>: Unpin,

Auto Trait Implementations§

Blanket Implementations§

source§

impl<T> Any for T
where T: 'static + ?Sized,

source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
source§

impl<T> Borrow<T> for T
where T: ?Sized,

source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
source§

impl<T> From<T> for T

source§

fn from(t: T) -> T

Returns the argument unchanged.

source§

impl<T> Instrument for T

source§

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more
source§

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more
source§

impl<T, U> Into<U> for T
where U: From<T>,

source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

source§

impl<Unshared, Shared> IntoShared<Shared> for Unshared
where Shared: FromUnshared<Unshared>,

source§

fn into_shared(self) -> Shared

Creates a shared type from an unshared type.
source§

impl<T> Same for T

§

type Output = T

Should always be Self
source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

§

type Error = Infallible

The type returned in the event of a conversion error.
source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
source§

impl<V, T> VZip<V> for T
where V: MultiLane<T>,

source§

fn vzip(self) -> V

source§

impl<T> WithSubscriber for T

source§

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more
source§

fn with_current_subscriber(self) -> WithDispatch<Self>

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more