Struct AppendStream 

pub struct AppendStream { /* private fields */ }

A cloneable, bounded producer handle for asynchronous table appends. Dropping every handle drains admitted rows in the background while the Tokio runtime remains alive. Use shutdown() to observe settlement before exiting.

Implementations

impl AppendStream

pub async fn send<T>(&self, row: &T) -> Result<(), Error>

where T: Serialize + ?Sized,

Serializes and admits one row, waiting for bounded local capacity.

Success means local admission only. Use Self::flush or Self::shutdown to wait for remote settlement.

Examples found in repository

examples/import_ndjson.rs (line 51)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let path = std::env::args()
        .nth(1)
        .ok_or("usage: import_ndjson <events.ndjson>")?;
    let mut input = BufReader::new(File::open(path)?);
    let client = common::client()?;
    let stream = common::write_table(&client)?.append_stream().build()?;
    let admitted: Result<(), Box<dyn std::error::Error>> = async {
        let mut line_number = 0;
        loop {
            // Bound a single input record as well as the stream's queued bytes.
            let mut line = String::new();
            let bytes = input
                .by_ref()
                .take(8 * 1024 * 1024 + 1)
                .read_line(&mut line)?;
            if bytes == 0 {
                break;
            }
            line_number += 1;
            if bytes > 8 * 1024 * 1024 {
                return Err(format!("line {line_number} exceeds 8 MiB").into());
            }
            if line.trim().is_empty() {
                continue;
            }
            let row: serde_json::Value = serde_json::from_str(&line)
                .map_err(|error| format!("line {line_number}: {error}"))?;
            stream.send(&row).await?;
        }
        Ok(())
    }
    .await;
    // Settle the accepted prefix even if parsing or admission failed midway.
    let settled = stream.shutdown().await;
    println!("delivery: {:?}", stream.stats());
    settled?;
    admitted?;
    Ok(())
}

pub fn try_send<T>(&self, row: &T) -> Result<(), AppendTrySendError>

where T: Serialize + ?Sized,

Attempts synchronous local admission without waiting for capacity. Returns an error if the circuit is open or local capacity is full.

Examples found in repository

examples/telemetry.rs (lines 26-31)

fn track(stream: &AppendStream, request: u64) -> bool {
    match stream.try_send(&serde_json::json!({
        "event_id": Uuid::now_v7(),
        "occurred_at": jiff::Timestamp::now(),
        "name": "request.completed",
        "attributes": {"request": request, "status": 200},
    })) {
        // This means local admission only, not a remote commit.
        Ok(()) => true,
        Err(error) => {
            // Route this diagnostic to another sink, never back into `stream`.
            eprintln!("telemetry row dropped locally: {error}");
            false
        }
    }
}

pub async fn send_all<I>(&self, rows: I) -> Result<AppendAdmissionResult, Error>

where I: IntoIterator, I::Item: Serialize,

Admits an iterator of rows sequentially with bounded memory.

If admission fails partway, the error includes the accepted count. Call Self::shutdown to settle those rows; Self::stats retains cumulative counts.

Examples found in repository

examples/append_stream.rs (lines 24-35)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;
    let stream = table.append_stream().build()?;

    let admitted = stream
        .send_all([
            serde_json::json!({
                "id": 1,
                "occurred_at": jiff::Timestamp::now(),
                "name": "first",
            }),
            serde_json::json!({
                "id": 2,
                "occurred_at": jiff::Timestamp::now(),
                "name": "second",
            }),
        ])
        .await?;
    println!("accepted locally: {} rows", admitted.accepted_rows);

    // send_all() confirms only local admission. With the default strict policy,
    // successful shutdown confirms at least one commit acknowledgement per row.
    // Retrying an unknown outcome can insert duplicates.
    let report = stream.shutdown().await?;
    println!("committed remotely: {} rows", report.committed_rows);

    Ok(())
}

