<commit_before>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Elasticsearch.Net.Serialization;
using Elasticsearch.Net.ConnectionPool;
using Elasticsearch.Net.Providers;
using Elasticsearch.Net.Connection.Configuration;
using Purify;
using System.IO;
using System.Collections.Specialized;
using System.Threading;
using Elasticsearch.Net.Extensions;

namespace Elasticsearch.Net.Connection
{
	public class Transport<TConnectionSettings> : ITransport<TConnectionSettings>
		where TConnectionSettings : IConnectionConfigurationValues
	{
		private SemaphoreSlim _semaphore;

		//TODO discuss which should be public
		public TConnectionSettings Settings { get; }
		public IDateTimeProvider DateTimeProvider { get; }
		public IMemoryStreamFactory MemoryStreamFactory { get; }
		public IRequestPipelineFactory PipelineProvider { get; private set; }

		/// <summary>
		/// Transport coordinates the client requests over the connection pool nodes and is in charge of falling over on different nodes 
		/// </summary>
		/// <param name="configurationValues">The connectionsettings to use for this transport</param>
		public Transport(TConnectionSettings configurationValues)
			: this(configurationValues, null, null, null)
		{ }

		/// <summary>
		/// Transport coordinates the client requests over the connection pool nodes and is in charge of falling over on different nodes 
		/// </summary>
		/// <param name="configurationValues">The connectionsettings to use for this transport</param>
		/// <param name="requestPipelineProvider">In charge of create a new pipeline, safe to pass null to use the default</param>
		/// <param name="dateTimeProvider">The date time proved to use, safe to pass null to use the default</param>
		/// <param name="memoryStreamFactory">The memory stream provider to use, safe to pass null to use the default</param>
		public Transport(
			TConnectionSettings configurationValues,
			IRequestPipelineFactory pipelineProvider,
			IDateTimeProvider dateTimeProvider,
			IMemoryStreamFactory memoryStreamFactory
			)
		{
			configurationValues.ThrowIfNull(nameof(configurationValues));
			configurationValues.ConnectionPool.ThrowIfNull(nameof(configurationValues.ConnectionPool));
			configurationValues.Connection.ThrowIfNull(nameof(configurationValues.Connection));
			configurationValues.Serializer.ThrowIfNull(nameof(configurationValues.Serializer));

			this.Settings = configurationValues;
			this.PipelineProvider = pipelineProvider ?? new RequestPipelineFactory();
			this.DateTimeProvider = dateTimeProvider ?? Providers.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);

				var requestData = new RequestData(method, path, data, this.Settings, requestParameters, this.MemoryStreamFactory);
				ElasticsearchResponse<TReturn> response = null;

				var exceptions = new List<ConnectionException>();
				foreach (var node in pipeline.NextNode())
				{
					requestData.Node = node;
					try
					{
						pipeline.SniffOnStaleCluster();
						pipeline.Ping(node);
						response = pipeline.CallElasticsearch<TReturn>(requestData);
					}
					catch (ConnectionException exception) when (!exception.Recoverable)
					{
						pipeline.MarkDead(node);
						exception.RethrowKeepingStackTrace();
					}
					catch (ConnectionException exception)
					{
						pipeline.MarkDead(node);
						exceptions.Add(exception);
					}
					if (response != null && response.SuccessOrKnownError)
					{
						if (this.Settings.ThrowOnElasticsearchServerExceptions && !response.Success)
							throw new ElasticsearchServerException(response.ErrorDetails.Info.Reason, response.ErrorDetails);
						pipeline.MarkAlive(node);
						return response;
					}
				}
				pipeline.BadResponse(ref response, requestData, exceptions);
				return response;
			}
		}

		public async Task<ElasticsearchResponse<TReturn>> RequestAsync<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))
			{
				await pipeline.FirstPoolUsageAsync(this._semaphore);

				var requestData = new RequestData(method, path, data, this.Settings, requestParameters, this.MemoryStreamFactory);
				ElasticsearchResponse<TReturn> response = null;

				var exceptions = new List<ConnectionException>();
				foreach (var node in pipeline.NextNode())
				{
					requestData.Node = node;
					try
					{
						await pipeline.SniffOnStaleClusterAsync();
						await pipeline.PingAsync(node);
						response = await pipeline.CallElasticsearchAsync<TReturn>(requestData);
					}
					catch (ConnectionException exception) when (!exception.Recoverable)
					{
						pipeline.MarkDead(node);
						exception.RethrowKeepingStackTrace();
					}
					catch (ConnectionException exception)
					{
						pipeline.MarkDead(node);
						exceptions.Add(exception);
					}
					if (response != null && response.SuccessOrKnownError)
					{
						if (this.Settings.ThrowOnElasticsearchServerExceptions)
							throw response.OriginalException;
						pipeline.MarkAlive(node);<commit_msg>Fix async path for throwing ElasticsearchServerException<commit_after>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Elasticsearch.Net.Serialization;
using Elasticsearch.Net.ConnectionPool;
using Elasticsearch.Net.Providers;
using Elasticsearch.Net.Connection.Configuration;
using Purify;
using System.IO;
using System.Collections.Specialized;
using System.Threading;
using Elasticsearch.Net.Extensions;

namespace Elasticsearch.Net.Connection
{
	public class Transport<TConnectionSettings> : ITransport<TConnectionSettings>
		where TConnectionSettings : IConnectionConfigurationValues
	{
		private SemaphoreSlim _semaphore;

		//TODO discuss which should be public
		public TConnectionSettings Settings { get; }
		public IDateTimeProvider DateTimeProvider { get; }
		public IMemoryStreamFactory MemoryStreamFactory { get; }
		public IRequestPipelineFactory PipelineProvider { get; private set; }

		/// <summary>
		/// Transport coordinates the client requests over the connection pool nodes and is in charge of falling over on different nodes 
		/// </summary>
		/// <param name="configurationValues">The connectionsettings to use for this transport</param>
		public Transport(TConnectionSettings configurationValues)
			: this(configurationValues, null, null, null)
		{ }

		/// <summary>
		/// Transport coordinates the client requests over the connection pool nodes and is in charge of falling over on different nodes 
		/// </summary>
		/// <param name="configurationValues">The connectionsettings to use for this transport</param>
		/// <param name="requestPipelineProvider">In charge of create a new pipeline, safe to pass null to use the default</param>
		/// <param name="dateTimeProvider">The date time proved to use, safe to pass null to use the default</param>
		/// <param name="memoryStreamFactory">The memory stream provider to use, safe to pass null to use the default</param>
		public Transport(
			TConnectionSettings configurationValues,
			IRequestPipelineFactory pipelineProvider,
			IDateTimeProvider dateTimeProvider,
			IMemoryStreamFactory memoryStreamFactory
			)
		{
			configurationValues.ThrowIfNull(nameof(configurationValues));
			configurationValues.ConnectionPool.ThrowIfNull(nameof(configurationValues.ConnectionPool));
			configurationValues.Connection.ThrowIfNull(nameof(configurationValues.Connection));
			configurationValues.Serializer.ThrowIfNull(nameof(configurationValues.Serializer));

			this.Settings = configurationValues;
			this.PipelineProvider = pipelineProvider ?? new RequestPipelineFactory();
			this.DateTimeProvider = dateTimeProvider ?? Providers.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);

				var requestData = new RequestData(method, path, data, this.Settings, requestParameters, this.MemoryStreamFactory);
				ElasticsearchResponse<TReturn> response = null;

				var exceptions = new List<ConnectionException>();
				foreach (var node in pipeline.NextNode())
				{
					requestData.Node = node;
					try
					{
						pipeline.SniffOnStaleCluster();
						pipeline.Ping(node);
						response = pipeline.CallElasticsearch<TReturn>(requestData);
					}
					catch (ConnectionException exception) when (!exception.Recoverable)
					{
						pipeline.MarkDead(node);
						exception.RethrowKeepingStackTrace();
					}
					catch (ConnectionException exception)
					{
						pipeline.MarkDead(node);
						exceptions.Add(exception);
					}
					if (response != null && response.SuccessOrKnownError)
					{
						if (this.Settings.ThrowOnElasticsearchServerExceptions && !response.Success)
							throw new ElasticsearchServerException(response.ErrorDetails.Info.Reason, response.ErrorDetails);
						pipeline.MarkAlive(node);
						return response;
					}
				}
				pipeline.BadResponse(ref response, requestData, exceptions);
				return response;
			}
		}

		public async Task<ElasticsearchResponse<TReturn>> RequestAsync<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))
			{
				await pipeline.FirstPoolUsageAsync(this._semaphore);

				var requestData = new RequestData(method, path, data, this.Settings, requestParameters, this.MemoryStreamFactory);
				ElasticsearchResponse<TReturn> response = null;

				var exceptions = new List<ConnectionException>();
				foreach (var node in pipeline.NextNode())
				{
					requestData.Node = node;
					try
					{
						await pipeline.SniffOnStaleClusterAsync();
						await pipeline.PingAsync(node);
						response = await pipeline.CallElasticsearchAsync<TReturn>(requestData);
					}
					catch (ConnectionException exception) when (!exception.Recoverable)
					{
						pipeline.MarkDead(node);
						exception.RethrowKeepingStackTrace();
					}
					catch (ConnectionException exception)
					{
						pipeline.MarkDead(node);
						exceptions.Add(exception);
					}
					if (response != null && response.SuccessOrKnownError)
					{
						if (this.Settings.ThrowOnElasticsearchServerExceptions && !response.Success)
							throw new ElasticsearchServerException(response.ErrorDetails.Info.Reason, response.ErrorDetails);
						pipeline.MarkAlive(node);
{'fraction_non_alphanumeric': 0.08817946505608283, 'fraction_numerical': 0.0003451251078515962, 'mean_word_length': 4.556567593480345, 'pattern_counts': {'":': 0, '<': 61, '<?xml version=': 0, '>': 61, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3823351', 'n_tokens_mistral': 3573, 'n_tokens_neox': 2982, 'n_words': 743}