NATS C Client

By default, the NATS library creates a dedicated thread for each connection to handle socket reads. To reduce thread overhead or integrate with existing event-driven architectures, you can use an event loop adapter (e.g., libuv or libevent).

Integration Steps

1. Create your event loop instance (e.g., uv_default_loop()).
2. Use natsOptions_SetEventLoop() to provide the loop and the necessary adapter functions (Attach, Read, Write, Detach).
3. Run your event loop.

Critical Warning: Publishing and Request-Reply

When using an event loop, publishing is asynchronous; data is placed in a buffer and sent when the event loop notifies the library that the socket is writable.

Do not call blocking or request-reply functions from the thread running the event loop, such as:

• natsConnection_Request()
• natsConnection_Flush()
• natsConnection_FlushTimeout()

If you call these from the event loop thread, the data may never be sent because the loop is blocked, causing the calls to timeout. For natsConnection_Request(), use natsConnection_PublishRequest() and register a subscriber for the response instead.

// Example using libuv
uv_loop_t *uvLoop = uv_default_loop();

natsOptions_SetEventLoop(opts,
                         (void*) uvLoop,
                         natsLibuv_Attach,
                         natsLibuv_Read,
                         natsLibuv_Write,
                         natsLibuv_Detach);

natsConnection_Connect(&conn, opts);
natsConnection_Subscribe(&sub, conn, subj, onMsg, NULL);

uv_run(uvLoop, UV_RUN_DEFAULT);

Durable queue groups combine the load-balancing benefits of queue groups with the persistence of durable subscriptions. This allows a group to maintain its state even when no members are currently connected.

Key Behaviors:

• Creation: Created by providing a DurableName via stanSubOptions_SetDurableName when calling stanConnection_QueueSubscribe.
• Naming: The server creates a group named DurableName:QueueName. Because of this, the : character is not allowed in the durable name.
• Persistence: Unlike non-durable groups, the group is not removed when the last member leaves. It is only destroyed when the last member calls Unsubscribe.
• Resuming: When a new member joins a durable queue group, it resumes from the last position recorded for that group, receiving any unacknowledged messages left by previous members.

// Create a durable queue subscriber on "foo" for group "bar"
stanSubOptions *subOpts = NULL;
stanSubOptions_Create(&subOpts);
stanSubOptions_SetDurableName(subOpts, "mydurablegroup");

stanConnection_QueueSubscribe(&qsub, "foo", "bar", onMsg, NULL, subOpts);

The library automatically attempts to reconnect by default.

• Reconnection Configuration: Use natsOptions_SetAllowReconnect(), natsOptions_SetMaxReconnect(), etc., to control the behavior.
• URL Randomization: If natsOptions_SetNoRadomize() is not set to true, the list of URLs is randomized on connection. On disconnect, the library tries the next URL in the list.
• Cluster Awareness: If the NATS server has connect URL advertise enabled, the client will automatically learn about new servers added to the cluster and add them to its connection pool.
• Buffering during Disconnect: While disconnected, publish and new subscription calls are buffered in memory. When reconnected, this buffer is flushed to the server.
• Lifecycle Callbacks: To monitor connection state, use:
  • natsOptions_SetDisconnectedCB()
  • natsOptions_SetReconnectedCB()
  • natsOptions_SetClosedCB() (This is a final event; the connection is no longer valid after this is called).

NATS Streaming provides At-Least-Once delivery. By default, the client library automatically acknowledges messages after the subscriber's handler is invoked.

To control exactly when a message is acknowledged (e.g., to defer acknowledgment until after heavy I/O is complete), you must:

1. Enable manual acknowledgment mode using stanSubOptions_SetManualAckMode(subOpts, true).
2. Set an acknowledgment wait timeout using stanSubOptions_SetAckWait(subOpts, timeout_ms).
3. Explicitly call stanSubscription_AckMsg(sub, msg) within your message handler.

