Add continuous example to validate our defaults (#31)

elastic · May 2, 2023 · 8eef0e1 · 8eef0e1
1 parent b00c267
commit 8eef0e1
Show file tree

Hide file tree

Showing 5 changed files with 69 additions and 11 deletions.
diff --git a/elastic-ingest-dotnet.sln b/elastic-ingest-dotnet.sln
@@ -55,6 +55,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elastic.Ingest.Apm.Example"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elastic.Ingest.OpenTelemetry.Example", "examples\Elastic.Ingest.OpenTelemetry.Example\Elastic.Ingest.OpenTelemetry.Example.csproj", "{F2F1FC18-C30F-4D9E-9EF0-F1E89EE71D17}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elastic.Channels.Continuous", "examples\Elastic.Channels.Continuous\Elastic.Channels.Continuous.csproj", "{42069F63-6AB0-43D1-AF8B-C7DD521F64C5}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -73,6 +75,7 @@ Global
 		{92F87F85-3028-4E98-A1C7-6CCEC4392AB4} = {A60DDBBB-4BF4-4B3B-A13A-E0B409917433}
 		{81F91C49-F590-47E2-AA69-AA713E840DD6} = {B67CBB46-74C1-47EB-9E41-D55C5E0E0D85}
 		{F2F1FC18-C30F-4D9E-9EF0-F1E89EE71D17} = {B67CBB46-74C1-47EB-9E41-D55C5E0E0D85}
+		{42069F63-6AB0-43D1-AF8B-C7DD521F64C5} = {B67CBB46-74C1-47EB-9E41-D55C5E0E0D85}
 	EndGlobalSection
 	GlobalSection(ProjectConfigurationPlatforms) = postSolution
 		{9F529525-E8E3-463D-A920-4D6E34150FC5}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
@@ -123,5 +126,9 @@ Global
 		{F2F1FC18-C30F-4D9E-9EF0-F1E89EE71D17}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{F2F1FC18-C30F-4D9E-9EF0-F1E89EE71D17}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{F2F1FC18-C30F-4D9E-9EF0-F1E89EE71D17}.Release|Any CPU.Build.0 = Release|Any CPU
+		{42069F63-6AB0-43D1-AF8B-C7DD521F64C5}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{42069F63-6AB0-43D1-AF8B-C7DD521F64C5}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{42069F63-6AB0-43D1-AF8B-C7DD521F64C5}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{42069F63-6AB0-43D1-AF8B-C7DD521F64C5}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 EndGlobal
diff --git a/examples/Elastic.Channels.Continuous/Elastic.Channels.Continuous.csproj b/examples/Elastic.Channels.Continuous/Elastic.Channels.Continuous.csproj
@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <TargetFramework>net6.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+      <ProjectReference Include="..\..\src\Elastic.Channels\Elastic.Channels.csproj" />
+    </ItemGroup>
+
+</Project>
diff --git a/examples/Elastic.Channels.Continuous/Program.cs b/examples/Elastic.Channels.Continuous/Program.cs
@@ -0,0 +1,37 @@
+// Licensed to Elasticsearch B.V under one or more agreements.
+// Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information
+
+using Elastic.Channels;
+using Elastic.Channels.Diagnostics;
+
+var options = new NoopBufferedChannel.NoopChannelOptions
+{
+	BufferOptions = new BufferOptions()
+	{
+		OutboundBufferMaxSize = 10_000,
+		InboundBufferMaxSize = 10_000_000,
+	},
+	ExportBufferCallback = () => Console.Write("."),
+	ExportExceptionCallback = e => Console.Write("!"),
+	PublishToInboundChannelFailureCallback  = () => Console.Write("I"),
+	PublishToOutboundChannelFailureCallback  = () => Console.Write("O"),
+
+};
+var channel = new DiagnosticsBufferedChannel(options);
+for (long i = 0; i < long.MaxValue; i++)
+{
+	var e = new NoopBufferedChannel.NoopEvent { Id = i };
+	var written = false;
+	var ready = await channel.WaitToWriteAsync();
+	if (ready) written = channel.TryWrite(e);
+	if (!written || channel.BufferMismatches > 0)
+	{
+		Console.WriteLine();
+		Console.WriteLine(channel);
+		Console.WriteLine(i);
+		Environment.Exit(1);
+	}
+
+}
+
diff --git a/src/Elastic.Channels/Diagnostics/DiagnosticsBufferedChannel.cs b/src/Elastic.Channels/Diagnostics/DiagnosticsBufferedChannel.cs
@@ -23,6 +23,12 @@ public DiagnosticsBufferedChannel(BufferOptions options, bool observeConcurrency
 	{
 	}
 
+	/// <inheritdoc cref="DiagnosticsBufferedChannel"/>
+	public DiagnosticsBufferedChannel(NoopChannelOptions options, string? name = null)
+		: base(options, new [] { new ChannelDiagnosticsListener<NoopEvent, NoopResponse>(name ?? nameof(DiagnosticsBufferedChannel)) })
+	{
+	}
+
 	private long _bufferMismatches;
 	/// <summary> Keeps track of the number of times the buffer size or the buffer offset was off</summary>
 	public long BufferMismatches => _bufferMismatches;

diff --git a/src/Elastic.Channels/Diagnostics/NoopBufferedChannel.cs b/src/Elastic.Channels/Diagnostics/NoopBufferedChannel.cs
@@ -21,7 +21,7 @@ public class NoopBufferedChannel
 	public class NoopEvent
 	{
 		/// <summary> An id marker for the noop event </summary>
-		public int? Id { get; set; }
+		public long? Id { get; set; }
 	}
 
 	/// <summary> Empty response for use with <see cref="NoopBufferedChannel"/> </summary>
@@ -34,22 +34,16 @@ public class NoopChannelOptions : ChannelOptionsBase<NoopEvent, NoopResponse>
 		public bool TrackConcurrency { get; set; }
 	}
 
-	/// <inheritdoc cref="NoopBufferedChannel"/>
-	public NoopBufferedChannel(NoopChannelOptions options) : base(options) { }
-
 	/// <inheritdoc cref="NoopBufferedChannel"/>
 	public NoopBufferedChannel(
-		BufferOptions options,
-		bool observeConcurrency = false
-	) : this(options, null, observeConcurrency)
-	{
-
-	}
+		NoopChannelOptions options,
+		ICollection<IChannelCallbacks<NoopEvent, NoopResponse>>? channelListeners = null
+	) : base(options, channelListeners) { }
 
 	/// <inheritdoc cref="NoopBufferedChannel"/>
 	public NoopBufferedChannel(
 		BufferOptions options,
-		ICollection<IChannelCallbacks<NoopEvent, NoopResponse>>? channelListeners,
+		ICollection<IChannelCallbacks<NoopEvent, NoopResponse>>? channelListeners = null,
 		bool observeConcurrency = false
 	) : base(new NoopChannelOptions { BufferOptions = options, TrackConcurrency = observeConcurrency }, channelListeners)
 	{