ThreadPooledClient

ThreadPooledClient creates multiple threads for each subscription. When the subscriber receives the stream data from the publisher, if there are idle threads available in the pool, one of them will be assigned to invoke the the callback function (handler). Therefore, in scenarios where data arrives faster than it is being processed, ThreadPooledClient offers an advantage over ThreadedClient.

This section presents a comprehensive overview of the ThreadPooledClient, covering its constructor, subscription and unsubscription mechanisms, and provides a practical usage example.

Constructing a ThreadPooledClient

The constructor declaration is as follows:

ThreadPooledClient(int listeningPort, int threadCount)

listeningPort: The subscription port on the client to subscribe to the data sent from the server.
threadCount: The number of working threads to be created.

DolphinDB server version 2.00.9 and later enable the publisher to push data through the existing connection requested by the subscriber. Therefore, the subscriber does not need to specify a specific port (filled in 0). If listeningPort is specified, it will be ignored by the API.

Creating Subscription

ThreadPooledClient offers the subscribe function to establish subscriptions to DolphinDB stream tables. The declaration for the subscribe function is as follows:

vector<ThreadSP> ThreadPooledClient::subscribe(
    string host,
    int port,
    const MessageHandler &handler,
    string tableName,
    string actionName,
    int64_t offset = -1,
    bool resub = true,
    const VectorSP &filter = nullptr,
    bool msgAsTable = false,
    bool allowExists = false, 
    string userName = "",
    string password = "",
    const StreamDeserializerSP &blobDeserializer = nullptr,
    const std::vector<std::string>& backupSites = std::vector<std::string>(),
    int resubTimeout = 100,
    bool subOnce = false
)

Arguments

host: The IP address of the publisher node.
port: The port number of the publisher node.
handler: A user-defined function to process the received data.
tableName: The name of the subscribed stream table.
actionName: The name of the subscription task. It can contain letters, digits, and underlines.
offset: The position of the first message where the subscription begins. A message is a row of the stream table. Offset is relative to the first row of the subscribed stream table when it was created. If some rows were cleared from memory due to cache size limit, they are still considered in determining where the subscription starts. Note that if offset is unspecified, negative numbers, or exceeds the row count, the subscription starts with the next new message.
resub: Whether to re-subscribe after a network disconnection.
filter: The filtering conditions. Only the messages with the specified filtering column values will be published.
msgAsTable: true means the subscribed data is ingested into handler as a Table. false means the subscribed data is ingested into handler as an AnyVector.
allowExists: true means multiple handlers can be specified. false means only a single handler can be specified.
userName:The username used to connect to the server.
password: The password used to connect to the server.
blobDeserializer: The deserializer for the subscribed heterogeneous stream table.
backupSites: A backup node list with each specified in host:port format, e.g., ["192.168.0.1:8848", "192.168.0.2:8849"].
- If specified, failover mechanism is automatically activated. If the subscribed node (i.e., primary node) becomes unavailable due to a disconnection or failover, the API attempts reconnection repeatedly across backup nodes until the subscription is successful.
- The stream table definition, including the name, schema, and data entries, must be identical across the primary node and all configured backup nodes to prevent potential data inconsistencies during failover.
- If the subscribed table is a high-availability stream table, the connection will be established based on the node set of backupSites.
- To cancel a subscription, specify the host and port of the primary node.
resubTimeout: A non-negative integer indicating how long (in milliseconds) to wait before attempting to resubscribe if the API detects a disconnection. The default value is 100.
subOnce: Whether to include the subscribed node in subsequent reconnection attempts following the node switch. Note that this parameter does not take effect for a single subscribed node when resub = true.

Return Values

A vector of pointers, where each pointer points to the thread that continuously invokes handler. These threads will stop calling handler when function unsubscribe is called on the same topic.

Canceling Subscription

ThreadPooledClient offers the unsubscribe function to cancel an existing subscription. The function declaration for unsubscribe is as follows:

void unsubscribe(
    string host,
    int port,
    string tableName,
    string actionName
)

Note: The arguments passed to the unsubscribe function must match the arguments provided when creating the subscription through the subscribe call.

Usage Example

The following example uses the DolphinDB server version 2.00.10.

DolphinDB script: Create a shared stream table “shared_stream_table“.

rt = streamTable(`XOM`GS`AAPL as id, 102.1 33.4 73.6 as x)
share rt as shared_stream_table

C++ Code: Create 10 threads for the subscription to table “shared_stream_table“. The subscription starts from the first record. Cancel this subscription after 10 seconds. During this 10-second window, the subscriber will receive any data appended to the "shared_stream_table".

#include <iostream>
#include "Streaming.h"
using namespace dolphindb;

int main(int argc, const char **argv)
{
    auto handler = [](Message msg){
        std::cout << msg->getString() << std::endl;
    };
    ThreadPooledClient client(0, 10);
    auto t = client.subscribe("127.0.0.1", 8848, handler, "shared_stream_table", "action1", 0);
    sleep(10);
    client.unsubscribe("127.0.0.1", 8848, "shared_stream_table", "action2");
    return 0;
}