Struct NatsCsvStreamer 

pub(crate) struct NatsCsvStreamer {
    pub(crate) client: Client,
    pub(crate) reply: Subject,
    pub(crate) ack_sub: Option<Subscriber>,
    pub(crate) chunks_since_flush: usize,
    pub(crate) unacked: usize,
    pub(crate) chunks_sent: usize,
    pub(crate) bytes_sent: usize,
    pub(crate) asset: u32,
    pub(crate) start: DateTime<Utc>,
    pub(crate) end: DateTime<Utc>,
    pub(crate) filters: Option<ExportFilters>,
}

CSV streamer for NATS worker exports.

Builds the CSV on a CSV thread (spawn_csv) and publishes it as chunk frames (raw bytes with the header Avena-Export-Frame: chunk) to the reply subject, each at least Self::CHUNK_SIZE (512 KiB) except the last, exceeding it by less than one row. After every Self::FLUSH_EVERY_CHUNKS (8) chunks it flushes the NATS client and, when the request named an ack_subject, waits until the client has published one ack message per chunk sent, allowing Self::ACK_TIMEOUT_SECS (30 s) for each. The content of ack messages is ignored, except that a cancel message (see is_cancel_message) stops the export: messages already on the ack subject are read before every chunk, so at most the chunk being published when the cancel arrives still goes out. Without an ack subject the worker only flushes, and nothing stops it from outrunning a slow client.

Fields

client: Client

Connected NATS client.

reply: Subject

Reply subject of the export request.

ack_sub: Option<Subscriber>

Subscription to the request’s ack_subject, if one was given.

chunks_since_flush: usize

Chunks published since the last forced client flush.

unacked: usize

Chunks published and not yet acknowledged by the client.

chunks_sent: usize

Chunks published so far.

bytes_sent: usize

CSV bytes published so far.

asset: u32

Asset number used to locate the Parquet partitions.

start: DateTime<Utc>

Inclusive start of the export range.

end: DateTime<Utc>

Inclusive end of the export range.

filters: Option<ExportFilters>

Filters asked for; None for a plain export.

Implementations

impl NatsCsvStreamer

pub(crate) const CHUNK_SIZE: usize

Buffer size in bytes at which a chunk frame is published (512 KiB).

pub(crate) const FLUSH_EVERY_CHUNKS: usize = 8

Number of chunks published between forced NATS flushes and between ack waits.

pub(crate) const ACK_TIMEOUT_SECS: u64 = 30

Maximum time to wait for each client chunk acknowledgement, in seconds.

pub(crate) fn new( client: Client, reply: Subject, ack_sub: Option<Subscriber>, asset: u32, start: DateTime<Utc>, end: DateTime<Utc>, ) -> Self

Creates a NATS CSV streamer.

Arguments

• client - Connected NATS client.
• reply - Reply subject of the export request.
• ack_sub - Subscription to the client’s ack subject, or None to skip ack waits.
• asset - Asset number used to locate the Parquet partitions.
• start - Inclusive start of the export range.
• end - Inclusive end of the export range.

pub(crate) fn with_filters(self, filters: Option<ExportFilters>) -> Self

Sets the filters of the export. With Some, the header gets a filtered_value column and every row a value for it.

pub(crate) async fn stream_channels( &mut self, parquet_root: &Path, channels: &[u8], ) -> Result<Vec<u8>>

Streams all requested channels, one after another, in the given order.

The CSV is built by spawn_csv and every chunk is published with Self::publish_chunk as soon as it is ready.

Arguments

• parquet_root - Root of the Parquet archive.
• channels - Channels to export.

Returns

The channels that produced no rows in the range, in request order.

Errors

Returns ExportCancelled if the client cancelled, an error if a publish, flush or ack wait fails, or the first channel error from spawn_csv, prefixed with channel NN:, after the chunks completed before it.

pub(crate) async fn publish_chunk(&mut self, data: Vec<u8>) -> Result<()>

Publishes one chunk frame.

First reads what has arrived on the ack subject (Self::poll_ack_subject), so a cancel stops the export before the next chunk. Adds the chunk to bytes_sent, flushes the NATS client every Self::FLUSH_EVERY_CHUNKS chunks, and calls Self::wait_for_acks once that many chunks are unacknowledged.

Errors

Returns ExportCancelled if the client cancelled, or an error if the publish or flush fails, or if Self::wait_for_acks fails.

pub(crate) fn cancelled(&self) -> Error

Returns the error for a cancel seen after the chunks sent so far.

pub(crate) fn handle_ack_message(&mut self, message: &Message) -> Result<()>

Counts one message from the ack subject: an ack, or a cancel.

Errors

Returns ExportCancelled if the message is a cancel.

pub(crate) fn poll_ack_subject(&mut self) -> Result<()>

Reads, without waiting, every message already received on the ack subject.

Errors

Returns ExportCancelled if one of them is a cancel.

pub(crate) async fn wait_for_acks(&mut self) -> Result<()>

Waits until the client has acknowledged every chunk sent so far.

Returns at once if there is no ack subscription or no unacknowledged chunk. Otherwise flushes the client so the chunks actually leave, then waits for messages on the ack subject until every chunk is acknowledged, allowing Self::ACK_TIMEOUT_SECS for each.

Errors

Returns ExportCancelled if a cancel arrives, or an error if the flush fails, if any ack takes longer than Self::ACK_TIMEOUT_SECS, or if the ack subscription closes.

pub(crate) async fn finish(self, missing_channels: Vec<u8>) -> Result<()>

Waits for outstanding acks and sends summary and completion frames.

Call after Self::stream_channels, which publishes every chunk. Waits for all outstanding acks, then publishes summary ({"type":"summary","bytesSent":N,"missingChannels":[...]} with the channels sorted and deduplicated) and complete ({"type":"complete"}), and flushes the client.

Arguments

• missing_channels - Channels that produced no rows.

Errors

Returns an error if a publish, flush or ack wait fails.