ChangeFeed : Adds ChangeFeedStartFrom to support StartTimes x FeedRanges

Microsoft.Azure.Cosmos/src/Resource/QueryResponses/ChangeFeedIteratorCore.cs → Microsoft.Azure.Cosmos/src/ChangeFeed/ChangeFeedIteratorCore.cs

@@ -2,11 +2,10 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 //------------------------------------------------------------
 
-namespace Microsoft.Azure.Cosmos
+namespace Microsoft.Azure.Cosmos.ChangeFeed
 {
  using System;
  using System.Collections.Generic;
- using System.Linq;
  using System.Net;
  using System.Threading;
  using System.Threading.Tasks;
@@ -26,12 +25,14 @@ internal sealed class ChangeFeedIteratorCore : FeedIteratorInternal
  private readonly CosmosClientContext clientContext;
  private readonly ChangeFeedRequestOptions changeFeedOptions;
  private readonly AsyncLazy<TryCatch<string>> lazyContainerRid;
+ private ChangeFeedStartFrom changeFeedStartFrom;
  private bool hasMoreResults;
 
  private FeedRangeContinuation FeedRangeContinuation;
 
  public ChangeFeedIteratorCore(
  ContainerInternal container,
+ ChangeFeedStartFrom changeFeedStartFrom,
  ChangeFeedRequestOptions changeFeedRequestOptions)
  {
  this.container = container ?? throw new ArgumentNullException(nameof(container));
@@ -43,24 +44,19 @@ public ChangeFeedIteratorCore(
  });
  this.hasMoreResults = true;
 
- if (this.changeFeedOptions?.From is ChangeFeedRequestOptions.StartFromContinuation startFromContinuation)
+ this.changeFeedStartFrom = changeFeedStartFrom;
+ if (this.changeFeedStartFrom is ChangeFeedStartFromContinuation startFromContinuation)
  {
  if (!FeedRangeContinuation.TryParse(startFromContinuation.Continuation, out FeedRangeContinuation feedRangeContinuation))
  {
  throw new ArgumentException(string.Format(ClientResources.FeedToken_UnknownFormat, startFromContinuation.Continuation));
  }
 
  this.FeedRangeContinuation = feedRangeContinuation;
- this.changeFeedOptions.FeedRange = feedRangeContinuation.GetFeedRange();
- string continuationToken = feedRangeContinuation.GetContinuation();
- if (continuationToken != null)
- {
- this.changeFeedOptions.From = ChangeFeedRequestOptions.StartFrom.CreateFromContinuation(continuationToken);
- }
- else
- {
- this.changeFeedOptions.From = ChangeFeedRequestOptions.StartFrom.CreateFromBeginning();
- }
+ FeedRange feedRange = feedRangeContinuation.GetFeedRange();
+ string etag = feedRangeContinuation.GetContinuation();
+
+ this.changeFeedStartFrom = new ChangeFeedStartFromContinuationAndFeedRange(etag, (FeedRangeInternal)feedRange);
  }
  }
 
