Transport does not own Semaphore, ConnectionSettings does

Author: Mpdreamz

src/Elasticsearch.Net/Configuration/ConnectionConfiguration.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Specialized;
 using System.ComponentModel;
+using System.Threading;
 
 namespace Elasticsearch.Net
 {
@@ -56,6 +57,9 @@ public ConnectionConfiguration(IConnectionPool connectionPool, IConnection conne
 	public abstract class ConnectionConfiguration<T> : IConnectionConfigurationValues, IHideObjectMembers
 		where T : ConnectionConfiguration<T>
 	{
+		private SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
+		SemaphoreSlim IConnectionConfigurationValues.BootstrapLock => this._semaphore;
+
 		private TimeSpan _requestTimeout;
 		TimeSpan IConnectionConfigurationValues.RequestTimeout => _requestTimeout;
 
@@ -316,6 +320,7 @@ protected virtual void DisposeManagedResources()
 		{
 			this._connectionPool?.Dispose();
 			this._connection?.Dispose();
+			this._semaphore?.Dispose();
 		}
 	}
 }

src/Elasticsearch.Net/Configuration/IConnectionConfigurationValues.cs

@@ -1,10 +1,14 @@
 using System;
 using System.Collections.Specialized;
+using System.Threading;
 
 namespace Elasticsearch.Net
 {
 	public interface IConnectionConfigurationValues : IDisposable
 	{
+		/// <summary> Provides a semaphoreslim to transport implementations that need to limit access to a resource</summary>
+		SemaphoreSlim BootstrapLock { get; }
+
 		/// <summary> The connection pool to use when talking with elasticsearch </summary>
 		IConnectionPool ConnectionPool { get; }
 

src/Elasticsearch.Net/Transport/Transport.cs

@@ -8,8 +8,6 @@ namespace Elasticsearch.Net
 	public class Transport<TConnectionSettings> : ITransport<TConnectionSettings>, IDisposable
 		where TConnectionSettings : IConnectionConfigurationValues
 	{
-		private readonly SemaphoreSlim _semaphore;
-
 		//TODO should all of these be public?
 		public TConnectionSettings Settings { get; }
 		public IDateTimeProvider DateTimeProvider { get; }
@@ -38,7 +36,6 @@ public Transport(
 			IMemoryStreamFactory memoryStreamFactory
 			)
 		{
-
 			configurationValues.ThrowIfNull(nameof(configurationValues));
 			configurationValues.ConnectionPool.ThrowIfNull(nameof(configurationValues.ConnectionPool));
 			configurationValues.Connection.ThrowIfNull(nameof(configurationValues.Connection));
@@ -48,15 +45,14 @@ IMemoryStreamFactory memoryStreamFactory
 			this.PipelineProvider = pipelineProvider ?? new RequestPipelineFactory();
 			this.DateTimeProvider = dateTimeProvider ?? Net.DateTimeProvider.Default;
 			this.MemoryStreamFactory = memoryStreamFactory ?? new MemoryStreamFactory();
-			this._semaphore = new SemaphoreSlim(1, 1);
 		}
 
 		public ElasticsearchResponse<TReturn> Request<TReturn>(HttpMethod method, string path, PostData<object> data = null, IRequestParameters requestParameters = null)
 			where TReturn : class
 		{
 			using (var pipeline = this.PipelineProvider.Create(this.Settings, this.DateTimeProvider, this.MemoryStreamFactory, requestParameters))
 			{
-				pipeline.FirstPoolUsage(this._semaphore);
+				pipeline.FirstPoolUsage(this.Settings.BootstrapLock);
 
 				var requestData = new RequestData(method, path, data, this.Settings, requestParameters, this.MemoryStreamFactory);
 				ElasticsearchResponse<TReturn> response = null;
@@ -113,7 +109,7 @@ public async Task<ElasticsearchResponse<TReturn>> RequestAsync<TReturn>(HttpMeth
 		{
 			using (var pipeline = this.PipelineProvider.Create(this.Settings, this.DateTimeProvider, this.MemoryStreamFactory, requestParameters))
 			{
-				await pipeline.FirstPoolUsageAsync(this._semaphore);
+				await pipeline.FirstPoolUsageAsync(this.Settings.BootstrapLock);
 
 				var requestData = new RequestData(method, path, data, this.Settings, requestParameters, this.MemoryStreamFactory);
 				ElasticsearchResponse<TReturn> response = null;
@@ -193,9 +189,6 @@ private static async Task PingAsync(IRequestPipeline pipeline, Node node, List<P
 
 		void IDisposable.Dispose() => this.DisposeManagedResources();
 
-		protected virtual void DisposeManagedResources()
-		{
-			this._semaphore?.Dispose();
-		}
+		protected virtual void DisposeManagedResources() { }
 	}
 }

src/Tests/ClientConcepts/ConnectionPooling/Sniffing/OnStartup.doc.cs

@@ -37,6 +37,35 @@ await audit.TraceCall(new ClientCall
 			});
 		}
 
+		[U] [SuppressMessage("AsyncUsage", "AsyncFixer001:Unnecessary async/await usage", Justification = "Its a test")]
+		public async Task ASniffOnStartupHappensOnce()
+		{
+			var audit = new Auditor(() => Framework.Cluster
+				.Nodes(10)
+				.Sniff(s => s.Fails(Always))
+				.Sniff(s => s.OnPort(9202).Succeeds(Always))
+				.SniffingConnectionPool()
+				.AllDefaults()
+			);
+
+			 await audit.TraceCalls(
+				 new ClientCall
+				 {
+					{ SniffOnStartup},
+					{ SniffFailure, 9200},
+					{ SniffFailure, 9201},
+					{ SniffSuccess, 9202},
+					{ PingSuccess , 9200},
+					{ HealthyResponse, 9200}
+				},
+				new ClientCall
+				{
+					{ PingSuccess, 9201},
+					{ HealthyResponse, 9201}
+				}
+			);
+		}
+
 		[U] [SuppressMessage("AsyncUsage", "AsyncFixer001:Unnecessary async/await usage", Justification = "Its a test")]
 		public async Task SniffOnStartUpTakesNewClusterState()
 		{

src/Tests/Framework/VirtualClustering/FixedPipelineFactory.cs

@@ -5,12 +5,14 @@ namespace Tests.Framework
 {
 	public class FixedPipelineFactory : IRequestPipelineFactory
 	{
-		public IConnectionSettingsValues Settings { get; }
-		public Transport<IConnectionSettingsValues> Transport { get; }
-		public IRequestPipeline Pipeline { get; }
+		private IConnectionSettingsValues Settings { get; }
+		private Transport<IConnectionSettingsValues> Transport =>
+			new Transport<IConnectionSettingsValues>(this.Settings, this, this.DateTimeProvider, this.MemoryStreamFactory);
+
+		private IDateTimeProvider DateTimeProvider { get; }
+		private MemoryStreamFactory MemoryStreamFactory { get; }
 
-		public IDateTimeProvider DateTimeProvider { get; }
-		public MemoryStreamFactory MemoryStreamFactory { get; }
+		public IRequestPipeline Pipeline { get; }
 
 		public ElasticClient Client => new ElasticClient(this.Transport);
 
@@ -21,7 +23,6 @@ public FixedPipelineFactory(IConnectionSettingsValues connectionSettings, IDateT
 
 			this.Settings = connectionSettings;
 			this.Pipeline = this.Create(this.Settings, this.DateTimeProvider, this.MemoryStreamFactory, new SearchRequestParameters());
-			this.Transport = new Transport<IConnectionSettingsValues>(this.Settings, this, this.DateTimeProvider, this.MemoryStreamFactory);
 		}
 
 		public IRequestPipeline Create(IConnectionConfigurationValues configurationValues, IDateTimeProvider dateTimeProvider, IMemoryStreamFactory memorystreamFactory, IRequestParameters requestParameters) => 

src/Tests/Framework/VirtualClustering/VirtualizedCluster.cs

@@ -1,42 +1,41 @@
-using System;
-using System.Threading.Tasks;
+using System;
+using System.Threading.Tasks;
 using Elasticsearch.Net;
 using Nest;
-using Tests.Framework.MockData;
-
-namespace Tests.Framework
-{
-	public class VirtualizedCluster
-	{
-		private readonly ElasticClient _client;
-		private readonly VirtualCluster _cluster;
-		private readonly IConnectionPool _connectionPool;
-		private readonly TestableDateTimeProvider _dateTimeProvider;
-		private readonly ConnectionSettings _settings;
-		public FixedPipelineFactory _fixedRequestPipeline;
-
-		public IConnectionPool ConnectionPool => this._client.ConnectionSettings.ConnectionPool;
-
-		public VirtualizedCluster(VirtualCluster cluster, IConnectionPool pool, TestableDateTimeProvider dateTimeProvider, ConnectionSettings settings)
-		{
-			this._dateTimeProvider = dateTimeProvider;
-			_settings = settings;
-			this._fixedRequestPipeline = new FixedPipelineFactory(settings, this._dateTimeProvider);
-			this._client = this._fixedRequestPipeline.Client;
-
-			this._cluster = cluster;
-			this._connectionPool = pool;
-		}
-
-		public ISearchResponse<Project> ClientCall(Func<RequestConfigurationDescriptor, IRequestConfiguration> requestOverrides = null) 
-			=> this._client.Search<Project>(s => s.RequestConfiguration(requestOverrides));
-
-		public async Task<ISearchResponse<Project>> ClientCallAsync(Func<RequestConfigurationDescriptor, IRequestConfiguration> requestOverrides = null) => 
-			await this._client.SearchAsync<Project>(s => s.RequestConfiguration(requestOverrides));
-
-		public void ChangeTime(Func<DateTime, DateTime> change) => _dateTimeProvider.ChangeTime(change);
-
-		public void ClientThrows(bool throws) => _settings.ThrowExceptions(throws);
-	}
-
+using Tests.Framework.MockData;
+
+namespace Tests.Framework
+{
+	public class VirtualizedCluster
+	{
+		private ElasticClient Client => this._fixedRequestPipeline?.Client;
+		private readonly VirtualCluster _cluster;
+		private readonly IConnectionPool _connectionPool;
+		private readonly TestableDateTimeProvider _dateTimeProvider;
+		private readonly ConnectionSettings _settings;
+		public FixedPipelineFactory _fixedRequestPipeline;
+
+		public IConnectionPool ConnectionPool => this.Client.ConnectionSettings.ConnectionPool;
+
+		public VirtualizedCluster(VirtualCluster cluster, IConnectionPool pool, TestableDateTimeProvider dateTimeProvider, ConnectionSettings settings)
+		{
+			this._dateTimeProvider = dateTimeProvider;
+			this._settings = settings;
+			this._fixedRequestPipeline = new FixedPipelineFactory(settings, this._dateTimeProvider);
+
+			this._cluster = cluster;
+			this._connectionPool = pool;
+		}
+
+		public ISearchResponse<Project> ClientCall(Func<RequestConfigurationDescriptor, IRequestConfiguration> requestOverrides = null) =>
+			this.Client.Search<Project>(s => s.RequestConfiguration(requestOverrides));
+
+		public async Task<ISearchResponse<Project>> ClientCallAsync(Func<RequestConfigurationDescriptor, IRequestConfiguration> requestOverrides = null) => 
+			await this.Client.SearchAsync<Project>(s => s.RequestConfiguration(requestOverrides));
+
+		public void ChangeTime(Func<DateTime, DateTime> change) => _dateTimeProvider.ChangeTime(change);
+
+		public void ClientThrows(bool throws) => _settings.ThrowExceptions(throws);
+	}
+
 }