nats clib Messaging / Pub-Sub

NATS cloud-native messaging for PascalAI. Pub/Sub, Request/Reply, Queue groups, and JetStream persistence — all backed by a single ~20 MB server binary with zero-config core operation.

ppm install nats

Overview

nats wraps libnatsc — the official NATS C client — to give PascalAI programs access to the full NATS messaging stack. NATS is designed for cloud-native applications: microservice communication, IoT telemetry, event streaming, and distributed task queues.

The core server is a single ~20 MB binary with no external dependencies. Subjects use dot notation (orders.us.new) with two wildcard tokens for flexible routing. For durability, JetStream adds persistence, replay, and at-least-once delivery on top of the same subject namespace.

CLib package

Requires libnatsc-dev. Build from source: https://github.com/nats-io/nats.c — follow the README cmake instructions, then sudo make install.

Core Patterns

PatternHow it worksDelivery guarantee
Pub/SubPublishers send to a subject; all matching subscribers receive the message independentlyAt-most-once (fire and forget)
Request/ReplyRequester publishes with an auto-generated reply subject; a responder sends back exactly one replyAt-most-once; timeout controlled by caller
Queue groupsMultiple subscribers join the same named queue; NATS delivers each message to exactly one member of the groupAt-most-once; load-balanced
JetStreamMessages stored in a server-side stream; consumers replay from any offset and acknowledge each messageAt-least-once or exactly-once

Quick Start

Pub/Sub

uses nats;

var conn := Connect('nats://localhost:4222');

{ Subscribe to all sensor temperature readings }
var sub := Subscribe(conn, 'sensors.*.temp');

{ Publish a reading from sensor-42 }
Publish(conn, 'sensors.42.temp', '{"v":23.5}');

var msg := NextMessage(sub, 2000);
if msg <> 0 then
begin
  var m := GetMessage(msg);
  WriteLn('Subject: ' + m.Subject);
  WriteLn('Data:    ' + m.Data);
  FreeMessage(msg);
end;

Unsubscribe(sub);
Close(conn);

Request/Reply

uses nats;

var conn := Connect('nats://localhost:4222');

{ Synchronous RPC — blocks until reply or timeout }
var reply := Request(conn, 'users.lookup', '{"id":99}', 3000);
WriteLn('Reply: ' + reply.Data);

Close(conn);

Queue groups (load balancing)

uses nats;

{ Two workers join the same queue group "workers".
  Each task message is delivered to exactly one of them. }
var conn1 := Connect('nats://localhost:4222');
var conn2 := Connect('nats://localhost:4222');

var sub1 := QueueSubscribe(conn1, 'tasks.process', 'workers');
var sub2 := QueueSubscribe(conn2, 'tasks.process', 'workers');

Publish(conn1, 'tasks.process', 'job-001');  { goes to sub1 OR sub2, never both }

Close(conn1);
Close(conn2);

JetStream persistence

uses nats;

var conn := Connect('nats://localhost:4222');
var js   := NewJetStream(conn);

{ Create (or update) a stream that captures all order events }
var cfg: TNATSJSConfig;
cfg.Name        := 'ORDERS';
cfg.Subjects    := ['orders.>'];
cfg.MaxMessages := -1;
cfg.MaxBytes    := -1;
cfg.Replicas    := 1;
cfg.Storage     := 0;  { 0 = file (persistent) }
cfg.AckWait     := 30000;
cfg.MaxDeliver  := 5;
AddStream(js, cfg);

{ Publish persistently — returns the stream sequence number }
var seq := JSPublish(js, 'orders.new', '{"order_id":"A-001"}');
WriteLn('Stored at sequence ' + IntToStr(seq));

{ Subscribe with a durable consumer — survives restarts }
var sub := JSSubscribe(js, 'orders.>', 'order-processor');
var msg := NextMessage(sub, 5000);
if msg <> 0 then
begin
  var m := GetMessage(msg);
  WriteLn('Order: ' + m.Data + ' seq=' + IntToStr(m.Sequence));
  JSAck(msg);      { acknowledge — prevents redelivery }
  FreeMessage(msg);
end;

FreeJetStream(js);
Close(conn);

Connection API

