added Reindex() implements #328

Author: Mpdreamz

src/Nest.Tests.Integration/Index/ReindexTests.cs

@@ -0,0 +1,55 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Nest.Tests.MockData;
+using Nest.Tests.MockData.Domain;
+using NUnit.Framework;
+
+
+namespace Nest.Tests.Integration.Index
+{
+	[TestFixture]
+	public class ReindexTests : IntegrationTests
+	{
+
+
+		[Test]
+		public void Reindex()
+		{
+			var toIndex = ElasticsearchConfiguration.NewUniqueIndexName();
+			var observable = this._client.Reindex<object>(r => r
+				.FromIndex(ElasticsearchConfiguration.DefaultIndex)
+				.ToIndex(toIndex)
+				.Query(q=>q.MatchAll())
+				.Scroll("10s")
+				.CreateIndex(c=>c
+					.NumberOfReplicas(0)
+					.NumberOfShards(1)
+				)
+			);
+			var observer = new ReindexObserver<object>(
+				onNext: (r) =>
+				{
+					var scrollResults = r.QueryResponse;
+					var bulkResults = r.BulkResponse;
+
+					Assert.NotNull(scrollResults);
+					Assert.NotNull(bulkResults);
+
+					Assert.True(scrollResults.IsValid);
+					Assert.True(bulkResults.IsValid);
+				},
+				onError: (e) => Assert.Fail(e.Message),
+				completed: () =>
+				{
+					var originalIndexCount = this._client.Count(new[] { ElasticsearchConfiguration.DefaultIndex }, q=>q.MatchAll());
+					var newIndexCount = this._client.Count(new[] { toIndex  }, q=>q.MatchAll());
+					Assert.AreEqual(originalIndexCount.Count, newIndexCount.Count);
+				}
+			);
+
+			observable.Subscribe(observer);
+		}
+
+	}
+}

src/Nest.Tests.Integration/Nest.Tests.Integration.csproj

@@ -60,6 +60,7 @@
     <Compile Include="ElasticsearchConfiguration.cs" />
     <Compile Include="Core\IndexTests.cs" />
     <Compile Include="Help\AliasTests.cs" />
+    <Compile Include="Index\ReindexTests.cs" />
     <Compile Include="Index\IndexUsingUrlIdTests.cs" />
     <Compile Include="Indices\Analysis\Analyzers\AnalyzerTest.cs" />
     <Compile Include="Indices\Analysis\Analyzers\AnalyzerTestResult.cs" />

src/Nest/DSL/ReindexDescriptor.cs

@@ -0,0 +1,78 @@
+using System;
+
+namespace Nest
+{
+	public class ReindexDescriptor<T> where T : class
+	{
+		internal string _ToIndexName { get; set; }
+		internal string _FromIndexName { get; set; }
+		internal string _Scroll { get; set; }
+		
+		internal Func<QueryDescriptor<T>, BaseQuery> _QuerySelector { get; set; }
+
+		internal Func<CreateIndexDescriptor, CreateIndexDescriptor> _CreateIndexSelector { get; set; }
+
+		internal RootObjectMappingDescriptor<T> _RootObjectMappingDescriptor { get; set; } 
+
+		/// <summary>
+		/// The index into which we're indexing
+		/// </summary>
+		public ReindexDescriptor<T> ToIndex(string name)
+		{
+			this._ToIndexName = name;
+			return this;
+		}
+
+		/// <summary>
+		/// The index from which we're reindexing 
+		/// </summary>
+		public ReindexDescriptor<T> FromIndex(string name)
+		{
+			this._FromIndexName = name;
+			return this;
+		}
+
+		/// <summary>
+		/// A search request can be scrolled by specifying the scroll parameter. The scroll parameter is a time value parameter (for example: scroll=5m), indicating for how long the nodes that participate in the search will maintain relevant resources in order to continue and support it. This is very similar in its idea to opening a cursor against a database.
+		/// </summary>
+		/// <param name="scrollTime">The scroll parameter is a time value parameter (for example: scroll=5m)</param>
+		/// <returns></returns>
+		public ReindexDescriptor<T> Scroll(string scrollTime)
+		{
+			scrollTime.ThrowIfNullOrEmpty("scrollTime");
+			this._Scroll = scrollTime;
+			return this;
+		}
+
+		/// <summary>
+		/// A query to optionally limit the documents to use for the reindex operation.  
+		/// </summary>
+		public ReindexDescriptor<T> Query(Func<QueryDescriptor<T>, BaseQuery> querySelector)
+		{
+			querySelector.ThrowIfNull("querySelector");
+			this._QuerySelector = querySelector;
+			return this;
+		}
+
+		/// <summary>
+		/// The new index name to reindex too. 
+		/// </summary>
+		public ReindexDescriptor<T> NewIndexName(string name)
+		{
+			this._ToIndexName = name;
+			return this;
+		}
+
+		/// <summary>
+		/// CreateIndex selector, will be passed the a descriptor initialized with the settings from
+		/// the index we're reindexing from
+		/// </summary>
+		public ReindexDescriptor<T> CreateIndex(Func<CreateIndexDescriptor, CreateIndexDescriptor> createIndexSelector)
+		{
+			createIndexSelector.ThrowIfNull("createIndexSelector");
+			this._CreateIndexSelector = createIndexSelector;
+			return this;
+		}
+
+	}
+}

