Connect > Client Libraries > Use Cases

Use case: market data to a live dashboard

Goal. A feeder program writes quotes into a Center table; the platform aggregates them; a dashboard program subscribes and shows the live picture. Three roles, feeder, server, subscriber, wired with the client library.

The shape

feeder (SQL INSERT → Center)  →  Center table + aggregate  →  dashboard (subscribe → Center)

You write the feeder and the dashboard. This guide focuses on the two client programs; the summary table and its aggregate are set up once on the server.

To fill a Center table, use SQL INSERT, not publish

Relay publish() / publish_object() sends an object to the Relay, a separate messaging plane. It does not put rows into a Center table on its own: a subscriber to the table will not see published objects unless the deployment has a Relay→Center feed configured. To write rows a Center subscriber can see, use SQL INSERT (or sql_insert for bulk). Confirmed on a live instance, published quotes never appeared in the table; INSERTed rows did. Reserve publish for the Relay-object pattern where a feed carries the objects onward.

Create the table

Create a CREATE PUBLIC TABLE so it is shared, live state subscribers can see. Run it over the JDBC (SQL) connection with client.sql(...), or provision it on the server:

CREATE PUBLIC TABLE IF NOT EXISTS QUOTES (SYM String, bid Double, ask Double, last Double, volume Long);

A DDL statement returns an empty result ({} in Python), no rows and no rows_affected; that is success, not a failure. The per-symbol summary the dashboard reads (QUOTES_SUMMARY) is a second table maintained by a server-side aggregate/trigger, defining tables, triggers, and persistence belongs to the Center and AMI SQL documentation, not the client.

1. Feed quotes into the table

The feeder connects with a JDBC (SQL) config and writes rows with INSERT / UPDATE. On a secured server, point the JDBC config's shard at your certificate first (see Configuration → Connecting to a secured server).

PythonC++

import pyami.rt.ami_client as ac

node = ac.AmiCluster("N1", "localhost")
node.jdbc_config = ac.AmiDatabaseConfig("3280", True, "demo", "demo123", 60000, 10000)
# node.jdbc_config.shards[0].ssl = ac.AmiSslOptions(...)   # on a secured server
client = ac.AmiClient()
client.add_cluster(node, sync=True)

def feed_quote(sym, bid, ask, last, volume):
    client.sql("N1",
        f'INSERT INTO QUOTES (SYM, bid, ask, last, volume) '
        f'VALUES ("{sym}", {bid}, {ask}, {last}, {volume})')

def update_bid(sym, bid):
    client.sql("N1", f'UPDATE QUOTES SET bid={bid} WHERE SYM == "{sym}"')

Cluster node("N1", "localhost");
node.database_config() = DatabaseConfig("3280", true, {}, "demo", "demo123");
Client client;
client.add_cluster(node, /*sync=*/true);

client.sql("N1",
    R"(INSERT INTO QUOTES (SYM, bid, ask, last, volume) VALUES ("IBM", 123.4, 123.5, 123.45, 1000))");
client.sql("N1", R"(UPDATE QUOTES SET bid=123.6 WHERE SYM == "IBM")");

Note the expression syntax in WHERE: equality is == and string literals use double quotes. For high volume, use bulk insert (a dict/DataFrame of columns), client.sql_insert("N1", "QUOTES", data); see Python client → Bulk insert. To remove the table when done: client.sql("N1", "DROP TABLE QUOTES").

2. Aggregate on the server

The live summary, say a per-symbol last price and total volume, is built on the server, not in the client, by a Center aggregate/trigger over the QUOTES table into a QUOTES_SUMMARY table. That keeps the computation next to the data and lets many dashboards share one result. Defining server-side tables and triggers is covered in the Center documentation; the client just subscribes to the result.

3. Subscribe (the dashboard)

The dashboard connects with a Center (subscribe) config, attaches a listener, and subscribes. Each change arrives in on_update; merge it into your view keyed by id: the stable numeric row identity. (object_id is empty on subscription updates; keying on it collapses every row under "".)

PythonC++

class Dashboard:
    def __init__(self):
        self.rows = {}                     # id -> row dict

    def on_update(self, client, name, update):
        key = update.id                    # merge on id, NOT object_id
        if update.action == 3:             # delete
            self.rows.pop(key, None)
        else:                              # add / update, merge changed fields
            self.rows.setdefault(key, {}).update(
                dict(zip(update.field_names, update.row_data)))
        self.render()

    def render(self):
        ...                                 # redraw from self.rows

node = ac.AmiCluster("N1", "localhost")
node.center_config = ac.AmiSubscriptionConfig("3270", True, "demo", "demo123")
# node.center_config.shards[0].ssl = ac.AmiSslOptions(...)   # on a secured server
client = ac.AmiClient()
client.add_client_listener(Dashboard())
client.add_cluster(node, sync=True)
client.subscribe("N1", ["QUOTES_SUMMARY"])

class Dashboard : public ClientListener {
    std::unordered_map<std::int64_t, Row> rows_;
public:
    void on_update(Client &, std::string_view, const CenterIncrementalUpdate & u) override {
        auto key = u.id_;                          // merge on id_, NOT object_id_
        if (u.action_ == 3) { rows_.erase(key); return; }
        auto & r = rows_[key];
        for (std::size_t i = 0; i < u.field_names_.size(); ++i)
            apply(r, u.field_names_[i], u.row_data_[i]);   // merge changed fields
    }
};

The key discipline: an update carries only the changed columns, so merge, do not replace: keyed by id. Replacing the whole row would blank the fields that did not change this tick.

What this shows

• Feeder + dashboard compose. The feeder never talks to the dashboard; they meet in the Center table.
• The server does the aggregation, so every dashboard sees the same summary with no duplicated client logic.
• Failover is free for the subscriber. If the dashboard drops, its subscription re-snapshots on reconnect.