A TNATSConn handle represents a connection to a NATS server. Keep one connection per process and reuse it for all operations — connections are multiplexed and cheap to share.

FunctionDescription
Connect(URL)Connect to nats://host:4222. Returns a connection handle.
ConnectEx(Opts)Connect with full TNATSOptions (credentials, TLS, reconnect policy, client name)
Close(Conn)Close the connection immediately
Drain(Conn)Wait for all in-flight messages to be delivered, then close gracefully
Status(Conn)Returns TNATSConnStatus: Disconnected, Connecting, Connected, Closed, Reconnecting, Draining
IsConnected(Conn)Quick boolean check; equivalent to Status = csConnected
ConnectedURL(Conn)Returns the URL of the server currently serving this connection

Publish API

Publishing is non-blocking and fire-and-forget in core NATS. Use Flush to ensure all buffered messages have been sent before checking results or closing the connection.

FunctionDescription
Publish(Conn, Subject, Data)Publish a string payload to a subject. Returns immediately.
PublishRequest(Conn, Subject, Reply, Data)Publish with an explicit reply subject. Used when implementing custom Request/Reply servers.
Flush(Conn, TimeoutMs)Block until all outgoing messages are flushed to the server, or the timeout elapses.

Subscribe API

Subscriptions filter messages by subject pattern. Wildcards make it easy to subscribe to families of subjects without registering each one.

FunctionDescription
Subscribe(Conn, Subject)Subscribe to a subject pattern. Returns a TNATSSub handle.
QueueSubscribe(Conn, Subject, Queue)Subscribe within a named queue group. Only one member of the group receives each message.
Unsubscribe(Sub)Remove the subscription immediately
AutoUnsubscribe(Sub, N)Automatically unsubscribe after receiving N messages (pass 0 to unsubscribe now)
Wildcard syntax

* matches exactly one subject token: sensors.*.temp matches sensors.42.temp but not sensors.us.east.temp.
> matches one or more tokens: orders.> matches orders.new, orders.us.new, and anything deeper.

Receiving Messages API

Pull messages from a subscription handle with NextMessage. Always free the handle after use to avoid memory leaks.

FunctionDescription
NextMessage(Sub, TimeoutMs)Wait up to TimeoutMs ms for the next message. Returns 0 on timeout.
GetMessage(Msg)Decode a message handle into a TNATSMsg_t record (Subject, Reply, Data, Headers, Sequence)
FreeMessage(Msg)Release the native message handle. Call after every non-zero NextMessage result.

Request/Reply API

NATS implements RPC-style calls with auto-generated inbox subjects. The server is a regular subscriber that calls Reply using the msg.Reply field.

FunctionDescription
Request(Conn, Subject, Data, TimeoutMs)Send a request and block until a reply arrives or the timeout elapses. Returns a TNATSMsg_t directly.
Reply(Conn, ReplySubject, Data)Send a reply to a request message. Use msg.Reply as the ReplySubject.
uses nats;

{ Server side — subscribe and reply to each request }
var srv  := Connect('nats://localhost:4222');
var sub  := Subscribe(srv, 'math.double');

var req := NextMessage(sub, 10000);
if req <> 0 then
begin
  var m := GetMessage(req);
  Reply(srv, m.Reply, '42');   { send answer back to inbox }
  FreeMessage(req);
end;

{ Client side }
var cli   := Connect('nats://localhost:4222');
var reply := Request(cli, 'math.double', '21', 2000);
WriteLn('Answer: ' + reply.Data);   { '42' }

Close(srv);
Close(cli);

JetStream API

JetStream turns NATS subjects into persistent, replayable streams. Messages survive server restarts and can be delivered to consumers that were offline when the message was published.

FunctionDescription
NewJetStream(Conn)Create a JetStream context from a connection. Required before any JS call.
FreeJetStream(JS)Release the JetStream context
AddStream(JS, Cfg)Create or update a stream with a TNATSJSConfig. Idempotent.
JSPublish(JS, Subject, Data)Publish a message to JetStream. Returns the stream sequence number confirming persistence.
JSSubscribe(JS, Subject, Durable)Subscribe with a named durable consumer. The server remembers the consumer position across restarts.
JSAck(Msg)Acknowledge a JetStream message. Prevents redelivery.
JSNak(Msg)Negatively acknowledge. Triggers redelivery after AckWait ms.