src/Nest/Domain/Responses/BaseResponse.cs

@@ -14,7 +14,7 @@ public BaseResponse()
 		{
 			this.IsValid = true;
 		}
-		public bool IsValid { get; internal set; }
+		public virtual bool IsValid { get; internal set; }
 		public ConnectionStatus ConnectionStatus { get; internal set; }
 
 		internal PropertyNameResolver PropertyNameResolver { get; set; }

src/Nest/Domain/Responses/IReindexResponse.cs

@@ -0,0 +1,28 @@
+namespace Nest
+{
+	/// <summary>
+	/// POCO representing the reindex response for a each step
+	/// </summary>
+	public interface IReindexResponse<T>  where T : class
+	{
+		/// <summary>
+		/// The bulk result indexing the search results into the new index.
+		/// </summary>
+		IBulkResponse BulkResponse { get; }
+
+		/// <summary>
+		/// The scroll result
+		/// </summary>
+		IQueryResponse<T> QueryResponse { get; }
+
+		/// <summary>
+		/// The no of scroll this result represents
+		/// </summary>
+		int Scroll { get;  }
+
+		/// <summary>
+		/// Whether both the scroll and reindex result are valid
+		/// </summary>
+		bool IsValid { get; }
+	}
+}

src/Nest/Domain/Responses/IndexResponse.cs

