feat: improve WriteApi performance

Client/Client.csproj

@@ -33,6 +33,7 @@
 
     <ItemGroup>
         <PackageReference Include="JsonSubTypes" Version="1.5.2" />
+        <PackageReference Include="Microsoft.Extensions.ObjectPool" Version="3.1.4" />
         <PackageReference Include="Microsoft.Net.Http.Headers" Version="2.1.1" />
         <PackageReference Include="System.Collections.Immutable" Version="1.7.1" />
         <PackageReference Include="System.Configuration.ConfigurationManager" Version="4.5.0" />

Client/WriteApi.cs

@@ -14,6 +14,7 @@
 using InfluxDB.Client.Core.Exceptions;
 using InfluxDB.Client.Internal;
 using InfluxDB.Client.Writes;
+using Microsoft.Extensions.ObjectPool;
 using RestSharp;
 
 namespace InfluxDB.Client
@@ -26,8 +27,11 @@ public class WriteApi : IDisposable
         private readonly MeasurementMapper _measurementMapper = new MeasurementMapper();
         private readonly InfluxDBClientOptions _options;
         private readonly Subject<BatchWriteData> _subject = new Subject<BatchWriteData>();
+        private static readonly ObjectPoolProvider _objectPoolProvider = new DefaultObjectPoolProvider();
+        private static readonly ObjectPool<StringBuilder> _stringBuilderPool = _objectPoolProvider.CreateStringBuilderPool();
         private readonly IDisposable _unsubscribeDisposeCommand;
 
+
         private bool _disposed;
         protected internal WriteApi(
             InfluxDBClientOptions options, 
@@ -52,26 +56,25 @@ protected internal WriteApi(
             // 
             // https://github.com/dotnet/reactive/issues/19
 
-            var tempBoundary = new Subject<IObservable<BatchWriteData>>();
 
-            _subject
+
+            IObservable<IObservable<BatchWriteRecord>> 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(tempBoundary)
-                        .Merge(Observable.Defer(() =>
-                        {
-                            connectedSource
-                                .Window(TimeSpan.FromMilliseconds(writeOptions.FlushInterval), writeOptions.BatchSize,
-                                    writeOptions.WriteScheduler)
-                                .Merge(_flush)
-                                .Subscribe(tempBoundary);
-
-                            return Observable.Empty<IObservable<BatchWriteData>>();
-                        }));
+                        .Window(trigger);
                 })
                 //
                 // Group by key - same bucket, same org
@@ -83,7 +86,7 @@ protected internal WriteApi(
                 .Select(grouped =>
                 {
                     var aggregate = grouped
-                        .Aggregate(new StringBuilder(""), (builder, batchWrite) =>
+                        .Aggregate(_stringBuilderPool.Get(), (builder, batchWrite) =>
                         {
                             var data = batchWrite.ToLineProtocol();
 
@@ -95,27 +98,34 @@ protected internal WriteApi(
                             }
 
                             return builder.Append(data);
-                        }).Select(builder => builder.ToString());
-
-                    return aggregate.Select(records => new BatchWriteRecord(grouped.Key, records));
-                })
-                //
-                // Jitter
-                //
-                .Select(source =>
-                {
-                    if (writeOptions.JitterInterval <= 0)
-                    {
-                        return source;
-                    }
+                        }).Select(builder =>
+                        {
+                            var result = builder.ToString();
+                            builder.Clear();
+                            _stringBuilderPool.Return(builder);
+                            return result;
+                        });
 
-                    return source.Delay(_ => Observable.Timer(TimeSpan.FromMilliseconds(JitterDelay(writeOptions)), writeOptions.WriteScheduler));
-                })
-                .Concat()
+                    return aggregate.Select(records => new BatchWriteRecord(grouped.Key, records))
+                                    .Where(batchWriteItem => !string.IsNullOrEmpty(batchWriteItem.ToLineProtocol()));
+                });
+
+            if (writeOptions.JitterInterval > 0)
+            {
+                batches = batches
+                    //
+                    // Jitter
+                    //
+                    .Select(source =>
+                    {
+                        return source.Delay(_ => Observable.Timer(TimeSpan.FromMilliseconds(JitterDelay(writeOptions)), writeOptions.WriteScheduler));
+                    });
+            }
+            var query = batches
+                .Concat() 
                 //
                 // Map to Async request
                 //
-                .Where(batchWriteItem => !string.IsNullOrEmpty(batchWriteItem.ToLineProtocol()))
                 .Select(batchWriteItem =>
                 {
                     var org = batchWriteItem.Options.OrganizationId;

Client/Writes/PointData.cs

@@ -35,6 +35,16 @@ private PointData(string measurementName)
             Precision = WritePrecision.Ns;
         }
 
+        /// <summary>
+        /// Create a new Point withe specified a measurement name.
+        /// </summary>
+        /// <param name="measurementName">the measurement name</param>
+        /// <returns>the new Point</returns>
+        public static PointData Measurement(string measurementName)
+        {
+            return new PointData(measurementName);
+        }
+
         private PointData(string measurementName,
                             WritePrecision precision,
                             BigInteger? time,
@@ -48,16 +58,6 @@ private PointData(string measurementName,
             _fields = fields;
         }
 
-        /// <summary>
-        /// Create a new Point withe specified a measurement name.
-        /// </summary>
-        /// <param name="measurementName">the measurement name</param>
-        /// <returns>the new Point</returns>
-        public static PointData Measurement(string measurementName)
-        {
-            return new PointData(measurementName);
-        }
-
         /// <summary>
         /// Adds or replaces a tag value for a point.
         /// </summary>