using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Reactive;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Reactive.Threading.Tasks;
using System.Text;
using System.Threading;
using InfluxDB.Client.Api.Domain;
using InfluxDB.Client.Api.Service;
using InfluxDB.Client.Core;
using InfluxDB.Client.Core.Exceptions;
using InfluxDB.Client.Core.Internal;
using InfluxDB.Client.Internal;
using InfluxDB.Client.Writes;
using Microsoft.Extensions.ObjectPool;
using RestSharp;
namespace InfluxDB.Client
{
public interface IWriteApi : IDisposable
{
///
/// 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 .
void WriteRecord(string record, WritePrecision precision = WritePrecision.Ns, string bucket = null,
string org = null);
///
/// 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 .
void WriteRecords(List records, WritePrecision precision = WritePrecision.Ns,
string bucket = null, string org = null);
///
/// 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 .
void WriteRecords(string[] records, WritePrecision precision = WritePrecision.Ns, string bucket = null,
string org = null);
///
/// 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 .
void WritePoint(PointData point, string bucket = null, string org = null);
///
/// 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 .
void WritePoints(List points, string bucket = null, string org = null);
///
/// 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 .
void WritePoints(PointData[] points, string bucket = null, string org = null);
///
/// 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 .
/// measurement type
void WriteMeasurement(TM measurement, WritePrecision precision = WritePrecision.Ns,
string bucket = null, string org = null);
///
/// 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 .
/// measurement type
void WriteMeasurements(List measurements, WritePrecision precision = WritePrecision.Ns,
string bucket = null, string org = null);
///
/// 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 .
/// measurement type
void WriteMeasurements(TM[] measurements, WritePrecision precision = WritePrecision.Ns,
string bucket = null, string org = null);
///
/// Forces the client to flush all pending writes from the buffer to the InfluxDB via HTTP.
///
void Flush();
}
public class WriteApi : IWriteApi
{
private readonly Subject> _flush = new Subject>();
private readonly InfluxDBClient _influxDbClient;
private readonly IDomainObjectMapper _mapper;
private readonly InfluxDBClientOptions _options;
private readonly Subject _subject = new Subject();
private static readonly ObjectPoolProvider ObjectPoolProvider = new DefaultObjectPoolProvider();
private static readonly ObjectPool StringBuilderPool =
ObjectPoolProvider.CreateStringBuilderPool();
private readonly IDisposable _unsubscribeDisposeCommand;
private bool _disposed;
protected internal WriteApi(
InfluxDBClientOptions options,
WriteService service,
WriteOptions writeOptions,
IDomainObjectMapper mapper,
InfluxDBClient influxDbClient,
IObservable disposeCommand)
{
Arguments.CheckNotNull(service, nameof(service));
Arguments.CheckNotNull(writeOptions, nameof(writeOptions));
Arguments.CheckNotNull(mapper, nameof(mapper));
Arguments.CheckNotNull(influxDbClient, nameof(_influxDbClient));
Arguments.CheckNotNull(disposeCommand, nameof(disposeCommand));
_options = options;
_mapper = mapper;
_influxDbClient = influxDbClient;
_unsubscribeDisposeCommand = disposeCommand.Subscribe(_ => Dispose());
// backpreasure - is not implemented in C#
//
// => use unbound buffer
//
// https://github.com/dotnet/reactive/issues/19
var batches = _subject
//
// Batching
//
.Publish(connectedSource =>
{
var trigger = Observable.Merge(
// triggered by time & count
connectedSource.Window(TimeSpan.FromMilliseconds(
writeOptions.FlushInterval),
writeOptions.BatchSize,
writeOptions.WriteScheduler),
// flush trigger
_flush
);
return connectedSource
.Window(trigger);
})
//
// Group by key - same bucket, same org
//
.SelectMany(it => it.GroupBy(batchWrite => batchWrite.Options))
//
// Create Write Point = bucket, org, ... + data
//
.SelectMany(grouped =>
{
var aggregate = grouped
.Aggregate(StringBuilderPool.Get(), (builder, batchWrite) =>
{
var data = batchWrite.ToLineProtocol();
if (string.IsNullOrEmpty(data))
{
return builder;
}
if (builder.Length > 0)
{
builder.Append("\n");
}
return builder.Append(data);
}).Select(builder =>
{
var result = builder.ToString();
builder.Clear();
StringBuilderPool.Return(builder);
return result;
});
return aggregate.Select(records => new BatchWriteRecord(grouped.Key, records))
.Where(batchWriteItem => !string.IsNullOrEmpty(batchWriteItem.ToLineProtocol()));
});
var unused = batches
//
// Map to Async request
//
.Select(batchWriteItem =>
{
var org = batchWriteItem.Options.OrganizationId;
var bucket = batchWriteItem.Options.Bucket;
var lineProtocol = batchWriteItem.ToLineProtocol();
var precision = batchWriteItem.Options.Precision;
return Observable
.Defer(() =>
{
var observable = service.PostWriteAsyncWithIRestResponse(org, bucket,
Encoding.UTF8.GetBytes(lineProtocol), null,
"identity", "text/plain; charset=utf-8", null, "application/json", null, precision)
.ToObservable();
if (writeOptions.JitterInterval > 0)
{
observable = observable
.Delay(_ => Observable.Timer(
TimeSpan.FromMilliseconds(RetryAttempt.JitterDelay(writeOptions)),
writeOptions.WriteScheduler));
}
return observable;
})
.RetryWhen(f => f
.Zip(Observable.Range(1, writeOptions.MaxRetries + 1), (exception, count)
=> new RetryAttempt(exception, count, writeOptions))
.SelectMany(attempt =>
{
if (attempt.IsRetry())
{
var retryInterval = attempt.GetRetryInterval();
var retryable = new WriteRetriableErrorEvent(org, bucket, precision, lineProtocol,
attempt.Error, retryInterval);
Publish(retryable);
return Observable.Timer(TimeSpan.FromMilliseconds(retryInterval),
writeOptions.WriteScheduler);
}
throw attempt.Error;
}))
.Select(result =>
{
// ReSharper disable once ConvertIfStatementToReturnStatement
if (result.IsSuccessful)
{
return Notification.CreateOnNext(result);
}
return Notification.CreateOnError(
HttpException.Create(result, result.Content));
})
.Catch, Exception>(ex =>
{
var error = new WriteErrorEvent(org, bucket, precision, lineProtocol, ex);
Publish(error);
return Observable.Return(Notification.CreateOnError(ex));
}).Do(res =>
{
if (res.Kind == NotificationKind.OnNext)
{
var success = new WriteSuccessEvent(org, bucket, precision, lineProtocol);
Publish(success);
}
});
})
.Concat()
.Subscribe(
notification =>
{
switch (notification.Kind)
{
case NotificationKind.OnNext:
Trace.WriteLine($"The batch item: {notification} was processed successfully.");
break;
case NotificationKind.OnError:
Trace.WriteLine(
$"The batch item wasn't processed successfully because: {notification.Exception}");
break;
default:
Trace.WriteLine($"The batch item: {notification} was processed");
break;
}
},
exception =>
{
Publish(new WriteRuntimeExceptionEvent(exception));
_disposed = true;
Trace.WriteLine($"The unhandled exception occurs: {exception}");
},
() =>
{
_disposed = true;
Trace.WriteLine("The WriteApi was disposed.");
});
}
public void Dispose()
{
ReleaseAndClose();
}
///
/// Release all resources and flush remaining data into database.
///
/// How much milliseconds wait to flush data.
internal void ReleaseAndClose(int millis = 30000)
{
_unsubscribeDisposeCommand.Dispose(); // avoid duplicate call to dispose
Trace.WriteLine("Flushing batches before shutdown.");
if (!_subject.IsDisposed)
{
_subject.OnCompleted();
}
if (!_flush.IsDisposed)
{
_flush.OnCompleted();
}
_subject.Dispose();
_flush.Dispose();
WaitToCondition(() => _disposed, millis);
}
public bool Disposed => _disposed;
public event EventHandler EventHandler;
///
/// 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 .
public void WriteRecord(string record, WritePrecision precision = WritePrecision.Ns, string bucket = null,
string org = null)
{
var options = new BatchWriteOptions(bucket ?? _options.Bucket, org ?? _options.Org, precision);
_subject.OnNext(new BatchWriteRecord(options, record));
}
///
/// 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 .
public void WriteRecords(List records, WritePrecision precision = WritePrecision.Ns,
string bucket = null, string org = null)
{
records.ForEach(record => WriteRecord(record, precision, bucket, org));
}
///
/// 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 .
public void WriteRecords(string[] records, WritePrecision precision = WritePrecision.Ns, string bucket = null,
string org = null)
{
foreach (var record in records) WriteRecord(record, precision, bucket, org);
}
///
/// 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 .
public void WritePoint(PointData point, string bucket = null, string org = null)
{
if (point == null)
{
return;
}
var options = new BatchWriteOptions(bucket ?? _options.Bucket, org ?? _options.Org, point.Precision);
_subject.OnNext(new BatchWritePoint(options, _options, point));
}
///
/// 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 .
public void WritePoints(List points, string bucket = null, string org = null)
{
foreach (var point in points) WritePoint(point, bucket, org);
}
///
/// 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 .
public void WritePoints(PointData[] points, string bucket = null, string org = null)
{
WritePoints(points.ToList(), bucket, org);
}
///
/// 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 .
/// measurement type
public void WriteMeasurement(TM measurement, WritePrecision precision = WritePrecision.Ns,
string bucket = null, string org = null)
{
if (measurement == null)
{
return;
}
var options = new BatchWriteOptions(bucket ?? _options.Bucket, org ?? _options.Org, precision);
_subject.OnNext(new BatchWriteMeasurement(options, _options, measurement, _mapper));
}
///
/// 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 .
/// measurement type
public void WriteMeasurements(List measurements, WritePrecision precision = WritePrecision.Ns,
string bucket = null, string org = null)
{
foreach (var measurement in measurements) WriteMeasurement(measurement, precision, bucket, org);
}
///
/// 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 .
/// measurement type
public void WriteMeasurements(TM[] measurements, WritePrecision precision = WritePrecision.Ns,
string bucket = null, string org = null)
{
WriteMeasurements(measurements.ToList(), precision, bucket, org);
}
///
/// Forces the client to flush all pending writes from the buffer to the InfluxDB via HTTP.
///
public void Flush()
{
if (!_flush.IsDisposed)
{
_flush.OnNext(Observable.Empty());
}
}
internal static void WaitToCondition(Func condition, int millis)
{
var start = DateTimeOffset.Now.ToUnixTimeMilliseconds();
while (!condition())
{
Thread.Sleep(25);
if (DateTimeOffset.Now.ToUnixTimeMilliseconds() - start > millis)
{
Trace.TraceError($"The WriteApi can't be gracefully dispose! - {millis}ms elapsed.");
break;
}
}
}
private void Publish(InfluxDBEventArgs eventArgs)
{
eventArgs.LogEvent();
EventHandler?.Invoke(this, eventArgs);
}
}
internal abstract class BatchWriteData
{
internal readonly BatchWriteOptions Options;
protected BatchWriteData(BatchWriteOptions options)
{
Arguments.CheckNotNull(options, "options");
Options = options;
}
internal abstract string ToLineProtocol();
}
internal class BatchWriteRecord : BatchWriteData
{
private readonly string _record;
internal BatchWriteRecord(BatchWriteOptions options, string record) : base(options)
{
Arguments.CheckNotNull(record, nameof(record));
_record = record;
}
internal override string ToLineProtocol()
{
return _record;
}
}
internal class BatchWritePoint : BatchWriteData
{
private readonly PointData _point;
private readonly InfluxDBClientOptions _clientOptions;
internal BatchWritePoint(BatchWriteOptions options, InfluxDBClientOptions clientOptions, PointData point) :
base(options)
{
Arguments.CheckNotNull(point, nameof(point));
_point = point;
_clientOptions = clientOptions;
}
internal override string ToLineProtocol()
{
if (!_point.HasFields())
{
Trace.WriteLine($"The point: ${_point} doesn't contains any fields, skipping");
return null;
}
return _point.ToLineProtocol(_clientOptions.PointSettings);
}
}
internal class BatchWriteMeasurement : BatchWriteData
{
private readonly TM _measurement;
private readonly IDomainObjectMapper _converter;
private readonly InfluxDBClientOptions _clientOptions;
internal BatchWriteMeasurement(BatchWriteOptions options, InfluxDBClientOptions clientOptions, TM measurement,
IDomainObjectMapper converter) :
base(options)
{
Arguments.CheckNotNull(measurement, nameof(measurement));
_clientOptions = clientOptions;
_measurement = measurement;
_converter = converter;
}
internal override string ToLineProtocol()
{
var point = _converter.ConvertToPointData(_measurement, Options.Precision);
if (!point.HasFields())
{
Trace.WriteLine($"The point: ${point} doesn't contains any fields, skipping");
return null;
}
return point.ToLineProtocol(_clientOptions.PointSettings);
}
}
internal class BatchWriteOptions
{
internal readonly string Bucket;
internal readonly string OrganizationId;
internal readonly WritePrecision Precision;
internal BatchWriteOptions(string bucket, string org, WritePrecision precision)
{
Arguments.CheckNonEmptyString(bucket, AbstractRestClient.BucketArgumentValidation);
Arguments.CheckNonEmptyString(org, AbstractRestClient.OrgArgumentValidation);
Arguments.CheckNotNull(precision, nameof(precision));
Bucket = bucket;
OrganizationId = org;
Precision = precision;
}
public override bool Equals(object obj)
{
if (ReferenceEquals(null, obj))
{
return false;
}
if (ReferenceEquals(this, obj))
{
return true;
}
if (obj.GetType() != GetType())
{
return false;
}
return Equals((BatchWriteOptions)obj);
}
public override int GetHashCode()
{
unchecked
{
var hashCode = Bucket != null ? Bucket.GetHashCode() : 0;
hashCode = (hashCode * 397) ^ (OrganizationId != null ? OrganizationId.GetHashCode() : 0);
hashCode = (hashCode * 397) ^ (int)Precision;
return hashCode;
}
}
private bool Equals(BatchWriteOptions other)
{
return string.Equals(Bucket, other.Bucket) && string.Equals(OrganizationId, other.OrganizationId) &&
Precision == other.Precision;
}
}
}