Stream Subscription

public ThreadedClient()
public ThreadedClient(String subscribeHost, int subscribePort)

public ThreadPooledClient()
public ThreadPooledClient(String subscribeHost, int subscribePort, int threadCount)

public PollingClient()
public PollingClient(String subscribeHost, int subscribePort)

The subscribe method is used for stream subscription in all three tools, with the same parameters.

Syntax

subscribe(String host, int port, String tableName, String actionName, MessageHandler handler, long offset, boolean reconnect, IVector filter, int batchSize, float throttle, StreamDeserializer deserializer, String user, String password, List<String> backupSites, int resubTimeout, boolean subOnce)

Parameters

• host: IP address of the publisher node.
• port: Port number of the publisher node.
• tableName: Name of the publisher stream table.
• actionName (optional, default "javaStreamingApi"): Name of the subscription task.
• handler: A user-defined function to process the subscribed data.
• offset (optional, default -1): An integer indicating the position of the first message where the subscription begins. A message is a row of the stream table. If offset is unspecified, negative or exceeding the number of rows in the stream table, the subscription starts with the next new message. offset is relative to the first row of the stream table when it is created. If some rows were cleared from memory due to cache size limit, they are still considered in determining where the subscription starts.
• reconnect (optional, default false): Whether to enable reconnection.
• filter (optional, default null): A vector of the filtering conditions. Only the rows with values of the filtering column in the vector specified by the parameter filter are published to the subscriber.
• batchSize (optional): The number of unprocessed messages to trigger the handler. If it is positive, the handler does not process messages until the number of unprocessed messages reaches batchSize. If it is unspecified or non-positive, the handler processes incoming messages as soon as they come in.
• throttle (optional): a float indicating the maximum waiting time (in seconds) before the handler processes the incoming messages. The default value is 1.

  Note that batchSize and throttle must be specified together.

• deserializer (optional, default null): Deserializer for the subscribed heterogeneous stream table.
• user (optional, default ""): Username used to connect to the server
• password (optional, default ""): Password.
• backupSites (optional): A String list representing the backup publisher nodes, 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 (optional, default 100): A non-negative integer indicating how long (in milliseconds) to wait before attempting to resubscribe if the API detects a disconnection.
• subOnce (optional, default false): 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.

Use the unsubscribe method to cancel an existing subscription.

public void unsubscribe(String host, int port, String tableName, String actionName)

Examples

client.unsubscribe("192.168.0.21", 9002, "outTables", "mutiSchema");

First define a handler which implements the com.xxdb.streaming.client.MessageHandler interface to be passed when constructing a subscription.

public static MessageHandler MessageHandler_handler = new MessageHandler() {
    @Override
    public void doEvent(IMessage msg) {
        try {
            String script = String.format("insert into Receive values(%d,%s,%f)", Integer.parseInt(msg.getEntity(0).getString()), msg.getEntity(1).getString(), Double.valueOf(msg.getEntity(2).toString()));
            conn.run(script);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
};

For single-threaded callback using ThreadedClient:

@Test
public void testThreadedClient() throws SocketException {
    // For versions before 2.00.9, start subscription with specified port 8000
    ThreadedClient client = new ThreadedClient(8000);
    // For version 2.00.9 and later, start subscription without specifying port
    ThreadedClient client = new ThreadedClient();
    client.subscribe("192.168.0.21", 8848, "Trades", handler, -1);
}

Incoming data will be passed through the msg parameter, and processed with MyHandler.

For multi-threaded callback using ThreadPooledClient:

@Test
public void testThreadPooledClient() throws SocketException {
    // For versions before 2.00.9, start subscription with specified port 8000
    ThreadPooledClient client = new ThreadPooledClient(8000);
    // For version 2.00.9 and later, start subscription without specifying port
    ThreadPooledClient client = new ThreadPooledClient();
    client.subscribe("192.168.0.21", 9002, "Trades", handler, -1);
}

For PollingClient that regularly queries the publisher table for new data through a Java client:

@Test
public void testPollingClient() {
    // For versions before 2.00.9, start subscription with specified port 8000
    PollingClient client = new PollingClient(subscribePort);
    // For version 2.00.9 and later, start subscription without specifying port
    PollingClient client = new PollingClient();
    TopicPoller poller1 = client.subscribe("192.168.0.21", 9002, "Trades", -1);

    while (true) {
        ArrayList<IMessage> msgs = poller1.poll(1000);
        if (msgs.size() > 0) {
            BasicInt value = (BasicInt) msgs.get(0).getEntity(2);
        }
    }
}

After poller1 observes that the stream data table receives new data, it will pull the new data. When there is no new data published, the Java program will be blocked in the poller1.poll method waiting.