@@ -2,28 +2,28 @@
 
 namespace Nest
 {
-  public interface IIndexResponse : IResponse
-  {
-    string Id { get; }
-    string Index { get; }
-    string Type { get; }
-    string Version { get; }
-    bool OK { get; }
-  }
+	public interface IIndexResponse : IResponse
+	{
+		string Id { get; }
+		string Index { get; }
+		string Type { get; }
+		string Version { get; }
+		bool OK { get; }
+	}
 
 	[JsonObject]
-  public class IndexResponse : BaseResponse, IIndexResponse
-	{	
-    [JsonProperty("_index")]
+	public class IndexResponse : BaseResponse, IIndexResponse
+	{
+		[JsonProperty("_index")]
 		public string Index { get; internal set; }
-    [JsonProperty("_type")]
-    public string Type { get; internal set; }
-    [JsonProperty("_id")]
-    public string Id { get; internal set; }
-    [JsonProperty("_version")]
-    public string Version { get; internal set; }
-    [JsonProperty("ok")]
-    public bool OK { get; internal set; }
-    
+		[JsonProperty("_type")]
+		public string Type { get; internal set; }
+		[JsonProperty("_id")]
+		public string Id { get; internal set; }
+		[JsonProperty("_version")]
+		public string Version { get; internal set; }
+		[JsonProperty("ok")]
+		public bool OK { get; internal set; }
+
 	}
 }

src/Nest/Domain/Responses/ReindexObservable.cs

@@ -0,0 +1,98 @@
+using System;
+
+namespace Nest
+{
+	public class ReindexObservable<T> : IDisposable, IObservable<IReindexResponse<T>> where T : class
+	{
+		private ReindexDescriptor<T> _reindexDescriptor;
+		internal IElasticClient CurrentClient { get; set; }
+		internal ReindexDescriptor<T> ReindexDescriptor { get; set; } 
+
+		public ReindexObservable(IElasticClient client, ReindexDescriptor<T> reindexDescriptor)
+		{
+			this._reindexDescriptor = reindexDescriptor;
+			this.CurrentClient = client;
+		}
+
+		public IDisposable Subscribe(IObserver<IReindexResponse<T>> observer)
+		{
+			observer.ThrowIfNull("observer");
+			try 
+			{
+				this.Reindex(observer);
+			}
+			catch (Exception e)
+			{
+				observer.OnError(e);
+			}
+			return this;
+
+		}
+
+		private void Reindex(IObserver<IReindexResponse<T>> observer)
+		{
+			var fromIndex = this._reindexDescriptor._FromIndexName;
+			var toIndex = this._reindexDescriptor._ToIndexName;
+			var scroll = this._reindexDescriptor._Scroll ?? "2m";
+
+			fromIndex.ThrowIfNullOrEmpty("fromIndex");
+			toIndex.ThrowIfNullOrEmpty("toIndex");
+
+			var indexSettings = this.CurrentClient.GetIndexSettings(this._reindexDescriptor._FromIndexName);
+			var createSettings = new CreateIndexDescriptor(this.CurrentClient.Settings).InitializeUsing(indexSettings.Settings);
+			var createIndexResponse = this.CurrentClient
+				.CreateIndex(toIndex, (c) => this._reindexDescriptor._CreateIndexSelector(createSettings));
+			if (!createIndexResponse.IsValid)
+				throw new ReindexException(createIndexResponse.ConnectionStatus);
+
+			var page = 0;
+			var searchResult = this.CurrentClient.Search<T>(
+				s => s
+					.Index(fromIndex)
+					.AllTypes()
+					.From(0)
+					.Take(100)
+					.Query(this._reindexDescriptor._QuerySelector)
+					.SearchType(SearchType.Scan)
+					.Scroll(scroll)
+				);
+			if (searchResult.Total <= 0)
+				throw new ReindexException(searchResult.ConnectionStatus, "index " + fromIndex + " has no documents!");
+			IBulkResponse indexResult = null;
+			do
+			{
+				searchResult = this.CurrentClient.Scroll<T>(scroll, searchResult.ScrollId);
+				if (searchResult.Documents.HasAny())
+					indexResult = this.IndexSearchResults(searchResult, observer, toIndex, page);
+				page++;
+			} while (searchResult.IsValid && indexResult != null && indexResult.IsValid && searchResult.Documents.HasAny());
+
+
+			observer.OnCompleted();
+		}
+
+		public IBulkResponse IndexSearchResults(IQueryResponse<T> searchResult,IObserver<IReindexResponse<T>> observer, string toIndex, int page)
+		{
+			if (!searchResult.IsValid)
+				throw new ReindexException(searchResult.ConnectionStatus, "reindex failed on scroll #" + page);
+
+			var indexResult = this.CurrentClient.IndexMany(searchResult.Documents, toIndex);
+			if (!indexResult.IsValid)
+				throw new ReindexException(indexResult.ConnectionStatus, "reindex failed when indexing page " + page);
+
+			observer.OnNext(new ReindexResponse<T>()
+			{
+				BulkResponse = indexResult,
+				QueryResponse = searchResult,
+				Scroll = page
+			});
+			return indexResult;
+		}
+
+
+		public void Dispose()
+		{
+
+		}
+	}
+}

src/Nest/Domain/Responses/ReindexObserver.cs

@@ -0,0 +1,37 @@
+using System;
+
+namespace Nest
+{
+	public class ReindexObserver<T> : IObserver<IReindexResponse<T>> where T : class
+	{
+		private readonly Action<IReindexResponse<T>> _onNext;
+		private readonly Action<Exception> _onError;
+		private readonly Action _completed;
+
+		public ReindexObserver(Action<IReindexResponse<T>> onNext, Action<Exception> onError, Action completed)
+		{
+			this._completed = completed;
+			this._onError = onError;
+			this._onNext = onNext;
+		}
+
+
+		public void OnCompleted()
+		{
+			if (this._completed != null)
+				this._completed();
+		}
+
+		public void OnError(Exception error)
+		{
+			if (this._onError != null)
+				this._onError(error);
+		}
+
+		public void OnNext(IReindexResponse<T> value)
+		{
+			if(this._onNext != null)
+				this._onNext(value);
+		}
+	}
+}

src/Nest/Domain/Responses/ReindexResponse.cs

@@ -0,0 +1,27 @@
+using System.Collections.Generic;
+using System.Collections.Concurrent;
+using Newtonsoft.Json;
+
+namespace Nest
+{
+	/// <summary>
+	/// POCO representing the reindex response for a each step
+	/// </summary>
+	[JsonObject]
+	public class ReindexResponse<T> : IReindexResponse<T> where T : class
+	{
+		public IBulkResponse BulkResponse { get; internal set; }
+		public IQueryResponse<T> QueryResponse { get; internal set; }
+
+		public int Scroll { get; internal set; }
+
+		public bool IsValid
+		{
+			get
+			{
+				return (this.BulkResponse != null && this.BulkResponse.IsValid
+					&& this.QueryResponse != null && this.QueryResponse.IsValid);
+			}
+		}
+	}
+}