Skip to main content

process_nats_request

exporter

Function process_nats_request 

Source 
pub(crate) async fn process_nats_request(
    nc: Client,
    reply: Subject,
    payload: Vec<u8>,
    state: AppState,
) -> Result<()>
Expand description

Validates and serves one NATS export request payload.

Every frame is published to reply with the header Avena-Export-Frame set to the frame name. The sequence is:

  1. meta: JSON {"type":"meta","fileName":...,"contentType":"text/csv"}.
  2. chunk (one or more): raw CSV bytes, not JSON. The first chunk starts with the header line. See NatsCsvStreamer for sizes and acks.
  3. summary: JSON {"type":"summary","bytesSent":N,"missingChannels":[...]}.
  4. complete: JSON {"type":"complete"}.

Validation failures (no channels, end before start, non-CSV format) publish only an error frame ({"type":"error","message":...}) and return Ok, with no complete frame. If ack_subject is set and not blank, the function subscribes to it before streaming. A cancel message on that subject stops the export before the next chunk with an ExportCancelled error.

§Arguments

  • nc - Connected NATS client.
  • reply - Reply subject of the request message.
  • payload - JSON-encoded ExportRequest.
  • state - Shared state holding the Parquet root.

§Errors

Returns an error if the payload is not a valid ExportRequest, if start or end is not RFC 3339, if a publish, flush or the ack subscription fails, if an ack does not arrive in time, or if a channel’s day directory cannot be listed. The caller (run_worker) turns these into an error frame, except ExportCancelled, which it only logs.