Small refactorings before releasing 0.3.0 (#19)

Author: Mpdreamz

examples/Elastic.Channels.Example/Drain.cs

@@ -47,10 +47,10 @@ public static class Drain
 		var bufferOptions = new BufferOptions
 		{
 			WaitHandle = new CountdownEvent(expectedSentBuffers),
-			MaxInFlightMessages = maxInFlight,
-			MaxConsumerBufferSize = bufferSize,
-			ConcurrentConsumers = concurrency,
-			MaxConsumerBufferLifetime = TimeSpan.FromSeconds(20)
+			InboundBufferMaxSize = maxInFlight,
+			OutboundBufferMaxSize = bufferSize,
+			ExportMaxConcurrency = concurrency,
+			OutboundBufferMaxLifetime = TimeSpan.FromSeconds(20)
 
 		};
 		var channel = new DiagnosticsBufferedChannel(bufferOptions, observeConcurrency: false);

examples/Elastic.Channels.Example/Program.cs

@@ -43,7 +43,7 @@
 Console.WriteLine(channel);
 Console.WriteLine();
 
-Console.WriteLine($"Written buffers: {channel.SentBuffersCount:N0}");
+Console.WriteLine($"Written buffers: {channel.ExportedBuffers:N0}");
 Console.WriteLine($"Written events: {writtenElastic:N0} events");
 Console.WriteLine($"ObservedConcurrency: {channel.ObservedConcurrency:N0}");
 Console.WriteLine($"Duration: {sw.Elapsed:g}");

examples/Elastic.Ingest.Apm.Example/Program.cs

@@ -41,13 +41,13 @@ private static int Main(string[] args)
 			var options =
 				new BufferOptions
 				{
-					ConcurrentConsumers = 1,
-					MaxConsumerBufferSize = 200,
-					MaxConsumerBufferLifetime = TimeSpan.FromSeconds(10),
+					ExportMaxConcurrency = 1,
+					OutboundBufferMaxSize = 200,
+					OutboundBufferMaxLifetime = TimeSpan.FromSeconds(10),
 					WaitHandle = handle,
-					MaxRetries = 3,
-					BackoffPeriod = times => TimeSpan.FromMilliseconds(1),
-					BufferExportedCallback = () => Console.WriteLine("Flushed"),
+					ExportMaxRetries = 3,
+					ExportBackoffPeriod = times => TimeSpan.FromMilliseconds(1),
+					ExportBufferCallback = () => Console.WriteLine("Flushed"),
 				};
 			var channelOptions = new ApmChannelOptions(transport)
 			{
@@ -61,7 +61,7 @@ private static int Main(string[] args)
 				},
 				ExportMaxRetriesCallback = (list) => Interlocked.Increment(ref _maxRetriesExceeded),
 				ExportRetryCallback = (list) => Interlocked.Increment(ref _retries),
-				ExceptionCallback = (e) => _exception = e
+				ExportExceptionCallback = (e) => _exception = e
 			};
 			var channel = new ApmChannel(channelOptions);
 

src/Elastic.Channels/BufferOptions.cs

