Struct pravega_client::byte::writer::ByteWriter

pub struct ByteWriter { /* private fields */ }

Allow for writing raw bytes directly to a segment.

ByteWriter does not frame, attach headers, or otherwise modify the bytes written to it in any way. So unlike EventWriter the data written cannot be split apart when read. As such, any bytes written by this API can ONLY be read using ByteReader.

Atomicity

If buf length is less than or equal to 8 MiB, the entire buffer will be written atomically. If buf length is greater than 8 MiB, only the first 8 MiB will be written, and it will be written atomically. In either case, the actual number of bytes written will be returned and those bytes are written atomically.

Parallelism

Multiple ByteWriters write to the same segment as this will result in interleaved data, which is not desirable in most cases. ByteWriter uses Conditional Append to make sure that writers are aware of the content in the segment. If another process writes data to the segment after this one began writing, all subsequent writes from this writer will not be written and flush will fail. This prevents data from being accidentally interleaved.

Backpressure

Write has a backpressure mechanism. Internally, it uses Channel to send event to Reactor for processing. Channel can has a limited capacity, when its capacity is reached, any further write will not be accepted until enough space has been freed in the Channel.

Retry

The ByteWriter implementation provides retry logic to handle connection failures and service host failures. Internal retries will not violate the exactly once semantic so it is better to rely on them than to wrap this with custom retry logic.

Examples

use pravega_client_config::ClientConfigBuilder;
use pravega_client::client_factory::ClientFactoryAsync;
use pravega_client_shared::ScopedStream;
use pravega_client::byte::ByteWriter;
use tokio::runtime::Handle;

#[tokio::main]
async fn main() {
    // assuming Pravega controller is running at endpoint `localhost:9090`
    let config = ClientConfigBuilder::default()
        .controller_uri("localhost:9090")
        .build()
        .expect("creating config");

    let handle = Handle::current();
    let client_factory = ClientFactoryAsync::new(config, handle);

    // assuming scope:myscope, stream:mystream exist.
    // notice that this stream should be a fixed sized single segment stream
    let stream = ScopedStream::from("myscope/mystream");

    let mut byte_writer = client_factory.create_byte_writer(stream).await;

    let payload = "hello world".to_string().into_bytes();

    // It doesn't mean the data is persisted on the server side
    // when write method returns Ok, user should call flush to ensure
    // all data has been acknowledged by the server.
    byte_writer.write(&payload).await.expect("write");
    byte_writer.flush().await.expect("flush");
}

Implementations

impl ByteWriter

pub async fn write(&mut self, buf: &[u8]) -> Result<usize, Error>

Writes the given data to the server asynchronously. It doesn’t mean the data is persisted on the server side when this method returns Ok, user should call flush to ensure all data has been acknowledged by the server.

Examples

let mut byte_writer = client_factory.create_byte_writer(segment).await;
let payload = vec![0; 8];
let size = byte_writer.write(&payload).await;

pub async fn flush(&mut self) -> Result<(), Error>

Flush data.

It will wait until all pending appends have acknowledgment.

Examples

let mut byte_writer = client_factory.create_byte_writer(segment).await;
let payload = vec![0; 8];
let size = byte_writer.write(&payload).await;
byte_writer.flush().await;

pub async fn seal(&mut self) -> Result<(), Error>

Seal the segment and no further writes are allowed.

Examples

let mut byte_writer = client_factory.create_byte_writer(segment).await;
byte_writer.seal().await.expect("seal segment");

pub async fn truncate_data_before(&self, offset: i64) -> Result<(), Error>

Truncate data before a given offset for the segment. No reads are allowed before truncation point after calling this method.

Examples

let byte_writer = client_factory.create_byte_writer(segment).await;
byte_writer.truncate_data_before(1024).await.expect("truncate segment");

pub fn current_offset(&self) -> u64

Track the current write position for this writer.

Examples

let byte_writer = client_factory.create_byte_writer(segment).await;
let offset = byte_writer.current_write_offset();

pub async fn seek_to_tail(&mut self)

Seek to the tail of the segment.

This method is useful for tail reads.

Examples

let mut byte_writer = client_factory.create_byte_writer(segment).await;
byte_writer.seek_to_tail_async().await;

pub async fn reset(&mut self) -> Result<(), Error>

Reset the internal Reactor, making it ready for new appends.

Use this method if you want to continue to append after ConditionalCheckFailure error. It will clear all pending events and set the Reactor ready.

Examples

if let Err(Error::ConditionalCheckFailure(_e)) = byte_writer.flush().await {
    byte_writer.reset().await.expect("reset");
    byte_writer.seek_to_tail().await;
}
byte_writer.write(&payload).await.expect("write");

Trait Implementations

impl Drop for ByteWriter

fn drop(&mut self)

Executes the destructor for this type.