@@ -76,7 +72,9 @@ public override async Task<ResponseMessage> ReadNextAsync(CancellationToken canc
  CosmosDiagnosticsContext diagnostics = CosmosDiagnosticsContext.Create(this.changeFeedOptions);
  using (diagnostics.GetOverallScope())
  {
- diagnostics.AddDiagnosticsInternal(new FeedRangeStatistics(this.changeFeedOptions.FeedRange));
+ diagnostics.AddDiagnosticsInternal(
+ new FeedRangeStatistics(
+ this.changeFeedStartFrom.Accept(ChangeFeedRangeExtractor.Singleton)));
  if (!this.lazyContainerRid.ValueInitialized)
  {
  using (diagnostics.CreateScope("InitializeContainerResourceId"))
@@ -130,33 +128,28 @@ private async Task<ResponseMessage> ReadNextInternalAsync(
  {
  cancellationToken.ThrowIfCancellationRequested();
 
- string continuation = this.FeedRangeContinuation.GetContinuation();
- if (continuation != null)
- {
- this.changeFeedOptions.From = ChangeFeedRequestOptions.StartFrom.CreateFromContinuation(continuation);
- }
-
- if ((this.changeFeedOptions.FeedRange == null) || this.changeFeedOptions.FeedRange is FeedRangeEpk)
- {
- // For now the backend does not support EPK Ranges if they don't line up with a PKRangeId
- // So if the range the user supplied is a logical pk value, then we don't want to overwrite it.
- this.changeFeedOptions.FeedRange = this.FeedRangeContinuation.GetFeedRange();
- }
-
  ResponseMessage responseMessage = await this.clientContext.ProcessResourceOperationStreamAsync(
  resourceUri: this.container.LinkUri,
  resourceType: ResourceType.Document,
  operationType: OperationType.ReadFeed,
  requestOptions: this.changeFeedOptions,
  cosmosContainerCore: this.container,
- requestEnricher: default,
+ requestEnricher: (request) =>
+ {
+ ChangeFeedStartFromRequestOptionPopulator visitor = new ChangeFeedStartFromRequestOptionPopulator(request);
+ this.changeFeedStartFrom.Accept(visitor);
+ },
  partitionKey: default,
  streamPayload: default,
  diagnosticsContext: diagnosticsScope,
  cancellationToken: cancellationToken);
 
  if (await this.ShouldRetryAsync(responseMessage, cancellationToken))
  {
+ string etag = this.FeedRangeContinuation.GetContinuation();
+ FeedRange feedRange = this.FeedRangeContinuation.GetFeedRange();
+ this.changeFeedStartFrom = new ChangeFeedStartFromContinuationAndFeedRange(etag, (FeedRangeInternal)feedRange);
+
  return await this.ReadNextInternalAsync(diagnosticsScope, cancellationToken);
  }
 
@@ -166,6 +159,11 @@ private async Task<ResponseMessage> ReadNextInternalAsync(
  // Change Feed read uses Etag for continuation
  this.hasMoreResults = responseMessage.IsSuccessStatusCode;
  this.FeedRangeContinuation.ReplaceContinuation(responseMessage.Headers.ETag);
+
+ string etag = this.FeedRangeContinuation.GetContinuation();
+ FeedRange feedRange = this.FeedRangeContinuation.GetFeedRange();
+ this.changeFeedStartFrom = new ChangeFeedStartFromContinuationAndFeedRange(etag, (FeedRangeInternal)feedRange);
+
  return FeedRangeResponse.CreateSuccess(
  responseMessage,
  this.FeedRangeContinuation);
@@ -216,13 +214,14 @@ private async Task InitializeFeedContinuationAsync(CancellationToken cancellatio
  {
  FeedRangePartitionKeyRangeExtractor feedRangePartitionKeyRangeExtractor = new FeedRangePartitionKeyRangeExtractor(this.container);
 
- IReadOnlyList<Documents.Routing.Range<string>> ranges = await ((FeedRangeInternal)this.changeFeedOptions.FeedRange).AcceptAsync(
+ FeedRange feedRange = this.changeFeedStartFrom.Accept(ChangeFeedRangeExtractor.Singleton);
+ IReadOnlyList<Documents.Routing.Range<string>> ranges = await ((FeedRangeInternal)feedRange).AcceptAsync(
  feedRangePartitionKeyRangeExtractor,
  cancellationToken);
 
  this.FeedRangeContinuation = new FeedRangeCompositeContinuation(
  containerRid: this.lazyContainerRid.Result.Result,
- feedRange: (FeedRangeInternal)this.changeFeedOptions.FeedRange,
+ feedRange: (FeedRangeInternal)feedRange,
  ranges: ranges);
  }
  else if (this.FeedRangeContinuation?.FeedRange is FeedRangePartitionKeyRange feedRangePartitionKeyRange)

Microsoft.Azure.Cosmos/src/Resource/QueryResponses/ChangeFeedPartitionKeyResultSetIteratorCore.cs → Microsoft.Azure.Cosmos/src/ChangeFeed/ChangeFeedPartitionKeyResultSetIteratorCore.cs

@@ -2,7 +2,7 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 //------------------------------------------------------------
 
-namespace Microsoft.Azure.Cosmos
+namespace Microsoft.Azure.Cosmos.ChangeFeed
 {
  using System;
  using System.Net;
@@ -18,16 +18,18 @@ internal sealed class ChangeFeedPartitionKeyResultSetIteratorCore : FeedIterator
  private readonly CosmosClientContext clientContext;
  private readonly ContainerInternal container;
  private readonly ChangeFeedRequestOptions changeFeedOptions;
+ private ChangeFeedStartFrom changeFeedStartFrom;
  private bool hasMoreResultsInternal;
- private string continuationToken;
 
  public ChangeFeedPartitionKeyResultSetIteratorCore(
  CosmosClientContext clientContext,
  ContainerInternal container,
+ ChangeFeedStartFrom changeFeedStartFrom,
  ChangeFeedRequestOptions options)
  {
  this.clientContext = clientContext ?? throw new ArgumentNullException(nameof(clientContext));
  this.container = container ?? throw new ArgumentNullException(nameof(container));
+ this.changeFeedStartFrom = changeFeedStartFrom;
  this.changeFeedOptions = options;
  }
 
@@ -42,27 +44,29 @@ public ChangeFeedPartitionKeyResultSetIteratorCore(
  /// <returns>A change feed response from cosmos service</returns>
  public override async Task<ResponseMessage> ReadNextAsync(CancellationToken cancellationToken = default(CancellationToken))
  {
- // Change Feed read uses Etag for continuation
- if (this.continuationToken != null)
- {
- this.changeFeedOptions.From = ChangeFeedRequestOptions.StartFrom.CreateFromContinuation(this.continuationToken);
- }
-
  ResponseMessage responseMessage = await this.clientContext.ProcessResourceOperationStreamAsync(
  cosmosContainerCore: this.container,
  resourceUri: this.container.LinkUri,
  resourceType: Documents.ResourceType.Document,
  operationType: Documents.OperationType.ReadFeed,
  requestOptions: this.changeFeedOptions,
- requestEnricher: default,
+ requestEnricher: (requestMessage) =>
+ {
+ ChangeFeedStartFromRequestOptionPopulator visitor = new ChangeFeedStartFromRequestOptionPopulator(requestMessage);
+ this.changeFeedStartFrom.Accept(visitor);
+ },
  partitionKey: default,
  streamPayload: default,
  diagnosticsContext: default,
  cancellationToken: cancellationToken);
 
- this.continuationToken = responseMessage.Headers.ETag;
+ // Change Feed uses etag as continuation token.
+ string etag = responseMessage.Headers.ETag;
  this.hasMoreResultsInternal = responseMessage.IsSuccessStatusCode;
- responseMessage.Headers.ContinuationToken = this.continuationToken;
+ responseMessage.Headers.ContinuationToken = etag;
+
+ FeedRangeInternal feedRange = (FeedRangeInternal)this.changeFeedStartFrom.Accept(ChangeFeedRangeExtractor.Singleton);
+ this.changeFeedStartFrom = new ChangeFeedStartFromContinuationAndFeedRange(etag, feedRange);
 
  return responseMessage;
  }

Microsoft.Azure.Cosmos/src/ChangeFeed/ChangeFeedRangeExtractor.cs

@@ -0,0 +1,28 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// ------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos.ChangeFeed
+{
+ using System;
+
+ internal sealed class ChangeFeedRangeExtractor : ChangeFeedStartFromVisitor<FeedRange>
+ {
+ public static readonly ChangeFeedRangeExtractor Singleton = new ChangeFeedRangeExtractor();
+
+ private ChangeFeedRangeExtractor()
+ {
+ }
+
+ public override FeedRange Visit(ChangeFeedStartFromNow startFromNow) => startFromNow.FeedRange;
+
+ public override FeedRange Visit(ChangeFeedStartFromTime startFromTime) => startFromTime.FeedRange;
+
+ public override FeedRange Visit(ChangeFeedStartFromContinuation startFromContinuation)
+ => throw new NotSupportedException($"{nameof(ChangeFeedStartFromContinuation)} does not have a feed range.");
+
+ public override FeedRange Visit(ChangeFeedStartFromBeginning startFromBeginning) => startFromBeginning.FeedRange;
+
+ public override FeedRange Visit(ChangeFeedStartFromContinuationAndFeedRange startFromContinuationAndFeedRange) => startFromContinuationAndFeedRange.FeedRange;
+ }
+}

Microsoft.Azure.Cosmos/src/ChangeFeed/ChangeFeedStartFrom.cs

@@ -0,0 +1,105 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// ------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos
+{
+ using System;
+ using Microsoft.Azure.Cosmos.ChangeFeed;
+
+ /// <summary>
+ /// Base class for where to start a ChangeFeed operation in <see cref="ChangeFeedRequestOptions"/>.
+ /// </summary>
+ /// <remarks>Use one of the static constructors to generate a StartFrom option.</remarks>
+#if PREVIEW
+ public
+#else
+ internal
+#endif
+ abstract class ChangeFeedStartFrom
+ {
+ /// <summary>
+ /// Initializes an instance of the <see cref="ChangeFeedStartFrom"/> class.
+ /// </summary>
+ internal ChangeFeedStartFrom()
+ {
+ // Internal so people can't derive from this type.
+ }
+
+ internal abstract void Accept(ChangeFeedStartFromVisitor visitor);
+
+ internal abstract TResult Accept<TResult>(ChangeFeedStartFromVisitor<TResult> visitor);
+
+ /// <summary>
+ /// Creates a <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from this moment onward.
+ /// </summary>
+ /// <returns>A <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from this moment onward.</returns>
+ public static ChangeFeedStartFrom Now() => Now(FeedRangeEpk.FullRange);
+
+ /// <summary>
+ /// Creates a <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from this moment onward.
+ /// </summary>
+ /// <param name="feedRange">The range to start from.</param>
+ /// <returns>A <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from this moment onward.</returns>
+ public static ChangeFeedStartFrom Now(FeedRange feedRange)
+ {
+ if (!(feedRange is FeedRangeInternal feedRangeInternal))
+ {
+ throw new ArgumentException($"{nameof(feedRange)} needs to be a {nameof(FeedRangeInternal)}.");
+ }
+
+ return new ChangeFeedStartFromNow(feedRangeInternal);
+ }
+
+ /// <summary>
+ /// Creates a <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from some point in time onward.
+ /// </summary>
+ /// <param name="dateTimeUtc">The time (in UTC) to start reading from.</param>
+ /// <returns>A <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from some point in time onward.</returns>
+ public static ChangeFeedStartFrom Time(DateTime dateTimeUtc) => Time(dateTimeUtc, FeedRangeEpk.FullRange);
+
+ /// <summary>
+ /// Creates a <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from some point in time onward.
+ /// </summary>
+ /// <param name="dateTimeUtc">The time to start reading from.</param>
+ /// <param name="feedRange">The range to start from.</param>
+ /// <returns>A <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from some point in time onward.</returns>
+ public static ChangeFeedStartFrom Time(DateTime dateTimeUtc, FeedRange feedRange)
+ {
+ if (!(feedRange is FeedRangeInternal feedRangeInternal))
+ {
+ throw new ArgumentException($"{nameof(feedRange)} needs to be a {nameof(FeedRangeInternal)}.");
+ }
+
+ return new ChangeFeedStartFromTime(dateTimeUtc, feedRangeInternal);
+ }
+
+ /// <summary>
+ /// Creates a <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from a save point.
+ /// </summary>
+ /// <param name="continuationToken">The continuation to resume from.</param>
+ /// <returns>A <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from a save point.</returns>
+ public static ChangeFeedStartFrom ContinuationToken(string continuationToken) => new ChangeFeedStartFromContinuation(continuationToken);
+
+ /// <summary>
+ /// Creates a <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start from the beginning of time.
+ /// </summary>
+ /// <returns>A <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from the beginning of time.</returns>
+ public static ChangeFeedStartFrom Beginning() => Beginning(FeedRangeEpk.FullRange);
+
+ /// <summary>
+ /// Creates a <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start from the beginning of time.
+ /// </summary>
+ /// <param name="feedRange">The range to start from.</param>
+ /// <returns>A <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from the beginning of time.</returns>
+ public static ChangeFeedStartFrom Beginning(FeedRange feedRange)
+ {
+ if (!(feedRange is FeedRangeInternal feedRangeInternal))
+ {
+ throw new ArgumentException($"{nameof(feedRange)} needs to be a {nameof(FeedRangeInternal)}.");
+ }
+
+ return new ChangeFeedStartFromBeginning(feedRangeInternal);
+ }
+ }
+}

Microsoft.Azure.Cosmos/src/ChangeFeed/ChangeFeedStartFromBeginning.cs

@@ -0,0 +1,33 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// ------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos.ChangeFeed
+{
+ using System;
+
+ /// <summary>
+ /// Derived instance of <see cref="ChangeFeedStartFrom"/> that tells the ChangeFeed operation to start reading changes from the beginning of time.
+ /// </summary>
+ internal sealed class ChangeFeedStartFromBeginning : ChangeFeedStartFrom
+ {
+ /// <summary>
+ /// Initializes an instance of the <see cref="ChangeFeedStartFromBeginning"/> class.
+ /// </summary>
+ /// <param name="feedRange">The (optional) range to start from.</param>
+ public ChangeFeedStartFromBeginning(FeedRangeInternal feedRange)
+ : base()
+ {
+ this.FeedRange = feedRange ?? throw new ArgumentNullException(nameof(feedRange));
+ }
+
+ /// <summary>
+ /// Gets the (optional) range to start from.
+ /// </summary>
+ public FeedRangeInternal FeedRange { get; }
+
+ internal override void Accept(ChangeFeedStartFromVisitor visitor) => visitor.Visit(this);
+
+ internal override TResult Accept<TResult>(ChangeFeedStartFromVisitor<TResult> visitor) => visitor.Visit(this);
+ }
+}