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;
}
}
}