Skip to content

Publish data with SAPI

This page explains how a SAPI client publishes sensor readings into KX Sensors, and the delivery trade-offs you choose between.

Publishing is how raw sensor data enters the system. A client registers as a publisher for a feed, publishes messages to it, and unregisters when it's finished. Messages travel over Reliable Transport (RT) to the Sensor Data Loaders (SDLs) that handle the feed.

Before you begin

The deployment needs an external EMS stream, and a feed mapped to that stream:

  • The stream is defined in ems.yaml with external: true.
  • The feed is defined in feed.yaml with a stream field naming the stream.

By convention the feed and its stream have the same name, but they don't have to. See Add an external feed.

Register as a publisher

registerPublisher takes the name of the feed you're publishing to, as registered in Discovery, and returns a handle. Registering connects the client to the RT endpoints that Discovery reports for that feed. The parameter is named topic or topicName in the language bindings, but it takes a feed name. It doesn't take an EMS stream name or an EMS topic.

KxsResponse response;
TopicHandle publisher = sapiRegisterPublisher(handle, "feed1", &response);

KxsResponse response = new();
IntPtr publisher = kxs.RegisterPublisher("feed1", response);

KxsResponse response = new KxsResponse();
long publisher = kxs.registerPublisher("feed1", response);

if (publisher <= 0) {
    LOG.error("RegisterPublisher failed, rc={} ac={}",
        response.getRC(), response.getAC());
    return;
}

response = KxsResponse()
publisher = sapi.register_publisher("feed1", response)

if not publisher:
    print(f"RegisterPublisher failed, rc={response.rc}")
    return

Register once and reuse the handle for the lifetime of the client. Don't register per message.

If the feed name doesn't resolve, registration fails and returns no handle. SAPI doesn't queue anything or retry. The response has rc and ac set to the error codes, with a message that starts "Unable to register as publisher". This happens when:

  • Discovery doesn't know the feed, or knows it but has no endpoints for it.
  • Every endpoint for the feed is filtered out because of a bridge mode or TLS mismatch between the client and the server.

If you add the feed to Discovery later, call registerPublisher again. SAPI applies Discovery updates only to publishers that have already registered successfully.

Publish a message

publish takes the publisher handle, a message type, the data payload, an optional header, an optional retry policy, and a response object.

Parameter Purpose
topicHandle Handle returned by registerPublisher.
msgType A 16-bit integer that identifies the kind of record you're publishing. SAPI copies it into the message header without validating it, and there's no default. Your KX Sensors environment defines the valid values.
data The payload to publish. SAPI puts no structure on it. See Build the payload.
header A dictionary of extra header fields. Optional. Pass only the fields your environment requires. SAPI fills in the standard header fields, and a key you supply overrides SAPI's value.
retryPolicy Retry policy. Its presence and timeout also determine whether the publish is synchronous — see Choose synchronous or asynchronous publishing.
response Object that receives the outcome.

Replace the msgType value of 100 in the examples with a value from your environment.

Build the payload

The payload is any object that the language binding can serialize: a K object in C, c.Flip or c.Dict in C# and Java, and a PyKX object or pandas DataFrame in Python. SAPI wraps whatever you pass in a message that the receiving service interprets, so the record layout comes from the environment you publish into, not from SAPI.

The following examples build one illustrative sensor reading as a single-row table. The real columns come from your environment's feed definition.

J now = 0;  // Nanoseconds since 2000.01.01D00:00:00. Replace with the current time.

K columns = ktn(KS, 4);
kS(columns)[0] = ss("equipmentID");
kS(columns)[1] = ss("ts");
kS(columns)[2] = ss("val");
kS(columns)[3] = ss("updateTS");

K equipmentID = ktn(KI, 1);
kI(equipmentID)[0] = 101;

K ts = ktn(KP, 1);
kJ(ts)[0] = now;

K val = ktn(KF, 1);
kF(val)[0] = 21.5;

K updateTS = ktn(KP, 1);
kJ(updateTS)[0] = now;

K values = knk(4, equipmentID, ts, val, updateTS);
K payload = xT(xD(columns, values));

string[] columns = { "equipmentID", "ts", "val", "updateTS" };
object[] columnData =
{
    new int[] { 101 },
    new DateTime[] { DateTime.UtcNow },
    new double[] { 21.5 },
    new DateTime[] { DateTime.UtcNow }
};

c.Flip payload = new(new c.Dict(columns, columnData));

String[] columns = { "equipmentID", "ts", "val", "updateTS" };
Object[] columnData = {
    new int[] { 101 },
    new Timestamp[] { new Timestamp(System.currentTimeMillis()) },
    new double[] { 21.5 },
    new Timestamp[] { new Timestamp(System.currentTimeMillis()) }
};

c.Flip payload = new c.Flip(new c.Dict(columns, columnData));

import pandas as pd
import pykx

now = pd.Timestamp.now()
payload = pykx.toq(pd.DataFrame({
    "equipmentID": pd.array([101], dtype="int32"),
    "ts": [now],
    "val": [21.5],
    "updateTS": [now],
}))

Publish the message

The following examples publish the payload without a header or a retry policy, which makes the publish asynchronous.

KxsResponse response;
bool ok = sapiPublish(publisher, 100, payload, NULL, NULL, &response);

if (!ok)
    printf("Publish failed, rc=%d ac=%d\n", response.rc, response.ac);

KxsResponse response = new();
bool ok = kxs.Publish(publisher, 100, payload, null, null, response);

if (!ok)
    Console.WriteLine($"Publish failed, rc={response.RC} ac={response.AC}");