examples/bulk_append.rs (line 47)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // The default Stop policy gives strict barriers. Local memory and request
    // concurrency are bounded independently; concurrent batches have no
    // defined commit order.
    let stream = table
        .append_stream()
        .target_batch_bytes(32 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(1024 * 1024)
        .max_concurrent_batches(4)
        .attempt_timeout(Duration::from_secs(30))
        .build()?;

    let rows = (1..=10_000).map(|id| {
        serde_json::json!({
            "id": id,
            "occurred_at": jiff::Timestamp::now(),
            "name": format!("example-{id}"),
        })
    });

    let admission_error = match stream.send_all(rows).await {
        Ok(result) => {
            println!("accepted locally: {} rows", result.accepted_rows);
            None
        }
        Err(error) => {
            // There is no stream-wide abort or rollback. Settle the already
            // accepted prefix even if row production or admission stopped.
            eprintln!("row admission stopped; settling the accepted prefix: {error}");
            Some(error)
        }
    };

    match stream.shutdown().await {
        Ok(report) => {
            println!("committed remotely: {} rows", report.committed_rows);
            println!("final stream stats: {:?}", stream.stats());
        }
        Err(error) => {
            if matches!(
                error.append_details().map(|details| details.status),
                Some(AppendState::Unknown)
            ) {
                eprintln!("commit outcome is unknown; reconcile before replaying");
            } else {
                eprintln!("bulk append failed: {error}");
            }
            eprintln!("final stream stats: {:?}", stream.stats());
            return Err(error.into());
        }
    }

    if let Some(error) = admission_error {
        return Err(error.into());
    }
    Ok(())
}

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

Dispatches rows admitted before this barrier and waits for their outcomes.

The report covers activity since the previous barrier. On a Stop-mode failure, partial counts remain available in Self::stats.

Committed rows have at least one successful acknowledgement; earlier unknown attempts may also have committed. A failed barrier may include successful batches, so replaying its entire input can duplicate rows.

Once enqueued, dropping this future does not cancel remote settlement. Keep the future alive to receive its interval report; the latest completed report also remains observable through Self::stats.

pub async fn shutdown(&self) -> Result<AppendDeliveryReport, Error>

Permanently closes the stream after settling every admitted row.

This method is idempotent across cloned producer handles. Once polled, dropping its future stops waiting but settlement continues in the worker. Call shutdown again to observe completion. Failure reports remain in stats().

Examples found in repository

examples/append_stream.rs (line 42)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;
    let stream = table.append_stream().build()?;

    let admitted = stream
        .send_all([
            serde_json::json!({
                "id": 1,
                "occurred_at": jiff::Timestamp::now(),
                "name": "first",
            }),
            serde_json::json!({
                "id": 2,
                "occurred_at": jiff::Timestamp::now(),
                "name": "second",
            }),
        ])
        .await?;
    println!("accepted locally: {} rows", admitted.accepted_rows);

    // send_all() confirms only local admission. With the default strict policy,
    // successful shutdown confirms at least one commit acknowledgement per row.
    // Retrying an unknown outcome can insert duplicates.
    let report = stream.shutdown().await?;
    println!("committed remotely: {} rows", report.committed_rows);

    Ok(())
}

examples/import_ndjson.rs (line 57)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let path = std::env::args()
        .nth(1)
        .ok_or("usage: import_ndjson <events.ndjson>")?;
    let mut input = BufReader::new(File::open(path)?);
    let client = common::client()?;
    let stream = common::write_table(&client)?.append_stream().build()?;
    let admitted: Result<(), Box<dyn std::error::Error>> = async {
        let mut line_number = 0;
        loop {
            // Bound a single input record as well as the stream's queued bytes.
            let mut line = String::new();
            let bytes = input
                .by_ref()
                .take(8 * 1024 * 1024 + 1)
                .read_line(&mut line)?;
            if bytes == 0 {
                break;
            }
            line_number += 1;
            if bytes > 8 * 1024 * 1024 {
                return Err(format!("line {line_number} exceeds 8 MiB").into());
            }
            if line.trim().is_empty() {
                continue;
            }
            let row: serde_json::Value = serde_json::from_str(&line)
                .map_err(|error| format!("line {line_number}: {error}"))?;
            stream.send(&row).await?;
        }
        Ok(())
    }
    .await;
    // Settle the accepted prefix even if parsing or admission failed midway.
    let settled = stream.shutdown().await;
    println!("delivery: {:?}", stream.stats());
    settled?;
    admitted?;
    Ok(())
}

examples/telemetry.rs (line 75)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

