Skip to content
Connect > Client Libraries

C++ client

The ami_client_common library. The entire public API is one header, include it and use the ami namespace.

1
2
#include "ami_client.hpp"
using namespace ami;

The library is C++20 and links ASIO and OpenSSL. ami::Client is the single entry point.

Connect

Create a cluster, fill in the configs your program uses, and add it. Each config's third positional argument is the failover-endpoints list, pass {} for a single server.

1
2
3
4
5
6
7
8
Cluster node("N1", "localhost");
node.database_config()     = DatabaseConfig("3280", /*ssl=*/true, {}, "demo", "demo123");
node.subscription_config() = SubscriptionConfig("3270", /*ssl=*/true);
node.publication_config()  = PublicationConfig("3289", /*ssl=*/true, {}, "cpp-client",
                                               PublicationConfig::ENABLE_SEND_SEQNUM);

Client client;
client.add_cluster(node, /*sync=*/true);

The full DatabaseConfig signature is DatabaseConfig(port, ssl, failover, username, password, query_timeout_ms = 30000, query_limit = 10000, center_name = "").

Run SQL

sql returns a std::unique_ptr<TableAccessor>. Check has_result_set(), then advance the cursor.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
auto result = client.sql("N1", "SELECT id, name FROM my_table");
if (result && result->has_result_set()) {
    const auto & cols = result->columns();
    while (result->cursor_advance()) {
        const std::int64_t row = result->cursor_row_number();
        for (std::size_t c = 0; c < cols.size(); ++c) {
            // read cols[c] at row
        }
    }
    result->print(std::cout);   // or inspect affected_rows() / error_message()
}

A column is an ami::Column variant; read a cell with std::visit (or std::get<T> when you know the type). The async form takes a callback:

1
2
client.sql_async("N1", "SELECT * FROM my_table",
                 [](const TableAccessor & accessor) { accessor.print(std::cout); });

One JDBC request at a time per cluster, capped at 10,000 rows

The SQL socket does not queue concurrent requests, so serialize them or use separate clusters. Results default to 10,000 rows; raise the limit on the config or narrow the query.

Subscribe

Subclass ClientListener, override the callbacks you need, and attach it before adding the cluster.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class MyListener : public ClientListener {
public:
    void on_connect(Client &, std::string_view name) override {
        std::cout << "[CONNECTED] " << name << "\n";
    }
    void on_subscribe(Client &, std::string_view name,
                      const std::vector<std::string> & tables) override {
        std::cout << "[SUBSCRIBED] " << name << "\n";
    }
    void on_update(Client &, std::string_view name,
                   const CenterIncrementalUpdate & u) override {
        if (u.action_ == 3 /* delete */) { /* erase by u.id_ */ return; }
        // add (1) or update (2): apply only the fields present
        for (std::size_t i = 0; i < u.field_names_.size(); ++i) {
            const auto & field = u.field_names_[i];
            const auto & value = u.row_data_[i];   // std::get<double>(value), etc.
        }
    }
};

client.listener_add(std::make_shared<MyListener>());
client.add_cluster(node, /*sync=*/true);
client.subscribe("N1", {"trades"}, /*sync=*/true);

CenterIncrementalUpdate fields: seqnum_, action_ (1=add, 2=update, 3=delete), id_, table_name_, object_id_, and the parallel field_names_ / row_data_. An update carries only the changed fields, merge by id_, the stable numeric row identity. object_id_ is the Relay-publish key and is empty on subscription updates for SQL-created rows, so do not merge on it.

Callbacks run on the client's I/O threads

The string_views and data in a CenterIncrementalUpdate are valid only for the duration of the callback: copy anything you keep. Keep the callback fast and run blocking work, and any same-connection synchronous call, on your own thread rather than in the callback. Guard your own shared state.

Knowing when the initial load is done. on_subscribe fires when the request reaches the server, before the initial rows, which arrive as on_update with action_ == 1 (add). No listener callback marks the snapshot complete. Pass sync=true to subscribe to block until the initial snapshot has been received.

Server-side filter and projection

Use a CenterSubscription to push a WHERE filter and a column list to the Center:

1
2
3
4
5
CenterSubscription sub;
sub.table   = "Trades";
sub.filters = {"price > 100"};       // server-side WHERE
sub.columns = {"symbol", "price"};   // empty = all columns
client.subscribe("N1", std::vector{sub}, /*sync=*/true);

unsubscribe(cluster, tables) and get_table_schema(cluster, table) are available in C++.

Publish

Build the field names and row, then publish. Publishing is asynchronous; an outbox replays across reconnects.

1
2
3
4
5
const RowFieldNames field_names = {"SYM", "bid", "ask"};
RowData row_data = {std::string("IBM"), 123.4, 123.5};
client.publish_object("N1", "QUOTES", "IBM", field_names, row_data);

client.remove_object("N1", "QUOTES", "IBM");   // delete

publish_object(cluster, table, object_id, fields, row, seqnum = 0, expires_in_ms = 0). For the pre-serialized hot path, use serialize_object(...) once and publish_raw(...) many times.

Bulk insert

Build a Table column-by-column with ColumnBuilder<T> (the type sets the wire type), then sql_insert or sql_replace.

1
2
3
4
5
6
7
8
9
Table table("my_table");

Table::ColumnBuilder<std::string>("name")
    .emplace_value("Earth").emplace_value("Mars").build_into(table);
Table::ColumnBuilder<std::int64_t>("radius_km")
    .emplace_value(6371).emplace_value(3389).build_into(table);

client.sql_insert("N1", std::move(table));
// client.sql_replace("N1", std::move(table), "PersistEngine=\"HISTORICAL\"");   // replace contents + persist

For types the compiler cannot map on its own, UUID, BigDecimal, BigInteger, declare them explicitly with add_column_as:

1
table.add_column_as("uid", ExplicitColumnType::Uuid, uuids);

sql_replace creates the table if it is missing, otherwise deletes all its rows, then inserts the rows passed; its suffix argument passes a persist-engine clause.

Commands (bidirectional)

Register a named command that the platform or another client can invoke; handle it in on_command and reply with send_command_response. See the risk and P&L use case.

TLS and mutual TLS

Set the SSL options on the config's shard before adding the cluster:

1
2
3
4
5
auto & ssl = node.database_config().shards[0].ssl;
ssl.trusted_server_pub_file = "keys/server.pub";   // pin the server's ssh-rsa .pub
ssl.client_cert_file        = "keys/client.crt.pem";
ssl.client_key_file         = "keys/client.key.pem";
// ssl.client_key_password  = "...";                // only if the key is encrypted

A secured server requires the client cert (mutual TLS). See Configuration → TLS.

Disconnect and teardown

remove_cluster(name, sync=false) is the counterpart to add_cluster: it drops that cluster's JDBC, Center, and Relay connections and stops its updates. Use it to stop a subscription (alongside unsubscribe, above) and to shut a cluster down cleanly.

1
2
client.remove_cluster("N1");                 // async
client.remove_cluster("N1", /*sync=*/true);  // block until torn down

What this does not support

  • No transactions: SQL runs statement by statement, with no multi-statement unit of work.
  • No concurrent JDBC requests on one cluster: the SQL socket does not queue them, so serialize requests or use a second cluster.
  • No wire compression: every protocol sends its data uncompressed.