diff --git a/docs/examples/c/tmq.c b/docs/examples/c/tmq.c new file mode 100644 index 0000000000..15ab4fcfc9 --- /dev/null +++ b/docs/examples/c/tmq.c @@ -0,0 +1,333 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * This program is free software: you can use, redistribute, and/or modify + * it under the terms of the GNU Affero General Public License, version 3 + * or later ("AGPL"), as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +#include +#include +#include +#include +#include +#include "taos.h" + +static int running = 1; +const char* topic_name = "topicname"; + +static int32_t msg_process(TAOS_RES* msg) { + char buf[1024]; + int32_t rows = 0; + + const char* topicName = tmq_get_topic_name(msg); + const char* dbName = tmq_get_db_name(msg); + int32_t vgroupId = tmq_get_vgroup_id(msg); + + printf("topic: %s\n", topicName); + printf("db: %s\n", dbName); + printf("vgroup id: %d\n", vgroupId); + + while (1) { + TAOS_ROW row = taos_fetch_row(msg); + if (row == NULL) break; + + TAOS_FIELD* fields = taos_fetch_fields(msg); + int32_t numOfFields = taos_field_count(msg); + // int32_t* length = taos_fetch_lengths(msg); + int32_t precision = taos_result_precision(msg); + rows++; + taos_print_row(buf, row, fields, numOfFields); + printf("precision: %d, row content: %s\n", precision, buf); + } + + return rows; +} + +static int32_t init_env() { + TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); + if (pConn == NULL) { + return -1; + } + + TAOS_RES* pRes; + // drop database if exists + printf("create database\n"); + pRes = taos_query(pConn, "drop topic topicname"); + if (taos_errno(pRes) != 0) { + printf("error in drop topicname, reason:%s\n", taos_errstr(pRes)); + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "drop database if exists tmqdb"); + if (taos_errno(pRes) != 0) { + printf("error in drop tmqdb, reason:%s\n", taos_errstr(pRes)); + } + taos_free_result(pRes); + + // create database + pRes = taos_query(pConn, "create database tmqdb precision 'ns' WAL_RETENTION_PERIOD 3600"); + if (taos_errno(pRes) != 0) { + printf("error in create tmqdb, reason:%s\n", taos_errstr(pRes)); + goto END; + } + taos_free_result(pRes); + + // create super table + printf("create super table\n"); + pRes = taos_query( + pConn, "create table tmqdb.stb (ts timestamp, c1 int, c2 float, c3 varchar(16)) tags(t1 int, t3 varchar(16))"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table stb, reason:%s\n", taos_errstr(pRes)); + goto END; + } + taos_free_result(pRes); + + // create sub tables + printf("create sub tables\n"); + pRes = taos_query(pConn, "create table tmqdb.ctb0 using tmqdb.stb tags(0, 'subtable0')"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table ctb0, reason:%s\n", taos_errstr(pRes)); + goto END; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create table tmqdb.ctb1 using tmqdb.stb tags(1, 'subtable1')"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table ctb1, reason:%s\n", taos_errstr(pRes)); + goto END; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create table tmqdb.ctb2 using tmqdb.stb tags(2, 'subtable2')"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table ctb2, reason:%s\n", taos_errstr(pRes)); + goto END; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create table tmqdb.ctb3 using tmqdb.stb tags(3, 'subtable3')"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table ctb3, reason:%s\n", taos_errstr(pRes)); + goto END; + } + taos_free_result(pRes); + + // insert data + printf("insert data into sub tables\n"); + pRes = taos_query(pConn, "insert into tmqdb.ctb0 values(now, 0, 0, 'a0')(now+1s, 0, 0, 'a00')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes)); + goto END; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "insert into tmqdb.ctb1 values(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes)); + goto END; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "insert into tmqdb.ctb2 values(now, 2, 2, 'a1')(now+1s, 22, 22, 'a22')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes)); + goto END; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "insert into tmqdb.ctb3 values(now, 3, 3, 'a1')(now+1s, 33, 33, 'a33')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes)); + goto END; + } + taos_free_result(pRes); + taos_close(pConn); + return 0; + +END: + taos_free_result(pRes); + taos_close(pConn); + return -1; +} + +int32_t create_topic() { + printf("create topic\n"); + TAOS_RES* pRes; + TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); + if (pConn == NULL) { + return -1; + } + + pRes = taos_query(pConn, "use tmqdb"); + if (taos_errno(pRes) != 0) { + printf("error in use tmqdb, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create topic topicname as select ts, c1, c2, c3, tbname from tmqdb.stb where c1 > 1"); + if (taos_errno(pRes) != 0) { + printf("failed to create topic topicname, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + taos_close(pConn); + return 0; +} + +void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) { + printf("tmq_commit_cb_print() code: %d, tmq: %p, param: %p\n", code, tmq, param); +} + +tmq_t* build_consumer() { + tmq_conf_res_t code; + tmq_t* tmq = NULL; + + tmq_conf_t* conf = tmq_conf_new(); + code = tmq_conf_set(conf, "enable.auto.commit", "true"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + code = tmq_conf_set(conf, "auto.commit.interval.ms", "1000"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + code = tmq_conf_set(conf, "group.id", "cgrpName"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + code = tmq_conf_set(conf, "client.id", "user defined name"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + code = tmq_conf_set(conf, "td.connect.user", "root"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + code = tmq_conf_set(conf, "td.connect.pass", "taosdata"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + code = tmq_conf_set(conf, "auto.offset.reset", "earliest"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + + tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL); + tmq = tmq_consumer_new(conf, NULL, 0); + +_end: + tmq_conf_destroy(conf); + return tmq; +} + +tmq_list_t* build_topic_list() { + tmq_list_t* topicList = tmq_list_new(); + int32_t code = tmq_list_append(topicList, topic_name); + if (code) { + tmq_list_destroy(topicList); + return NULL; + } + return topicList; +} + +void basic_consume_loop(tmq_t* tmq) { + int32_t totalRows = 0; + int32_t msgCnt = 0; + int32_t timeout = 5000; + while (running) { + TAOS_RES* tmqmsg = tmq_consumer_poll(tmq, timeout); + if (tmqmsg) { + msgCnt++; + totalRows += msg_process(tmqmsg); + taos_free_result(tmqmsg); + } else { + break; + } + } + + fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows); +} + +void consume_repeatly(tmq_t* tmq) { + int32_t numOfAssignment = 0; + tmq_topic_assignment* pAssign = NULL; + + int32_t code = tmq_get_topic_assignment(tmq, topic_name, &pAssign, &numOfAssignment); + if (code != 0) { + fprintf(stderr, "failed to get assignment, reason:%s", tmq_err2str(code)); + } + + // seek to the earliest offset + for(int32_t i = 0; i < numOfAssignment; ++i) { + tmq_topic_assignment* p = &pAssign[i]; + + code = tmq_offset_seek(tmq, topic_name, p->vgId, p->begin); + if (code != 0) { + fprintf(stderr, "failed to seek to %d, reason:%s", (int)p->begin, tmq_err2str(code)); + } + } + + tmq_free_assignment(pAssign); + + // let's do it again + basic_consume_loop(tmq); +} + +int main(int argc, char* argv[]) { + int32_t code; + + if (init_env() < 0) { + return -1; + } + + if (create_topic() < 0) { + return -1; + } + + tmq_t* tmq = build_consumer(); + if (NULL == tmq) { + fprintf(stderr, "build_consumer() fail!\n"); + return -1; + } + + tmq_list_t* topic_list = build_topic_list(); + if (NULL == topic_list) { + return -1; + } + + if ((code = tmq_subscribe(tmq, topic_list))) { + fprintf(stderr, "Failed to tmq_subscribe(): %s\n", tmq_err2str(code)); + } + + tmq_list_destroy(topic_list); + + basic_consume_loop(tmq); + + consume_repeatly(tmq); + + code = tmq_consumer_close(tmq); + if (code) { + fprintf(stderr, "Failed to close consumer: %s\n", tmq_err2str(code)); + } else { + fprintf(stderr, "Consumer closed\n"); + } + + return 0; +} diff --git a/docs/examples/rust/nativeexample/examples/schemaless.rs b/docs/examples/rust/nativeexample/examples/schemaless.rs index 2da99400d3..298dcd3890 100644 --- a/docs/examples/rust/nativeexample/examples/schemaless.rs +++ b/docs/examples/rust/nativeexample/examples/schemaless.rs @@ -10,7 +10,7 @@ async fn put() -> anyhow::Result<()> { std::env::set_var("RUST_LOG", "taos=debug"); pretty_env_logger::init(); let dsn = - std::env::var("TDENGINE_ClOUD_DSN").unwrap_or("http://localhost:6041".to_string()); + std::env::var("TDENGINE_ClOUD_DSN").unwrap_or("taos://localhost:6030".to_string()); log::debug!("dsn: {:?}", &dsn); let client = TaosBuilder::from_dsn(dsn)?.build().await?; diff --git a/docs/zh/08-develop/04-schemaless.md b/docs/zh/08-develop/04-schemaless.md index afb622e157..073e36c66e 100644 --- a/docs/zh/08-develop/04-schemaless.md +++ b/docs/zh/08-develop/04-schemaless.md @@ -232,7 +232,12 @@ writer.write(lineDemo, SchemalessProtocolType.LINE, SchemalessTimestampType.NANO ``` -除 DSN 不同，其余同 Websocket 代码示例。 + +```rust +{{#include docs/examples/rust/nativeexample/examples/schemaless.rs}} +``` + + ```csharp diff --git a/docs/zh/08-develop/05-stmt.md b/docs/zh/08-develop/05-stmt.md index e24e502a4b..d2ee8a3a2d 100644 --- a/docs/zh/08-develop/05-stmt.md +++ b/docs/zh/08-develop/05-stmt.md @@ -86,7 +86,11 @@ import TabItem from "@theme/TabItem"; ``` -除 DSN 不同，其余同 Websocket 代码示例。 + +```rust +{{#include docs/examples/rust/nativeexample/examples/stmt.rs}} +``` + ```csharp diff --git a/docs/zh/14-reference/05-connector/14-java.mdx b/docs/zh/14-reference/05-connector/14-java.mdx index 4877fe69d1..81b550ed16 100644 --- a/docs/zh/14-reference/05-connector/14-java.mdx +++ b/docs/zh/14-reference/05-connector/14-java.mdx @@ -68,7 +68,7 @@ JDBC 连接器可能报错的错误码包括 4 种： - JDBC driver 本身的报错（错误码在 0x2301 到 0x2350 之间） - 原生连接方法的报错（错误码在 0x2351 到 0x2360 之间） - 数据订阅的报错（错误码在 0x2371 到 0x2380 之间） -- TDengine 其他功能模块的报错。 +- TDengine 其他功能模块的报错，请参考 [错误码](../cpp/error-code) 具体的错误码请参考： diff --git a/docs/zh/14-reference/05-connector/26-rust.mdx b/docs/zh/14-reference/05-connector/26-rust.mdx index 54ff21090d..9991b638fe 100644 --- a/docs/zh/14-reference/05-connector/26-rust.mdx +++ b/docs/zh/14-reference/05-connector/26-rust.mdx @@ -61,6 +61,7 @@ match conn.exec(sql) { } } ``` +错误信息的错误码可以参考：[错误码](../cpp/error-code) ## 数据类型映射