How do I use Azure TableClient 2.0 BeginExecuteQue

Question

I'm trying to get all entries in my table asynchronously but am unable to figure out how to work with the continuation token. I suspect I need to take my anonymous method and convert it to a delegate, then recursively call it with the continuation token.

How do I take the following code and perform an Async call and fetch all 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

This person made some extensions methods for ExecuteQueryAsync that wrap the segmented methods

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

Answer 2

Here is an alternate implementation of @Gaurav Mantri's code.

public class AzureTableQueryState
{
    public CloudTable CloudTable { get; set; }
    public TableContinuationToken Token { get; set; }
    public ManualResetEvent Evt { get; set; } //i'm guessing that's 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()
                }
            }

Answer 3

Please try this:

    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 I noticed is that if I execute a query which returns a dynamic table entity, I'm getting an error related to DateTimeOffset. That's why I ended up creating a temporary entity.

Hope this helps.

Answer 4

Here is a another alternative, but this time in 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);
            });
        }