// Subscribe with manual ack mode, and set AckWait to 60 seconds
stanSubOptions_Create(&subOpts);
stanSubOptions_SetManualAckMode(subOpts, true);
stanSubOptions_SetAckWait(subOpts, 60000);
stanConnection_Subscribe(&sub, sc, "foo", onMsg, NULL, subOpts);

// In the callback
void
onMsg(stanConnection *sc, stanSubscription *sub, const char *channel, stanMsg *msg, void *closure)
{
    // ack message before performing I/O intensive operation
    stanSubscription_AckMsg(sub, msg);

    printf("Received a message on %s: %.*s\n",
        channel,
        stanMsg_GetDataLength(msg),
        stanMsg_GetData(msg));
}

NATS Streaming allows clients to start receiving messages from different points in the stream using stanSubOptions. You can create these options using stanSubOptions_Create and then apply one of the following strategies:

• Last Received: Start with the most recently published value.
• Deliver All: Receive all messages currently stored in the stream.
• Specific Sequence: Start at a specific message sequence number.
• Time Delta: Start at messages that were stored a certain amount of time ago (value in milliseconds).

After configuring the options, pass the stanSubOptions pointer to stanConnection_Subscribe.

// Create a Subscription Options:
stanSubOptions *subOpts = NULL;
stanSubOptions_Create(&subOpts);

// Subscribe starting with most recently published value
stanSubOptions_StartWithLastReceived(subOpts);

// OR: Receive all stored messages
stanSubOptions_DeliverAllAvailable(subOpts);

// OR: Receive messages starting at a specific sequence number
stanSubOptions_StartAtSequence(subOpts, 22);

// OR: Start at messages that were stored 30 seconds ago. Value is expressed in milliseconds.
stanSubOptions_StartAtTimeDelta(subOpts, 30000);

// Create the subscription with options
stanConnection_Subscribe(&sub, sc, "foo", onMsg, NULL, subOpts);

Queue groups allow you to distribute messages across multiple subscribers. When multiple subscribers join the same queue group (using the same queue name), each message is delivered to only one member of the group.

Key Behaviors:

• Automatic Creation: A group is created when the first member joins.
• Start Position: When a new member joins an existing group, its requested start position is ignored. The member starts receiving messages from the last position in the group.
• Leaving a Group: Members leave by calling stanSubscription_Unsubscribe or closing their connection. Unacknowledged messages are reassigned to remaining members.
• Closing a Group: A group is removed from the server once all members have left. The next QueueSubscribe with that name will create a brand new group where start positions will be respected.

stanConnection_Connect(&sc, "test-cluster", "clientid", NULL);

// Create a queue subscriber on "foo" for group "bar"
stanConnection_QueueSubscribe(&qsub1, "foo", "bar", onMsg, NULL, NULL);

// Add a second member to the same group
stanConnection_QueueSubscribe(&qsub2, "foo", "bar", onMsg, NULL, NULL);

// Normal subscribers still work on the same subject
stanConnection_Subscribe(&sub, "foo", onMsg, NULL, NULL);

Durable subscriptions allow a client to pick up exactly where it left off in a stream without manually tracking sequence numbers. By assigning a DurableName via stanSubOptions_SetDurableName, the NATS Streaming server tracks the last acknowledged message for that specific clientID + durableName combination.

Workflow:

1. Connect with a clientID.
2. Create subscription options and set a durable name.
3. Subscribe to a subject.
4. If the client disconnects, reconnecting with the same clientID and using the same durableName will cause the server to deliver only the messages published since the last acknowledgment.

stanConnection_Connect(&sc, "test-cluster", "client-123", NULL);

// Create subscription options
stanSubOptions *subOpts = NULL;
stanSubOptions_Create(&subOpts);

// Set a durable name
stanSubOptions_SetDurableName(subOpts, "my-durable");

// Subscribe
stanConnection_Subscribe(&sub, sc, "foo", onMsg, NULL, subOpts);