using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Text; using System.Threading; using System.Threading.Tasks; using InfluxDB.Client.Api.Domain; using InfluxDB.Client.Api.Service; using InfluxDB.Client.Core; using InfluxDB.Client.Core.Internal; using InfluxDB.Client.Writes; using RestSharp; namespace InfluxDB.Client { public interface IWriteApiAsync { /// /// Write Line Protocol record into specified bucket. /// /// specifies the record in InfluxDB Line Protocol. The is considered as one batch unit. /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests Task WriteRecordAsync(string record, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default); /// /// Write Line Protocol records into specified bucket. /// /// specifies the record in InfluxDB Line Protocol /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests Task WriteRecordsAsync(List records, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default); /// /// Write Line Protocol records into specified bucket. /// /// specifies the record in InfluxDB Line Protocol /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests Task WriteRecordsAsync(string[] records, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default); /// /// Write Line Protocols records into specified bucket. /// /// specifies the record in InfluxDB Line Protocol /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests /// Write Task with IRestResponse Task WriteRecordsAsyncWithIRestResponse(IEnumerable records, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default); /// /// Write a Data point into specified bucket. /// /// specifies the Data point to write into bucket /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests Task WritePointAsync(PointData point, string bucket = null, string org = null, CancellationToken cancellationToken = default); /// /// Write Data points into specified bucket. /// /// specifies the Data points to write into bucket /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests Task WritePointsAsync(List points, string bucket = null, string org = null, CancellationToken cancellationToken = default); /// /// Write Data points into specified bucket. /// /// specifies the Data points to write into bucket /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests Task WritePointsAsync(PointData[] points, string bucket = null, string org = null, CancellationToken cancellationToken = default); /// /// Write Data points into specified bucket. /// /// specifies the Data points to write into bucket /// specifies the destination bucket for writes. /// If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests /// Write Tasks with IRestResponses. Task WritePointsAsyncWithIRestResponse(IEnumerable points, string bucket = null, string org = null, CancellationToken cancellationToken = default); /// /// Write a Measurement into specified bucket. /// /// specifies the Measurement to write into bucket /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests /// measurement type Task WriteMeasurementAsync(TM measurement, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default); /// /// Write Measurements into specified bucket. /// /// specifies Measurements to write into bucket /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests /// measurement type Task WriteMeasurementsAsync(List measurements, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default); /// /// Write Measurements into specified bucket. /// /// specifies Measurements to write into bucket /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests /// measurement type Task WriteMeasurementsAsync(TM[] measurements, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default); /// /// Write Measurements into specified bucket. /// /// specifies Measurements to write into bucket /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests /// measurement type /// Write Task with IRestResponse Task WriteMeasurementsAsyncWithIRestResponse(IEnumerable measurements, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default); } public class WriteApiAsync : IWriteApiAsync { private readonly InfluxDBClient _influxDbClient; private readonly InfluxDBClientOptions _options; private readonly WriteService _service; private readonly IDomainObjectMapper _mapper; private const string PostHeaderAccept = "application/json"; private const string PostHeaderEncoding = "identity"; private const string PostHeaderContentType = "text/plain; charset=utf-8"; protected internal WriteApiAsync(InfluxDBClientOptions options, WriteService service, IDomainObjectMapper mapper, InfluxDBClient influxDbClient) { Arguments.CheckNotNull(service, nameof(service)); Arguments.CheckNotNull(mapper, nameof(mapper)); Arguments.CheckNotNull(influxDbClient, nameof(_influxDbClient)); _options = options; _mapper = mapper; _influxDbClient = influxDbClient; _service = service; } /// /// Write Line Protocol record into specified bucket. /// /// specifies the record in InfluxDB Line Protocol. The is considered as one batch unit. /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests public Task WriteRecordAsync(string record, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default) { return WriteRecordsAsync(new List { record }, precision, bucket, org, cancellationToken); } /// /// Write Line Protocol records into specified bucket. /// /// specifies the record in InfluxDB Line Protocol /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests public Task WriteRecordsAsync(List records, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default) { var options = new BatchWriteOptions(bucket ?? _options.Bucket, org ?? _options.Org, precision); var list = records.Select(record => new BatchWriteRecord(options, record)).ToList(); return WriteData(options.OrganizationId, options.Bucket, precision, list, cancellationToken); } /// /// Write Line Protocol records into specified bucket. /// /// specifies the record in InfluxDB Line Protocol /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests public Task WriteRecordsAsync(string[] records, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default) { return WriteRecordsAsync(records.ToList(), precision, bucket, org, cancellationToken); } /// /// Write Line Protocols records into specified bucket. /// /// specifies the record in InfluxDB Line Protocol /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests /// Write Task with IRestResponse public Task WriteRecordsAsyncWithIRestResponse(IEnumerable records, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default) { var options = new BatchWriteOptions(bucket ?? _options.Bucket, org ?? _options.Org, precision); var batch = records .Select(it => new BatchWriteRecord(options, it)); return WriteDataAsyncWithIRestResponse(batch, options.Bucket, options.OrganizationId, precision, cancellationToken); } /// /// Write a Data point into specified bucket. /// /// specifies the Data point to write into bucket /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests public Task WritePointAsync(PointData point, string bucket = null, string org = null, CancellationToken cancellationToken = default) { if (point == null) { return Task.CompletedTask; } return WritePointsAsync(new List { point }, bucket, org, cancellationToken); } /// /// Write Data points into specified bucket. /// /// specifies the Data points to write into bucket /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests public async Task WritePointsAsync(List points, string bucket = null, string org = null, CancellationToken cancellationToken = default) { foreach (var grouped in points.GroupBy(it => it.Precision)) { var options = new BatchWriteOptions(bucket ?? _options.Bucket, org ?? _options.Org, grouped.Key); var groupedPoints = grouped .Select(it => new BatchWritePoint(options, _options, it)) .ToList(); await WriteData(options.OrganizationId, options.Bucket, grouped.Key, groupedPoints, cancellationToken) .ConfigureAwait(false); } } /// /// Write Data points into specified bucket. /// /// specifies the Data points to write into bucket /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests public Task WritePointsAsync(PointData[] points, string bucket = null, string org = null, CancellationToken cancellationToken = default) { return WritePointsAsync(points.ToList(), bucket, org, cancellationToken); } /// /// Write Data points into specified bucket. /// /// specifies the Data points to write into bucket /// specifies the destination bucket for writes. /// If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests /// Write Tasks with IRestResponses. public Task WritePointsAsyncWithIRestResponse(IEnumerable points, string bucket = null, string org = null, CancellationToken cancellationToken = default) { var tasks = new List>(); foreach (var grouped in points.GroupBy(it => it.Precision)) { var options = new BatchWriteOptions(bucket ?? _options.Bucket, org ?? _options.Org, grouped.Key); var groupedPoints = grouped .Select(it => new BatchWritePoint(options, _options, it)) .ToList(); tasks.Add(WriteDataAsyncWithIRestResponse(groupedPoints, options.Bucket, options.OrganizationId, grouped.Key, cancellationToken)); } return Task.WhenAll(tasks); } /// /// Write a Measurement into specified bucket. /// /// specifies the Measurement to write into bucket /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests /// measurement type public Task WriteMeasurementAsync(TM measurement, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default) { if (measurement == null) { return Task.CompletedTask; } return WriteMeasurementsAsync(new List { measurement }, precision, bucket, org, cancellationToken); } /// /// Write Measurements into specified bucket. /// /// specifies Measurements to write into bucket /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests /// measurement type public Task WriteMeasurementsAsync(List measurements, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default) { var list = new List(); var options = new BatchWriteOptions(bucket ?? _options.Bucket, org ?? _options.Org, precision); foreach (var measurement in measurements) { BatchWriteData data = new BatchWriteMeasurement(options, _options, measurement, _mapper); list.Add(data); } return WriteData(options.OrganizationId, options.Bucket, precision, list, cancellationToken); } /// /// Write Measurements into specified bucket. /// /// specifies Measurements to write into bucket /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests /// measurement type public Task WriteMeasurementsAsync(TM[] measurements, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default) { return WriteMeasurementsAsync(measurements.ToList(), precision, bucket, org, cancellationToken); } /// /// Write Measurements into specified bucket. /// /// specifies Measurements to write into bucket /// specifies the precision for the unix timestamps within the body line-protocol; default Nanoseconds /// specifies the destination bucket for writes. If the bucket is not specified then is used config from . /// specifies the destination organization for writes. If the org is not specified then is used config from . /// specifies the token to monitor for cancellation requests /// measurement type /// Write Task with IRestResponse public Task WriteMeasurementsAsyncWithIRestResponse(IEnumerable measurements, WritePrecision precision = WritePrecision.Ns, string bucket = null, string org = null, CancellationToken cancellationToken = default) { var options = new BatchWriteOptions(bucket ?? _options.Bucket, org ?? _options.Org, precision); var batch = measurements .Select(it => new BatchWriteMeasurement(options, _options, it, _mapper)); return WriteDataAsyncWithIRestResponse(batch, bucket, org, precision, cancellationToken); } private Task WriteData(string org, string bucket, WritePrecision precision, IEnumerable data, CancellationToken cancellationToken) { var sb = ToLineProtocolBody(data); if (sb.Length == 0) { Trace.WriteLine($"The writes: {data} doesn't contains any Line Protocol, skipping"); return Task.CompletedTask; } return _service.PostWriteAsync(org, bucket, Encoding.UTF8.GetBytes(sb.ToString()), null, PostHeaderEncoding, PostHeaderContentType, null, PostHeaderAccept, null, precision, cancellationToken); } private Task WriteDataAsyncWithIRestResponse(IEnumerable batch, string bucket = null, string org = null, WritePrecision precision = WritePrecision.Ns, CancellationToken cancellationToken = default) { var localBucket = bucket ?? _options.Bucket; var localOrg = org ?? _options.Org; Arguments.CheckNonEmptyString(localBucket, AbstractRestClient.BucketArgumentValidation); Arguments.CheckNonEmptyString(localOrg, AbstractRestClient.OrgArgumentValidation); Arguments.CheckNotNull(precision, nameof(precision)); var sb = ToLineProtocolBody(batch); return _service.PostWriteAsyncWithIRestResponse(org, bucket, Encoding.UTF8.GetBytes(sb.ToString()), null, PostHeaderEncoding, PostHeaderContentType, null, PostHeaderAccept, null, precision, cancellationToken); } private static StringBuilder ToLineProtocolBody(IEnumerable data) { var sb = new StringBuilder(""); foreach (var item in data) { var lineProtocol = item.ToLineProtocol(); if (string.IsNullOrEmpty(lineProtocol)) { continue; } sb.Append(lineProtocol); sb.Append("\n"); } if (sb.Length != 0) { // remove last \n sb.Remove(sb.Length - 1, 1); } return sb; } } }