Enable DisableDirectStreaming on a per request basis

Closes #2166
(cherry picked from commit 8196a60)

Author: russcam

src/Elasticsearch.Net/Configuration/RequestConfiguration.cs

@@ -48,6 +48,11 @@ public interface IRequestConfiguration
 		/// </summary>
 		bool? DisablePing { get; set; }
 
+		/// <summary>
+		/// Whether to buffer the request and response bytes for the call
+		/// </summary>
+		bool? DisableDirectStreaming { get; set; }
+
 		/// <summary>
 		/// Treat the following statuses (on top of the 200 range) NOT as error.
 		/// </summary>
@@ -81,6 +86,7 @@ public class RequestConfiguration : IRequestConfiguration
 		public Uri ForceNode { get; set; }
 		public bool? DisableSniff { get; set; }
 		public bool? DisablePing { get; set; }
+		public bool? DisableDirectStreaming { get; set; }
 		public IEnumerable<int> AllowedStatusCodes { get; set; }
 		public BasicAuthenticationCredentials BasicAuthenticationCredentials { get; set; }
 		public bool EnableHttpPipelining { get; set; } = true;
@@ -94,7 +100,6 @@ public class RequestConfiguration : IRequestConfiguration
 
 	public class RequestConfigurationDescriptor : IRequestConfiguration
 	{
-
 		private IRequestConfiguration Self => this;
 		TimeSpan? IRequestConfiguration.RequestTimeout { get; set; }
 		TimeSpan? IRequestConfiguration.PingTimeout { get; set; }
@@ -105,6 +110,7 @@ public class RequestConfigurationDescriptor : IRequestConfiguration
 		Uri IRequestConfiguration.ForceNode { get; set; }
 		bool? IRequestConfiguration.DisableSniff { get; set; }
 		bool? IRequestConfiguration.DisablePing { get; set; }
+		bool? IRequestConfiguration.DisableDirectStreaming { get; set; }
 		IEnumerable<int> IRequestConfiguration.AllowedStatusCodes { get; set; }
 		BasicAuthenticationCredentials IRequestConfiguration.BasicAuthenticationCredentials { get; set; }
 		bool IRequestConfiguration.EnableHttpPipelining { get; set; } = true;
@@ -120,6 +126,7 @@ public RequestConfigurationDescriptor(IRequestConfiguration config)
 			Self.ForceNode = config?.ForceNode;
 			Self.DisableSniff = config?.DisableSniff;
 			Self.DisablePing = config?.DisablePing;
+			Self.DisableDirectStreaming = config?.DisableDirectStreaming;
 			Self.AllowedStatusCodes = config?.AllowedStatusCodes;
 			Self.BasicAuthenticationCredentials = config?.BasicAuthenticationCredentials;
 			Self.EnableHttpPipelining = config?.EnableHttpPipelining ?? true;
@@ -184,6 +191,12 @@ public RequestConfigurationDescriptor DisablePing(bool? disable = true)
 			return this;
 		}
 
+		public RequestConfigurationDescriptor DisableDirectStreaming(bool? disable = true)
+		{
+			Self.DisableDirectStreaming = disable;
+			return this;
+		}
+
 		public RequestConfigurationDescriptor ForceNode(Uri uri)
 		{
 			Self.ForceNode = uri;

src/Elasticsearch.Net/Configuration/Security/BasicAuthenticationCredentials.cs

@@ -8,9 +8,6 @@ public class BasicAuthenticationCredentials
 
 		public string Password { get; set; }
 
-		public override string ToString()
-		{
-			return this.Username + ":" + this.Password;
-		}
+		public override string ToString() => $"{this.Username}:{this.Password}";
 	}
 }

src/Elasticsearch.Net/Domain/RequestParameters/IRequestParameters.cs

@@ -27,7 +27,5 @@ public interface IRequestParameters
 		TOut GetQueryStringValue<TOut>(string name);
 
 		void AddQueryStringValue(string name, object value);
-
-
 	}
 }

src/Elasticsearch.Net/Transport/Pipeline/RequestData.cs

