C# API

The C# API connects to the DolphinDB server via TCP/IP protocol. To connect to a local DolphinDB server with port number 8848:

using dolphindb;
using dolphindb.data;
using dolphindb.io;

public void Test_Connect(){
      DBConnection conn=new DBConnection();
      Assert.AreEqual(true,conn.connect("localhost",8848));
}

Starting from API 1.30.17, the following optional parameters can be specified for connection: asynchronousTask, useSSL, compress, and usePython. The default values of these parameters are false. Currently, only linux stable version >= 1.10.17, and latest version >= 1.20.6 are supported.

The following example establishes a connection to the server. It enables SSL and compression but disables asynchronous communication. Note that the configuration parameter enableHTTPS=true must be specified on the server side.

DBConnection conn = new DBConnection(false,true,true)

In the following example, the connection disables SSL and enables asynchronous communication. In this case, only DolphinDB scripts and functions can be executed and no values are returned. This feature is for asynchronous writes.

DBConnection conn = new DBConnection(true,false)

Establish a connection with a username and password:

boolean success = conn.connect("localhost", 8848, "admin", "123456");

To define and use user-defined functions in a C# program, you can pass in the user-defined scripts to the parameter startup. The advantages are: (1) These functions don't need to be defined repeatedly every time run is called; (2) The API client can automatically connect to the server after disconnection. If the parameter startup is specified, the C# API will automatically execute the script and register the functions. The parameter can be very useful for scenarios where the network is not stable but the program needs to run continuously.

boolean success = conn.connect("localhost", 8848, "admin", "123456", "");

Multiple DBconnection objects can be reused by ExclusiveDBConnectionPool. You can either execute command ExclusiveDBConnectionPool.run, or execute a task with execute and then obtain the results with getResults method of BasicDBTask.

Note: If the current ExclusiveDBConnectionPool is idle, C# API will automatically close the connection after a while. To release the connection resources, call shutdown() upon the completion of thread tasks.

BasicDBTask encapsulates the functions and arguments to be executed.

Build a connection pool with 10 connections.

ExclusiveDBConnectionPool pool = new ExclusiveDBConnectionPool("192.168.1.38", 8902, "admin", "123456", 10, false, true);

//run the script
IEntity ret = pool.run("1 + 1");
Console.Out.WriteLine(ret.getString());

//run the specified function
ret = pool.run("abs", new List<IEntity> { new BasicInt(-3) });
Console.Out.WriteLine(ret.getString());

Create a task.

BasicDBTask task = new BasicDBTask("1..10");
pool.execute(task);

Check whether the task is executed successfully. If successful, returns the results; otherwise returns the error messages.

BasicIntVector data = null;
if (task.isSuccessful())
{
      data = (BasicIntVector)task.getResults();
}
else
{
      throw new Exception(task.getErrorMsg());
}
System.Console.Out.WriteLine(data.getString());

Output:

[1,2,3,4,5,6,7,8,9,10]

Create multiple tasks and call these tasks concurrently in ExclusiveDBConnectionPool.

List<IDBTask> tasks = new List<IDBTask>();
for (int i = 0; i < 10; ++i){
      //call function log
      tasks.Add(new BasicDBTask("log", new List<IEntity> { data.get(i) }));
}
pool.execute(tasks);

Check whether the task is executed successfully. If successful, returns the results; otherwise returns the error messages.

for (int i = 0; i < 10; ++i)
{
      if (tasks[i].isSuccessful())
      {
            logData.append((IScalar)tasks[i].getResults());
      }
      else
      {
            throw new Exception(tasks[i].getErrorMsg());
      }
}
System.Console.Out.WriteLine(logData.getString());

Output:

[0,0.693147,1.098612,1.386294,1.609438,1.791759,1.94591,2.079442,2.197225,2.302585]

DolphinDB offers several ways to write to an in-memory table:

• Insert a single row of data with insert into
• Insert multiple rows of data in bulk with function tableInsert
• Insert a table object with function tableInsert

It is not recommended to save data with function append!, as append! returns the schema of a table and unnecessarily increases the network traffic.

The table in the following examples has 4 columns. Their data types are STRING, INT, TIMESTAMP and DOUBLE. The column names are cstring, cint, ctimestamp and cdouble, respectively.