@@ -11,44 +11,51 @@ public class BufferOptions
 	{
 		/// <summary>
 		/// The maximum number of in flight instances that can be queued in memory. If this threshold is reached, events will be dropped
+		/// <para>Defaults to <c>100_000</c></para>
 		/// </summary>
-		public int MaxInFlightMessages { get; set; } = 100_000;
+		public int InboundBufferMaxSize { get; set; } = 100_000;
 
 		/// <summary>
-		/// The number of events a local buffer should reach before sending the events in a single call to Elasticsearch.
+		/// The maximum size to export to <see cref="BufferedChannelBase{TChannelOptions,TEvent,TResponse}.Export"/> at once.
+		/// <para>Defaults to <c>1_000</c></para>
 		/// </summary>
-		public int MaxConsumerBufferSize { get; set; } = 1_000;
+		public int OutboundBufferMaxSize { get; set; } = 1_000;
 
 		/// <summary>
-		/// The maximum number of times that an item that returns with a retryable status code is retried to be stored in Elasticsearch.
-		/// <see cref="BackoffPeriod"/> to implement a backoff period of your choosing. MaxRetries default to 3.
+		/// The maximum lifetime of a buffer to export to <see cref="BufferedChannelBase{TChannelOptions,TEvent,TResponse}.Export"/>.
+		/// If a buffer is older then the configured <see cref="OutboundBufferMaxLifetime"/> it will be flushed to
+		/// <see cref="BufferedChannelBase{TChannelOptions,TEvent,TResponse}.Export"/> regardless of it's current size
+		/// <para>Defaults to <c>5 seconds</c></para>
 		/// </summary>
-		public int MaxRetries { get; set; } = 3;
+		public TimeSpan OutboundBufferMaxLifetime { get; set; } = TimeSpan.FromSeconds(5);
 
 		/// <summary>
-		/// A consumer builds up a local buffer until <see cref="MaxConsumerBufferSize"/> is reached. If events come in too slow, these
-		/// events could end up taking forever to be sent to Elasticsearch. This controls how long a buffer may exist before a flush is triggered.
+		/// The maximum number of consumers allowed to poll for new events on the channel.
+		/// <para>Defaults to <c>1</c>, increase to introduce concurrency.</para>
 		/// </summary>
-		public TimeSpan MaxConsumerBufferLifetime { get; set; } = TimeSpan.FromSeconds(5);
+		public int ExportMaxConcurrency { get; set; } = 1;
 
 		/// <summary>
-		/// The maximum number of consumers allowed to poll for new events on the channel. Defaults to 1, increase to introduce concurrency.
+		/// The times to retry an export if <see cref="BufferedChannelBase{TChannelOptions,TEvent,TResponse}.RetryBuffer"/> yields items to retry.
+		/// <para>Whether or not items are selected for retrying depends on the actual channel implementation</para>
+		/// <see cref="ExportBackoffPeriod"/> to implement a backoff period of your choosing.
+		/// <para>Defaults to <c>3</c>, when <see cref="BufferedChannelBase{TChannelOptions,TEvent,TResponse}.RetryBuffer"/> yields any items</para>
 		/// </summary>
-		public int ConcurrentConsumers { get; set; } = 1;
+		public int ExportMaxRetries { get; set; } = 3;
 
 
 		/// <summary>
 		/// A function to calculate the backoff period, gets passed the number of retries attempted starting at 0.
 		/// By default backs off in increments of 2 seconds.
 		/// </summary>
-		public Func<int, TimeSpan> BackoffPeriod { get; set; } = (i) => TimeSpan.FromSeconds(2 * (i + 1));
+		public Func<int, TimeSpan> ExportBackoffPeriod { get; set; } = (i) => TimeSpan.FromSeconds(2 * (i + 1));
 
 		/// <summary>
 		/// Called once after a buffer has been flushed, if the buffer is retried this callback is only called once
-		/// all retries have been exhausted. Its called regardless of whether the call to <see cref="BufferedChannelBase{TChannelOptions,TEvent,TResponse}.Send"/>
+		/// all retries have been exhausted. Its called regardless of whether the call to <see cref="BufferedChannelBase{TChannelOptions,TEvent,TResponse}.Export"/>
 		/// succeeded.
 		/// </summary>
-		public Action? BufferExportedCallback { get; set; }
+		public Action? ExportBufferCallback { get; set; }
 
 		/// <summary>
 		/// Allows you to inject a <see cref="CountdownEvent"/> to wait for N number of buffers to flush.

src/Elastic.Channels/BufferedChannelBase.cs

@@ -7,6 +7,7 @@
 using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
+using Elastic.Channels.Buffers;
 
 namespace Elastic.Channels;
 
@@ -48,10 +49,10 @@ protected BufferedChannelBase(TChannelOptions options)
 	{
 		TokenSource = new CancellationTokenSource();
 		Options = options;
-		var maxConsumers = Math.Max(1, BufferOptions.ConcurrentConsumers);
+		var maxConsumers = Math.Max(1, BufferOptions.ExportMaxConcurrency);
 		_throttleTasks = new SemaphoreSlim(maxConsumers, maxConsumers);
 		_signal = options.BufferOptions.WaitHandle;
-		var maxIn = Math.Max(1, BufferOptions.MaxInFlightMessages);
+		var maxIn = Math.Max(1, BufferOptions.InboundBufferMaxSize);
 		InChannel = Channel.CreateBounded<TEvent>(new BoundedChannelOptions(maxIn)
 		{
 			SingleReader = false,
@@ -63,8 +64,8 @@ protected BufferedChannelBase(TChannelOptions options)
 			// DropWrite will make `TryWrite` always return true, which is not what we want.
 			FullMode = BoundedChannelFullMode.Wait
 		});
-		// The minimum out buffer the max of (1 or MaxConsumerBufferSize) as long as it does not exceed MaxInFlightMessages
-		var maxOut = Math.Min(BufferOptions.MaxInFlightMessages, Math.Max(1, BufferOptions.MaxConsumerBufferSize));
+		// The minimum out buffer the max of (1 or OutboundBufferMaxSize) as long as it does not exceed InboundBufferMaxSize
+		var maxOut = Math.Min(BufferOptions.InboundBufferMaxSize, Math.Max(1, BufferOptions.OutboundBufferMaxSize));
 		OutChannel = Channel.CreateBounded<IOutboundBuffer<TEvent>>(
 			new BoundedChannelOptions(maxOut)
 			{
@@ -78,12 +79,12 @@ protected BufferedChannelBase(TChannelOptions options)
 				FullMode = BoundedChannelFullMode.Wait
 			});
 
-		InboundBuffer = new InboundBuffer<TEvent>(maxOut, BufferOptions.MaxConsumerBufferLifetime);
+		InboundBuffer = new InboundBuffer<TEvent>(maxOut, BufferOptions.OutboundBufferMaxLifetime);
 
 		_outThread = Task.Factory.StartNew(async () => await ConsumeOutboundEvents().ConfigureAwait(false),
 			TaskCreationOptions.LongRunning | TaskCreationOptions.PreferFairness);
 		_inThread = Task.Factory.StartNew(async () =>
-				await ConsumeInboundEvents(maxOut, BufferOptions.MaxConsumerBufferLifetime)
+				await ConsumeInboundEvents(maxOut, BufferOptions.OutboundBufferMaxLifetime)
 					.ConfigureAwait(false)
 			, TaskCreationOptions.LongRunning | TaskCreationOptions.PreferFairness
 		);
@@ -105,9 +106,12 @@ await ConsumeInboundEvents(maxOut, BufferOptions.MaxConsumerBufferLifetime)
 
 	public override bool TryWrite(TEvent item)
 	{
-		if (InChannel.Writer.TryWrite(item)) return true;
-
-		Options.PublishRejectionCallback?.Invoke(item);
+		if (InChannel.Writer.TryWrite(item))
+		{
+			Options.PublishToInboundChannelCallback?.Invoke();
+			return true;
+		}
+		Options.PublishToInboundChannelFailureCallback?.Invoke();
 		return false;
 	}
 
@@ -117,16 +121,14 @@ public virtual async Task<bool> WaitToWriteAsync(TEvent item, CancellationToken
 		if (await InChannel.Writer.WaitToWriteAsync(ctx).ConfigureAwait(false) &&
 		    InChannel.Writer.TryWrite(item))
 		{
-			Options.PublishToInboundChannel?.Invoke();
+			Options.PublishToInboundChannelCallback?.Invoke();
 			return true;
 		}
-		Options.PublishToInboundChannelFailure?.Invoke();
-
-		Options.PublishRejectionCallback?.Invoke(item);
+		Options.PublishToInboundChannelFailureCallback?.Invoke();
 		return false;
 	}
 
-	protected abstract Task<TResponse> Send(IReadOnlyCollection<TEvent> buffer, CancellationToken ctx = default);
+	protected abstract Task<TResponse> Export(IReadOnlyCollection<TEvent> buffer, CancellationToken ctx = default);
 
 	private static readonly IReadOnlyCollection<TEvent> DefaultRetryBuffer = new TEvent[] { };
 
@@ -138,9 +140,9 @@ IWriteTrackingBuffer statistics
 
 	private async Task ConsumeOutboundEvents()
 	{
-		Options.OutboundChannelStarted?.Invoke();
+		Options.OutboundChannelStartedCallback?.Invoke();
 
-		var maxConsumers = Options.BufferOptions.ConcurrentConsumers;
+		var maxConsumers = Options.BufferOptions.ExportMaxConcurrency;
 		var taskList = new List<Task>(maxConsumers);
 
 		while (await OutChannel.Reader.WaitToReadAsync().ConfigureAwait(false))
@@ -165,12 +167,12 @@ private async Task ConsumeOutboundEvents()
 			}
 		}
 		await Task.WhenAll(taskList).ConfigureAwait(false);
-		Options.OutboundChannelExited?.Invoke();
+		Options.OutboundChannelExitedCallback?.Invoke();
 	}
 
 	private async Task ExportBuffer(IReadOnlyCollection<TEvent> items, IWriteTrackingBuffer buffer)
 	{
-		var maxRetries = Options.BufferOptions.MaxRetries;
+		var maxRetries = Options.BufferOptions.ExportMaxRetries;
 		for (var i = 0; i <= maxRetries && items.Count > 0; i++)
 		{
 			if (TokenSource.Token.IsCancellationRequested) break;
@@ -180,12 +182,12 @@ private async Task ExportBuffer(IReadOnlyCollection<TEvent> items, IWriteTrackin
 			TResponse? response;
 			try
 			{
-				response = await Send(items, TokenSource.Token).ConfigureAwait(false);
+				response = await Export(items, TokenSource.Token).ConfigureAwait(false);
 				Options.ExportResponseCallback?.Invoke(response, buffer);
 			}
 			catch (Exception e)
 			{
-				Options.ExceptionCallback?.Invoke(e);
+				Options.ExportExceptionCallback?.Invoke(e);
 				break;
 			}
 
@@ -195,21 +197,21 @@ private async Task ExportBuffer(IReadOnlyCollection<TEvent> items, IWriteTrackin
 			var atEndOfRetries = i == maxRetries;
 			if (items.Count > 0 && !atEndOfRetries)
 			{
-				await Task.Delay(Options.BufferOptions.BackoffPeriod(i), TokenSource.Token).ConfigureAwait(false);
+				await Task.Delay(Options.BufferOptions.ExportBackoffPeriod(i), TokenSource.Token).ConfigureAwait(false);
 				Options.ExportRetryCallback?.Invoke(items);
 			}
 			// otherwise if retryable items still exist and the user wants to be notified notify the user
 			else if (items.Count > 0 && atEndOfRetries)
 				Options.ExportMaxRetriesCallback?.Invoke(items);
 		}
-		Options.BufferOptions.BufferExportedCallback?.Invoke();
+		Options.BufferOptions.ExportBufferCallback?.Invoke();
 		if (_signal is { IsSet: false })
 			_signal.Signal();
 	}
 
 	private async Task ConsumeInboundEvents(int maxQueuedMessages, TimeSpan maxInterval)
 	{
-		Options.InboundChannelStarted?.Invoke();
+		Options.InboundChannelStartedCallback?.Invoke();
 		while (await InboundBuffer.WaitToReadAsync(InChannel.Reader).ConfigureAwait(false))
 		{
 			if (TokenSource.Token.IsCancellationRequested) break;
@@ -223,7 +225,7 @@ private async Task ConsumeInboundEvents(int maxQueuedMessages, TimeSpan maxInter
 					break;
 			}
 
-			Options.PublishToOutboundChannel?.Invoke();
+			Options.PublishToOutboundChannelCallback?.Invoke();
 			if (InboundBuffer.NoThresholdsHit) continue;
 
 			//:w
@@ -234,24 +236,21 @@ private async Task ConsumeInboundEvents(int maxQueuedMessages, TimeSpan maxInter
 
 			if (await PublishAsync(outboundBuffer).ConfigureAwait(false))
 				continue;
-
-			foreach (var e in InboundBuffer.Buffer)
-				Options.PublishRejectionCallback?.Invoke(e);
+			Options.PublishToOutboundChannelFailureCallback?.Invoke();
 		}
 	}
 
 	private ValueTask<bool> PublishAsync(IOutboundBuffer<TEvent> buffer)
 	{
 		async Task<bool> AsyncSlowPath(IOutboundBuffer<TEvent> b)
 		{
-			var maxRetries = Options.BufferOptions.MaxRetries;
+			var maxRetries = Options.BufferOptions.ExportMaxRetries;
 			for (var i = 0; i <= maxRetries; i++)
 				while (await OutChannel.Writer.WaitToWriteAsync().ConfigureAwait(false))
 				{
 					if (OutChannel.Writer.TryWrite(b))
 						return true;
 				}
-			Options.PublishToOutboundChannelFailure?.Invoke();
 			return false;
 		}
 

src/Elastic.Channels/IWriteTrackingBuffer.cs renamed to src/Elastic.Channels/Buffers/IWriteTrackingBuffer.cs

@@ -4,7 +4,7 @@
 
 using System;
 
-namespace Elastic.Channels;
+namespace Elastic.Channels.Buffers;
 
 /// <summary>
 /// Represents a buffer that tracks the <see cref="DurationSinceFirstWrite"/> and it's current <see cref="Count"/>

src/Elastic.Channels/InboundBuffer.cs renamed to src/Elastic.Channels/Buffers/InboundBuffer.cs

@@ -8,7 +8,7 @@
 using System.Threading.Channels;
 using System.Threading.Tasks;
 
-namespace Elastic.Channels;
+namespace Elastic.Channels.Buffers;
 
 /// <summary>
 /// <see cref="InboundBuffer{TEvent}"/> is a buffer that will block <see cref="WaitToReadAsync"/> until

src/Elastic.Channels/OutboundBuffer.cs renamed to src/Elastic.Channels/Buffers/OutboundBuffer.cs

@@ -5,7 +5,7 @@
 using System;
 using System.Collections.Generic;
 
-namespace Elastic.Channels;
+namespace Elastic.Channels.Buffers;
 
 public interface IOutboundBuffer<out TEvent> : IWriteTrackingBuffer
 {

src/Elastic.Channels/ChannelOptionsBase.cs

@@ -6,6 +6,7 @@
 using System.IO;
 using System.Threading;
 using System.Threading.Tasks;
+using Elastic.Channels.Buffers;
 
 namespace Elastic.Channels
 {
@@ -20,40 +21,40 @@ public abstract class ChannelOptionsBase<TEvent, TResponse>
 
 		public Func<Stream, CancellationToken, TEvent, Task> WriteEvent { get; set; } = null!;
 
-		/// <summary>
-		/// If <see cref="Elastic.Channels.BufferOptions.MaxInFlightMessages"/> is reached, <see cref="TEvent"/>'s will fail to be published to the channel. You can be notified of dropped
-		/// events with this callback
-		/// </summary>
-		public Action<TEvent>? PublishRejectionCallback { get; set; }
-
-		public Action<Exception>? ExceptionCallback { get; set; }
+		/// <summary> Called if the call to <see cref="BufferedChannelBase{TChannelOptions,TEvent,TResponse}.Export"/> throws. </summary>
+		public Action<Exception>? ExportExceptionCallback { get; set; }
 
+		/// <summary> Called with (number of retries) (number of items to be exported) </summary>
 		public Action<int, int>? ExportItemsAttemptCallback { get; set; }
 
-		/// <summary> Subscribe to be notified of events that are retryable but did not store correctly withing the boundaries of <see cref="Elastic.Channels.BufferOptions.MaxRetries"/></summary>
+		/// <summary> Subscribe to be notified of events that are retryable but did not store correctly withing the boundaries of <see cref="Channels.BufferOptions.ExportMaxRetries"/></summary>
 		public Action<IReadOnlyCollection<TEvent>>? ExportMaxRetriesCallback { get; set; }
 
-		/// <summary> Subscribe to be notified of events that are retryable but did not store correctly within the number of configured <see cref="Elastic.Channels.BufferOptions.MaxRetries"/></summary>
+		/// <summary> Subscribe to be notified of events that are retryable but did not store correctly within the number of configured <see cref="Channels.BufferOptions.ExportMaxRetries"/></summary>
 		public Action<IReadOnlyCollection<TEvent>>? ExportRetryCallback { get; set; }
 
 		/// <summary> A generic hook to be notified of any bulk request being initiated by <see cref="InboundBuffer{TEvent}"/> </summary>
 		public Action<TResponse, IWriteTrackingBuffer>? ExportResponseCallback { get; set; }
 
 		/// <summary>Called everytime an event is written to the inbound channel </summary>
-		public Action? PublishToInboundChannel { get; set; }
+		public Action? PublishToInboundChannelCallback { get; set; }
 
 		/// <summary>Called everytime an event is not written to the inbound channel </summary>
-		public Action? PublishToInboundChannelFailure { get; set; }
+		public Action? PublishToInboundChannelFailureCallback { get; set; }
 
 		/// <summary>Called everytime the inbound channel publishes to the outbound channel. </summary>
-		public Action? PublishToOutboundChannel { get; set; }
+		public Action? PublishToOutboundChannelCallback { get; set; }
+
+		/// <summary> Called when the thread to read the outbound channel is started </summary>
+		public Action? OutboundChannelStartedCallback { get; set; }
+		/// <summary> Called when the thread to read the outbound channel has exited</summary>
+		public Action? OutboundChannelExitedCallback { get; set; }
 
-		public Action? OutboundChannelStarted { get; set; }
-		public Action? OutboundChannelExited { get; set; }
-		public Action? InboundChannelStarted { get; set; }
+		/// <summary> Called when the thread to read the inbound channel has started</summary>
+		public Action? InboundChannelStartedCallback { get; set; }
 
 		/// <summary>Called everytime the inbound channel fails to publish to the outbound channel. </summary>
-		public Action? PublishToOutboundChannelFailure { get; set; }
+		public Action? PublishToOutboundChannelFailureCallback { get; set; }
 	}
 
 }