How to use Azure TableClient 2.0 BeginExecuteQuerySegmented

Question

I am trying to get all the entries in my table asynchronously, but I cannot figure out how to work with the continuation token . I suspect that I need to accept my anonymous method and convert it to a delegate, and then call it recursively using the continue token.

How to take the following code and make an Async call and get all the entries in the new API?

Task<string[]> GetAllTableEntries(CloudTable tbl, string[] urls, string name, CancellationToken token) { TableRequestOptions reqOptions = new TableRequestOptions() { }; OperationContext ctx = new OperationContext() { ClientRequestID = "" }; object state = null; // Register Cancelation Token ICancellableAsyncResult result = null; TableQuery qry = new TableQuery(); TableContinuationToken tok = null; result = tbl.BeginExecuteQuerySegmented(qry, tok, reqOptions, ctx, (o) => { var response = (o.AsyncState as CloudTable).EndExecuteQuerySegmented(o); Console.WriteLine("Found " + response.Results.Count + " records"); // The following code was used in the previous version of the SDK // //26: // add first segment of data //27: pageData.CompletedList.AddRange( //28: from wu in response.Results //29: select new CompletedWorkUnit(wu)); //30: //31: // continue fetching segments to complete page //32: while (response.HasMoreResults) //33: { //34: response = response.GetNext(); //35: pageData.CompletedList.AddRange( //36: from wu in response.Results //37: select new CompletedWorkUnit(wu)); //38: } //39: //40: // set continuation token for next page request //41: pageData.ContinuationToken = response.ContinuationToken; //42: evt.Set(); }, state); // Add cancellation token according to guidance from Table Client 2.0 Breaking Changes blog entry token.Register((o) => result.Cancel(), state); 

Answer 1

Please try the following:

  static void ExecuteQuery() { TableContinuationToken token = null; TableRequestOptions reqOptions = new TableRequestOptions() { }; OperationContext ctx = new OperationContext() { ClientRequestID = "" }; long totalEntitiesRetrieved = 0; while (true) { CloudTable table = cloudTableClient.GetTableReference("MyTable"); TableQuery<TempEntity> query = (new TableQuery<TempEntity>()).Take(100); System.Threading.ManualResetEvent evt = new System.Threading.ManualResetEvent(false); var result = table.BeginExecuteQuerySegmented<TempEntity>(query, token, reqOptions, ctx, (o) => { var response = (o.AsyncState as CloudTable).EndExecuteQuerySegmented<TempEntity>(o); token = response.ContinuationToken; int recordsRetrieved = response.Count(); totalEntitiesRetrieved += recordsRetrieved; Console.WriteLine("Records retrieved in this attempt = " + recordsRetrieved + " | Total records retrieved = " + totalEntitiesRetrieved); evt.Set(); }, table); evt.WaitOne(); if (token == null) { break; } } } 

One thing that I noticed is that if I execute a query that returns a dynamic table object, I get an error related to DateTimeOffset. Therefore, I created a temporary entity.

Hope this helps.

Answer 2

Here is another alternative, but this time in the TPL / Task Parallel Library. Full source code is available here.

  /* The following overloads of ExecuteQuerySegmentedAsync is executed like this) */ CloudTableClient client = acct.CreateCloudTableClient(); CloudTable tableSymmetricKeys = client.GetTableReference("SymmetricKeys5"); TableContinuationToken token = new TableContinuationToken() { }; TableRequestOptions opt = new TableRequestOptions() { }; OperationContext ctx = new OperationContext() { ClientRequestID = "ID" }; CancellationToken cancelToken = new CancellationToken(); List<Task> taskList = new List<Task>(); while (true) { Task<TableQuerySegment<DynamicTableEntity>> task3 = tableSymmetricKeys.ExecuteQuerySegmentedAsync(query, token, opt, ctx, cancelToken); // Run the method task3.Wait(); token = task3.Result.ContinuationToken; Console.WriteLine("Records retrieved in this attempt = " + task3.Result.Count ()); if (token == null) { break; } else { // persist token // token.WriteXml() } } */ // Overload #4 public static Task<TableQuerySegment<DynamicTableEntity>> ExecuteQuerySegmentedAsync(this CloudTable tbl, TableQuery query, TableContinuationToken continuationToken, TableRequestOptions opt, OperationContext ctx ,CancellationToken token ) { ICancellableAsyncResult result = null; if (opt == null && ctx == null) result = tbl.BeginExecuteQuerySegmented(query, continuationToken, null, tbl); else result = tbl.BeginExecuteQuerySegmented(query, continuationToken, opt, ctx, null, tbl); // Add cancellation token according to guidance from Table Client 2.0 Breaking Changes blog entry var cancellationRegistration = token.Register(result.Cancel); return Task.Factory.FromAsync(result, iAsyncResult => { CloudTable currentTable = iAsyncResult.AsyncState as CloudTable; //cancellationRegistration.Dispose(); return currentTable.EndExecuteQuerySegmented(result); }); } 

Answer 3

This user has created several extension methods for ExecuteQueryAsync that wrap segmented methods.

https://github.com/glueckkanja/tasync/blob/master/StorageExtensions.cs

Answer 4

Here is an alternative implementation of @Gaurav Mantri code.

 public class AzureTableQueryState { public CloudTable CloudTable { get; set; } public TableContinuationToken Token { get; set; } public ManualResetEvent Evt { get; set; } //i'm guessing that what this is //any other variables you were using public int TotalEntitiesRetrieved { get; set; } } // snip.... while (true) { // Initialize variables TableQuery query = (new TableQuery()).Take(100); AzureTableQueryState queryState = new AzureTableQueryState(); queryState.Evt = new System.Threading.ManualResetEvent(false); queryState.TotalEntitiesRetrieved = 0; AsyncCallback asyncCallback = (iAsyncResult) => { AzureTableQueryState state = iAsyncResult.AsyncState as AzureTableQueryState; var response = state.CloudTable.EndExecuteQuerySegmented(iAsyncResult); token = response.ContinuationToken; int recordsRetrieved = response.Results.Count; state.TotalEntitiesRetrieved += recordsRetrieved; Console.WriteLine("Records retrieved in this attempt = " + recordsRetrieved + " | Total records retrieved = " + state.TotalEntitiesRetrieved); state.Evt.Set(); }; // Run the method var result = tableSymmetricKeys.BeginExecuteQuerySegmented(query, token, opt, ctx, asyncCallback, tableSymmetricKeys); // Add cancellation token according to guidance from Table Client 2.0 Breaking Changes blog entry cancelToken.Register((o) => result.Cancel(), null); queryState.Evt.WaitOne(); if (token == null) { break; } else { // persist token // token.WriteXml() } }