t = table(10000:0,`cstring`cint`ctimestamp`cdouble,[STRING,INT,TIMESTAMP,DOUBLE])
share t as sharedTable

By default, an in-memory table is not shared among sessions. To access it in a different session, share it among sessions with share.

DFS table is recommended in production environment. It supports snapshot isolation and ensures data consistency. With data replication, DFS tables offer fault tolerance and load balancing.

In the C# API, a table is saved as a BasicTable object. Since BasicTable is column-based, to retrieve rows, you need to get the necessary columns first and then get the rows.

In the example below, the BasicTable has 4 columns with data types STRING, INT, TIMESTAMP and DOUBLE. The column names are cstring, cint, ctimestamp and cdouble.

public void test_loop_basicTable(BasicTable table1)
{
      BasicStringVector stringv = (BasicStringVector) table1.getColumn("cstring");
      BasicIntVector intv = (BasicIntVector)table1.getColumn("cint");
      BasicTimestampVector timestampv = (BasicTimestampVector)table1.getColumn("ctimestamp");
      BasicDoubleVector doublev = (BasicDoubleVector)table1.getColumn("cdouble");
      for(int ri=0; ri<table1.rows(); ri++){
            Console.WriteLine(stringv.getString(ri));
            Console.WriteLine(intv.getInt(ri));
            DateTime timestamp = timestampv.getTimestamp(ri);
            Console.WriteLine(timestamp);
            Console.WriteLine(doublev.getDouble(ri));
      }
}

DolphinDB C# API provides MultithreadedTableWriter that supports concurrent writes in multiple threads. You can use methods of MultithreadedTableWriter class to asynchronously append data to a DolphinDB in-memory table, dimension table, or a DFS table.

The methods of MultithreadedTableWriter object are introduced as follows:

MultithreadedTableWriter(string hostName, int port, string userId, string password,string dbName, string tableName, bool useSSL, bool enableHighAvailability = false, string[] pHighAvailabilitySites = null,int batchSize = 1, float throttle = 0.01f, int threadCount = 1, string partitionCol = "", int[] pCompressMethods = null, Mode mode = Mode.M_Append, string[] pModeOption = null, Callback callbackHandler = null);

Parameters:

• hostName: a string indicating host name
• port: an integer indicating port number
• userId / password: a string indicating username and password
• dbPath: a string indicating the DFS database path. Leave it unspecified for an in-memory table.
• tableName: a string indicating the DFS database path. Leave it unspecified for an in-memory table. Note: For API 1.30.17 or lower versions, when writing to an in-memory table, please specify the in-memory table name for dbPath and leave tableName empty.
• useSSL: a Boolean value indicating whether to enable SSL. The default value is false.
• enableHighAvailability: a Boolean value indicating whether to enable high availability. The default value is false.
• pHighAvailabilitySites: a list of ip:port of all available nodes
• batchSize: an integer indicating the number of messages in batch processing. The default value is 1, indicating the server processes the data as soon as they are written. If it is greater than 1, only when the number of data reaches batchSize, the client will send the data to the server.
• throttle: a positive floating-point number indicating the waiting time (in seconds) before the server processes the incoming data if the number of data written from the client does not reach batchSize.
• threadCount: an integer indicating the number of working threads to be created. The default value is 1, indicating single-threaded process. It must be 1 for a dimension table.
• partitionCol: a string indicating the partitioning column. It is None by default, and only takes effect when threadCount is greater than 1. For a partitioned table, it must be the partitioning column; for a stream table, it must be a column name; for a dimension table, the parameter does not take effect.
• pCompressMethods: an array of the compression methods used for each column. If unspecified, the columns are not compressed. The compression methods (case-insensitive) include: * "Vector_Fields.COMPRESS_LZ4": LZ4 algorithm * "Vector_Fields.COMPRESS_DELTA": Delta-of-delta encoding
• mode: indicates how data is written. It can be specified as Mode.M_Append or Mode.M_Upsert.
  • Mode.M_Append: the data is appended by [tableInsert](https://www.dolphindb.com/help/FunctionsandCommands/FunctionReferences/t/tableInsert.html).
  • Mode.M_Upsert: the data is updated by [upsert!](https://www.dolphindb.com/help/FunctionsandCommands/FunctionReferences/u/upsert%21.html).
• pModeOption: Currently, it takes effect only when mode is specified as M_Upsert. It is a STRING array consisting of the optional parameters of upsert!, indicating the expansion option of mode. For example, it can be specified as new String[] { "ignoreNull=false", "keyColNames=`volume" }.
• callbackHandler: the Callback class. The default value is None, indicating that no callback is used. With callback enabled, the Callback class is inherited and method writeCompletion is reloaded. If parameter callbackHandler is specified, note that:

• The first parameter of insert must be of type STRING, indicating the id of the row.
• The getUnwrittenData method will not be available.

The following part introduces methods of MultithreadedTableWriter object.

(1) insert

ErrorCodeInfo insert(params Object[] args)

Details:

Insert a single record. Return a class ErrorCodeInfo containing errorCode and errorInfo. If errorCode is not "", MultithreadedTableWriter has failed to insert the data, and errorInfo displays the error message.

The class ErrorCodeInfo provides methods hasError() and succeed() to check whether the data is written properly. hasError() returns true if an error occurred, otherwise false. succeed() returns true if the data is written successfully, otherwise false.

Parameters:

• args: a variable-length argument (varargs) indicating the record to be inserted.

(2) getUnwrittenData

List<List<IEntity>> getUnwrittenData();

Details:

Return a nested list of data that has not been written to the server.

Note: Data obtained by this method will be released by MultithreadedTableWriter.

(3) insertUnwrittenData

ErrorCodeInfo insertUnwrittenData(List<List<IEntity>> data);

Details:

Insert unwritten data. The result is in the same format as insert. The difference is that insertUnwrittenData can insert multiple records at a time.

Parameters:

• data: the data that has not been written to the server. You can obtain the object with method getUnwrittenData.

(4) getStatus

Status getStatus()

Details:

Get the current status of the MultithreadedTableWriter object.

Parameters:

• status: the MultithreadedTableWriter.Status class with the following attributes and methods:

Attributes:

• isExiting: whether the threads are exiting
• errorCode: error code
• errorInfo: error message
• sentRows: number of sent rows
• unsentRows: number of rows to be sent
• sendFailedRows: number of rows failed to be sent
• threadStatus: a list of the thread status
  • threadId: thread ID
  • sentRows: number of rows sent by the thread
  • unsentRows: number of rows to be sent by the thread
  • sendFailedRows: number of rows failed to be sent by the thread

(5) waitForThreadCompletion

waitForThreadCompletion()

Details:

After calling the method, MultithreadedTableWriter will wait until all working threads complete their tasks.

The methods of MultithreadedTableWriter are usually used in the following way:

//build a connection and initialize the environment
string HOST = "192.168.1.38";
int PORT = 18848;
string USER = "admin";
string PASSWD = "123456";
DBConnection dBConnection = new DBConnection();
dBConnection.connect(HOST, PORT, USER, PASSWD);
Random random = new Random();
string script =
"dbName = 'dfs://valuedb3'" +
"if (exists(dbName))" +
"{" +
      "dropDatabase(dbName);" +
"}" +
"datetest = table(1000:0,`date`symbol`id,[DATE, SYMBOL, LONG]);" +
"db = database(directory= dbName, partitionType= HASH, partitionScheme=[INT, 10]);" +
"pt=db.createPartitionedTable(datetest,'pdatetest','id');";
dBConnection.run(script);

ErrorCodeInfo ret;
MultithreadedTableWriter.Status writeStatus;
MultithreadedTableWriter writer = new MultithreadedTableWriter(HOST, PORT, USER, PASSWD, "dfs://valuedb3", "pdatetest", false, false, null, 10000, 1, 5, "id", new int[] { Vector_Fields.COMPRESS_LZ4, Vector_Fields.COMPRESS_LZ4, Vector_Fields.COMPRESS_DELTA });
try
{
      //insert 100 rows of records with correct data types and column count
      for (int i = 0; i < 100; ++i)
      {
            ret = writer.insert(new DateTime(2022, 3, 23), "AAAAAAAB", i);
            //this line will not be executed
            if (ret.errorCode != "")
            Console.WriteLine(string.Format("insert wrong format data: {0}\n", ret.ToString()));
      }
      Thread.Sleep(2000);

      //An error message is returned at once if insert one row with wrong data type.
      ret = writer.insert(new DateTime(2022, 3, 23), random.Next() % 10000, random.Next() % 10000);
      if (ret.errorCode != "")
            Console.WriteLine("insert wrong format data: {0}\n", ret.ToString());
      /*
      insert wrong format data: code = A1 info = Failed to insert data. Cannot convert int to DT_SYMBOL
      */

      //If a disconnection occurs, MTW will fail the next time it writes data to the server.
      //Write one row of data first to trigger an error.
      ret = writer.insert(new DateTime(2022, 3, 23), "AAAAAAAB", 1);

      Thread.Sleep(1000);

      //Insert 9 more rows of correct data types, MTW raises an exception because the work thread terminates and the row will not be written to MTW.
      //an exception will be thrown directly here
      for (int i = 0; i < 9; ++i)
      {
            ret = writer.insert(new DateTime(2022, 3, 23), "AAAAAAAB", random.Next() % 10000);
      }

}
catch (Exception e)
{
      Console.WriteLine(e.Message);
      //Thread is exiting.
}
writer.waitForThreadCompletion();
writeStatus = writer.getStatus();
if (writeStatus.errorCode != "")
      //an error occurred when writing
      Console.WriteLine(string.Format("error in writing:\n {0}", writeStatus.ToString()));
Console.WriteLine(((BasicLong)dBConnection.run("exec count(*) from pt")).getLong());

/*
      error in writing: Cause of write failure
sentRows: 100
unsentRows: 3
sendFailedRows: 7
threadId : 3 sentRows : 20 unsentRows : 0 sendFailedRows : 5
threadId : 4 sentRows : 20 unsentRows : 2 sendFailedRows : 1
threadId : 5 sentRows : 20 unsentRows : 1 sendFailedRows : 0
threadId : 6 sentRows : 20 unsentRows : 0 sendFailedRows : 0
threadId : 7 sentRows : 20 unsentRows : 0 sendFailedRows : 1

100
      */

for (int i = 0; i < 30; ++i)
      Console.Write('-');
Console.WriteLine();

List<List<IEntity>> unwriterdata = new List<List<IEntity>>();
if (writeStatus.sentRows != 110)
{
      Console.WriteLine("error after write complete:" + writeStatus.errorInfo);
      unwriterdata = writer.getUnwrittenData();
      Console.WriteLine("unwriterdata {0}", unwriterdata.Count);

      //To rewrite this data, a new MTW needs to be created because the original MTW is no longer available due to an exception.
      MultithreadedTableWriter newWriter = new MultithreadedTableWriter(HOST, PORT, USER, PASSWD, "dfs://valuedb3", "pdatetest", false, false, null, 10000, 1, 5, "id", new int[] { Vector_Fields.COMPRESS_LZ4, Vector_Fields.COMPRESS_LZ4, Vector_Fields.COMPRESS_DELTA });
      try
      {
            //insert unwritten data
            if (newWriter.insertUnwrittenData(unwriterdata).errorCode != "")
            {
            //wait until MTW finishes writing
            newWriter.waitForThreadCompletion();
            writeStatus = newWriter.getStatus();
            if (writeStatus.errorCode != "")
            {
                  System.Console.Out.WriteLine("error in write again:" + writeStatus.errorInfo);
            }
            }

      }
      finally
      {
            newWriter.waitForThreadCompletion();
            writeStatus = newWriter.getStatus();
            Console.WriteLine(string.Format("write again result:\n{0}", writeStatus));
      }
}
else
      Console.WriteLine("write complete : \n {0}", writeStatus.ToString());
//check the status
Console.WriteLine(((BasicLong)dBConnection.run("exec count(*) from pt")).getLong());
/*
      unwriterdata 10
write again result:

isExiting: True
sentRows: 10
unsentRows: 0
sendFailedRows: 0
threadId : 8 sentRows : 5 unsentRows : 0 sendFailedRows : 0
threadId : 9 sentRows : 3 unsentRows : 0 sendFailedRows : 0
threadId : 10 sentRows : 1 unsentRows : 0 sendFailedRows : 0
threadId : 11 sentRows : 0 unsentRows : 0 sendFailedRows : 0
threadId : 12 sentRows : 1 unsentRows : 0 sendFailedRows : 0

110
      */

Enable the Callback class for MultithreadedTableWriter

With callback enabled, the Callback class is inherited and method writeCompletion is reloaded to obtain the callback data.

A table of type BasicTable will be returned using the callback method, which consists of two columns:

• The first column (of String type) holds the id of each row added with MultithreadedTableWriter.insert.
• The second column (of Boolean type) indicates whether each row was written successfully or not. True means it was written successfully; False means it failed.

Example:

public class CallbackHandler : Callback
{
    public void writeCompletion(ITable callbackTable)
    {
        List<String> failedIdList = new List<string>();
        BasicStringVector idVec = (BasicStringVector)callbackTable.getColumn(0);
        BasicBooleanVector successVec = (BasicBooleanVector)callbackTable.getColumn(1);
        for (int i = 0; i < successVec.rows(); i++)
        {
            if (!successVec.getBoolean(i))
            {
                failedIdList.Add(idVec.getString(i));
            }
        }
    }
}

Example:

MultithreadedTableWriter mtw = new MultithreadedTableWriter(host, port, userName, password, dbName, tbName, useSSL,
        enableHighAvailability, null, 10000, 1, 1, "price", null,MultithreadedTableWriter.Mode.M_Append,null, new CallbackHandler());

Use the insert method of MultithreadedTableWriter and write the id for each row in the first column.

String theme = "theme1";
for (int id = 0; id < 1000000; id++){
    mtw.insert(theme + id, code, price); //theme+id is the id of each row, which will be returned in the callback
}

DolphinDB C# API provides AutoFitTableUpsert class to update and write to DolphinDB tables. AutoFitTableUpsert has the same functions as MultithreadedTableWriter when its parameter mode is specified as Mode.M_Upsert. The difference is that AutoFitTableUpsert is single-threaded and MultithreadedTableWriter writes with multiple threads.

The methods of AutoFitTableUpsert object are introduced as follows:

AutoFitTableUpsert(string dbUrl, string tableName, DBConnection connection, bool ignoreNull, string[] pkeyColNames, string[] psortColumns)

Parameters:

• dbUrl: a string indicating the path of the DFS database. It is specified as None for in-memory table.
• tableName: a STRING indicating the in-memory or DFS table name.
• connection: a DBConnection object to connect to the DolphinDB server and upsert data. Note that asynchronousTask must be false when creating the DBConnection object for AutoFitTableUpsert.
• ignoreNull: a Boolean value. It is used to specify the parameter ignoreNull of upsert!, indicating whether to update the NULL values of the target table.
• pkeyColNames: a STRING array. It is used to specify the parameter keyColNames of upsert!, i.e., specifing the key columns.
• psortColumnsa STRING array. It is used to specify the parameter sortColumns of upsert!. The updated partitions will be sorted on sortColumns (only within each partition, not across partitions).

The following introduces the upsert method of AutoFitTableUpsert object.

int upsert(BasicTable table)

Details:

Update a BasicTable object to the target table, and return an integer indicating the number of rows that have been updated.

Example:

DBConnection conn = new DBConnection(false, false, false);
conn.connect("192.168.1.116", 18999, "admin", "123456");
String dbName = "dfs://upsertTable";
String tableName = "pt";
String script = "dbName = \"dfs://upsertTable\"\n" +
"if(exists(dbName)){\n" +
"\tdropDatabase(dbName)\t\n" +
"}\n" +
"db  = database(dbName, RANGE,1 10000,,'TSDB')\n" +
"t = table(1000:0, `id`value,[ INT, INT[]])\n" +
"pt = db.createPartitionedTable(t,`pt,`id,,`id)";
conn.run(script);

BasicIntVector v1 = new BasicIntVector(3);
v1.setInt(0, 1);
v1.setInt(1, 100);
v1.setInt(2, 9999);

BasicArrayVector ba = new BasicArrayVector(DATA_TYPE.DT_INT_ARRAY);
ba.append(v1);
ba.append(v1);
ba.append(v1);

List<String> colNames = new List<string>();
colNames.Add("id");
colNames.Add("value");
List<IVector> cols = new List<IVector>();
cols.Add(v1);
cols.Add(ba);
BasicTable bt = new BasicTable(colNames, cols);
String[] keyColName = new String[] { "id" };
AutoFitTableUpsert aftu = new AutoFitTableUpsert(dbName, tableName, conn, false, keyColName, null);
aftu.upsert(bt);
BasicTable res = (BasicTable)conn.run("select * from pt;");
System.Console.Out.WriteLine(res.getString());

The corresponding interfaces of subscribe are:

(1) Subscribe using ThreadedClient:

subscribe(string host, int port, string tableName, string actionName, MessageHandler handler, long offset, bool reconnect, IVector filter, int batchSize, float throttle = 0.01f, StreamDeserializer deserializer = null, string user = "", string password = "", bool msgAsTable = false)

Parameters:

• host: the IP address of the publisher node.
• port: the port number of the publisher node.
• tableName: a string indicating the name of the publishing stream table.
• actionName: a string indicating the name of the subscription task.
• handler: a user-defined function to process the subscribed data.
• offset: 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: a Boolean value indicating whether to resubscribe after network disconnection.
• filter: a vector indicating 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: an integer indicating 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: a floating-point number indicating the maximum waiting time (in seconds) before the handler processes the incoming messages. The default value is 1. This optional parameter has no effect if batchSize is not specified.
• deserializer: the deserializer for the subscribed heterogeneous stream table.
• user: a string indicating the username used to connect to the server.
• password: a string indicating the password used to connect to the server.
• msgAsTable: a Boolean value indicating whether the subscribed data is ingested into handler as a Table. true means the subscribed data is ingested into handler as a Table.false (default) means the subscribed data is ingested into handler as an AnyVector. Note: Do not set batchSize and deserializer when msgAsTable=true. Setting msgAsTable=true is not recommended when subscribing to heterogeneous stream tables.

(2) Subscribe using ThreadPooledClient:

subscribe(string host, int port, string tableName, string actionName, MessageHandler handler, long offset, bool reconnect, IVector filter, StreamDeserializer deserializer = null, string user = "", string password = "", bool msgAsTable = false)

(3) Subscribe using PollingClient:

subscribe(string host, int port, string tableName, string actionName, long offset, bool reconnect, IVector filter, StreamDeserializer deserializer = null, string user = "", string password = "", bool msgAsTable = false)

The following examples introduce how to subscribe to stream table:

• The application on the client periodically checks if new data has been added to the streaming table. If yes, the application will acquire and consume the new data.

PollingClient client = new PollingClient(subscribeHost, subscribePort);
TopicPoller poller1 = client.subscribe(serverIP, serverPort, tableName, offset);

while (true)
{
      List<IMessage> msgs = poller1.poll(1000);

      if (msgs.Count > 0)
      {
            foreach(IMessage msg in msgs)
            System.Console.Out.WriteLine(string.Format("receive: {0}, {1}, {2}", msg.getEntity(0).getString(), msg.getEntity(1).getString(), msg.getEntity(2).getString()));
      }
      /*
      Successfully subscribed table 192.168.1.38:18848:local8848/Trades/csharpStreamingApi
      receive: 1, 2022.05.26T10:39:22.105, 1.5
      */
}

• The API uses MessageHandler to get new data

First you need to define the message handler, which needs to implement dolphindb.streaming.MessageHandler interface.

public class MyHandler : MessageHandler
{
      public void doEvent(IMessage msg)
      {
            System.Console.Out.WriteLine(string.Format("receive: {0}, {1}, {2}", msg.getEntity(0).getString(), msg.getEntity(1).getString(), msg.getEntity(2).getString()));
      }

      public void batchHandler(List<IMessage> msgs)
      {
            throw new NotImplementedException();
      }
}

You can pass the handler instance into function subscribe as a parameter with single-thread or multi-thread callbacks.

(1) ThreadedClient

ThreadedClient client = new ThreadedClient(subscribeHost, subscribePort);
client.subscribe(serverIP, serverPort, tableName, new MyHandler());
Thread.Sleep(10000);
//To cancel the subscription, you can use function close.
client.close();

(2) ThreadPooledClient: Handler mode client (multithreading)

ThreadPooledClient client = new ThreadPooledClient(subscribeHost, subscribePort);
client.subscribe(serverIP, serverPort, tableName, new MyHandler());
//To cancel the subscription, you can use function close.
Thread.Sleep(10000);
client.close();

Parameter reconnect is a Boolean value indicating whether to automatically resubscribe after the subscription experiences an unexpected interruption. The default value is false.

When reconnect=true:

• If the publisher and the subscriber both stay on but the network connection is interrupted, then after network is restored, the subscriber resumes subscription from where the network interruption occurs.
• If the publisher crashes, the subscriber will keep attempting to resume subscription after the publisher restarts.
  • If persistence was enabled on the publisher, the publisher starts to read the persisted data on disk after restarting. Automatic resubscription would fail until the publisher has read the data for the time when the publisher crashed.
  • If persistence was not enabled on the publisher, the automatic subscription will fail.
• If the subscriber crashes, the subscriber won't automatically resume the subscription after it restarts. In this case, we need to execute function subscribe again.

Parameter reconnect is set to be true for the following example:

PollingClient client = new PollingClient(subscribePort);
TopicPoller poller1 = client.subscribe(serverIP, serverPort, tableName, offset, true);

Parameter filter is a vector. It is used together with function setStreamTableFilterColumn at the publisher node. Function setStreamTableFilterColumn specifies the filtering column in the streaming table. Only the rows with filtering column values in filter are published.

In the following example, parameter filter is assigned an INT vector [1,2]:

BasicIntVector filter = new BasicIntVector(2);
filter.setInt(0, 1);
filter.setInt(1, 2);

PollingClient client = new PollingClient(subscribePort);
TopicPoller poller1 = client.subscribe(serverIP, serverPort, tableName, actionName, offset, filter);

Since DolphinDB server version 1.30.17/2.00.5, the replay function supports replaying (serializing) multiple stream tables with different schemata into a single stream table (known as "heterogeneous stream table"). Starting from DolphinDB C# API version 1.30.19, a new class streamDeserializer has been introduced for the subscription and deserialization of heterogeneous stream table.

Each subscription is identified with a subscription topic. Subscription fails if a topic with the same name already exists. You can cancel the subscription with unsubscribe.

client.unsubscribe(serverIP, serverPort, tableName,actionName);

EventSchema can be used to define events on API client.

Constructor

public EventSchema(string eventType, List<string> fieldNames, 
  List<DATA_TYPE> fieldTypes, List<DATA_FORM> fieldForms, 
  List<int> fieldExtraParams = null)

Arguments

• eventType: A String indicating the event.
• fieldNames: A String indicating the field names.
• fieldTypes: Data types of fields.
• fieldForms: Data forms of fields.
• fieldExtraParams: A non-negative int used to specify the scale when the field type is DECIMAL.

EventSender can be used to write serialized event data into heterogeneous stream tables, which are subscribed by CEP engines in DolphinDB to capture and process the events.

The API provides the EventClient tool for subscribing to events in heterogeneous stream tables. These tables typically receive event data processed and serialized by the DolphinDB CEP engine.

This example introduces how to create events on the server, interfaces for serializing events, heterogeneous stream tables for writing and outputting events, and the CEP engine for processing events. Then use EventSender on the API to send events to the heterogeneous stream table. After the CEP engine processes and outputs messages to another heterogeneous stream table through subscription, use EventClient on the API side to subscribe to the output events.

The following script will create two heterogeneous stream tables "input" and "output", two streamEventSerializer (serialization interfaces) objects, a CEP engine, and a "MarketData" event on the DolphinDB server.

class MarketData{
  market :: STRING
  code::STRING
  price :: DOUBLE
  qty :: INT
  eventTime :: TIMESTAMP
  def MarketData(m,c,p,q){
    market = m
    code = c
    price = p
    qty = q
    eventTime = now()   
  }
}   

class MainMonitor{

  def MainMonitor(){
  }
  def updateMarketData(event)

  def onload(){
    addEventListener(updateMarketData,'MarketData',,'all')
      }

  def updateMarketData(event){
    emitEvent(event)
  }
}
dummy = table(array(TIMESTAMP, 0) as eventTime, array(STRING, 0) as eventType, array(BLOB, 0) as blobs)
share streamTable(array(TIMESTAMP, 0) as eventTime, array(STRING, 0) as eventType, array(BLOB, 0) as blobs) as intput
share streamTable(array(TIMESTAMP, 0) as eventTime, array(STRING, 0) as eventType, array(BLOB, 0) as blobs) as output
schema = table(1:0, `eventType`eventKeys`eventValuesTypeString`eventValueTypeID`eventValuesFormID, [STRING, STRING, STRING, INT[], INT[]])
insert into schema values("MarketData", "market,code,price,qty,eventTime", "STRING,STRING,DOUBLE,INT,TIMESTAMP", [18 18 16 4 12], [0 0 0 0 0]) 
inputSerializer = streamEventSerializer(name=`serInput, eventSchema=schema, outputTable=intput, eventTimeField = "eventTime")
outputSerializer = streamEventSerializer(name=`serOutput, eventSchema=schema, outputTable=output, eventTimeField = "eventTime")

engine = createCEPEngine('cep1', <MainMonitor()>, dummy, [MarketData], 1, 'eventTime', 10000, outputSerializer)
subscribeTable(,`intput, `subopt, 0, getStreamEngine('cep1'),true)

Then write data to input via EventSender on API.

// Define schema
List<string> fieldNames = new List<string> { "market", "code", "price", "qty", "eventTime" };
List<DATA_TYPE> fieldTypes = new List<DATA_TYPE> {
  DATA_TYPE.DT_STRING, DATA_TYPE.DT_STRING, DATA_TYPE.DT_DOUBLE, DATA_TYPE.DT_INT, DATA_TYPE.DT_TIMESTAMP };
List<DATA_FORM> fieldForms = new List<DATA_FORM> {
  DATA_FORM.DF_SCALAR, DATA_FORM.DF_SCALAR, DATA_FORM.DF_SCALAR, DATA_FORM.DF_SCALAR, DATA_FORM.DF_SCALAR };
EventSchema schema = new EventSchema("MarketData", fieldNames, fieldTypes, fieldForms);
List<EventSchema> eventSchemas = new List<EventSchema> { schema };
List<string> eventTimeKeys = new List<string> { "eventTime" };

// Create EventSender
DBConnection conn = new DBConnection();
conn.connect(HOST, PORT, "admin", "123456");
EventSender sender = new EventSender(conn, "intput", eventSchemas, eventTimeKeys);

// Prepare data
List<IEntity> attributes = new List<IEntity> { new BasicString("sz"), new BasicString("s001") ,
  new BasicDouble(9.8), new BasicInt(100), new BasicTimestamp(DateTime.Now)};
sender.sendEvent("MarketData", attributes);

The stream table input is subscribed by a CEP engine, which generates events to the output table.

Then use EventClient to subscribe to the output table.

// user-defined function called by handler
class MyEventMessageHandler : EventMessageHandler
{
    public void doEvent(string eventType, List<IEntity> attribute)
    {// user-defined processing
        System.Console.WriteLine("eventType: " + eventType);
        foreach (IEntity entity in attribute)
        {
            System.Console.WriteLine(entity.getString());
        }
        System.Console.WriteLine();
    }
}

// Define schema
List<string> fieldNames = new List<string> { "market", "code", "price", "qty", "eventTime" };
List<DATA_TYPE> fieldTypes = new List<DATA_TYPE> {
    DATA_TYPE.DT_STRING, DATA_TYPE.DT_STRING, DATA_TYPE.DT_DOUBLE, DATA_TYPE.DT_INT, DATA_TYPE.DT_TIMESTAMP };
List<DATA_FORM> fieldForms = new List<DATA_FORM> {
    DATA_FORM.DF_SCALAR, DATA_FORM.DF_SCALAR, DATA_FORM.DF_SCALAR, DATA_FORM.DF_SCALAR, DATA_FORM.DF_SCALAR };
EventSchema schema = new EventSchema("MarketData", fieldNames, fieldTypes, fieldForms);
List<EventSchema> eventSchemas = new List<EventSchema> { schema };
List<string> eventTimeKeys = new List<string> { "eventTime" };

// create EventClient
EventClient client = new EventClient(eventSchemas, eventTimeKeys);

// create subscription
EventMessageHandler handler = new MyEventMessageHandler();
client.subscribe(HOST, PORT, "output", "CSharpClient", handler, 0, true, "admin", "123456");
Thread.Sleep(10000);

// cancel subscription
client.unsubscribe(HOST, PORT, "output", "CSharpClient");