Add reindex method

Author: giometrix

src/TableStorage.Abstractions.POCO.SecondaryIndexes/PocoStoreIndexer.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Azure.Documents;
 using TableStorage.Abstractions.Models;
@@ -436,5 +437,70 @@ public static IPocoTableStore<T, TIndexPartitionKey, TIndexRowKey> Index<T, TPar
 		}
 
 
+		/// <summary>
+		/// Reindexes a table.  You will need to call this for existing tables that have never had an index, or if things get out of sync.  This can take a while.
+		/// </summary>
+		/// <typeparam name="T"></typeparam>
+		/// <typeparam name="TPartitionKey">The type of the t partition key.</typeparam>
+		/// <typeparam name="TRowKey">The type of the t row key.</typeparam>
+		/// <param name="tableStore">The table store.</param>
+		/// <param name="indexName">Name of the index.</param>
+		/// <param name="maxDegreeOfParallelism">The maximum degree of parallelism.</param>
+		/// <param name="recordsIndexedCallback">The records indexed callback.</param>
+		/// <param name="failedIndexCallback">The failed index callback.</param>
+		/// <exception cref="ArgumentException"></exception>
+		public static async Task ReindexAsync<T, TPartitionKey, TRowKey>(this IPocoTableStore<T, TPartitionKey, TRowKey> tableStore, string indexName, int? maxDegreeOfParallelism = null, Action<int> recordsIndexedCallback = null, Action<T, Exception> failedIndexCallback = null)
+		{
+			maxDegreeOfParallelism = maxDegreeOfParallelism ?? Environment.ProcessorCount * 20;
+
+			string pageToken = null;
+			int count = 0;
+			using (var semaphore = new SemaphoreSlim(maxDegreeOfParallelism.Value, maxDegreeOfParallelism.Value))
+			{
+				try
+				{
+					dynamic indexStore = _indexes[indexName];
+					do
+					{
+						var result = await indexStore.GetAllRecordsPagedAsync(1000, pageToken);
+						pageToken = result.ContinuationToken;
+
+						if (result.Items.Count > 0)
+						{
+							foreach (var record in result.Items)
+							{
+								await semaphore.WaitAsync(TimeSpan.FromSeconds(20));
+								Task task = indexStore.InsertOrReplaceAsync(record);
+								task.ContinueWith(r =>
+								{
+									if (r.IsFaulted)
+									{
+										failedIndexCallback?.Invoke(record, r.Exception);
+									}
+									Interlocked.Increment(ref count);
+									semaphore.Release(1);
+								});
+
+							}
+						}
+
+						recordsIndexedCallback?.Invoke(count);
+					} while (pageToken != null);
+
+					while (semaphore.CurrentCount < maxDegreeOfParallelism)
+					{
+
+						await Task.Delay(5);
+					}
+					recordsIndexedCallback?.Invoke(count);
+				}
+				catch (KeyNotFoundException e)
+				{
+					throw new ArgumentException($"{indexName} is not a defined secondary index");
+				}
+			}
+		}
+
+
 	}
 }

tests/TableStorage.Abstractions.POCO.SecondaryIndexes.Tests/IndexTests.cs

@@ -405,5 +405,32 @@ public async Task get_by_index_partition_key_string_paged_async()
 			Assert.IsNull(p.ContinuationToken);
 
 		}
+
+		[TestMethod]
+		public async Task reindex()
+		{
+			var employee = new Employee
+			{
+				Name = "Test",
+				CompanyId = 99,
+				Id = 99,
+				Department = new Department { Id = 5, Name = "Test" }
+			};
+			var employee2 = new Employee
+			{
+				Name = "Test2",
+				CompanyId = 99,
+				Id = 100,
+				Department = new Department { Id = 5, Name = "Test" }
+			};
+			await TableStore.InsertAsync(employee);
+			await TableStore.InsertAsync(employee2);
+
+			int count = 0;
+
+			await TableStore.ReindexAsync("Name", maxDegreeOfParallelism: 20, recordsIndexedCallback: i=>count = i);
+
+			Assert.AreEqual(2, count);
+		}
 	}
 }