Ontology SDK

Subscriptions

Get a callback when objects of a type change.

A subscription tells you when objects of one type are created, changed or deleted. You do not have to poll.

TypeScript

const sub = client.objects["Customer"]
  .where({ tier: "gold" })
  .subscribe({
    onSuccessfulSubscription: () => console.log("listening"),
    onChange: (u) => {
      if (u.state === "ADDED_OR_UPDATED") console.log("changed", u.object.$rid);
      else console.log("gone", u.primaryKey);                // state "DELETED"; the $rid, e.g. "customer/c1"
    },
    onOutOfDate: () => reload(),                             // events were dropped: re-read
    onError: (e) => console.error(e),
  });

// later
sub.unsubscribe();

Pass { properties: ["name", "tier"] } as the second argument of subscribe to re-read only those properties. For each change, the client re-reads the object as you. So onChange only ever gets objects you are allowed to see. An object that no longer matches where, or that you can no longer see, arrives as DELETED. Callbacks run in order, one at a time. For a DELETED update, primaryKey holds the object's $rid (customer/c1); the primary key itself is the part after the /.

The client does not reconnect. When the socket closes, onError runs once, and your code must call subscribe again (and re-read, since changes may have been missed). In Node.js before version 22, or in tests, pass a WebSocket constructor: createClient({ ..., webSocket: WebSocketClass }).

Raw WebSocket

You can use any WebSocket client:

wss://cnvrg.si/api/v2/ontologies/<ontology>/objectTypes/<ObjectType>/subscribe

Send your token in the Authorization header. Browsers cannot set that header, so they send it as the subprotocol Bearer-<token> instead. The server sends JSON messages:

If the server refuses the subscription, it accepts the connection and then closes it at once with one of these codes:

There is one socket per object type, and the type is part of the URL. You send nothing after connecting, and the server ignores anything you send. To watch several types, or a filtered set, on one socket, use streamSubscriptions (below).

Several subscriptions on one socket

streamSubscriptions follows Palantir's WebSocket API: one socket carries many subscriptions, each with its own filter and properties.

wss://cnvrg.si/api/v2/ontologySubscriptions/ontologies/<ontology>/streamSubscriptions

It authenticates like the per-type socket (Authorization header, or the Bearer-<token> subprotocol) and is refused with the same close codes (4401, 4403, 4404). After connecting, send a subscribe message:

{
  "id": "6f1c0e2a-0000-4000-8000-000000000001",
  "requests": [
    { "objectSet": { "type": "base", "objectType": "Order" } },
    {
      "objectSet": {
        "type": "filter",
        "objectSet": { "type": "base", "objectType": "Customer" },
        "where": { "type": "eq", "field": "tier", "value": "gold" }
      },
      "propertySet": ["name", "tier"]
    }
  ]
}

objectSet is either a base set of one type, or a filter of a base set with a where written as in loadObjects. propertySet limits the properties each change carries; leave it out to get all you may see. To change what you watch, send another message with the same id. The server closes that id's subscriptions and opens the new ones. Send "requests": [] to stop watching. Messages with other ids add subscriptions next to the ones you have.

The server answers each subscribe message with one response per request, in order:

{ "type": "subscribeResponses", "id": "6f1c0e2a-…",
  "responses": [
    { "type": "success", "id": "<subscription id>" },
    { "type": "error", "errors": [ { "error": "PERMISSION_DENIED", "args": [ { "name": "message", "value": "…" } ] } ] }
  ] }

Each request is checked on its own, so one bad request does not fail the others. The error codes are INVALID_OBJECT_TYPE (the type does not exist or you cannot see it), PERMISSION_DENIED (the where or propertySet uses a property that is masked for you), INVALID_REQUEST (a malformed object set or where, or a referenceSet, which Converge does not support) and TOO_MANY_SUBSCRIPTIONS.

After that, the server sends these messages. id is the subscription id from the response:

