InfluxDB.Flurl 1.0.8

influxdbflurl

介绍

c# influxdb2.0 SDK

使用说明

using InfluxDB.Flurl;

namespace Test
{
    public class People
    {
        public DateTimeOffset? Time { get; set; }

        [InfluxTag]
        public string app { get; set; }

        public int id { get; set; }

        public string name { get; set; }

        public decimal money { get; set; }

        public string remark { get; set; }
    }
}

using InfluxDB.Flurl;
using System.Diagnostics;
using System.Text.Json;

namespace Test
{
    internal class Program
    {
        static async Task Main(string[] args)
        {
            var url = "http://localhost:8086";
            var token = "yx02F7HEede3MRVn0XknwHphSVmDQ8sc47JXh3bVlToaTFv-Z5oWpb9RLQEQVCjL5hYCgGnyNMGfS4m6Fdm5Ig==";

            //org1
            //token = "QCXwqSVWqvYa4Rh4PiWrfcT9SN7GIqgVk9Wp8TBQF-BpVv5ejmH2FTrgfWIV2uWAmfZ-vwtf83D8LF4_ShmitA==";

            //client must be singleton mode(必须是单例模式)
            var client = new InfluxDbClient(url, token);

            var org = "test";
            var bucketName = "bucket-1";
            var measurement = "measurement-1";

            //await client.DeleteBucketAsync(bucketName, org);
            if (!await client.ExistsBucketAsync(bucketName, org))
            {
                await client.CreateBucketAsync(bucketName, org, 1);
            }

            //write stream by InfluxDbWriter
            await client.WriteAsync(async stream =>
            {
                var writer = new InfluxDbWriter(stream);

                var line = """
                measurement-1,app=qms id=1i,name="abcd",money=10.8
                """;
                await writer.WriteLineAsync(line);

                Thread.Sleep(1);
                await writer.WriteNAsync(); //row by row must write \n

                line = """
                measurement-1,app=qms id=2i,name="abcd",money=10.8
                """;
                await writer.WriteLineAsync(line);

                //await writer.WriteAsync  TEntity
                //await writer.WritePointDataAsync PointData

            }, bucketName, org);

            //writeLines
            List<string> lineList = [];
            for (int i = 0; i < 10; i++)
            {
                var line = $"""
                measurement-1,app=qms id={i}i,name="abcd",money=10.8
                """;
                lineList.Add(line);
                Thread.Sleep(1);
            }
            await client.WriteLinesAsync(lineList, bucketName, org);

            //Write PointData
            List<PointData> pointList = [];
            for (int i = 0; i < 10; i++)
            {
                var point = new PointData(measurement, DateTimeOffset.Now);

                point.AddTag("app", "qms");

                point.AddField("id", i + 10);
                point.AddField("name", $"asds王五{i}\"");
                point.AddField("money", Convert.ToDecimal(Random.Shared.Next(1, 2000) + 0.1));
                point.AddField("remark", JsonSerializer.Serialize(new { a = 1, b = 2, c = "李四" }));
                pointList.Add(point);
                Thread.Sleep(1);
            }
            await client.WritePointDataAsync(pointList, bucketName, org);


            //Write POCO
            List<People> peopleList = [];
            for (int i = 0; i < 10; i++)
            {
                var people = new People
                {
                    app = "qms",
                    id = i,
                    name = "李四" + i,
                    money = 20.22M,
                    remark = JsonSerializer.Serialize(new { a = 1, b = 2, c = "你好" + i }),
                    Time = DateTimeOffset.Now
                };
                peopleList.Add(people);
                Thread.Sleep(1);
            }
            var start = Stopwatch.GetTimestamp();
            await client.WriteAsync(peopleList, measurement, bucketName, org);
            Console.WriteLine(Stopwatch.GetElapsedTime(start).TotalMilliseconds);

            //flux query
            var flux = $"""
                from(bucket:"{bucketName}")
                  |> range(start: -1h, stop:2h)
                  |> filter(fn: (r) => r._measurement == "{measurement}")
                  //|> filter(fn: (r) => r._field == "id" or r._field == "money")
                  //|> group()
                  //|> mean()
                """;

            //query IAsyncEnumerable record
            var unBuffer = client.QueryUnbufferAsync(flux, org);
            await foreach (var item in unBuffer)
            {
                Console.WriteLine(item.GetValue());
            }

            //query list record
            var recordList = await client.QueryAsync(flux, org);
            var opt = new JsonSerializerOptions()
            {
                Encoder = System.Text.Encodings.Web.JavaScriptEncoder.UnsafeRelaxedJsonEscaping,
                WriteIndented = true
            };
            Console.WriteLine(JsonSerializer.Serialize(recordList, opt));

            //query POCO
            var listModel = await client.QueryAsync<People>(flux, org);
            foreach (var item in listModel)
            {
                Console.WriteLine($"{item.id}==>{item.Time.Value:yyyy-MM-dd HH:mm:ss.fff}");
            }

            //Queue
            var queue = new InfluxDbQueue(client);
            queue.OnError += errList =>
            {
                Console.WriteLine(errList.Count);
            };
            _ = Task.Run(async () =>
            {
                while (true)
                {
                    var point = new PointData(measurement, DateTimeOffset.Now);
                    point.AddTag("app", "qms");
                    point.AddField("id", DateTime.Now.Second);
                    point.AddField("name", $"asds王五\"");
                    point.AddField("money", Convert.ToDecimal(Random.Shared.Next(1, 2000) + 0.1));
                    point.AddField("remark", JsonSerializer.Serialize(new { a = 1, b = 2, c = "李四" }));
                    queue.Add(org, bucketName, point);


                    var people = new People
                    {
                        app = "qms",
                        id = 1,
                        name = "李四",
                        money = 20.22M,
                        remark = JsonSerializer.Serialize(new { a = 1, b = 2, c = "你好" }),
                        Time = DateTimeOffset.Now
                    };
                    queue.Add(org, bucketName, people, measurement);


                    await Task.Delay(10);
                }
            });

            Console.WriteLine("Hello, World!");
            Console.ReadKey();
            await queue.StopAsync();

        }
    }
}


//https://flurl.dev/docs/fluent-http/#responses-and-error-handling
//try
//{
//    var result = await url.PostJsonAsync(requestObj).ReceiveJson<T>();
//}
//catch (FlurlHttpException ex)
//{
//    var err = await ex.GetResponseJsonAsync<TError>(); // or GetResponseStringAsync(), etc.
//    logger.Write($"Error returned from {ex.Call.Request.Url}: {err.SomeDetails}");
//    var resContent = await ex.GetResponseStringAsync();
//}