ThreadedClient

The constructor declaration is as follows:

ThreadedClient(int listeningPort = 0)

• listeningPort: The subscription port on the client to subscribe to the data sent from the server.

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.

Starting from version 3.00.2.1, a new constructor for threaded clients is available:

ThreadedClient(StreamingClientConfig config);

where StreamingClientConfig is the configuration for streaming subscription. The default setup is defined as follows:

enum class TransportationProtocol {
    TCP, UDP,
};
struct StreamingClientConfig {
    TransportationProtocol protocol{TransportationProtocol::TCP};
};

• protocol: Defines the transport protocol for the client.
  • TCP: Functions identically to previous versions.
  • UDP: Enables UDP multicast communication with the DolphinDB server via the Aeron library.

Note (for Using UDP Protocol):

• The current UDP communication method only supports the first subscribe interface (which processes each message as it arrives) and does not support parameters resub, filter, backupSites, resubTimeout, and subOnce.
• The C++ API requires the Aeron library (of any version) for compilation on Linux only.
• The API operates with an embedded Aeron Media Driver, so users do not need to run a separate Aeron thread.
• When using UDP for streaming subscriptions, the C++ API creates a folder named dolphindb_udp_<pid> in the /dev/shm directory for Aeron’s use. This folder is automatically deleted when the subscription ends. If the client process terminates unexpectedly, delete the relevant folder manually.
• The DolphinDB server must be Linux 3.00.0 or later.
• The maximum message size that the server can publish at one time is 2 MB. If data is not received during subscription, check if this limit has been exceeded. Adjusting the maxMsgNumPerBlock configuration parameter may resolve this issue.

ThreadedClient offers the subscribe function to establish subscriptions to DolphinDB stream tables. Concerning the handling of incoming messages, two types of interfaces are provided.

(1) Processing each message as it arrives individually. Accessing the Message object must be completed within the handler.

using MessageHandler = std::function<void(Message)>;
ThreadSP subscribe(
    string host, int port,
    const MessageHandler &handler,
    string tableName,
    string actionName = DEFAULT_ACTION_NAME,
    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 resubscribeInterval = 100,
    bool subOnce = false,    
    int resubscribeTimeout = 0
)

(2) Processing messages in batches using a caching mechanism. The handler is triggered only when specified conditions (batchSizeand throttle) are met.

using MessageBatchHandler = std::function<void(vector<Message>)>;
ThreadSP subscribe(
    string host, int port,
    const MessageBatchHandler &handler,
    string tableName,
    string actionName = DEFAULT_ACTION_NAME,
    int64_t offset = -1,
    bool resub = true,
    const VectorSP &filter = nullptr,
    bool allowExists = false,
    int batchSize = 1,
    double throttle = 1,
    bool msgAsTable = false,
    string userName = "",
    string password = "",
    const StreamDeserializerSP &blobDeserializer = nullptr,
    const std::vector<std::string>& backupSites = std::vector<std::string>(),
    int resubscribeInterval = 100,
    bool subOnce = false,    
    int resubscribeTimeout = 0
)

ThreadedClient 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.

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 a subscription to table “shared_stream_table“, starting 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 batchHandler = [](std::vector<Message> msgs){
        std::cout << "receive " << msgs.size() << " msgs\n";
        for(auto& msg : msgs){
            std::cout << msg->getString() << std::endl;
        }
    };
    ThreadedClient client(0);
    ThreadSP t = client.subscribe("127.0.0.1", 8848, batchHandler, "shared_stream_table", "action1", 0, true, nullptr, false, 3, 1.0);
    sleep(10);
    client.unsubscribe("127.0.0.1", 8848, "shared_stream_table", "action1");
    return 0;
}