--- sidebar_label: Subscription description: "Lightweight service for data subscription and pushing, the time series data inserted into TDengine continuously can be pushed automatically to the subscribing clients." title: Data Subscription --- import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; import Java from "./_sub_java.mdx"; import Python from "./_sub_python.mdx"; import Go from "./_sub_go.mdx"; import Rust from "./_sub_rust.mdx"; import Node from "./_sub_node.mdx"; import CSharp from "./_sub_cs.mdx"; import CDemo from "./_sub_c.mdx"; ## Introduction According to the time series nature of the data, data inserting in TDengine is similar to data publishing in message queues, they both can be considered as a new data record with timestamp is inserted into the system. Data is stored in ascending order of timestamp inside TDengine, so essentially each table in TDengine can be considered as a message queue. Lightweight service for data subscription and pushing is built in TDengine. With the API provided by TDengine, client programs can used `select` statement to subscribe the data from one or more tables. The subscription and and state maintenance is performed on the client side, the client programs polls the server to check whether there is new data, and if so the new data will be pushed back to the client side. If the client program is restarted, where to start for retrieving new data is up to the client side. There are 3 major APIs related to subscription provided in the TDengine client driver. ```c taos_subscribe taos_consume taos_unsubscribe ``` For more details about these API please refer to [C/C++ Connector](/reference/connector/cpp). Their usage will be introduced below using the use case of meters, in which the schema of STable and sub tables please refer to the previous section "continuous query". Full sample code can be found [here](https://github.com/taosdata/TDengine/blob/master/examples/c/subscribe.c). If we want to get notification and take some actions if the current exceeds a threshold, like 10A, from some meters, there are two ways: The first way is to query on each sub table and record the last timestamp matching the criteria, then after some time query on the data later than recorded timestamp and repeat this process. The SQL statements for this way are as below. ```sql select * from D1001 where ts > {last_timestamp1} and current > 10; select * from D1002 where ts > {last_timestamp2} and current > 10; ... ``` The above way works, but the problem is that the number of `select` statements increases with the number of meters grows. Finally the performance of both client side and server side will be unacceptable once the number of meters grows to a big enough number. A better way is to query on the STable, only one `select` is enough regardless of the number of meters, like below: ```sql select * from meters where ts > {last_timestamp} and current > 10; ``` However, how to choose `last_timestamp` becomes a new problem if using this way. Firstly, the timestamp when the data is generated is different from the timestamp when the data is inserted into the database, sometimes the difference between them may be very big. Secondly, the time when the data from different meters may arrives at the database may be different too. If the timestamp of the "slowest" meter is used as `last_timestamp` in the query, the data from other meters may be selected repeatedly; but if the timestamp of the "fasted" meters is used as `last_timestamp`, some data from other meters may be missed. All the problems mentioned above can be resolved thoroughly using subscription provided by TDengine. The first step is to create subscription using `taos_subscribe`. ```c TAOS_SUB* tsub = NULL; if (async) { 　　// create an asynchronous subscription, the callback function will be called every 1s 　　tsub = taos_subscribe(taos, restart, topic, sql, subscribe_callback, &blockFetch, 1000); } else { 　　// create an synchronous subscription, need to call 'taos_consume' manually 　　tsub = taos_subscribe(taos, restart, topic, sql, NULL, NULL, 0); } ``` The subscription in TDengine can be either synchronous or asynchronous. In the above sample code, the value of variable `async` is determined from the CLI input, then it's used to create either an async or sync subscription. Sync subscription means the client program needs to invoke `taos_consume` to retrieve data, and async subscription means another thread created by `taos_subscribe` internally invokes `taos_consume` to retrieve data and pass the data to `subscribe_callback` for processing, `subscribe_callback` is a call back function provided by the client program and it's suggested not to do time consuming operation in the call back function. The parameter `taos` is an established connection. There is nothing special in sync subscription mode. In async subscription, it should be exclusively by current thread, otherwise unpredictable error may occur. The parameter `sql` is a `select` statement in which `where` clause can be used to specify filter conditions. In our example, the data whose current exceeds 10A needs to be subscribed like below SQL statement: ```sql select * from meters where current > 10; ``` Please be noted that, all the data will be processed because no start time is specified. If only the data from one day ago needs to be processed, a time related condition can be added: ```sql select * from meters where ts > now - 1d and current > 10; ``` The parameter `topic` is the name of the subscription, it needs to be guaranteed unique in the client program, but it's not necessary to be globally unique because subscription is implemented in the APIs on client side. If the subscription named as `topic` doesn't exist, parameter `restart` would be ignored. If the subscription named as `topic` has been created before by the client program which then exited, when the client program is restarted to use this `topic`, parameter `restart` is used to determine retrieving data from beginning or from the last point where the subscription was broken. If the value of `restart` is **true** (i.e. a non-zero value), the data will be retrieved from beginning, or if it is **false** (i.e. zero), the data already consumed before will not be processed again. The last parameter of `taos_subscribe` is the polling interval in unit of millisecond. In sync mode, if the time difference between two continuous invocations to `taos_consume` is smaller than the interval specified by `taos_subscribe`, `taos_consume` would be blocked until the interval is reached. In async mode, this interval is the minimum interval between two invocations to the call back function. The last second parameter of `taos_subscribe` is used to pass arguments to the call back function. `taos_subscribe` doesn't process this parameter and simply passes it to the call back function. This parameter is simply ignored in sync mode. After a subscription is created, its data can be consumed and processed, below is the sample code of how to consume data in sync mode, in the else part if `if (async)`. ```c if (async) { 　　getchar(); } else while(1) { 　　TAOS_RES* res = taos_consume(tsub); 　　if (res == NULL) { 　　　　printf("failed to consume data."); 　　　　break; 　　} else { 　　　　print_result(res, blockFetch); 　　　　getchar(); 　　} } ``` In the above sample code, there is an infinite loop, each time carriage return is entered `taos_consume` is invoked, the return value of `taos_consume` is the selected result set, exactly as the input of `taos_use_result`, in the above sample `print_result` is used instead to simplify the sample. Below is the implementation of `print_result`. ```c void print_result(TAOS_RES* res, int blockFetch) { 　　TAOS_ROW row = NULL; 　　int num_fields = taos_num_fields(res); 　　TAOS_FIELD* fields = taos_fetch_fields(res); 　　int nRows = 0; 　　if (blockFetch) { 　　　　nRows = taos_fetch_block(res, &row); 　　　　for (int i = 0; i < nRows; i++) { 　　　　　　char temp[256]; 　　　　　　taos_print_row(temp, row + i, fields, num_fields); 　　　　　　puts(temp); 　　　　} 　　} else { 　　　　while ((row = taos_fetch_row(res))) { 　　　　　　char temp[256]; 　　　　　　taos_print_row(temp, row, fields, num_fields); 　　　　　　puts(temp); 　　　　　　nRows++; 　　　　} 　　} 　　printf("%d rows consumed.\n", nRows); } ``` In the above code `taos_print_row` is used to process the data consumed. All the matching rows will be printed. In async mode, the data consuming is simpler as below. ```c void subscribe_callback(TAOS_SUB* tsub, TAOS_RES *res, void* param, int code) { 　　print_result(res, *(int*)param); } ``` `taos_unsubscribe` can be invoked to terminate a subscription. ```c taos_unsubscribe(tsub, keep); ``` The second parameter `keep` is used to specify whether to keep the subscription progress on the client sde. If it is **false**, i.e. **0**, then subscription will be restarted from beginning regardless of the `restart` parameter's value in when `taos_subscribe` is invoked again. The subscription progress information is stored in _{DataDir}/subscribe/_ , under which there is a file with same name as `topic` for each subscription, the subscription will be restarted from beginning if the corresponding progress file is removed. Now let's see the effect of the above sample code, assuming below prerequisites have been done. - The sample code has been downloaded to local system - TDengine has been installed and launched properly on same system - The database, STable, sub tables required in the sample code have been ready It's ready to launch below command in the directory where the sample code resides to compile and start the program. ```bash make ./subscribe -sql='select * from meters where current > 10;' ``` After the program is started, open another terminal and launch TDengine CLI `taos`, then use below SQL commands to insert a row whose current is 12A into table **D1001**. ```sql use test; insert into D1001 values(now, 12, 220, 1); ``` Then, this row of data will be shown by the example program on the first terminal because its current exceeds 10A. More data can be inserted for you to observe the output of the example program. ## Examples Below example program demonstrates how to subscribe the data rows whose current exceeds 10A using connectors. ### Prepare Data ```bash # create database "power" taos> create database power; # use "power" as the database in following operations taos> use power; # create super table "meters" taos> create table meters(ts timestamp, current float, voltage int, phase int) tags(location binary(64), groupId int); # create tabes using the schema defined by super table "meters" taos> create table d1001 using meters tags ("Beijing.Chaoyang", 2); taos> create table d1002 using meters tags ("Beijing.Haidian", 2); # insert some rows taos> insert into d1001 values("2020-08-15 12:00:00.000", 12, 220, 1),("2020-08-15 12:10:00.000", 12.3, 220, 2),("2020-08-15 12:20:00.000", 12.2, 220, 1); taos> insert into d1002 values("2020-08-15 12:00:00.000", 9.9, 220, 1),("2020-08-15 12:10:00.000", 10.3, 220, 1),("2020-08-15 12:20:00.000", 11.2, 220, 1); # filter out the rows in which current is bigger than 10A taos> select * from meters where current > 10; ts | current | voltage | phase | location | groupid | =========================================================================================================== 2020-08-15 12:10:00.000 | 10.30000 | 220 | 1 | Beijing.Haidian | 2 | 2020-08-15 12:20:00.000 | 11.20000 | 220 | 1 | Beijing.Haidian | 2 | 2020-08-15 12:00:00.000 | 12.00000 | 220 | 1 | Beijing.Chaoyang | 2 | 2020-08-15 12:10:00.000 | 12.30000 | 220 | 2 | Beijing.Chaoyang | 2 | 2020-08-15 12:20:00.000 | 12.20000 | 220 | 1 | Beijing.Chaoyang | 2 | Query OK, 5 row(s) in set (0.004896s) ``` ### Example Programs {/* */} {/* */} ### Run the Examples The example programs firstly consume all historical data matching the criteria. ```bash ts: 1597464000000 current: 12.0 voltage: 220 phase: 1 location: Beijing.Chaoyang groupid : 2 ts: 1597464600000 current: 12.3 voltage: 220 phase: 2 location: Beijing.Chaoyang groupid : 2 ts: 1597465200000 current: 12.2 voltage: 220 phase: 1 location: Beijing.Chaoyang groupid : 2 ts: 1597464600000 current: 10.3 voltage: 220 phase: 1 location: Beijing.Haidian groupid : 2 ts: 1597465200000 current: 11.2 voltage: 220 phase: 1 location: Beijing.Haidian groupid : 2 ``` Next, use TDengine CLI to insert a new row. ``` # taos taos> use power; taos> insert into d1001 values(now, 12.4, 220, 1); ``` Because the current in inserted row exceeds 10A, it will be consumed by the example program. ``` ts: 1651146662805 current: 12.4 voltage: 220 phase: 1 location: Beijing.Chaoyang groupid: 2 ```