Optimize allocations on emission of single point

Fixes influxdata#60

(cherry picked from commit f9d678f)

src/InfluxDB.Collector/Configuration/AggregateEmitter.cs

@@ -4,7 +4,7 @@
 
 namespace InfluxDB.Collector.Configuration
 {
-    class AggregateEmitter : IPointEmitter
+    class AggregateEmitter : IPointEmitter, ISinglePointEmitter
     {
         readonly List<IPointEmitter> _emitters;
 
@@ -19,5 +19,20 @@ public void Emit(PointData[] points)
             foreach (var emitter in _emitters)
                 emitter.Emit(points);
         }
+
+        public void Emit(PointData point)
+        {
+            foreach (var emitter in _emitters)
+            {
+                if (emitter is ISinglePointEmitter singlePointEmitter)
+                {
+                    singlePointEmitter.Emit(point);
+                }
+                else
+                {
+                    emitter.Emit(new[] { point });
+                }
+            }
+        }
     }
 }

src/InfluxDB.Collector/MetricsCollector.cs

@@ -5,7 +5,7 @@
 
 namespace InfluxDB.Collector
 {
-    public abstract class MetricsCollector : IPointEmitter, IDisposable
+    public abstract class MetricsCollector : IPointEmitter, ISinglePointEmitter, IDisposable
     {
         readonly Util.ITimestampSource _timestampSource = new Util.PseudoHighResTimestampSource();
 
@@ -34,14 +34,16 @@ public void Dispose()
             Dispose(true);
         }
 
-        protected virtual void Dispose(bool disposing) { }
+        protected virtual void Dispose(bool disposing)
+        {
+        }
 
         public void Write(string measurement, IReadOnlyDictionary<string, object> fields, IReadOnlyDictionary<string, string> tags = null, DateTime? timestamp = null)
         {
             try
             {
                 var point = new PointData(measurement, fields, tags, timestamp ?? _timestampSource.GetUtcNow());
-                Emit(new[] { point });
+                Emit(point);
             }
             catch (Exception ex)
             {
@@ -54,6 +56,16 @@ void IPointEmitter.Emit(PointData[] points)
             Emit(points);
         }
 
+        void ISinglePointEmitter.Emit(PointData point)
+        {
+            Emit(point);
+        }
+
         protected abstract void Emit(PointData[] points);
+
+        protected virtual void Emit(PointData point)
+        {
+            Emit(new[] { point });
+        }
     }
-}
+}

src/InfluxDB.Collector/Pipeline/Common/IntervalEmitterBase.cs

@@ -6,7 +6,7 @@
 
 namespace InfluxDB.Collector.Pipeline.Common
 {
-    internal abstract class IntervalEmitterBase : IPointEmitter, IDisposable
+    internal abstract class IntervalEmitterBase : IPointEmitter, ISinglePointEmitter, IDisposable
     {
         readonly object _queueLock = new object();
         Queue<PointData> _queue = new Queue<PointData>();
@@ -77,22 +77,45 @@ protected Task OnTick()
         }
 
         public void Emit(PointData[] points)
+        {
+            if (!CheckState())
+            {
+                return;
+            }
+
+            lock (_queueLock)
+            {
+                foreach (var point in points)
+                    _queue.Enqueue(point);
+            }
+        }
+
+        public void Emit(PointData point)
+        {
+            if (!CheckState())
+            {
+                return;
+            }
+
+            lock (_queueLock)
+            {
+                _queue.Enqueue(point);
+            }
+        }
+
+        private bool CheckState()
         {
             lock (_stateLock)
             {
-                if (_unloading) return;
+                if (_unloading) return false;
                 if (!_started)
                 {
                     _started = true;
                     _timer.Start(TimeSpan.Zero);
                 }
             }
 
-            lock (_queueLock)
-            {
-                foreach (var point in points)
-                    _queue.Enqueue(point);
-            }
+            return true;
         }
 
         protected abstract void HandleBatch(IReadOnlyCollection<PointData> batch);

src/InfluxDB.Collector/Pipeline/Emit/HttpLineProtocolEmitter.cs

@@ -5,7 +5,7 @@
 
 namespace InfluxDB.Collector.Pipeline.Emit
 {
-    class HttpLineProtocolEmitter : IDisposable, IPointEmitter
+    class HttpLineProtocolEmitter : IDisposable, IPointEmitter, ISinglePointEmitter
     {
         readonly ILineProtocolClient _client;
 
@@ -29,9 +29,23 @@ public void Emit(PointData[] points)
                 payload.Add(new LineProtocolPoint(point.Measurement, point.Fields, point.Tags, point.UtcTimestamp));
             }
 
+            SendPayload(payload);
+        }
+
+        public void Emit(PointData point)
+        {
+            var payload = new LineProtocolPayload();
+
+            payload.Add(new LineProtocolPoint(point.Measurement, point.Fields, point.Tags, point.UtcTimestamp));
+
+            SendPayload(payload);
+        }
+
+        private void SendPayload(LineProtocolPayload payload)
+        {
             var influxResult = _client.WriteAsync(payload).Result;
             if (!influxResult.Success)
                 CollectorLog.ReportError(influxResult.ErrorMessage, null);
         }
     }
-}
+}

src/InfluxDB.Collector/Pipeline/ISinglePointEmitter.cs

@@ -0,0 +1,7 @@
+namespace InfluxDB.Collector.Pipeline
+{
+    interface ISinglePointEmitter
+    {
+        void Emit(PointData point);
+    }
+}

src/InfluxDB.Collector/Pipeline/NullMetricsCollector.cs

@@ -5,5 +5,9 @@ class NullMetricsCollector : MetricsCollector
         protected override void Emit(PointData[] points)
         {
         }
+
+        protected override void Emit(PointData point)
+        {
+        }
     }
 }

src/InfluxDB.Collector/Pipeline/PipelinedMetricsCollector.cs

@@ -1,5 +1,4 @@
-
-using System;
+using System;
 
 namespace InfluxDB.Collector.Pipeline
 {
@@ -24,10 +23,24 @@ protected override void Emit(PointData[] points)
             _emitter.Emit(points);
         }
 
+        protected override void Emit(PointData point)
+        {
+            _enricher.Enrich(point);
+
+            if (_emitter is ISinglePointEmitter singlePointEmitter)
+            {
+                singlePointEmitter.Emit(point);
+            }
+            else
+            {
+                _emitter.Emit(new[] { point });
+            }
+        }
+
         protected override void Dispose(bool disposing)
         {
             if (disposing)
                 _dispose();
         }
     }
-}
+}