xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'docs/sheyj-3.0' of github.com:taosdata/TDengine into docs/sheyj-3.0

This commit is contained in:
sheyanjie-qq 2024-08-16 15:18:58 +08:00
parent 1590a51e57 d6284424b6
commit c33705e4c9
7 changed files with 737 additions and 495 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

112
docs/examples/csharp/subscribe/Program.cs
View File

@ -6,6 +6,11 @@ namespace TMQExample
{ {
internal class SubscribeDemo internal class SubscribeDemo
{ {
private static string _host = "";
private static string _groupId = "";
private static string _clientId = "";
private static string _topic = "";
public static void Main(string[] args) public static void Main(string[] args)
{ {
try try
@ -64,9 +69,9 @@ namespace TMQExample
{ {
// ANCHOR: create_consumer // ANCHOR: create_consumer
// consumer config // consumer config
var host = "127.0.0.1"; _host = "127.0.0.1";
var groupId = "group1"; _groupId = "group1";
var clientId = "client1"; _clientId = "client1";
var cfg = new Dictionary<string, string>() var cfg = new Dictionary<string, string>()
{ {
{ "td.connect.port", "6030" }, { "td.connect.port", "6030" },
@ -74,9 +79,9 @@ namespace TMQExample
{ "msg.with.table.name", "true" }, { "msg.with.table.name", "true" },
{ "enable.auto.commit", "true" }, { "enable.auto.commit", "true" },
{ "auto.commit.interval.ms", "1000" }, { "auto.commit.interval.ms", "1000" },
{ "group.id", groupId }, { "group.id", _groupId },
{ "client.id", clientId }, { "client.id", _clientId },
{ "td.connect.ip", host }, { "td.connect.ip", _host },
{ "td.connect.user", "root" }, { "td.connect.user", "root" },
{ "td.connect.pass", "taosdata" }, { "td.connect.pass", "taosdata" },
}; };
@ -85,20 +90,32 @@ namespace TMQExample
{ {
// create consumer // create consumer
consumer = new ConsumerBuilder<Dictionary<string, object>>(cfg).Build(); consumer = new ConsumerBuilder<Dictionary<string, object>>(cfg).Build();
Console.WriteLine("Create consumer successfully, host: " + host + ", groupId: " + groupId + Console.WriteLine(
", clientId: " + clientId); $"Create consumer successfully, " +
$"host: {_host}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}");
} }
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to create native consumer, host: " + host + ", ErrCode:" + e.Code + Console.WriteLine(
", ErrMessage: " + e.Error); $"Failed to create native consumer, " +
$"host: {_host}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"ErrCode: {e.Code}, " +
$"ErrMessage: {e.Error}");
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to create native consumer, host: " + host + ", ErrMessage: " + e.Message); Console.WriteLine($"Failed to create native consumer, " +
$"host: {_host}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"ErrMessage: {e.Message}");
throw; throw;
} }
@ -109,11 +126,12 @@ namespace TMQExample
static void Consume(IConsumer<Dictionary<string, object>> consumer) static void Consume(IConsumer<Dictionary<string, object>> consumer)
{ {
// ANCHOR: subscribe // ANCHOR: subscribe
_topic = "topic_meters";
try try
{ {
// subscribe // subscribe
consumer.Subscribe(new List<string>() { "topic_meters" }); consumer.Subscribe(new List<string>() { _topic });
Console.WriteLine("subscribe topics successfully"); Console.WriteLine("Subscribe topics successfully");
for (int i = 0; i < 50; i++) for (int i = 0; i < 50; i++)
{ {
// consume message with using block to ensure the result is disposed // consume message with using block to ensure the result is disposed
@ -133,13 +151,23 @@ namespace TMQExample
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to poll data, ErrCode: " + e.Code + ", ErrMessage: " + e.Error); Console.WriteLine(
$"Failed to poll data, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"ErrCode: {e.Code}, " +
$"ErrMessage: {e.Error}");
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to poll data, ErrMessage: " + e.Message); Console.WriteLine($"Failed to poll data, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"ErrMessage: {e.Message}");
throw; throw;
} }
// ANCHOR_END: subscribe // ANCHOR_END: subscribe
@ -152,7 +180,7 @@ namespace TMQExample
{ {
// get assignment // get assignment
var assignment = consumer.Assignment; var assignment = consumer.Assignment;
Console.WriteLine($"now assignment: {assignment}"); Console.WriteLine($"Now assignment: {assignment}");
// seek to the beginning // seek to the beginning
foreach (var topicPartition in assignment) foreach (var topicPartition in assignment)
{ {
@ -163,13 +191,25 @@ namespace TMQExample
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to execute seek example, ErrCode: " + e.Code + ", ErrMessage: " + e.Error); Console.WriteLine(
$"Failed to execute seek example, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"offset: 0, " +
$"ErrCode: {e.Code}, " +
$"ErrMessage: {e.Error}");
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to execute seek example, ErrMessage: " + e.Message); Console.WriteLine($"Failed to execute seek example, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"offset: 0, " +
$"ErrMessage: {e.Message}");
throw; throw;
} }
// ANCHOR_END: seek // ANCHOR_END: seek
@ -180,6 +220,7 @@ namespace TMQExample
// ANCHOR: commit_offset // ANCHOR: commit_offset
for (int i = 0; i < 5; i++) for (int i = 0; i < 5; i++)
{ {
TopicPartitionOffset topicPartitionOffset = null;
try try
{ {
// consume message with using block to ensure the result is disposed // consume message with using block to ensure the result is disposed
@ -187,9 +228,10 @@ namespace TMQExample
{ {
if (cr == null) continue; if (cr == null) continue;
// commit offset // commit offset
topicPartitionOffset = cr.TopicPartitionOffset;
consumer.Commit(new List<TopicPartitionOffset> consumer.Commit(new List<TopicPartitionOffset>
{ {
cr.TopicPartitionOffset, topicPartitionOffset,
}); });
Console.WriteLine("Commit offset manually successfully."); Console.WriteLine("Commit offset manually successfully.");
} }
@ -197,13 +239,26 @@ namespace TMQExample
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to execute commit example, ErrCode:" + e.Code + ", ErrMessage: " + e.Error); Console.WriteLine(
$"Failed to execute commit example, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"offset: {topicPartitionOffset}, " +
$"ErrCode: {e.Code}, " +
$"ErrMessage: {e.Error}");
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to execute commit example, ErrMessage: " + e.Message); Console.WriteLine(
$"Failed to execute commit example, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"offset: {topicPartitionOffset}, " +
$"ErrMessage: {e.Message}");
throw; throw;
} }
} }
@ -221,13 +276,24 @@ namespace TMQExample
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to unsubscribe consumer, ErrCode: " + e.Code + ", ErrMessage: " + e.Error); Console.WriteLine(
$"Failed to unsubscribe consumer, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"ErrCode: {e.Code}, " +
$"ErrMessage: {e.Error}");
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to unsubscribe consumer, ErrMessage: " + e.Message); Console.WriteLine(
$"Failed to execute commit example, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"ErrMessage: {e.Message}");
throw; throw;
} }
finally finally

63
docs/examples/csharp/wsInsert/Program.cs
View File

@ -16,10 +16,10 @@ namespace Examples
var builder = new ConnectionStringBuilder(connectionString); var builder = new ConnectionStringBuilder(connectionString);
using (var client = DbDriver.Open(builder)) using (var client = DbDriver.Open(builder))
{ {
CreateDatabaseAndTable(client,connectionString); CreateDatabaseAndTable(client, connectionString);
InsertData(client,connectionString); InsertData(client, connectionString);
QueryData(client,connectionString); QueryData(client, connectionString);
QueryWithReqId(client,connectionString); QueryWithReqId(client, connectionString);
} }
} }
catch (TDengineError e) catch (TDengineError e)
@ -52,7 +52,8 @@ namespace Examples
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to create database power or stable meters, ErrCode: " + e.Code + ", ErrMessage: " + e.Error); Console.WriteLine("Failed to create database power or stable meters, ErrCode: " + e.Code +
", ErrMessage: " + e.Error);
throw; throw;
} }
catch (Exception e) catch (Exception e)
@ -64,40 +65,43 @@ namespace Examples
// ANCHOR_END: create_db_and_table // ANCHOR_END: create_db_and_table
} }
private static void InsertData(ITDengineClient client,string connectionString) private static void InsertData(ITDengineClient client, string connectionString)
{ {
// ANCHOR: insert_data // ANCHOR: insert_data
// insert data, please make sure the database and table are created before
var insertQuery = "INSERT INTO " +
"power.d1001 USING power.meters TAGS(2,'California.SanFrancisco') " +
"VALUES " +
"(NOW + 1a, 10.30000, 219, 0.31000) " +
"(NOW + 2a, 12.60000, 218, 0.33000) " +
"(NOW + 3a, 12.30000, 221, 0.31000) " +
"power.d1002 USING power.meters TAGS(3, 'California.SanFrancisco') " +
"VALUES " +
"(NOW + 1a, 10.30000, 218, 0.25000) ";
try try
{ {
// insert data, please make sure the database and table are created before
var insertQuery = "INSERT INTO " +
"power.d1001 USING power.meters TAGS(2,'California.SanFrancisco') " +
"VALUES " +
"(NOW + 1a, 10.30000, 219, 0.31000) " +
"(NOW + 2a, 12.60000, 218, 0.33000) " +
"(NOW + 3a, 12.30000, 221, 0.31000) " +
"power.d1002 USING power.meters TAGS(3, 'California.SanFrancisco') " +
"VALUES " +
"(NOW + 1a, 10.30000, 218, 0.25000) ";
var affectedRows = client.Exec(insertQuery); var affectedRows = client.Exec(insertQuery);
Console.WriteLine("Successfully inserted " + affectedRows + " rows to power.meters."); Console.WriteLine("Successfully inserted " + affectedRows + " rows to power.meters.");
} }
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to insert data to power.meters, ErrCode: " + e.Code + ", ErrMessage: " + e.Error); Console.WriteLine("Failed to insert data to power.meters, sql: " + insertQuery + ", ErrCode: " +
e.Code + ", ErrMessage: " +
e.Error);
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to insert data to power.meters, ErrMessage: " + e.Message); Console.WriteLine("Failed to insert data to power.meters, sql: " + insertQuery + ", ErrMessage: " +
e.Message);
throw; throw;
} }
// ANCHOR_END: insert_data // ANCHOR_END: insert_data
} }
private static void QueryData(ITDengineClient client,string connectionString) private static void QueryData(ITDengineClient client, string connectionString)
{ {
// ANCHOR: select_data // ANCHOR: select_data
// query data, make sure the database and table are created before // query data, make sure the database and table are created before
@ -108,6 +112,7 @@ namespace Examples
{ {
while (rows.Read()) while (rows.Read())
{ {
// Add your data processing logic here
var ts = (DateTime)rows.GetValue(0); var ts = (DateTime)rows.GetValue(0);
var current = (float)rows.GetValue(1); var current = (float)rows.GetValue(1);
var location = Encoding.UTF8.GetString((byte[])rows.GetValue(2)); var location = Encoding.UTF8.GetString((byte[])rows.GetValue(2));
@ -119,28 +124,30 @@ namespace Examples
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to query data from power.meters, sql: " + query + ", ErrCode: " + e.Code + ", ErrMessage: " + e.Error); Console.WriteLine("Failed to query data from power.meters, sql: " + query + ", ErrCode: " + e.Code +
", ErrMessage: " + e.Error);
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to query data from power.meters, sql: " + query + ", ErrMessage: " + e.Message); Console.WriteLine(
"Failed to query data from power.meters, sql: " + query + ", ErrMessage: " + e.Message);
throw; throw;
} }
// ANCHOR_END: select_data // ANCHOR_END: select_data
} }
private static void QueryWithReqId(ITDengineClient client,string connectionString) private static void QueryWithReqId(ITDengineClient client, string connectionString)
{ {
// ANCHOR: query_id // ANCHOR: query_id
var reqId = (long)3; var reqId = (long)3;
// query data
var query = "SELECT ts, current, location FROM power.meters limit 1";
try try
{ {
// query data
var query = "SELECT ts, current, location FROM power.meters limit 1";
// query with request id 3 // query with request id 3
using (var rows = client.Query(query,reqId)) using (var rows = client.Query(query, reqId))
{ {
while (rows.Read()) while (rows.Read())
{ {
@ -155,13 +162,15 @@ namespace Examples
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to execute sql with reqId: " + reqId + ", ErrCode: " + e.Code + ", ErrMessage: " + e.Error); Console.WriteLine("Failed to execute sql with reqId: " + reqId + ", sql: " + query + ", ErrCode: " +
e.Code + ", ErrMessage: " + e.Error);
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to execute sql with reqId: " + reqId + ", ErrMessage: " + e.Message); Console.WriteLine("Failed to execute sql with reqId: " + reqId + ", sql: " + query + ", ErrMessage: " +
e.Message);
throw; throw;
} }
// ANCHOR_END: query_id // ANCHOR_END: query_id

108
docs/examples/csharp/wssubscribe/Program.cs
View File

@ -6,6 +6,11 @@ namespace TMQExample
{ {
internal class SubscribeDemo internal class SubscribeDemo
{ {
private static string _host = "";
private static string _groupId = "";
private static string _clientId = "";
private static string _topic = "";
public static void Main(string[] args) public static void Main(string[] args)
{ {
try try
@ -68,9 +73,9 @@ namespace TMQExample
{ {
// ANCHOR: create_consumer // ANCHOR: create_consumer
// consumer config // consumer config
var host = "127.0.0.1"; _host = "127.0.0.1";
var groupId = "group1"; _groupId = "group1";
var clientId = "client1"; _clientId = "client1";
var cfg = new Dictionary<string, string>() var cfg = new Dictionary<string, string>()
{ {
{ "td.connect.type", "WebSocket" }, { "td.connect.type", "WebSocket" },
@ -79,9 +84,9 @@ namespace TMQExample
{ "msg.with.table.name", "true" }, { "msg.with.table.name", "true" },
{ "enable.auto.commit", "true" }, { "enable.auto.commit", "true" },
{ "auto.commit.interval.ms", "1000" }, { "auto.commit.interval.ms", "1000" },
{ "group.id", groupId }, { "group.id", _groupId },
{ "client.id", clientId }, { "client.id", _clientId },
{ "td.connect.ip", host }, { "td.connect.ip", _host },
{ "td.connect.user", "root" }, { "td.connect.user", "root" },
{ "td.connect.pass", "taosdata" }, { "td.connect.pass", "taosdata" },
}; };
@ -90,20 +95,32 @@ namespace TMQExample
{ {
// create consumer // create consumer
consumer = new ConsumerBuilder<Dictionary<string, object>>(cfg).Build(); consumer = new ConsumerBuilder<Dictionary<string, object>>(cfg).Build();
Console.WriteLine("Create consumer successfully, host: " + host + ", groupId: " + groupId + Console.WriteLine(
", clientId: " + clientId); $"Create consumer successfully, " +
$"host: {_host}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}");
} }
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to create websocket consumer, host: " + host + ", ErrCode: " + e.Code + Console.WriteLine(
", ErrMessage: " + e.Error); $"Failed to create native consumer, " +
$"host: {_host}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"ErrCode: {e.Code}, " +
$"ErrMessage: {e.Error}");
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to create websocket consumer, host: " + host + ", ErrMessage: " + e.Message); Console.WriteLine($"Failed to create native consumer, " +
$"host: {_host}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"ErrMessage: {e.Message}");
throw; throw;
} }
@ -114,10 +131,11 @@ namespace TMQExample
static void Consume(IConsumer<Dictionary<string, object>> consumer) static void Consume(IConsumer<Dictionary<string, object>> consumer)
{ {
// ANCHOR: subscribe // ANCHOR: subscribe
_topic = "topic_meters";
try try
{ {
// subscribe // subscribe
consumer.Subscribe(new List<string>() { "topic_meters" }); consumer.Subscribe(new List<string>() { _topic });
Console.WriteLine("Subscribe topics successfully"); Console.WriteLine("Subscribe topics successfully");
for (int i = 0; i < 50; i++) for (int i = 0; i < 50; i++)
{ {
@ -138,13 +156,23 @@ namespace TMQExample
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to poll data, ErrCode: " + e.Code + ", ErrMessage: " + e.Error); Console.WriteLine(
$"Failed to poll data, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"ErrCode: {e.Code}, " +
$"ErrMessage: {e.Error}");
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to poll data, ErrMessage: " + e.Message); Console.WriteLine($"Failed to poll data, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"ErrMessage: {e.Message}");
throw; throw;
} }
// ANCHOR_END: subscribe // ANCHOR_END: subscribe
@ -168,13 +196,25 @@ namespace TMQExample
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to execute seek example, ErrCode: " + e.Code + ", ErrMessage: " + e.Error); Console.WriteLine(
$"Failed to execute seek example, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"offset: 0, " +
$"ErrCode: {e.Code}, " +
$"ErrMessage: {e.Error}");
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to execute seek example, ErrMessage: " + e.Message); Console.WriteLine($"Failed to execute seek example, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"offset: 0, " +
$"ErrMessage: {e.Message}");
throw; throw;
} }
// ANCHOR_END: seek // ANCHOR_END: seek
@ -185,6 +225,7 @@ namespace TMQExample
// ANCHOR: commit_offset // ANCHOR: commit_offset
for (int i = 0; i < 5; i++) for (int i = 0; i < 5; i++)
{ {
TopicPartitionOffset topicPartitionOffset = null;
try try
{ {
// consume message with using block to ensure the result is disposed // consume message with using block to ensure the result is disposed
@ -192,9 +233,10 @@ namespace TMQExample
{ {
if (cr == null) continue; if (cr == null) continue;
// commit offset // commit offset
topicPartitionOffset = cr.TopicPartitionOffset;
consumer.Commit(new List<TopicPartitionOffset> consumer.Commit(new List<TopicPartitionOffset>
{ {
cr.TopicPartitionOffset, topicPartitionOffset,
}); });
Console.WriteLine("Commit offset manually successfully."); Console.WriteLine("Commit offset manually successfully.");
} }
@ -202,13 +244,26 @@ namespace TMQExample
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to execute commit example, ErrCode: " + e.Code + ", ErrMessage: " + e.Error); Console.WriteLine(
$"Failed to execute commit example, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"offset: {topicPartitionOffset}, " +
$"ErrCode: {e.Code}, " +
$"ErrMessage: {e.Error}");
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to execute commit example, ErrMessage: " + e.Message); Console.WriteLine(
$"Failed to execute commit example, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"offset: {topicPartitionOffset}, " +
$"ErrMessage: {e.Message}");
throw; throw;
} }
} }
@ -226,13 +281,24 @@ namespace TMQExample
catch (TDengineError e) catch (TDengineError e)
{ {
// handle TDengine error // handle TDengine error
Console.WriteLine("Failed to unsubscribe consumer, ErrCode :" + e.Code + ", ErrMessage: " + e.Error); Console.WriteLine(
$"Failed to unsubscribe consumer, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"ErrCode: {e.Code}, " +
$"ErrMessage: {e.Error}");
throw; throw;
} }
catch (Exception e) catch (Exception e)
{ {
// handle other exceptions // handle other exceptions
Console.WriteLine("Failed to unsubscribe consumer, ErrMessage: " + e.Message); Console.WriteLine(
$"Failed to execute commit example, " +
$"topic: {_topic}, " +
$"groupId: {_groupId}, " +
$"clientId: {_clientId}, " +
$"ErrMessage: {e.Message}");
throw; throw;
} }
finally finally

10
docs/examples/go/queryreqid/main.go
View File

@ -14,7 +14,7 @@ func main() {
taosDSN := "root:taosdata@tcp(localhost:6030)/" taosDSN := "root:taosdata@tcp(localhost:6030)/"
db, err := sql.Open("taosSql", taosDSN) db, err := sql.Open("taosSql", taosDSN)
if err != nil { if err != nil {
log.Fatalln("Failed to connect to " + taosDSN + "; ErrMessage: " + err.Error()) log.Fatalln("Failed to connect to " + taosDSN + ", ErrMessage: " + err.Error())
} }
defer db.Close() defer db.Close()
initEnv(db) initEnv(db)
@ -23,11 +23,13 @@ func main() {
reqId := int64(3) reqId := int64(3)
ctx := context.WithValue(context.Background(), "taos_req_id", reqId) ctx := context.WithValue(context.Background(), "taos_req_id", reqId)
// execute query with context // execute query with context
rows, err := db.QueryContext(ctx, "SELECT ts, current, location FROM power.meters limit 1") querySql := "SELECT ts, current, location FROM power.meters limit 1"
rows, err := db.QueryContext(ctx, querySql)
if err != nil { if err != nil {
log.Fatalf("Failed to execute sql with reqId: %d, url: %s; ErrMessage: %s\n", reqId, taosDSN, err.Error()) log.Fatalf("Failed to execute sql with reqId: %d, url: %s, sql: %s, ErrMessage: %s\n", reqId, taosDSN, querySql, err.Error())
} }
for rows.Next() { for rows.Next() {
// Add your data processing logic here
var ( var (
ts time.Time ts time.Time
current float32 current float32
@ -35,7 +37,7 @@ func main() {
) )
err = rows.Scan(&ts, &current, &location) err = rows.Scan(&ts, &current, &location)
if err != nil { if err != nil {
log.Fatal("Scan error: ", err) log.Fatalf("Failed to scan data, reqId: %d, url:%s, sql: %s, ErrMessage: %s\n", reqId, taosDSN, querySql, err)
} }
fmt.Printf("ts: %s, current: %f, location: %s\n", ts, current, location) fmt.Printf("ts: %s, current: %f, location: %s\n", ts, current, location)
} }

10
docs/examples/go/sqlquery/main.go
View File

@ -53,11 +53,11 @@ func main() {
"(NOW + 1a, 10.30000, 218, 0.25000) " "(NOW + 1a, 10.30000, 218, 0.25000) "
res, err = db.Exec(insertQuery) res, err = db.Exec(insertQuery)
if err != nil { if err != nil {
log.Fatal("Failed to insert data to power.meters, ErrMessage: " + err.Error()) log.Fatalf("Failed to insert data to power.meters, sql: %s, ErrMessage: %s\n", insertQuery, err.Error())
} }
rowsAffected, err = res.RowsAffected() rowsAffected, err = res.RowsAffected()
if err != nil { if err != nil {
log.Fatal("Failed to get insert rowsAffected, ErrMessage: " + err.Error()) log.Fatalf("Failed to get insert rowsAffected, sql: %s, ErrMessage: %s\n", insertQuery, err.Error())
} }
// you can check affectedRows here // you can check affectedRows here
fmt.Printf("Successfully inserted %d rows to power.meters.\n", rowsAffected) fmt.Printf("Successfully inserted %d rows to power.meters.\n", rowsAffected)
@ -67,9 +67,10 @@ func main() {
sql := "SELECT ts, current, location FROM power.meters limit 100" sql := "SELECT ts, current, location FROM power.meters limit 100"
rows, err := db.Query(sql) rows, err := db.Query(sql)
if err != nil { if err != nil {
log.Fatal("Failed to query data from power.meters, ErrMessage: " + err.Error()) log.Fatalf("Failed to query data from power.meters, sql: %s, ErrMessage: %s\n", sql, err.Error())
} }
for rows.Next() { for rows.Next() {
// Add your data processing logic here
var ( var (
ts time.Time ts time.Time
current float32 current float32
@ -77,9 +78,8 @@ func main() {
) )
err = rows.Scan(&ts, &current, &location) err = rows.Scan(&ts, &current, &location)
if err != nil { if err != nil {
log.Fatal("Failed to scan data, sql:" + sql + ", ErrMessage: " + err.Error()) log.Fatalf("Failed to scan data, sql: %s, ErrMessage: %s\n", sql, err)
} }
// you can check data here
fmt.Printf("ts: %s, current: %f, location: %s\n", ts, current, location) fmt.Printf("ts: %s, current: %f, location: %s\n", ts, current, location)
} }
// ANCHOR_END: select_data // ANCHOR_END: select_data

73
docs/examples/go/tmq/native/main.go
View File

@ -12,13 +12,17 @@ import (
) )
var done = make(chan struct{}) var done = make(chan struct{})
var groupID string
var clientID string
var host string
var topic string
func main() { func main() {
// init env // init env
taosDSN := "root:taosdata@tcp(127.0.0.1:6030)/" taosDSN := "root:taosdata@tcp(127.0.0.1:6030)/"
conn, err := sql.Open("taosSql", taosDSN) conn, err := sql.Open("taosSql", taosDSN)
if err != nil { if err != nil {
log.Fatalln("Failed to connect to " + taosDSN + "; ErrMessage: " + err.Error()) log.Fatalln("Failed to connect to " + taosDSN + ", ErrMessage: " + err.Error())
} }
defer func() { defer func() {
conn.Close() conn.Close()
@ -26,9 +30,9 @@ func main() {
initEnv(conn) initEnv(conn)
// ANCHOR: create_consumer // ANCHOR: create_consumer
// create consumer // create consumer
groupID := "group1" groupID = "group1"
clientID := "client1" clientID = "client1"
host := "127.0.0.1" host = "127.0.0.1"
consumer, err := tmq.NewConsumer(&tmqcommon.ConfigMap{ consumer, err := tmq.NewConsumer(&tmqcommon.ConfigMap{
"td.connect.user": "root", "td.connect.user": "root",
"td.connect.pass": "taosdata", "td.connect.pass": "taosdata",
@ -40,15 +44,28 @@ func main() {
"client.id": clientID, "client.id": clientID,
}) })
if err != nil { if err != nil {
log.Fatalln("Failed to create native consumer, host : " + host + "; ErrMessage: " + err.Error()) log.Fatalf(
"Failed to create native consumer, host: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
host,
groupID,
clientID,
err.Error(),
)
} }
log.Println("Create consumer successfully, host: " + host + ", groupId: " + groupID + ", clientId: " + clientID) log.Printf("Create consumer successfully, host: %s, groupId: %s, clientId: %s\n", host, groupID, clientID)
// ANCHOR_END: create_consumer // ANCHOR_END: create_consumer
// ANCHOR: subscribe // ANCHOR: subscribe
err = consumer.Subscribe("topic_meters", nil) topic = "topic_meters"
err = consumer.Subscribe(topic, nil)
if err != nil { if err != nil {
log.Fatalln("Failed to subscribe topic_meters, ErrMessage: " + err.Error()) log.Fatalf(
"Failed to subscribe topic_meters, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
topic,
groupID,
clientID,
err.Error(),
)
} }
log.Println("Subscribe topics successfully") log.Println("Subscribe topics successfully")
for i := 0; i < 50; i++ { for i := 0; i < 50; i++ {
@ -62,13 +79,19 @@ func main() {
// commit offset // commit offset
_, err = consumer.CommitOffsets([]tmqcommon.TopicPartition{e.TopicPartition}) _, err = consumer.CommitOffsets([]tmqcommon.TopicPartition{e.TopicPartition})
if err != nil { if err != nil {
log.Fatalln("Failed to commit offset, ErrMessage: " + err.Error()) log.Fatalf(
"Failed to commit offset, topic: %s, groupId: %s, clientId: %s, offset %s, ErrMessage: %s\n",
topic,
groupID,
clientID,
e.TopicPartition,
err.Error(),
)
} }
log.Println("Commit offset manually successfully.") log.Println("Commit offset manually successfully.")
// ANCHOR_END: commit_offset // ANCHOR_END: commit_offset
case tmqcommon.Error: case tmqcommon.Error:
fmt.Printf("%% Error: %v: %v\n", e.Code(), e) log.Fatalf("Failed to poll data, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n", topic, groupID, clientID, e.Error())
log.Fatalln("Failed to poll data, ErrMessage: " + err.Error())
} }
} }
} }
@ -77,7 +100,7 @@ func main() {
// get assignment // get assignment
partitions, err := consumer.Assignment() partitions, err := consumer.Assignment()
if err != nil { if err != nil {
log.Fatal("Failed to get assignment, ErrMessage: " + err.Error()) log.Fatalf("Failed to get assignment, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n", topic, groupID, clientID, err.Error())
} }
fmt.Println("Now assignment:", partitions) fmt.Println("Now assignment:", partitions)
for i := 0; i < len(partitions); i++ { for i := 0; i < len(partitions); i++ {
@ -88,7 +111,15 @@ func main() {
Offset: 0, Offset: 0,
}, 0) }, 0)
if err != nil { if err != nil {
log.Fatalln("Failed to execute seek example, ErrMessage: " + err.Error()) log.Fatalf(
"Failed to execute seek example, topic: %s, groupId: %s, clientId: %s, partition: %d, offset: %d, ErrMessage: %s\n",
topic,
groupID,
clientID,
partitions[i].Partition,
0,
err.Error(),
)
} }
} }
fmt.Println("Assignment seek to beginning successfully") fmt.Println("Assignment seek to beginning successfully")
@ -97,13 +128,25 @@ func main() {
// unsubscribe // unsubscribe
err = consumer.Unsubscribe() err = consumer.Unsubscribe()
if err != nil { if err != nil {
log.Fatal("Failed to unsubscribe consumer, ErrMessage: " + err.Error()) log.Fatalf(
"Failed to unsubscribe consumer, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
topic,
groupID,
clientID,
err.Error(),
)
} }
fmt.Println("Consumer unsubscribed successfully.") fmt.Println("Consumer unsubscribed successfully.")
// close consumer // close consumer
err = consumer.Close() err = consumer.Close()
if err != nil { if err != nil {
log.Fatal("Failed to close consumer, ErrMessage: " + err.Error()) log.Fatalf(
"Failed to close consumer, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
topic,
groupID,
clientID,
err.Error(),
)
} }
fmt.Println("Consumer closed successfully.") fmt.Println("Consumer closed successfully.")
// ANCHOR_END: close // ANCHOR_END: close

84
docs/examples/go/tmq/ws/main.go
View File

@ -13,13 +13,17 @@ import (
) )
var done = make(chan struct{}) var done = make(chan struct{})
var groupID string
var clientID string
var host string
var topic string
func main() { func main() {
// init env // init env
taosDSN := "root:taosdata@ws(127.0.0.1:6041)/" taosDSN := "root:taosdata@ws(127.0.0.1:6041)/"
conn, err := sql.Open("taosWS", taosDSN) conn, err := sql.Open("taosWS", taosDSN)
if err != nil { if err != nil {
log.Fatalln("Failed to connect to " + taosDSN + "; ErrMessage: " + err.Error()) log.Fatalln("Failed to connect to " + taosDSN + ", ErrMessage: " + err.Error())
} }
defer func() { defer func() {
conn.Close() conn.Close()
@ -28,8 +32,9 @@ func main() {
// ANCHOR: create_consumer // ANCHOR: create_consumer
// create consumer // create consumer
wsUrl := "ws://127.0.0.1:6041" wsUrl := "ws://127.0.0.1:6041"
groupID := "group1" groupID = "group1"
clientID := "client1" clientID = "client1"
host = "127.0.0.1"
consumer, err := tmq.NewConsumer(&tmqcommon.ConfigMap{ consumer, err := tmq.NewConsumer(&tmqcommon.ConfigMap{
"ws.url": wsUrl, "ws.url": wsUrl,
"ws.message.channelLen": uint(0), "ws.message.channelLen": uint(0),
@ -45,15 +50,28 @@ func main() {
"client.id": clientID, "client.id": clientID,
}) })
if err != nil { if err != nil {
log.Fatalln("Failed to create websocket consumer, host : " + wsUrl + "; ErrMessage: " + err.Error()) log.Fatalf(
"Failed to create websocket consumer, host: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
host,
groupID,
clientID,
err.Error(),
)
} }
log.Println("Create consumer successfully, host: " + wsUrl + ", groupId: " + groupID + ", clientId: " + clientID) log.Printf("Create consumer successfully, host: %s, groupId: %s, clientId: %s\n", host, groupID, clientID)
// ANCHOR_END: create_consumer // ANCHOR_END: create_consumer
// ANCHOR: subscribe // ANCHOR: subscribe
err = consumer.Subscribe("topic_meters", nil) topic = "topic_meters"
err = consumer.Subscribe(topic, nil)
if err != nil { if err != nil {
log.Fatalln("Failed to subscribe topic_meters, ErrMessage: " + err.Error()) log.Fatalf(
"Failed to subscribe topic_meters, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
topic,
groupID,
clientID,
err.Error(),
)
} }
log.Println("Subscribe topics successfully") log.Println("Subscribe topics successfully")
for i := 0; i < 50; i++ { for i := 0; i < 50; i++ {
@ -67,13 +85,25 @@ func main() {
// commit offset // commit offset
_, err = consumer.CommitOffsets([]tmqcommon.TopicPartition{e.TopicPartition}) _, err = consumer.CommitOffsets([]tmqcommon.TopicPartition{e.TopicPartition})
if err != nil { if err != nil {
log.Fatalln("Failed to commit offset, ErrMessage: " + err.Error()) log.Fatalf(
"Failed to commit offset, topic: %s, groupId: %s, clientId: %s, offset %s, ErrMessage: %s\n",
topic,
groupID,
clientID,
e.TopicPartition,
err.Error(),
)
} }
log.Println("Commit offset manually successfully.") log.Println("Commit offset manually successfully.")
// ANCHOR_END: commit_offset // ANCHOR_END: commit_offset
case tmqcommon.Error: case tmqcommon.Error:
fmt.Printf("%% Error: %v: %v\n", e.Code(), e) log.Fatalf(
log.Fatalln("Failed to poll data, ErrMessage: " + err.Error()) "Failed to poll data, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
topic,
groupID,
clientID,
e.Error(),
)
} }
} }
} }
@ -82,7 +112,13 @@ func main() {
// get assignment // get assignment
partitions, err := consumer.Assignment() partitions, err := consumer.Assignment()
if err != nil { if err != nil {
log.Fatal("Failed to get assignment, ErrMessage: " + err.Error()) log.Fatalf(
"Failed to get assignment, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
topic,
groupID,
clientID,
err.Error(),
)
} }
fmt.Println("Now assignment:", partitions) fmt.Println("Now assignment:", partitions)
for i := 0; i < len(partitions); i++ { for i := 0; i < len(partitions); i++ {
@ -93,7 +129,15 @@ func main() {
Offset: 0, Offset: 0,
}, 0) }, 0)
if err != nil { if err != nil {
log.Fatalln("Failed to execute seek example, ErrMessage: " + err.Error()) log.Fatalf(
"Failed to execute seek example, topic: %s, groupId: %s, clientId: %s, partition: %d, offset: %d, ErrMessage: %s\n",
topic,
groupID,
clientID,
partitions[i].Partition,
0,
err.Error(),
)
} }
} }
fmt.Println("Assignment seek to beginning successfully") fmt.Println("Assignment seek to beginning successfully")
@ -102,13 +146,25 @@ func main() {
// unsubscribe // unsubscribe
err = consumer.Unsubscribe() err = consumer.Unsubscribe()
if err != nil { if err != nil {
log.Fatal("Failed to unsubscribe consumer, ErrMessage: " + err.Error()) log.Fatalf(
"Failed to unsubscribe consumer, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
topic,
groupID,
clientID,
err.Error(),
)
} }
fmt.Println("Consumer unsubscribed successfully.") fmt.Println("Consumer unsubscribed successfully.")
// close consumer // close consumer
err = consumer.Close() err = consumer.Close()
if err != nil { if err != nil {
log.Fatal("Failed to close consumer, ErrMessage: " + err.Error()) log.Fatalf(
"Failed to close consumer, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
topic,
groupID,
clientID,
err.Error(),
)
} }
fmt.Println("Consumer closed successfully.") fmt.Println("Consumer closed successfully.")
// ANCHOR_END: close // ANCHOR_END: close