For each change the server reads the object again as you, through the same checks as loadObjects. So an ADDED_OR_UPDATED object holds only what you may see: masked properties are left out. The object always carries __apiName, __primaryKey and $rid. Without a propertySet it holds everything loadObjects returns, version and other $ fields included. With a propertySet it holds only the listed properties; add version to the list if you need it.

REMOVED is sent only for an object this subscription already sent you as ADDED_OR_UPDATED: one that was deleted, no longer matches the filter, or that you can no longer read. It carries only __apiName, __primaryKey and $rid, with no property values. A change to an object you could never read, or that never matched the filter, is not reported, so such an object is never named. Objects you loaded over HTTP but that the subscription never sent are not tracked either, so they do not get REMOVED. A subscription remembers up to 2,000 objects it sent you; past that it sends refreshObjectSet and starts over, so after a refresh it can only report REMOVED for objects it sends you again.

The server checks your token again at least every 5 minutes, and before it delivers a change when the last check is older than that. Once the token's expiry time has passed, the next change or subscribe message closes the socket, without waiting for the 5 minutes. If the token has expired, been revoked or rotated, or now belongs to another user, it closes the socket with 4401. If you can no longer see a type, it closes that type's subscriptions with subscriptionClosed.

A client may send "qos": "high" in the subscribe message. Converge accepts it but handles it the same as normal priority.

A malformed or oversized subscribe message closes the socket: code 1007 if it is not a valid subscribe message, 1009 if it is larger than 64 KiB.

From Python, with the websockets package (version 14 or later; older versions call the argument extra_headers):

import asyncio, json, os, websockets

token = os.environ["FOUNDRY_TOKEN"]

async def watch():
    url = "wss://cnvrg.si/api/v2/ontologies/<ontology>/objectTypes/Customer/subscribe"
    async with websockets.connect(url, additional_headers={"Authorization": f"Bearer {token}"}) as ws:
        async for msg in ws:
            print(json.loads(msg))

asyncio.run(watch())

Limits

• On the per-type socket, a subscription covers one object type. The where filter is applied by the TypeScript client, so that socket reports every change to the type and you filter yourself. On streamSubscriptions the server applies the filter.
• Subscriptions live in the memory of one server process. A server restart or deploy closes them, and the client does not reconnect: subscribe again and re-read.
• Changes are reported for edits made through actions on the server that holds your socket. A change written another way, for example by a pipeline, may not be reported, so do not rely on a subscription as the only way to see changes.
• Each subscription buffers 64 events. When it falls behind, further events are dropped and you get outOfDate (or refreshObjectSet on streamSubscriptions): re-read the objects.
• A server process holds at most 200 subscriptions in total, shared by every user and both endpoints, and one user holds at most 20 of them. Past either limit the per-type socket closes with 4429 and a streamSubscriptions request gets TOO_MANY_SUBSCRIPTIONS. One streamSubscriptions socket holds at most 50 subscriptions, so in practice a user's 20 is the limit you reach first.
• One user can have at most 5 streamSubscriptions sockets open on a server process. A sixth closes with 4429.
• A streamSubscriptions socket accepts at most 10 subscribe messages in 10 seconds. The 11th closes the socket with code 1008.
• A streamSubscriptions client must keep reading. If a message cannot be sent within 10 seconds, the server closes the socket with 1008.
• A filtered subscription checks only the objects in each batch of changes against the filter, so the size of the type does not matter.
• If the server cannot check your access because a backend is down, it closes the socket with 1013 (try again later). An unexpected server error closes it with 1011.
• On the per-type socket, a change message never carries the object's values. It names the object, and you read it with your own token, so properties you may not see stay hidden. An optional property you may see can still be absent from the object.
• The per-type socket names every changed object of a type you can see by its $rid, including objects you cannot read. The values stay hidden. It checks your token only when it connects. streamSubscriptions does neither of these.