Update Serilog.Sinks.PeriodicBatching package and reimplent `Elasti…

…csearchSink` so that it does not use obsolete `PeriodicBatchingSink` constructor.

src/Serilog.Sinks.Elasticsearch/Serilog.Sinks.Elasticsearch.csproj

@@ -39,7 +39,7 @@
     <PackageReference Include="Serilog" Version="2.12.0" />
     <PackageReference Include="Serilog.Formatting.Compact" Version="1.1.0" />
     <PackageReference Include="Serilog.Sinks.File" Version="5.0.0" />
-    <PackageReference Include="Serilog.Sinks.PeriodicBatching" Version="2.1.1" />
+    <PackageReference Include="Serilog.Sinks.PeriodicBatching" Version="3.1.0" />
     <PackageReference Include="Elasticsearch.Net" Version="7.17.5" />
     <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="6.0.0" />
   </ItemGroup>

src/Serilog.Sinks.Elasticsearch/Sinks/ElasticSearch/Durable/Elasticsearch/ElasticsearchLogClient.cs

@@ -86,7 +86,7 @@ private InvalidResult GetInvalidPayloadAsync(DynamicResponse baseResult, List<st
             bool hasErrors = false;
             foreach (dynamic item in items)
             {
-                var itemIndex = item?[ElasticsearchSink.BulkAction(_elasticOpType)];
+                var itemIndex = item?[BatchedElasticsearchSink.BulkAction(_elasticOpType)];
                 long? status = itemIndex?["status"];
                 i++;
                 if (!status.HasValue || status < 300)

src/Serilog.Sinks.Elasticsearch/Sinks/ElasticSearch/Durable/Elasticsearch/ElasticsearchPayloadReader.cs

@@ -92,7 +92,7 @@ protected override List<string> FinishPayLoad()
         protected override void AddToPayLoad(string nextLine)
         {
             var indexName = _getIndexForEvent(nextLine, _date);
-            var action = ElasticsearchSink.CreateElasticAction(
+            var action = BatchedElasticsearchSink.CreateElasticAction(
                 opType: _elasticOpType, 
                 indexName: indexName, pipelineName: _pipelineName,
                 id: _count + "_" + Guid.NewGuid(),

src/Serilog.Sinks.Elasticsearch/Sinks/ElasticSearch/ElasticSearchSink.cs

@@ -25,26 +25,40 @@
 
 namespace Serilog.Sinks.Elasticsearch
 {
+    public sealed class ElasticsearchSink : PeriodicBatchingSink
+    {
+        public ElasticsearchSink(ElasticsearchSinkOptions options)
+            : this(
+                  new BatchedElasticsearchSink(options),
+                  new PeriodicBatchingSinkOptions
+                    {
+                        BatchSizeLimit = options.BatchPostingLimit,
+                        Period = options.Period,
+                        EagerlyEmitFirstEvent = true,
+                        QueueLimit = (options.QueueSizeLimit == -1) ? null : new int?(options.QueueSizeLimit)
+                    }
+                  )
+        {
+        }
+
+        private ElasticsearchSink(IBatchedLogEventSink batchedSink, PeriodicBatchingSinkOptions options)
+            : base(batchedSink, options)
+        {
+        }
+    }
     /// <summary>
     /// Writes log events as documents to ElasticSearch.
     /// </summary>
-    public class ElasticsearchSink : PeriodicBatchingSink
+    internal sealed class BatchedElasticsearchSink : IBatchedLogEventSink
     {
-
         private readonly ElasticsearchSinkState _state;
 
-        /// <summary>
-        /// Creates a new ElasticsearchSink instance with the provided options
-        /// </summary>
-        /// <param name="options">Options configuring how the sink behaves, may NOT be null</param>
-        public ElasticsearchSink(ElasticsearchSinkOptions options)
-            : base(options.BatchPostingLimit, options.Period, options.QueueSizeLimit)
+        public BatchedElasticsearchSink(ElasticsearchSinkOptions options)
         {
             _state = ElasticsearchSinkState.Create(options);
             _state.DiscoverClusterVersion();
             _state.RegisterTemplateIfNeeded();
         }
-
         /// <summary>
         /// Emit a batch of log events, running to completion synchronously.
         /// </summary>
@@ -54,7 +68,7 @@ public ElasticsearchSink(ElasticsearchSinkOptions options)
         ///  or <see cref="M:Serilog.Sinks.PeriodicBatching.PeriodicBatchingSink.EmitBatchAsync(System.Collections.Generic.IEnumerable{Serilog.Events.LogEvent})" />,
         /// not both.
         /// </remarks>
-        protected override async Task EmitBatchAsync(IEnumerable<LogEvent> events)
+        public async Task EmitBatchAsync(IEnumerable<LogEvent> events)
         {
             DynamicResponse result;
 
@@ -71,42 +85,32 @@ protected override async Task EmitBatchAsync(IEnumerable<LogEvent> events)
             HandleResponse(events, result);
         }
 
-        /// <summary>
-        /// Emit a batch of log events, running to completion synchronously.
-        /// </summary>
-        /// <param name="events">The events to emit.</param>
-        /// <returns>Response from Elasticsearch</returns>
-        protected virtual Task<T> EmitBatchCheckedAsync<T>(IEnumerable<LogEvent> events) where T : class, IElasticsearchResponse, new()
+        public Task OnEmptyBatchAsync()
         {
-            // ReSharper disable PossibleMultipleEnumeration
-            if (events == null || !events.Any())
-                return Task.FromResult<T>(default(T));
-
-            var payload = CreatePayload(events);
-            return _state.Client.BulkAsync<T>(PostData.MultiJson(payload));
+            return Task.CompletedTask;
         }
 
         /// <summary>
         /// Emit a batch of log events, running to completion synchronously.
         /// </summary>
         /// <param name="events">The events to emit.</param>
         /// <returns>Response from Elasticsearch</returns>
-        protected virtual T EmitBatchChecked<T>(IEnumerable<LogEvent> events) where T : class, IElasticsearchResponse, new()
+        private Task<T> EmitBatchCheckedAsync<T>(IEnumerable<LogEvent> events) where T : class, IElasticsearchResponse, new()
         {
             // ReSharper disable PossibleMultipleEnumeration
             if (events == null || !events.Any())
-                return null;
+                return Task.FromResult<T>(default(T));
 
             var payload = CreatePayload(events);
-            return _state.Client.Bulk<T>(PostData.MultiJson(payload));
+            return _state.Client.BulkAsync<T>(PostData.MultiJson(payload));
         }
 
         /// <summary>
         /// Handles the exceptions.
         /// </summary>
         /// <param name="ex"></param>
         /// <param name="events"></param>
-        protected virtual void HandleException(Exception ex, IEnumerable<LogEvent> events)
+        private void HandleException(Exception ex, IEnumerable<LogEvent> events)
         {
             if (_state.Options.EmitEventFailure.HasFlag(EmitEventFailureHandling.WriteToSelfLog))
             {
@@ -153,18 +157,6 @@ protected virtual void HandleException(Exception ex, IEnumerable<LogEvent> event
                 throw ex;
         }
 
-        // Helper function: checks if a given dynamic member / dictionary key exists at runtime
-        private static bool HasProperty(dynamic settings, string name)
-        {
-            if (settings is IDictionary<string, object>)
-                return ((IDictionary<string, object>)settings).ContainsKey(name);
-
-            if (settings is System.Dynamic.DynamicObject)
-                return ((System.Dynamic.DynamicObject)settings).GetDynamicMemberNames().Contains(name);
-
-            return settings.GetType().GetProperty(name) != null;
-        }
-
         private IEnumerable<string> CreatePayload(IEnumerable<LogEvent> events)
         {
             if (!_state.TemplateRegistrationSuccess && _state.Options.RegisterTemplateFailure == RegisterTemplateRecovery.FailSink)
@@ -280,7 +272,7 @@ internal static object CreateElasticAction(ElasticOpType opType, string indexNam
                 : new ElasticIndexAction(actionPayload);
             return action;
         }
-        
+
         sealed class ElasticCreateAction
         {
             public ElasticCreateAction(ElasticActionPayload payload)