Group pubsub¶

Modules > pubsub

More...

Public Functions¶

Detailed Description¶

The Publisher publishes to a subject and fans out every message to all connected Subscribers. Subscribers subscribe and receive every frame published after they connect — a slow or absent subscriber simply misses frames (core NATS PUB/SUB has no queuing/replay).

Public Functions Documentation¶

function dp_pub_create¶

Create a Publisher and connect to endpoint .

dp_pub_t * dp_pub_create (
    const char * endpoint,
    dp_sample_type_t sample_type
) 

Parameters:

• endpoint NATS endpoint, e.g. "nats://127.0.0.1:4222/iq".
• sample_type Sample format that will be sent.

Returns:

Non-NULL context on success, NULL on failure.

function dp_pub_create_tlm¶

Create a Publisher that emits telemetry frames.

dp_pub_t * dp_pub_create_tlm (
    const char * endpoint
) 

A separate constructor because DP_KIND_TLM is a frame kind rather than a sample format: there is no BLUE code to pass to dp_pub_create(), and a publisher that emits records does not also emit I/Q. Send with dp_pub_send_tlm16().

Parameters:

• endpoint nats://host:port/subject.

Returns:

Publisher handle, or NULL on failure.

function dp_pub_destroy¶

Destroy a Publisher context and release all resources.

void dp_pub_destroy (
    dp_pub_t * ctx
) 

Parameters:

• ctx Publisher context (may be NULL).

function dp_pub_flush¶

Wait until the server has everything published so far.

int dp_pub_flush (
    dp_pub_t * ctx,
    int timeout_ms
) 

dp_pub_send_* hands the frame to the client and returns; the client writes it in the background. That is what makes publishing fast, and it means "the send returned" is not "the server has it". This waits for a round trip, so when it returns DP_OK everything published before it has arrived.

You do NOT need this before destroy: the NATS client flushes what is buffered when the connection closes. But it does so best-effort with a 500 ms cap and no way to report failure, so a backlog that cannot drain in half a second is dropped silently — and on a link slower than loopback that is not a large backlog. Call this when losing the tail would matter, and you get a budget you chose and an answer you can act on.

It is also the only way to ask the question WITHOUT closing: a long-lived publisher that wants "everything up to here is on the server" has nothing else to call. (PUSH does not need it — the JetStream publish is server-acked before it returns — and REQ/REP flush on every message already.)

A drained shutdown does not need one. dp_stream_drain() ends with this same flush as its final phase, so flush belongs at checkpoints a drain does not cover — confirming a batch is on the server before treating it as complete — not in front of a drain.

Parameters:

• ctx Any send-capable context (dp_pub_t / dp_push_t / dp_req_t / dp_rep_t are the same underlying type).
• timeout_ms How long to wait; <= 0 uses 2000 ms.

Returns:

DP_OK when the server has it, DP_ERR_TIMEOUT if the budget ran out with data still pending, DP_ERR_INVALID for a NULL context.

function dp_pub_send_cf32¶

Send an array of CF32 samples via a Publisher.

int dp_pub_send_cf32 (
    dp_pub_t * ctx,
    const float _Complex * samples,
    size_t num_samples,
    double sample_rate,
    double center_freq
) 

Parameters:

• ctx Publisher context.
• samples Interleaved int32_t I/Q pairs; length 2×num_samples.
• num_samples Number of complex samples.
• sample_rate Sample rate in Hz.
• center_freq Centre frequency in Hz.

Returns:

DP_OK (0) on success, negative error code on failure.

function dp_pub_send_cf64¶

Send an array of CF64 samples via a Publisher.

int dp_pub_send_cf64 (
    dp_pub_t * ctx,
    const double _Complex * samples,
    size_t num_samples,
    double sample_rate,
    double center_freq
) 

Parameters:

• ctx Publisher context.
• samples Interleaved int32_t I/Q pairs; length 2×num_samples.
• num_samples Number of complex samples.
• sample_rate Sample rate in Hz.
• center_freq Centre frequency in Hz.

Returns:

DP_OK (0) on success, negative error code on failure.

function dp_pub_send_ci16¶

Send an array of CI16 samples via a Publisher.

int dp_pub_send_ci16 (
    dp_pub_t * ctx,
    const int16_t * samples,
    size_t num_samples,
    double sample_rate,
    double center_freq
) 

Parameters:

• ctx Publisher context.
• samples Interleaved int32_t I/Q pairs; length 2×num_samples.
• num_samples Number of complex samples.
• sample_rate Sample rate in Hz.
• center_freq Centre frequency in Hz.

Returns:

DP_OK (0) on success, negative error code on failure.

function dp_pub_send_ci32¶

Send an array of CI32 samples via a Publisher.

int dp_pub_send_ci32 (
    dp_pub_t * ctx,
    const int32_t * samples,
    size_t num_samples,
    double sample_rate,
    double center_freq
) 

Parameters:

• ctx Publisher context.
• samples Interleaved int32_t I/Q pairs; length 2×num_samples.
• num_samples Number of complex samples.
• sample_rate Sample rate in Hz.
• center_freq Centre frequency in Hz.

Returns:

DP_OK (0) on success, negative error code on failure.

function dp_pub_send_ci8¶

Send an array of CI8 samples via a Publisher.

int dp_pub_send_ci8 (
    dp_pub_t * ctx,
    const int8_t * samples,
    size_t num_samples,
    double sample_rate,
    double center_freq
) 

Parameters:

• ctx Publisher context.
• samples Interleaved int32_t I/Q pairs; length 2×num_samples.
• num_samples Number of complex samples.
• sample_rate Sample rate in Hz.
• center_freq Centre frequency in Hz.

