nats.c
v3.9.2
A C client for the NATS messaging system.
Go here for the online documentation, and check the frequently asked questions.
This NATS Client implementation is heavily based on the NATS GO Client. There is support for Mac OS/X, Linux and Windows (although we don't have specific platform support matrix).
There are several package managers with NATS C client library available. If you know one that is not in this list, please submit a PR to add it!
First, download the source code:
git clone [email protected]:nats-io/nats.c.git .
To build the library, use CMake. Note that by default the NATS Streaming API will be built and included in the NATS library. See below if you do not want to build the Streaming related APIs.
Make sure that CMake is added to your path. If building on Windows, open a command shell from the Visual Studio Tools menu, and select the appropriate command shell (x64 or x86 for 64 or 32 bit builds respectively). You will also probably need to run this with administrator privileges.
Create a build directory (any name would work) from the root source tree, and cd into it. Then issue this command for the first time:
cmake ..
In some architectures, you may experience a compilation error for mutex.c.o because there is no support
for the assembler instruction that we use to yield when spinning trying to acquire a lock.
You may get this sort of build error:
/tmp/cc1Yp7sD.s: Assembler messages:
/tmp/cc1Yp7sD.s:302: Error: selected processor does not support ARM mode `yield'
src/CMakeFiles/nats_static.dir/build.make:542: recipe for target 'src/CMakeFiles/nats_static.dir/unix/mutex.c.o' failed
If that's the case, you can solve this by enabling the NATS_BUILD_NO_SPIN flag (or use -DNATS_NO_SPIN if you compile without CMake):
cmake .. -DNATS_BUILD_NO_SPIN=ON
If you had previously built the library, you may need to do a make clean, or simply delete and re-create the build directory before executing the cmake command.
To build on Windows, you would need to select the build generator. For instance, to select nmake, you would run:
cmake .. -G "NMake Makefiles"
Running cmake -h would give you the list of possible options and all the generator names.
Alternatively, you can run the GUI version. From that same build command shell, start the GUI:
c:program files (x86)CMakebincmake-gui.exe
If you started with an empty build directory, you would need to select the source and build directory, then click Configure. Here, you will be able to select from the drop-down box the name of the build generator. When done, click Generate. Then you can go back to your command shell, or Visual Studio and build.
To modify some of the build options, you need to edit the cache and rebuild.
make edit_cache
Note that if you build on Windows and have selected "NMake Makefiles", replace all following references to make with nmake.
Editing the cache allows you to select the build type (Debug, Release, etc), the architecture (64 or 32bit), and so on.
The default target will build everything, that is, the static and shared NATS libraries and also the examples and the test program. Each are located in their respective directories under your build directory: src, examples and test.
make install
Will copy both the static and shared libraries in the folder install/lib and the public headers in install/include.
By default, the library is built with TLS support. You can disable this from the cmake gui make edit_cache and switch the NATS_BUILD_WITH_TLS option to OFF, or pass the option directly to the cmake command:
cmake .. -DNATS_BUILD_WITH_TLS=OFF
Starting 2.0.0, when building with TLS/SSL support, the server certificate's expected hostname is always verified. It means that the hostname provided in the URL(s) or through the option natsOptions_SetExpectedHostname() will be used to check the hostname present in the certificate. Prior to 2.0.0, the hostname would be verified only if the option natsOptions_SetExpectedHostname() was invoked.
Although we recommend leaving the new default behavior, you can restore the previous behavior by building the library with this option off:
cmake .. -DNATS_BUILD_TLS_FORCE_HOST_VERIFY=OFF
The NATS C client is built using APIs from the OpenSSL library. By default we use 3.0+ APIs. Since OpenSSL 1.0.2 is no longer supported, starting with NATS C Client v3.6.0 version, the CMake variable NATS_BUILD_TLS_USE_OPENSSL_1_1_API is now set to ON by default (if you are setting up a new environment) and will use OpenSSL APIs from 1.1+/3.0+ APIs. You will still be able to compile with the OpenSSL 1.0.2 library by setting this CMake option to OFF:
cmake .. -DNATS_BUILD_TLS_USE_OPENSSL_1_1_API=OFF
The variable NATS_BUILD_TLS_USE_OPENSSL_1_1_API is deprecated, meaning that in the future this option will simply be removed and only OpenSSL 3.0+ APIs will be used. The code in the library using older OpenSSL APIs will be removed too.
Note that the variable NATS_BUILD_WITH_TLS_CLIENT_METHOD that was deprecated in v2.0.0 has now been removed.
Since the NATS C client dynamically links to the OpenSSL library, you need to make sure that you are then running your application against an OpenSSL 1.1+/3.0+ library.
If you want to link to the static OpenSSL library, you need to delete the CMakeCache.txt and regenerate it with the additional option:
rm CMakeCache.txt
cmake .. -DNATS_BUILD_OPENSSL_STATIC_LIBS=ON
Then call make (or equivalent depending on your platform) and this should ensure that the library (and examples and/or test suite executable) are linked against the OpenSSL library, if it was found by CMake.
When building the library with Streaming support, the NATS library uses the libprotobuf-c library.
When cmake runs for the first time (or after removing CMakeCache.txt and calling cmake .. again), it is looking for the libprotobuf-c library. If it does not find it, a message is printed and the build process fails.
CMake searches for the library in directories where libraries are usually found. However, if you want to specify a specific directory where the library is located, you need to do this:
cmake .. -DNATS_PROTOBUF_DIR=
The static library will be used by default. If you want to change that, or if the library has not the expected name, you need to do this:
# Use the library named mylibproto.so located at /my/location
cmake .. -DNATS_PROTOBUF_LIBRARY=/my/location/mylibproto.so
The two could be combined if the include header is located in a different directory
# Use the library named mylibproto.so located at /my/location and the directory protobuf-c/ containing protobuf-c.h located at /my/other/location
cmake .. -DNATS_PROTOBUF_LIBRARY=/my/location/mylibproto.so -DNATS_PROTOBUF_DIR=/my/other/location
If you don't want to build the NATS Streaming APIs to be included in the NATS library:
cmake .. -DNATS_BUILD_STREAMING=OFF
When using the new NATS 2.0 security features, the library needs to sign some "nonce" sent by the server during a connect or reconnect.
We use Ed25519 public-key signature. The library comes with some code to perform the signature.
In most case, it will be fine, but if performance is an issue (especially if you plan to use the natsConnection_Sign() function a lot), you will have the option to build with the Libsodium library.
Follow instructions on how to install the libsodium library here.
On macOS, you could use brew:
brew install libsodium
On Linux, you could use apt-get
apt-get install libsodium-dev
Once installed, you can rebuild the NATS C client by first enabling the use of the libsodium library:
cmake .. -DNATS_BUILD_USE_SODIUM=ON
If you have the libsodium library installed in a non standard location that CMake cannot find, you can specify the location of this directory:
cmake .. -DNATS_BUILD_USE_SODIUM=ON -DNATS_SODIUM_DIR=/my/path/to/libsodium
On platforms where valgrind is available, you can run the tests with memory checks.
Here is an example:
make test ARGS="-T memcheck"
Or, you can invoke directly the ctest program:
ctest -T memcheck -V -I 1,4
The above command would run the tests with valgrind (-T memcheck), with verbose output (-V), and run the tests from 1 to 4 (-I 1,4).
If you add a test to test/test.c, you need to add it into the list_test.txt
file. Each entry contains just the test name, the function must be named
identically, with a test_ prefix. The list is in alphabetical order, but it
does not need to be, you can add anywhere.
If you are adding a benchmark, it should be added to the list_bench.txt. These
tests are labeled differently (-L 'bench') and executed separately on CI.
You need to re-run cmake for the changes to take effect:
cmake ..
-- Configuring done
-- Generating done
-- Build files have been written to: /home/ivan/nats.c/build
You can use the following environment variables to influence the testsuite behavior.
When running with memory check, timing changes and overall performance is slower. The following variable allows the testsuite to adjust some of values used during the test:
export NATS_TEST_VALGRIND=yes
On Windows, it would be set instead of export.
When running the tests in verbose mode, the following environment variable allows you to see the server output from within the test itself. Without this option, the server output is silenced:
export NATS_TEST_KEEP_SERVER_OUTPUT=yes
If you want to change the default server executable name (nats-server.exe) or specify a specific location, use this environment variable:
set NATS_TEST_SERVER_EXE=c:testnats-server.exe
The public API has been documented using Doxygen.
To generate the documentation, go to the doc directory and type the following command:
doxygen DoxyFile.NATS.Client
If you toggle the build of the Streaming APIs, and the documentation is no longer matching
what is being built, you can update the documentation by switching the NATS_UPDATE_DOC build flag and rebuild the documentation.
From the build directory:
cmake .. -DNATS_UPDATE_DOC=ON
make
cd /doc
doxygen DoxyFile.NATS.Client
The generated documentation will be located in the html directory. To see the documentation, point your browser to the file index.html in that directory.
Go here for the online documentation.
The source code is also quite documented.
This section lists important changes such as deprecation notices, etc...
2.0.0
This version introduces the security concepts used by NATS Server 2.0.0 and therefore aligns with the server version. There have been new APIs introduced, but the most important change is the new default behavior with TLS connections:
When establishing a secure connection, the server certificate's hostname is now always verified, regardless if the user has invoked natsOptions_SetExpectedHostname(). This may break applications that were for instance using an IP to connect to a server that had only the hostname in the certificate. This can be solved by changing your application to use the hostname in the URL or make use of natsOptions_SetExpectedHostname(). If this is not possible, you can restore the old behavior by building the library with the new behavior disabled. See #tls-support for more information.
This repository used to include precompiled libraries of libprotobuf-c for macOS, Linux and Windows along with the header files (in the /pbuf directory).
We have now removed this directory and require that the user installs the libprotobuf-c library separately. See the building instructions to specify the library location if CMake cannot find it directly.
1.8.0
natsConnStatus enum values have been prefixed with NATS_CONN_STATUS_. If your application is
not using referencing any original value, such as CONNECTED or CLOSED, etc.. then there is nothing
that you need to do. If you do, you have two options:
NATS_BUILD_NO_PREFIX_CONNSTS. This can be done this way from the build directory:
cmake .. -DNATS_BUILD_NO_PREFIX_CONNSTS=ON
The examples/getstarted directory has a set of simple examples that are fully functional, yet very simple.
The goal is to demonstrate how easy to use the API is.
A more complex set of examples are in examples/ directory and can also be used to benchmark the client library.
Note that for simplicity, error checking is not performed here.
natsConnection *nc = NULL;
natsSubscription *sub = NULL;
natsMsg *msg = NULL;
// Connects to the default NATS Server running locally
natsConnection_ConnectTo(&nc, NATS_DEFAULT_URL);
// Connects to a server with username and password
natsConnection_ConnectTo(&nc, "nats://ivan:secret@localhost:4222");
// Connects to a server with token authentication
natsConnection_ConnectTo(&nc, "nats://myTopSecretAuthenticationToken@localhost:4222");
// Simple publisher, sending the given string to subject "foo"
natsConnection_PublishString(nc, "foo", "hello world");
// Publish binary data. Content is not interpreted as a string.
char data[] = {1, 2, 0, 4, 5};
natsConnection_Publish(nc, "foo", (const void*) data, 5);
// Simple asynchronous subscriber on subject foo, invoking message
// handler 'onMsg' when messages are received, and not providing a closure.
natsConnection_Subscribe(&sub, nc, "foo", onMsg, NULL);
// Simple synchronous subscriber
natsConnection_SubscribeSync(&sub, nc, "foo");
// Using a synchronous subscriber, gets the first message available, waiting
// up to 1000 milliseconds (1 second)
natsSubscription_NextMsg(&msg, sub, 1000);
// Destroy any message received (asynchronously or synchronously) or created
// by your application. Note that if 'msg' is NULL, the call has no effect.
natsMsg_Destroy(msg);
// Unsubscribing
natsSubscription_Unsubscribe(sub);
// Destroying the subscription (this will release the object, which may
// result in freeing the memory). After this call, the object must no
// longer be used.
natsSubscription_Destroy(sub);
// Publish requests to the given reply subject:
natsConnection_PublishRequestString(nc, "foo", "bar", "help!");
// Sends a request (internally creates an inbox) and Auto-Unsubscribe the
// internal subscriber, which means that the subscriber is unsubscribed
// when receiving the first response from potentially many repliers.
// This call will wait for the reply for up to 1000 milliseconds (1 second).
natsConnection_RequestString(&reply, nc, "foo", "help", 1000);
// Closing a connection (but not releasing the connection object)
natsConnection_Close(nc);
// When done with the object, free the memory. Note that this call
// closes the connection first, in other words, you could have simply
// this call instead of natsConnection_Close() followed by the destroy
// call.
natsConnection_Destroy(nc);
// Message handler
void
onMsg(natsConnection *nc, natsSubscription *sub, natsMsg *msg, void *closure)
{
// Prints the message, using the message getters:
printf("Received msg: %s - %.*sn",
natsMsg_GetSubject(msg),
natsMsg_GetDataLength(msg),
natsMsg_GetData(msg));
// Don't forget to destroy the message!
natsMsg_Destroy(msg);
}Support for JetStream starts with the version v3.0.0 of the library and NATS Server v2.2.0+, although getting JetStream
specific error codes requires the server at version v2.3.0+. Some of the configuration options are only available starting v2.3.3,
so we recommend that you use the latest NATS Server release to have a better experience.
Look at examples named js-xxx.c in the examples directory for examples on how to use the API.
The new objects and APIs are full documented in the online documentation.
// Connect to NATS
natsConnection_Connect(&conn, opts);
// Initialize and set some JetStream options
jsOptions jsOpts;
jsOptions_Init(&jsOpts);
jsOpts.PublishAsync.MaxPending = 256;
// Create JetStream Context
natsConnection_JetStream(&js, conn, &jsOpts);
// Simple Stream Publisher
js_Publish(&pa, js, "ORDERS.scratch", (const void*) "hello", 5, NULL, &jerr);
// Simple Async Stream Publisher
for (i=0; i < 500; i++)
{
js_PublishAsync(js, "ORDERS.scratch", (const void*) "hello", 5, NULL);
}
// Wait for up to 5 seconds to receive all publish acknowledgments.
jsPubOptions_Init(&jsPubOpts);
jsPubOpts.MaxWait = 5000;
js_PublishAsyncComplete(js, &jsPubOpts);
// One can get the list of all pending publish async messages,
// to either resend them or simply destroy them.
natsMsgList pending;
s = js_PublishAsyncGetPendingList(&pending, js);
if (s == NATS_OK)
{
int i;
for (i=0; i<pending.Count; i++)
{
// There could be a decision to resend these messages or not.
if (your_decision_to_resend(pending.Msgs[i]))
{
// If the call is successful, pending.Msgs[i] will be set
// to NULL so destroying the pending list will not destroy
// this message since the library has taken ownership back.
js_PublishMsgAsync(js, &(pending.Msgs[i]), NULL);
}
}
// Destroy the pending list object and all messages still in that list.
natsMsgList_Destroy(&pending);
}
// To create an asynchronous ephemeral consumer
js_Subscribe(&sub, js, "foo", myMsgHandler, myClosure, &jsOpts, NULL, &jerr);
// Same but use a subscription option to ask the callback to not do auto-ack.
jsSubOptions so;
jsSubOptions_Init(&so);
so.ManualAck = true;
js_Subscribe(&sub, js, "foo", myMsgHandler, myClosure, &jsOpts, &so, &jerr);
// Or to bind to an existing specific stream/durable:
jsSubOptions_Init(&so);
so.Stream = "MY_STREAM";
so.Consumer = "my_durable";
js_Subscribe(&sub, js, "foo", myMsgHandler, myClosure, &jsOpts, &so, &jerr);
// Synchronous subscription:
js_SubscribeSync(&sub, js, "foo", &jsOpts, &so, &jerr);jsStreamConfig cfg;
// Connect to NATS
natsConnection_Connect(&conn, opts);
// Create JetStream Context
natsConnection_JetStream(&js, conn, NULL);
// Initialize the configuration structure.
jsStreamConfig_Init(&cfg);
// Provide a name
cfg.Name = "ORDERS";
// Array of subjects and its size
cfg.Subjects = (const char*[1]){"ORDERS.*"};
cfg.SubjectsLen = 1;
// Create a Stream. If you are not interested in the returned jsStreamInfo object,
// you can pass NULL.
js_AddStream(NULL, js, &cfg, NULL, &jerr);
// Update a Stream
cfg.MaxBytes = 8;
js_UpdateStream(NULL, js, &cfg, NULL, &jerr);
// Delete a Stream
js_DeleteStream(js, "ORDERS", NULL, &jerr);EXPERIMENTAL FEATURE! We reserve the right to change the API without necessarily bumping the major version of the library.
A KeyValue store is a materialized view based on JetStream. A bucket is a stream and keys are subjects within that stream.
Some features require NATS Server v2.6.2, so we recommend that you use the latest NATS Server release to have a better experience.
The new objects and APIs are full documented in the online documentation.
Example of how to create a KeyValue store:
jsCtx *js = NULL;
kvStore *kv = NULL;
kvConfig kvc;
// Assume we got a JetStream context in `js`...
kvConfig_Init(&kvc);
kvc.Bucket = "KVS";
kvc.History = 10;
s = js_CreateKeyValue(&kv, js, &kvc);
// Do some stuff...
// This is to free the memory used by `kv` object,
// not delete the KeyValue store in the server
kvStore_Destroy(kv);This shows how to "bind" to an existing one:
jsCtx *js = NULL;
kvStore *kv = NULL;
// Assume we got a JetStream context in `js`...
s = js_KeyValue(&kv, ks, "KVS");
// Do some stuff...
// This is to free the memory used by `kv` object,
// not delete the KeyValue store in the server
kvStore_Destroy(kv);This is how to delete a KeyValue store in the server:
jsCtx *js = NULL;
// Assume we got a JetStream context in `js`...
s = js_DeleteKeyValue(js, "KVS");This is how to put a value for a given key:
kvStore *kv = NULL;
uint64_t rev = 0;
// Assume we got a kvStore...
s = kvStore_PutString(&rev, kv, "MY_KEY", "my value");
// If the one does not care about getting the revision, pass NULL:
s = kvStore_PutString(NULL, kv, "MY_KEY", "my value");The above places a value for a given key, but if instead one wants to make sure that the value is placed for the key only if it never existed before, one would call:
// Same note than before: if "rev" is not needed, pass NULL:
s = kvStore_CreateString(&rev, kv, "MY_KEY", "my value");One can update a key if and only if the last revision in the server matches the one passed to this API:
// This would update the key "MY_KEY" with the value "my updated value" only if the current revision (sequence number) for this key is 10.
s = kvStore_UpdateString(&rev, kv, "MY_KEY", "my updated value", 10);This is how to get a key:
kvStore *kv = NULL;
kvEntry *e = NULL;
// Assume we got a kvStore...
s = kvStore_Get(&e, kv, "MY_KEY");
// Then we can get some fields from the entry:
printf("Key value: %sn", kvEntry_ValueString(e));
// Once done with the entry, we need to destroy it to release memory.
// This is NOT deleting the key from the server.
kvEntry_Destroy(e);This is how to purge a key:
kvStore *kv = NULL;
// Assume we got a kvStore...
s = kvStore_Purge(kv, "MY_KEY");This will delete the key in the server:
kvStore *kv = NULL;
// Assume we got a kvStore...
s = kvStore_Delete(kv, "MY_KEY");To create a "watcher" for a given key:
kvWatcher *w = NULL;
kvWatchOptions o;
// Assume we got a kvStore...
// Say that we are not interested in getting the
// delete markers...
// Initialize a kvWatchOptions object:
kvWatchOptions_Init(&o);
o.IgnoreDeletes = true;
// Create the watcher
s = kvStore_Watch(&w, kv, "foo.*", &o);
// Check for error..
// Now get updates:
while (some_condition)
{
kvEntry *e = NULL;
// Wait for the next update for up to 5 seconds
s = kvWatcher_Next(&e, w, 5000);
// Do something with the entry...
// Destroy to release memory
kvEntry_Destroy(e);
}
// When done with the watcher, it needs to be destroyed to release memory:
kvWatcher_Destroy(w);To get the history of a key:
kvEntryList l;
int i;
// The list is defined on the stack and will be initilized/updated by this call:
s = kvStore_History(&l, kv, "MY_KEY", NULL);
for (i=0; i<l.Count; i++)
{
kvEntry *e = l.Entries[i];
// Do something with the entry...
}
// When done with the list, call this to free entries and the content of the list.
kvEntryList_Destroy(&l);
// In order to set a timeout to get the history, we need to do so through options:
kvWatchOptions o;
kvWatchOptions_Init(&o);
o.Timeout = 5000; // 5 seconds.
s = kvStore_History(&l, kv, "MY_KEY", &o);This is how you would get the keys of a bucket:
kvKeysList l;
int i;
// If no option is required, pass NULL as the last argument.
s = kvStore_Keys(&l, kv, NULL);
// Check error..
// Go over all keys:
for (i=0; i<l.Count; i++)
printf("Key: %sn", l.Keys[i]);
// When done, list need to be destroyed.
kvKeysList_Destroy(&l);
// If option need to be specified:
kvWatchOptions o;
kvWatchOptions_Init(&o);
o.Timeout = 5000; // 5 seconds.
s = kvStore_Keys(&l, kv, &o);Headers are available when connecting to servers at version 2.2.0+.
They closely resemble http headers. They are a map of key/value pairs, the value being an array of strings.
Headers allow users to add meta information about a message without interfering with the message payload.
Note that if an application attempts to send a message with a header when connected to a server that does not understand them, the publish call will return the error NATS_NO_SERVER_SUPPORT.
There is an API to know if the server currently connected to supports headers:
natsStatus s = natsConnection_HasHeaderSupport(conn);
if (s == NATS_NO_SERVER_SUPPORT)
// deal with server not supporting this feature.If the server understands headers but is about to deliver the message to a client that doesn't, the headers are stripped off so that the older clients can still receive the messsage. It is important to have all client and servers to a version that support headers if applications rely on headers.
For more details on the headers API, please get the example: examples/getstarted/headers.c.
The * wildcard matches any token, at any level of the subject:
natsConnection_Subscribe(&sub, nc, "foo.*.baz", onMsg, NULL);This subscriber would receive messages sent to:
It would not, however, receive messages on:
The > wildcard matches any length of the fail of a subject, and can only be the last token.
natsConnection_Subscribe(&sub, nc, "foo.>", onMsg, NULL);This subscriber would receive any message sent to:
However, it would not receive messages sent on:
Publishing on this subject would cause the two above subscriber to receive the message:
natsConnection_PublishString(nc, "foo.bar.baz", "got it?");All subscriptions with the same queue name will form a queue group. Each message will be delivered to only one subscriber per queue group, using queue sematics. You can have as many queue groups as you wish. Normal subscribers will continue to work as expected.
natsConnection_QueueSubscribe(&sub, nc, "foo", "job_workers", onMsg, NULL);(Note that the library needs to be built with TLS support - which is by default - for these APIs to work. See the Build chapter on how to build with or without TLS for more details).
An SSL/TLS connection is configured through the use of natsOptions. Depending on the level of security you desire, it can be as simple as setting the secure boolean to true on the natsOptions_SetSecure() call.
Even with full security (client verifying server certificate, and server requiring client certificates), the setup involves only a few calls.
// Here is the minimum to create a TLS/SSL connection:
// Create an options object.
natsOptions_Create(&opts);
// Set the secure flag.
natsOptions_SetSecure(opts, true);
// You may not need this, but suppose that the server certificate
// is self-signed and you would normally provide the root CA, but
// don't want to. You can disable the server certificate verification
// like this:
natsOptions_SkipServerVerification(opts, true);
// Connect now...
natsConnection_Connect(&nc, opts);
// That's it! On success you will have a secure connection with the server!
(...)
// This example shows what it takes to have a full SSL configuration,
// including server expected's hostname, root CA, client certificates
// and specific ciphers to use.
// Create an options object.
natsOptions_Create(&opts);
// Set the secure flag.
natsOptions_SetSecure(opts, true);
// For a server with a trusted chain built into the client host,
// simply designate the server name that is expected. Without this
// call, the server certificate is still verified, but not the
// hostname.
natsOptions_SetExpectedHostname(opts, "localhost");
// Instead, if you are using a self-signed cert and need to load in the CA.
natsOptions_LoadCATrustedCertificates(opts, caCertFileName);
// If the server requires client certificates, provide them along with the
// private key, all in one call.
natsOptions_LoadCertificatesChain(opts, certChainFileName, privateKeyFileName);
// You can also specify preferred ciphers if you want.
natsOptions_SetCiphers(opts, "-ALL:HIGH");
// Then simply pass the options object to the connect call:
natsConnection_Connect(&nc, opts);
// That's it! On success you will have a secure connection with the server!