TNATSOptions

Pass to ConnectEx for full control over connection behaviour.

var opts: TNATSOptions;
opts.URL          := 'nats://user:secret@nats.internal:4222';
opts.TLSCACert    := '/etc/ssl/nats-ca.pem';
opts.MaxReconnect := -1;     { infinite reconnect }
opts.ReconnectWait:= 2000;   { 2 s between attempts }
opts.Timeout      := 5000;   { connect timeout }
opts.Name         := 'order-service';  { visible in NATS monitoring }
var conn := ConnectEx(opts);
FieldTypeDescription
URLstringServer URL. Supports nats://host:port and nats://user:pass@host:port
Username / PasswordstringBasic authentication credentials
TokenstringToken-based authentication
TLSCACertstringPEM CA certificate for TLS verification
TLSClientCert / TLSClientKeystringMutual TLS client certificate and key
MaxReconnectIntegerMax reconnect attempts; -1 = infinite
ReconnectWaitIntegerMilliseconds between reconnect attempts
PingInterval / MaxPingsOutIntegerServer keepalive ping frequency and max unanswered pings
TimeoutIntegerInitial connect timeout in milliseconds
NamestringClient name, shown in NATS server monitoring endpoints

TNATSJSConfig

Configuration record for AddStream. Define retention limits and replication for a JetStream stream.

FieldDescription
NameUnique stream name (e.g. ORDERS). Required.
SubjectsArray of subject patterns this stream captures (e.g. ['orders.>'])
MaxMessagesMaximum messages retained; -1 = unlimited
MaxBytesMaximum storage in bytes; -1 = unlimited
MaxAgeMessage TTL in seconds; 0 = never expire
ReplicasNumber of replicas in a clustered NATS deployment
Storage0 = file (survives restarts), 1 = memory only
AckWaitMilliseconds before unacknowledged messages are redelivered
MaxDeliverMaximum redelivery attempts before a message is discarded

TNATSMsg_t

Decoded message record returned by GetMessage and Request.

FieldDescription
SubjectThe subject this message was published to
ReplyReply-to inbox subject. Non-empty for Request/Reply messages. Pass to Reply().
DataMessage payload as a string
HeadersArray of Key/Value header pairs (NATS 2.2+)
SequenceJetStream stream sequence number (0 for core NATS messages)

Comparison

vsDifference
KafkaNATS is simpler to operate and has lower end-to-end latency. Kafka has higher sustained throughput and stronger ordering guarantees at scale.
MQTTNATS is faster and has richer subject-based routing. MQTT has QoS levels (0/1/2); use JetStream instead of QoS 2 with NATS.
ZeroMQNATS has a central broker which makes clustering, monitoring, and ops much simpler. ZeroMQ is brokerless and requires you to manage topology manually.

Utility

FunctionDescription
NATSVersion()Returns the version string of the underlying libnatsc C library

Package Info

Version1.0.0
Typeclib
C Librarylibnatsc-dev
Authorgustavo

System Requirements

libnatsc-dev

Build from source:
github.com/nats-io/nats.c

NATS server:
~20 MB single binary
nats.io/download

Types

TNATSConn — connection handle
TNATSSub — subscription handle
TNATSMsg — raw message handle
TNATSJSContext — JetStream context
TNATSStream — JetStream stream
TNATSOptions — connect options
TNATSMsg_t — decoded message
TNATSJSConfig — stream config
TNATSConnStatus — connection state

Functions

  • Connect / ConnectEx
  • Close / Drain
  • Status / IsConnected
  • ConnectedURL
  • Publish / PublishRequest
  • Flush
  • Subscribe / QueueSubscribe
  • Unsubscribe / AutoUnsubscribe
  • NextMessage / GetMessage
  • FreeMessage
  • Request / Reply
  • NewJetStream / FreeJetStream
  • AddStream
  • JSPublish / JSSubscribe
  • JSAck / JSNak
  • NATSVersion

Wildcard Syntax

* — one token
> — one or more tokens

Examples:
sensors.*.temp
orders.>

See Also