Returns:

DP_OK (0) on success, negative error code on failure.

function dp_pub_send_eos¶

Tell subscribers the stream has ended.

int dp_pub_send_eos (
    dp_pub_t * ctx
) 

Publishes a zero-payload DP_KIND_EOS frame. A receiving *_recv reports DP_ERR_EOF instead of handing back an empty frame, so a consumer learns the sender finished rather than inferring it from silence — which is the inference this whole contract exists to remove.

Send it before dp_stream_drain(), not after. A drain cannot be reversed and refuses sends once it reaches its publish-flushing phase, so an EOS issued after one may simply not go. The ordered shutdown is: stop producing, send EOS, drain, destroy.

What it does NOT promise. PUB/SUB is at-most-once (§9), so this frame can be dropped like any other: it turns the common case from "wait forever" into "finish promptly", not from unreliable into guaranteed, and a subscriber that must not hang on a lost marker still needs a timeout. PUSH/PULL delivers it at-least-once, so it may arrive more than once and a handler must be idempotent.

Parameters:

• ctx Any send-capable context.

Returns:

DP_OK once handed to the client, DP_ERR_INVALID for a NULL context, DP_ERR_CLOSED if the context is already draining or closed.

function dp_pub_send_tlm16¶

Send an array of 16-byte telemetry records via a Publisher.

int dp_pub_send_tlm16 (
    dp_pub_t * ctx,
    const void * records,
    size_t num_records,
    double sample_rate,
    double center_freq
) 

The payload is num_records packed dp_tlm_rec_t (see dp_tlm/dp_tlm_core.h) — the header's num_samples counts records and sample_type is TLM16. Kept const void * so the wire layer stays decoupled from the telemetry component; the dp_tlm_sink_* helper (stream/tlm_sink.h) is the intended caller.

Parameters:

• ctx Publisher context.
• records Packed 16-byte records.
• num_records Record count.
• sample_rate Wire-header field; 0.0 if not meaningful.
• center_freq Wire-header field; 0.0 if not meaningful.

Returns:

DP_OK (0) on success, negative error code on failure.

function dp_stream_drain¶

Shut a context down gracefully: drain, then closed.

int dp_stream_drain (
    dp_pub_t * ctx,
    int timeout_ms
) 

The ordered shutdown, and the one a signal handler's exit path wants. The client stops accepting new deliveries, lets what is in flight finish, flushes everything pending, and then closes.

It waits for the connection to reach CLOSED before returning, and that is the part worth having in the library rather than in every caller: natsConnection_Drain returns immediately and does the work in the background, so a process that exits when it returns abandons exactly the work the drain was for. Getting that wrong looks like success.

Against dp_pub_flush(): flush answers "does the server have what I published", and the context keeps working afterwards. Drain answers "let everything finish, then stop", and the context is finished when it returns — call the matching *_destroy next, which is then just the free.

Drain last, after your application has stopped producing. A drain cannot be reversed, and a send issued while one is in progress is racing its phases: it may slip through while subscriptions drain, or be refused once the connection reaches its publish-flushing phase. Do not publish a "shutting down" notice after calling this and assume it went.

Because this waits for CLOSED, a single-threaded caller does not have to reason about that race: once it has returned, a send is refused with DP_ERR_CLOSED, deterministically. The race is real only for a thread still publishing while another drains.

Size timeout_ms to the slowest thing the drain has to wait for, with margin: cutting a drain off mid-write every deploy is worse than waiting. doppler's own receive is synchronous — there is no message handler to finish — so the wait is dominated by flushing whatever is still buffered, and the 5 s default is generous for a link that is keeping up. A slow or congested link, or a large backlog, wants more.

Parameters:

• ctx Any context.
• timeout_ms How long to wait for CLOSED; <= 0 uses 5000 ms.

Returns:

DP_OK once closed, DP_ERR_TIMEOUT if the budget ran out with the drain still in progress (the context is still safe to destroy), DP_ERR_INVALID for a NULL context.

function dp_sub_create¶

Create a Subscriber and connect to endpoint .

dp_sub_t * dp_sub_create (
    const char * endpoint
) 

Subscribes to all topics (empty topic filter).

Parameters:

• endpoint NATS endpoint, e.g. "nats://127.0.0.1:4222/iq".

Returns:

Non-NULL context on success, NULL on failure.

function dp_sub_destroy¶

Destroy a Subscriber context and release all resources.

void dp_sub_destroy (
    dp_sub_t * ctx
) 

Parameters:

• ctx Subscriber context (may be NULL).

function dp_sub_recv¶

Receive one frame from a Subscriber socket (zero-copy).

int dp_sub_recv (
    dp_sub_t * ctx,
    dp_msg_t ** msg,
    dp_header_t * header
) 

On success, *msg is set to a message handle whose data buffer is valid until dp_msg_free() is called. Use dp_msg_data() to access the sample pointer.

Parameters:

• ctx Subscriber context.
• msg Set to a zero-copy message handle.
• header Set to the frame metadata.

Returns:

DP_OK on success, DP_ERR_TIMEOUT on timeout, DP_ERR_EOF when the sender has finished (no message is produced, so there is nothing to free), negative on error.

function dp_sub_set_timeout¶

Set receive timeout for a Subscriber socket.

void dp_sub_set_timeout (
    dp_sub_t * ctx,
    int timeout_ms
) 

Parameters:

• ctx Subscriber context.
• timeout_ms Timeout in milliseconds (-1 = infinite, 0 = non-blocking).