KxsResponse response = new KxsResponse();
boolean ok = kxs.publish(publisher, (short) 100, payload, null, null, response);

if (!ok) {
    LOG.warn("Publish failed, rc={} ac={}", response.getRC(), response.getAC());
}

response = KxsResponse()
ok = sapi.publish(publisher, req_corr=0, msg_type=100, payload=payload,
                  header=None, retry_policy=None, response=response)

if not ok:
    print(f"Publish failed, rc={response.rc} ac={response.ac} ai={response.ai}")

Choose synchronous or asynchronous publishing

Synchronous. Pass a retry policy with a timeout greater than 0. The call returns when the SDL sends an explicit acknowledgement, which means the message passed SDL validation and was reliably persisted downstream. Nothing further is needed to guarantee delivery. If the acknowledgement doesn't arrive within the timeout, SAPI returns a timeout response and applies the retry policy.

Asynchronous. Pass no retry policy. SAPI queues the message and the call returns success immediately. Success means the client accepted the message, not that the server has processed it. Delivery to the SDL may still fail, and the SDL may still reject the message in validation.

Warning

A retry policy with a timeout of 0 publishes asynchronously, the same as passing no retry policy. Don't rely on this behavior. Pass no retry policy for an asynchronous publish.

Acknowledgements for synchronous publishes arrive over UDP, which is why the client binds a UDP port at connect time.

Synchronous publishing costs latency on every message. Where that latency is unacceptable but delivery must still be guaranteed, the usual approach is to publish asynchronously and confirm delivery out of band: keep the data upstream, and implement a custom API that reports which messages the server has received. Call that API every n messages and release the upstream data up to the confirmed point.

Choose bridge or non-bridge mode

RT can either persist messages to a local disk on the client host before transmitting them, or publish them over qIPC with no local persistence. You choose with useBridge in the RT configuration.

Both sides must agree. The client's useBridge must match the server's bridge setting for the stream, which is false by default. If you pass an RT configuration without setting useBridge, the client runs in non-bridge mode.

Non-bridge mode

RT writes every published message to log files on a disk local to the publishing client, and a separate push_client process transmits them to the server. Once a rolled log file has been transmitted and merged, it is truncated to zero bytes.

Messages that reached the local disk are transmitted eventually, as long as the disk, the participating processes, and the network are available. A network outage interrupts transmission but doesn't lose the data: publishing resumes when connectivity returns. If the client process crashes, a message may or may not have been flushed to disk.

Consider the following before choosing non-bridge mode:

  • Disk capacity. Asynchronous publishes keep succeeding during an outage, and the local logs keep growing. Size the disk for the longest outage you expect to tolerate.
  • Disk I/O. The local disk is in the ingestion path, so slow I/O becomes ingestion latency. Locally attached SSDs, or equivalent, suit most cases.
  • Client identity. Local log data belongs to a specific publishing client, identified by its instance name. If clients start and stop dynamically, you need a way to deal with unpublished data left in the logs of a client that has gone.
  • Process count. One push_client process starts as a child of the publishing client for each RT endpoint. With two ingestion feeds and three-node RT clusters, that is six push_client processes per publishing client.

For sizing the client-side log directory and the errors you see when it's misconfigured, see Client-side RT log files.

Bridge mode

Messages are held in memory only and published over qIPC to a bridge endpoint on the server. They are still persisted on the server. This removes the local disk requirement and the push_client processes, and resembles how publishing worked in v2.

In exchange, the client takes on three risks:

  • Messages can be lost.
  • Messages can be duplicated.
  • Messages can arrive out of order. Order is guaranteed once the message is sequenced into the RT stream output, but not before.

Out-of-order arrival is not hypothetical in a multi-node system. If a client publishes message one to rt-bridge-0, loses the connection, reconnects to rt-bridge-1, and publishes message two, the two messages sit in different input directories and can be sequenced in either order. Whichever order is established is then consistent for all subscribers.

Address these risks the same way as asynchronous publishing: confirm delivery out of band, or accept that the server may process duplicates. SAPI doesn't support message deduplication. See Deduplication.

Bridge mode requires the server to enable it: set bridge: true for the stream in ems.yaml. On the client, set useBridge to true in the RT configuration.

If the client and the server disagree, the result depends on which side is which:

  • Bridge client, non-bridge stream. Registration fails immediately, with a "Failed to allocate RT handle" error, and SAPI doesn't fall back to non-bridge mode.
  • Non-bridge client, bridge stream. There is no error. The client keeps writing to its local logs, and the data piles up without reaching the server.

Warning

The C library enables bridge mode by default only when you pass no RT configuration at all. When you pass an RT configuration, your value wins, and in the C#, Java, and Python libraries an unset useBridge is false. Set useBridge explicitly in your RT configuration.

Deduplication

SAPI doesn't deduplicate published messages. In bridge mode, the client generates a message identifier itself, as a counter that starts at 1 for each client process and increases by one with every message. Because the counter restarts whenever the client restarts, it can't identify the same message across a crash and republish, which deduplication needs. In non-bridge mode, published messages carry no identifier at all.

The dedupId setting in the RT configuration is a label that tells publishers apart. It doesn't identify messages, so setting it doesn't turn deduplication on.

To avoid duplicates, confirm delivery out of band, or accept that the server may process duplicates.

Unregister the publisher

Call unregisterPublisher with the handle when the client stops publishing, then disconnect. In non-bridge mode, check the client's local RT log directory afterwards to confirm the logs have been truncated, which indicates everything was transmitted.

The rtFlushTimeoutSecs RT configuration field sets how long the client waits for local logs to flush to the server on close. By default it doesn't time out.

Next steps