Merge pull request Azure#320 from ealsur/master

Syncing Change Feed Processor

Author: rnagpal

samples/ChangeFeedProcessor/DocumentDB.ChangeFeedProcessor/ChangeFeedEventHost.cs

@@ -83,6 +83,8 @@ public class ChangeFeedEventHost : IPartitionObserver<DocumentServiceLease>
     {
         const string DefaultUserAgentSuffix = "changefeed-0.2";
         const string LeaseContainerName = "docdb-changefeed";
+        const string LSNPropertyName = "_lsn";
+
         readonly DocumentCollectionInfo collectionLocation;
 
         string leasePrefix;
@@ -170,6 +172,60 @@ public async Task RegisterObserverFactoryAsync(IChangeFeedObserverFactory factor
             await this.StartAsync();
         }
 
+        /// <summary>
+        /// Asynchronously checks the current existing leases and calculates an estimate of remaining work per leased partitions.
+        /// </summary>
+        /// <returns>An estimate amount of remaining documents to be processed</returns>
+        public async Task<long> GetEstimatedRemainingWork()
+        {
+            await this.InitializeAsync();
+
+            long remaining = 0;
+            ChangeFeedOptions options = new ChangeFeedOptions
+            {
+                MaxItemCount = 1
+            };
+
+            foreach (DocumentServiceLease existingLease in await this.leaseManager.ListLeases())
+            {
+                options.PartitionKeyRangeId = existingLease.PartitionId;
+                options.RequestContinuation = existingLease.ContinuationToken;
+                IDocumentQuery<Document> query = this.documentClient.CreateDocumentChangeFeedQuery(this.collectionSelfLink, options);
+                FeedResponse<Document> response = null;
+
+                try
+                {
+                    response = await query.ExecuteNextAsync<Document>();
+                    long parsedLSNFromSessionToken = TryConvertToNumber(ParseAmountFromSessionToken(response.SessionToken));
+                    long lastSequenceNumber = response.Count > 0 ? TryConvertToNumber(response.First().GetPropertyValue<string>(LSNPropertyName)) : parsedLSNFromSessionToken;
+                    long partitionRemaining = parsedLSNFromSessionToken - lastSequenceNumber;
+                    remaining += partitionRemaining < 0 ? 0 : partitionRemaining;
+                }
+                catch (DocumentClientException ex)
+                {
+                    ExceptionDispatchInfo exceptionDispatchInfo = ExceptionDispatchInfo.Capture(ex);
+                    DocumentClientException dcex = (DocumentClientException)exceptionDispatchInfo.SourceException;
+                    if ((StatusCode.NotFound == (StatusCode)dcex.StatusCode && SubStatusCode.ReadSessionNotAvailable != (SubStatusCode)GetSubStatusCode(dcex))
+                        || StatusCode.Gone == (StatusCode)dcex.StatusCode)
+                    {
+                        // We are not explicitly handling Splits here to avoid any collision with an Observer that might have picked this up and managing the split
+                        TraceLog.Error(string.Format("GetEstimateWork > Partition {0}: resource gone (subStatus={1}).", existingLease.PartitionId, GetSubStatusCode(dcex)));
+                    }
+                    else if (StatusCode.TooManyRequests == (StatusCode)dcex.StatusCode ||
+                                    StatusCode.ServiceUnavailable == (StatusCode)dcex.StatusCode)
+                    {
+                        TraceLog.Warning(string.Format("GetEstimateWork > Partition {0}: retriable exception : {1}", existingLease.PartitionId, dcex.Message));
+                    }
+                    else
+                    {
+                        TraceLog.Error(string.Format("GetEstimateWork > Partition {0}: Unhandled exception", ex.Error.Message));
+                    }
+                }
+            }
+
+            return remaining;
+        }
+
         /// <summary>Asynchronously shuts down the host instance. This method maintains the leases on all partitions currently held, and enables each 
         /// host instance to shut down cleanly by invoking the method with object.</summary> 
         /// <returns>A task that indicates the host instance has stopped.</returns>
@@ -759,6 +815,34 @@ private bool IsCheckpointNeeded(DocumentServiceLease lease, CheckpointStats chec
             return isCheckpointNeeded;
         }
 
+        private static long TryConvertToNumber(string number)
+        {
+            if (string.IsNullOrEmpty(number))
+            {
+                return 0;
+            }
+
+            long parsed = 0;
+            if (!long.TryParse(number, NumberStyles.Any, CultureInfo.InvariantCulture, out parsed))
+            {
+                TraceLog.Warning(string.Format(CultureInfo.InvariantCulture, "Cannot parse number '{0}'.", number));
+                return 0;
+            }
+
+            return parsed;
+        }
+
+        private static string ParseAmountFromSessionToken(string sessionToken)
+        {
+            if (string.IsNullOrEmpty(sessionToken))
+            {
+                return string.Empty;
+            }
+
+            int separatorIndex = sessionToken.IndexOf(':');
+            return sessionToken.Substring(separatorIndex + 1);
+        }
+
         private static int GetDocumentCount(ResourceResponse<DocumentCollection> response)
         {
             Debug.Assert(response != null);