@@ -3,7 +3,6 @@
 using System.Collections.Specialized;
 using System.IO;
 using System.Linq;
-using System.Threading;
 using Purify;
 
 namespace Elasticsearch.Net
@@ -39,16 +38,15 @@ public class RequestData
 		public string ProxyUsername { get; }
 		public string ProxyPassword { get; }
 		public bool DisableAutomaticProxyDetection { get; }
+
 		public BasicAuthenticationCredentials BasicAuthorizationCredentials { get; }
 		public IEnumerable<int> AllowedStatusCodes { get; }
 		public Func<IApiCallDetails, Stream, object> CustomConverter { get; private set; }
 		public IConnectionConfigurationValues ConnectionSettings { get; }
 		public IMemoryStreamFactory MemoryStreamFactory { get; }
 
 		public RequestData(HttpMethod method, string path, PostData<object> data, IConnectionConfigurationValues global, IRequestParameters local, IMemoryStreamFactory memoryStreamFactory)
-#pragma warning disable CS0618 // Type or member is obsolete
-			: this(method, path, data, global, (IRequestConfiguration)local?.RequestConfiguration, memoryStreamFactory)
-#pragma warning restore CS0618 // Type or member is obsolete
+			: this(method, path, data, global, local?.RequestConfiguration, memoryStreamFactory)
 		{
 			this.CustomConverter = local?.DeserializationOverride;
 			this.Path = this.CreatePathWithQueryStrings(path, this.ConnectionSettings, local);
@@ -66,13 +64,17 @@ private RequestData(
 			this.MemoryStreamFactory = memoryStreamFactory;
 			this.Method = method;
 			this.PostData = data;
+
+			if (data != null)
+				data.DisableDirectStreaming = local?.DisableDirectStreaming ?? global.DisableDirectStreaming;
+
 			this.Path = this.CreatePathWithQueryStrings(path, this.ConnectionSettings, null);
 
-			this.Pipelined = global.HttpPipeliningEnabled || (local?.EnableHttpPipelining).GetValueOrDefault(false);
+			this.Pipelined = local?.EnableHttpPipelining ?? global.HttpPipeliningEnabled;
 			this.HttpCompression = global.EnableHttpCompression;
 			this.ContentType = local?.ContentType ?? MimeType;
 			this.Accept = local?.Accept ?? MimeType;
-			this.Headers = new NameValueCollection(global.Headers ?? new NameValueCollection());
+			this.Headers = global.Headers != null ? new NameValueCollection(global.Headers) : new NameValueCollection();
 			this.RunAs = local?.RunAs;
 
 			this.RequestTimeout = local?.RequestTimeout ?? global.RequestTimeout;

src/Elasticsearch.Net/Transport/Pipeline/ResponseBuilder.cs

@@ -1,8 +1,5 @@
 using System;
-using System.Collections.Generic;
 using System.IO;
-using System.Linq;
-using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -12,18 +9,23 @@ public class ResponseBuilder<TReturn>
 		where TReturn : class
 	{
 		public const int BufferSize = 8096;
+		private static readonly VoidResponse Void = new VoidResponse();
+		private static readonly IDisposable EmptyDisposable = new MemoryStream();
+
+		private readonly RequestData _requestData;
+		private readonly CancellationToken _cancellationToken;
+		private readonly bool _disableDirectStreaming;
 
 		public Exception Exception { get; set; }
 		public int? StatusCode { get; set; }
 		public Stream Stream { get; set; }
 
-		private RequestData _requestData;
-		private CancellationToken _cancellationToken;
-
 		public ResponseBuilder(RequestData requestData, CancellationToken cancellationToken = default(CancellationToken))
 		{
 			_requestData = requestData;
 			_cancellationToken = cancellationToken;
+			_disableDirectStreaming =
+				this._requestData.PostData?.DisableDirectStreaming ?? this._requestData.ConnectionSettings.DisableDirectStreaming;
 		}
 
 		public ElasticsearchResponse<TReturn> ToResponse()
@@ -56,8 +58,6 @@ private ElasticsearchResponse<TReturn> Initialize(int? statusCode, Exception exc
 			return response;
 		}
 
-		private static IDisposable EmptyDisposable = new MemoryStream();
-
 		private void SetBody(ElasticsearchResponse<TReturn> response, Stream stream)
 		{
 			byte[] bytes = null;
@@ -84,7 +84,7 @@ private void SetBody(ElasticsearchResponse<TReturn> response, Stream stream)
 					ServerError serverError;
 					if (ServerError.TryCreate(stream, out serverError))
 						response.ServerError = serverError;
-					if (this._requestData.ConnectionSettings.DisableDirectStreaming)
+					if (_disableDirectStreaming)
 						response.ResponseBodyInBytes = bytes;
 				}
 			}
@@ -114,7 +114,7 @@ private async Task SetBodyAsync(ElasticsearchResponse<TReturn> response, Stream
 				else if (response.HttpStatusCode != null)
 				{
 					response.ServerError = await ServerError.TryCreateAsync(stream, this._cancellationToken).ConfigureAwait(false);
-					if (this._requestData.ConnectionSettings.DisableDirectStreaming)
+					if (_disableDirectStreaming)
 						response.ResponseBodyInBytes = bytes;
 				}
 			}
@@ -130,7 +130,7 @@ private void Finalize(ElasticsearchResponse<TReturn> response)
 		}
 
 		private bool NeedsToEagerReadStream() =>
-			this._requestData.ConnectionSettings.DisableDirectStreaming || typeof(TReturn) == typeof(string) || typeof(TReturn) == typeof(byte[]);
+			_disableDirectStreaming || typeof(TReturn) == typeof(string) || typeof(TReturn) == typeof(byte[]);
 
 		private byte[] SwapStreams(ref Stream responseStream, ref MemoryStream ms)
 		{
@@ -139,13 +139,12 @@ private byte[] SwapStreams(ref Stream responseStream, ref MemoryStream ms)
 			responseStream = ms;
 			responseStream.Position = 0;
 			return bytes;
-
 		}
 
 		private bool SetSpecialTypes(Stream responseStream, ElasticsearchResponse<TReturn> cs, byte[] bytes)
 		{
 			var setSpecial = true;
-			if (this._requestData.ConnectionSettings.DisableDirectStreaming)
+			if (_disableDirectStreaming)
 				cs.ResponseBodyInBytes = bytes;
 			var returnType = typeof(TReturn);
 			if (returnType == typeof(string))
@@ -167,12 +166,10 @@ private bool SetSpecialTypes(Stream responseStream, ElasticsearchResponse<TRetur
 
 		private void SetStreamResult(ElasticsearchResponse<Stream> result, Stream response) => result.Body = response;
 
-		private static VoidResponse _void = new VoidResponse();
-
 		private void SetVoidResult(ElasticsearchResponse<VoidResponse> result, Stream response)
 		{
 			response.Dispose();
-			result.Body = _void;
+			result.Body = Void;
 		}
 	}
 }

src/Elasticsearch.Net/Transport/PostData.cs

@@ -1,7 +1,5 @@
 using System.Collections.Generic;
-using Purify;
 using System.IO;
-using System.Reflection;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -29,6 +27,7 @@ public class PostData<T> : IPostData<T>
 		private readonly IEnumerable<object> _enumerableOfObject;
 		private readonly T _serializable;
 
+		public bool? DisableDirectStreaming { get; set; }
 		public byte[] WrittenBytes { get; private set; }
 		public PostType Type { get; }
 
@@ -64,7 +63,8 @@ public PostData(T item)
 		public void Write(Stream writableStream, IConnectionConfigurationValues settings)
 		{
 			var indent = settings.PrettyJson ? SerializationFormatting.Indented : SerializationFormatting.None;
-			MemoryStream ms = null; Stream stream = null;
+			MemoryStream ms = null;
+			Stream stream = null;
 			switch (Type)
 			{
 				case PostType.ByteArray:
@@ -79,7 +79,7 @@ public void Write(Stream writableStream, IConnectionConfigurationValues settings
 				case PostType.EnumerableOfObject:
 					if (!_enumerableOfObject.HasAny()) return;
 
-					if (settings.DisableDirectStreaming)
+					if (this.DisableDirectStreaming ?? settings.DisableDirectStreaming)
 					{
 						ms = new MemoryStream();
 						stream = ms;
@@ -93,7 +93,7 @@ public void Write(Stream writableStream, IConnectionConfigurationValues settings
 					break;
 				case PostType.Serializable:
 					stream = writableStream;
-					if (settings.DisableDirectStreaming)
+					if (this.DisableDirectStreaming ?? settings.DisableDirectStreaming)
 					{
 						ms = new MemoryStream();
 						stream = ms;
@@ -127,7 +127,7 @@ public async Task WriteAsync(Stream writableStream, IConnectionConfigurationValu
 					break;
 				case PostType.EnumerableOfObject:
 					if (!_enumerableOfObject.HasAny()) return;
-					if (settings.DisableDirectStreaming)
+					if (this.DisableDirectStreaming ?? settings.DisableDirectStreaming)
 					{
 						ms = new MemoryStream();
 						stream = ms;
@@ -141,7 +141,7 @@ public async Task WriteAsync(Stream writableStream, IConnectionConfigurationValu
 					break;
 				case PostType.Serializable:
 					stream = writableStream;
-					if (settings.DisableDirectStreaming)
+					if (this.DisableDirectStreaming ?? settings.DisableDirectStreaming)
 					{
 						ms = new MemoryStream();
 						stream = ms;

src/Nest/CommonAbstractions/Request/RequestBase.cs

@@ -69,7 +69,6 @@ public IRequestConfiguration RequestConfiguration
 		}
 	}
 
-
 	public abstract class RequestDescriptorBase<TDescriptor, TParameters, TInterface> : RequestBase<TParameters>, IDescriptor
 		where TDescriptor : RequestDescriptorBase<TDescriptor, TParameters, TInterface>, TInterface
 		where TParameters : FluentRequestParameters<TParameters>, new()

src/Tests/ClientConcepts/LowLevel/DirectStreaming.cs

@@ -1,11 +1,6 @@
-using Elasticsearch.Net;
-using FluentAssertions;
+using FluentAssertions;
 using Nest;
 using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
 using Tests.Framework;
 
 namespace Tests.ClientConcepts.LowLevel
@@ -79,5 +74,39 @@ public void EnableDirectStreamingOnSuccess()
 			response = client.SearchAsync<object>(s => s).Result;
 			assert(response);
 		}
+
+		[U]
+		public void DisableDirectStreamingOnRequest()
+		{
+			Action<IResponse> assert = r =>
+			{
+				r.ApiCall.Should().NotBeNull();
+				r.ApiCall.RequestBodyInBytes.Should().NotBeNull();
+				r.ApiCall.ResponseBodyInBytes.Should().NotBeNull();
+			};
+
+			var client = TestClient.GetFixedReturnClient(new { });
+			var response = client.Search<object>(s => s.RequestConfiguration(r => r.DisableDirectStreaming()));
+			assert(response);
+			response = client.SearchAsync<object>(s => s.RequestConfiguration(r => r.DisableDirectStreaming())).Result;
+			assert(response);
+		}
+
+		[U]
+		public void EnableDirectStreamingOnRequest()
+		{
+			Action<IResponse> assert = r =>
+			{
+				r.ApiCall.Should().NotBeNull();
+				r.ApiCall.RequestBodyInBytes.Should().BeNull();
+				r.ApiCall.ResponseBodyInBytes.Should().BeNull();
+			};
+
+			var client = TestClient.GetFixedReturnClient(new { }, 200, c => c.DisableDirectStreaming());
+			var response = client.Search<object>(s => s.RequestConfiguration(r => r.DisableDirectStreaming(false)));
+			assert(response);
+			response = client.SearchAsync<object>(s => s.RequestConfiguration(r => r.DisableDirectStreaming(false))).Result;
+			assert(response);
+		}
 	}
 }