Support specifying one or many types as part of Reindexing

Current implementation takes a generic ReindexDescriptor<T> but specifies all types when performing the scan and scroll for retrieving documents to reindex. The type of T should be used as the type of documents to target in the reindex operation, with methods exposed to allow the consumer to specify other other types / all types.
See #1433

Author: russcam

src/Nest/Document/Multiple/Reindex/ReindexObservable.cs

@@ -54,7 +54,7 @@ private void Reindex(IObserver<IReindexResponse<T>> observer)
 			var searchResult = this._client.Search<T>(
 				s => s
 					.Index(fromIndex)
-					.Type(Types.All)
+					.Type(this._reindexRequest.Type)
 					.From(0)
 					.Size(size)
 					.Query(q=>this._reindexRequest.Query)

src/Nest/Document/Multiple/Reindex/ReindexRequest.cs

@@ -6,6 +6,7 @@ public interface IReindexRequest
 	{
 		IndexName To { get; set; }
 		IndexName From { get; set; }
+		Types Type { get; set; }
 		Time Scroll { get; set; }
 		int? Size { get; set; }
 
@@ -15,26 +16,30 @@ public interface IReindexRequest
 
 		IPutMappingRequest PutMappingRequest { get; set; } 
 	}
+
 	public class ReindexRequest : IReindexRequest
 	{
 		public IndexName To { get; set; }
 		public IndexName From { get; set; }
+		public Types Type { get; set; }
 		public Time Scroll { get; set; }
 		public int? Size { get; set; }
 		public QueryContainer Query { get; set; }
 		public ICreateIndexRequest CreateIndexRequest { get; set; }
 		public IPutMappingRequest PutMappingRequest { get; set; } 
-		public ReindexRequest(IndexName from, IndexName to)
+		public ReindexRequest(IndexName from, IndexName to, Types type)
 		{
 			this.To = to;
 			this.From = from;
+			this.Type = type;
 		}
 	}
 
 	public class ReindexDescriptor<T> : DescriptorBase<ReindexDescriptor<T>, IReindexRequest>, IReindexRequest where T : class
 	{
 		IndexName IReindexRequest.To { get; set; }
 		IndexName IReindexRequest.From { get; set; }
+		Types IReindexRequest.Type { get; set; }
 		Time IReindexRequest.Scroll { get; set; }
 		int? IReindexRequest.Size { get; set; }
 		QueryContainer IReindexRequest.Query { get; set; }
@@ -43,7 +48,9 @@ public class ReindexDescriptor<T> : DescriptorBase<ReindexDescriptor<T>, IReinde
 
 		public ReindexDescriptor(IndexName from, IndexName to)
 		{
-			Assign(a => a.From = from).Assign(a => a.To = to);
+			Assign(a => a.From = from)
+			.Assign(a => a.To = to)
+			.Assign(a => a.Type = typeof(T));
 		}
 
 		/// <summary>
@@ -75,6 +82,16 @@ public ReindexDescriptor<T> Query(Func<QueryContainerDescriptor<T>, QueryContain
 		/// </summary>
 		public ReindexDescriptor<T> Query(QueryContainer query) => Assign(a => a.Query = query);
 
+		/// <summary>
+		/// Specify the document types to reindex. By default, will be <typeparamref name="T"/>  
+		/// </summary>
+		public ReindexDescriptor<T> Type(Types type) => Assign(a => a.Type = type);
+
+		/// <summary>
+		/// Reindex all document types.
+		/// </summary>
+		public ReindexDescriptor<T> AllTypes() => this.Type(Types.AllTypes); 
+
 		/// <summary>
 		/// CreateIndex selector, will be passed the a descriptor initialized with the settings from
 		/// the index we're reindexing from

src/Nest/Document/Multiple/Reindex/ReindexResponse.cs

@@ -39,13 +39,8 @@ public class ReindexResponse<T> : IReindexResponse<T> where T : class
 
 		public int Scroll { get; internal set; }
 
-		public bool IsValid
-		{
-			get
-			{
-				return (this.BulkResponse != null && this.BulkResponse.IsValid
-					&& this.SearchResponse != null && this.SearchResponse.IsValid);
-			}
-		}
+		public bool IsValid => 
+			this.BulkResponse != null && this.BulkResponse.IsValid
+			&& this.SearchResponse != null && this.SearchResponse.IsValid;
 	}
 }

src/Tests/Document/Multiple/Reindex/ReindexApiTests.cs

@@ -1,14 +1,14 @@
 using System;
 using System.Linq;
 using System.Threading;
+using Elasticsearch.Net;
 using FluentAssertions;
 using Nest;
 using Tests.Framework;
 using Tests.Framework.Integration;
 using Tests.Framework.MockData;
 using Xunit;
 using static Nest.Infer;
-using Elasticsearch.Net;
 
 namespace Tests.Document.Multiple.Reindex
 {
@@ -26,10 +26,13 @@ public override void Boostrap()
 	[Collection(IntegrationContext.Reindex)]
 	public class ReindexApiTests : SerializationTestBase
 	{
-		private readonly IObservable<IReindexResponse<ILazyDocument>> _reindexResult;
+		private readonly IObservable<IReindexResponse<ILazyDocument>> _reindexManyTypesResult;
+		private readonly IObservable<IReindexResponse<Project>> _reindexSingleTypeResult;
 		private readonly IElasticClient _client;
 
-		private static string NewIndexName { get; } = $"project-copy-{Guid.NewGuid().ToString("N").Substring(8)}";
+		private static string NewManyTypesIndexName { get; } = $"project-copy-{Guid.NewGuid().ToString("N").Substring(8)}";
+
+		private static string NewSingleTypeIndexName { get; } = $"project-copy-{Guid.NewGuid().ToString("N").Substring(8)}";
 
 		private static string IndexName { get; } = "project";
 
@@ -64,27 +67,33 @@ public ReindexApiTests(ReindexCluster cluster, EndpointUsage usage)
 			bulkResult.IsValid.Should().BeTrue();
 
 			this._client.Refresh(IndexName);
-			this._reindexResult = this._client.Reindex<ILazyDocument>(IndexName, NewIndexName, r=>r);
+
+			this._reindexManyTypesResult = this._client.Reindex<ILazyDocument>(IndexName, NewManyTypesIndexName, r => r.AllTypes());
+			this._reindexSingleTypeResult = this._client.Reindex<Project>(IndexName, NewSingleTypeIndexName);
 		}
 
 		[I] public void ReturnsExpectedResponse()
 		{
-			var handle = new ManualResetEvent(false);
-			var observer = new ReindexObserver<ILazyDocument>(
+			var handles = new[]
+			{
+				new ManualResetEvent(false),
+				new ManualResetEvent(false)
+			};
+
+			var manyTypesObserver = new ReindexObserver<ILazyDocument>(
 					onError: (e) => { throw e; },
 					completed: () =>
 					{
-						var refresh = this._client.Refresh(NewIndexName);
-
+						var refresh = this._client.Refresh(NewManyTypesIndexName);
 						var originalIndexCount = this._client.Count<CommitActivity>(c => c.Index(IndexName));
-						var newIndexCount = this._client.Count<CommitActivity>(c => c.Index(NewIndexName));
+						var newIndexCount = this._client.Count<CommitActivity>(c => c.Index(NewManyTypesIndexName));
 
 						originalIndexCount.Count.Should().BeGreaterThan(0).And.Be(newIndexCount.Count);
 
 						var scroll = "20s";
 
 						var searchResult = this._client.Search<CommitActivity>(s => s
-							.Index(NewIndexName)
+							.Index(NewManyTypesIndexName)
 							.From(0)
 							.Size(100)
 							.Query(q => q.MatchAll())
@@ -103,12 +112,31 @@ [I] public void ReturnsExpectedResponse()
 								hit.Routing.Should().NotBeNullOrEmpty();
 							}
 						} while (searchResult.IsValid && searchResult.Documents.Any());
-						handle.Set();
+						handles[0].Set();
 					}
 				);
 
-			this._reindexResult.Subscribe(observer);
-			handle.WaitOne(TimeSpan.FromMinutes(3));
+			this._reindexManyTypesResult.Subscribe(manyTypesObserver);
+
+			var singleTypeObserver = new ReindexObserver<Project>(
+					onError: (e) => { throw e; },
+					completed: () =>
+					{
+						var refresh = this._client.Refresh(NewSingleTypeIndexName);
+						var originalIndexCount = this._client.Count<Project>(c => c.Index(IndexName));
+
+						// new index should only contain project document types
+						var newIndexCount = this._client.Count<Project>(c => c.Index(NewSingleTypeIndexName).AllTypes());
+
+						originalIndexCount.Count.Should().BeGreaterThan(0).And.Be(newIndexCount.Count);
+
+						handles[1].Set();
+					}
+				);
+
+			this._reindexSingleTypeResult.Subscribe(singleTypeObserver);
+
+			WaitHandle.WaitAll(handles, TimeSpan.FromMinutes(3));
 		}
 	}
 }