examples/bulk_append.rs (line 60)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // The default Stop policy gives strict barriers. Local memory and request
    // concurrency are bounded independently; concurrent batches have no
    // defined commit order.
    let stream = table
        .append_stream()
        .target_batch_bytes(32 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(1024 * 1024)
        .max_concurrent_batches(4)
        .attempt_timeout(Duration::from_secs(30))
        .build()?;

    let rows = (1..=10_000).map(|id| {
        serde_json::json!({
            "id": id,
            "occurred_at": jiff::Timestamp::now(),
            "name": format!("example-{id}"),
        })
    });

    let admission_error = match stream.send_all(rows).await {
        Ok(result) => {
            println!("accepted locally: {} rows", result.accepted_rows);
            None
        }
        Err(error) => {
            // There is no stream-wide abort or rollback. Settle the already
            // accepted prefix even if row production or admission stopped.
            eprintln!("row admission stopped; settling the accepted prefix: {error}");
            Some(error)
        }
    };

    match stream.shutdown().await {
        Ok(report) => {
            println!("committed remotely: {} rows", report.committed_rows);
            println!("final stream stats: {:?}", stream.stats());
        }
        Err(error) => {
            if matches!(
                error.append_details().map(|details| details.status),
                Some(AppendState::Unknown)
            ) {
                eprintln!("commit outcome is unknown; reconcile before replaying");
            } else {
                eprintln!("bulk append failed: {error}");
            }
            eprintln!("final stream stats: {:?}", stream.stats());
            return Err(error.into());
        }
    }

    if let Some(error) = admission_error {
        return Err(error.into());
    }
    Ok(())
}

pub fn stats(&self) -> AppendStreamStats

Returns a synchronous snapshot suitable for metrics and shutdown diagnostics.

Examples found in repository

examples/import_ndjson.rs (line 58)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let path = std::env::args()
        .nth(1)
        .ok_or("usage: import_ndjson <events.ndjson>")?;
    let mut input = BufReader::new(File::open(path)?);
    let client = common::client()?;
    let stream = common::write_table(&client)?.append_stream().build()?;
    let admitted: Result<(), Box<dyn std::error::Error>> = async {
        let mut line_number = 0;
        loop {
            // Bound a single input record as well as the stream's queued bytes.
            let mut line = String::new();
            let bytes = input
                .by_ref()
                .take(8 * 1024 * 1024 + 1)
                .read_line(&mut line)?;
            if bytes == 0 {
                break;
            }
            line_number += 1;
            if bytes > 8 * 1024 * 1024 {
                return Err(format!("line {line_number} exceeds 8 MiB").into());
            }
            if line.trim().is_empty() {
                continue;
            }
            let row: serde_json::Value = serde_json::from_str(&line)
                .map_err(|error| format!("line {line_number}: {error}"))?;
            stream.send(&row).await?;
        }
        Ok(())
    }
    .await;
    // Settle the accepted prefix even if parsing or admission failed midway.
    let settled = stream.shutdown().await;
    println!("delivery: {:?}", stream.stats());
    settled?;
    admitted?;
    Ok(())
}

examples/telemetry.rs (line 79)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

examples/bulk_append.rs (line 63)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // The default Stop policy gives strict barriers. Local memory and request
    // concurrency are bounded independently; concurrent batches have no
    // defined commit order.
    let stream = table
        .append_stream()
        .target_batch_bytes(32 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(1024 * 1024)
        .max_concurrent_batches(4)
        .attempt_timeout(Duration::from_secs(30))
        .build()?;

    let rows = (1..=10_000).map(|id| {
        serde_json::json!({
            "id": id,
            "occurred_at": jiff::Timestamp::now(),
            "name": format!("example-{id}"),
        })
    });

    let admission_error = match stream.send_all(rows).await {
        Ok(result) => {
            println!("accepted locally: {} rows", result.accepted_rows);
            None
        }
        Err(error) => {
            // There is no stream-wide abort or rollback. Settle the already
            // accepted prefix even if row production or admission stopped.
            eprintln!("row admission stopped; settling the accepted prefix: {error}");
            Some(error)
        }
    };

    match stream.shutdown().await {
        Ok(report) => {
            println!("committed remotely: {} rows", report.committed_rows);
            println!("final stream stats: {:?}", stream.stats());
        }
        Err(error) => {
            if matches!(
                error.append_details().map(|details| details.status),
                Some(AppendState::Unknown)
            ) {
                eprintln!("commit outcome is unknown; reconcile before replaying");
            } else {
                eprintln!("bulk append failed: {error}");
            }
            eprintln!("final stream stats: {:?}", stream.stats());
            return Err(error.into());
        }
    }

    if let Some(error) = admission_error {
        return Err(error.into());
    }
    Ok(())
}

Trait Implementations

impl Clone for AppendStream

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.