Internal Pagination: Fixes APIs to be cleaner.

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

@@ -10,39 +10,39 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed
     using Microsoft.Azure.Cosmos.ChangeFeed.Pagination;
     using Microsoft.Azure.Cosmos.Pagination;
     using Microsoft.Azure.Cosmos.Query.Core.Monads;
+    using Microsoft.Azure.Cosmos.Serializer;
 
     internal sealed class ChangeFeedCrossFeedRangeAsyncEnumerable : IAsyncEnumerable<TryCatch<ChangeFeedPage>>
     {
         private readonly IDocumentContainer documentContainer;
-        private readonly ChangeFeedMode changeFeedMode;
-        private readonly ChangeFeedRequestOptions changeFeedRequestOptions;
+        private readonly ChangeFeedPaginationOptions changeFeedPaginationOptions;
         private readonly ChangeFeedCrossFeedRangeState state;
+        private readonly JsonSerializationFormatOptions jsonSerializationFormatOptions;
 
         public ChangeFeedCrossFeedRangeAsyncEnumerable(
             IDocumentContainer documentContainer,
-            ChangeFeedMode changeFeedMode,
-            ChangeFeedRequestOptions changeFeedRequestOptions,
-            ChangeFeedCrossFeedRangeState state)
+            ChangeFeedCrossFeedRangeState state,
+            ChangeFeedPaginationOptions changeFeedPaginationOptions,
+            JsonSerializationFormatOptions jsonSerializationFormatOptions = null)
         {
             this.documentContainer = documentContainer ?? throw new ArgumentNullException(nameof(documentContainer));
-            this.changeFeedMode = changeFeedMode ?? throw new ArgumentNullException(nameof(changeFeedMode));
-            this.changeFeedRequestOptions = changeFeedRequestOptions;
+            this.changeFeedPaginationOptions = changeFeedPaginationOptions ?? ChangeFeedPaginationOptions.Default;
             this.state = state;
+            this.jsonSerializationFormatOptions = jsonSerializationFormatOptions;
         }
 
         public IAsyncEnumerator<TryCatch<ChangeFeedPage>> GetAsyncEnumerator(CancellationToken cancellationToken = default)
         {
             CrossFeedRangeState<ChangeFeedState> innerState = new CrossFeedRangeState<ChangeFeedState>(this.state.FeedRangeStates);
             CrossPartitionChangeFeedAsyncEnumerator innerEnumerator = CrossPartitionChangeFeedAsyncEnumerator.Create(
                 this.documentContainer,
-                this.changeFeedMode,
-                this.changeFeedRequestOptions,
                 innerState,
+                this.changeFeedPaginationOptions,
                 cancellationToken);
 
             return new ChangeFeedCrossFeedRangeAsyncEnumerator(
-                innerEnumerator, 
-                this.changeFeedRequestOptions?.JsonSerializationFormatOptions);
+                innerEnumerator,
+                this.jsonSerializationFormatOptions);
         }
     }
 }

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

@@ -26,7 +26,7 @@ internal sealed class ChangeFeedIteratorCore : FeedIteratorInternal
         public ChangeFeedIteratorCore(
             IDocumentContainer documentContainer,
             ChangeFeedMode changeFeedMode,
-            ChangeFeedRequestOptions changeFeedRequestOptions,            
+            ChangeFeedRequestOptions changeFeedRequestOptions,
             ChangeFeedStartFrom changeFeedStartFrom)
         {
             if (changeFeedStartFrom == null)
@@ -153,11 +153,38 @@ public ChangeFeedIteratorCore(
                                 innerException: monadicChangeFeedCrossFeedRangeState.Exception));
                     }
 
+                    Dictionary<string, string> additionalHeaders;
+                    if (changeFeedRequestOptions?.Properties != null)
+                    {
+                        additionalHeaders = new Dictionary<string, string>();
+                        Dictionary<string, object> nonStringHeaders = new Dictionary<string, object>();
+                        foreach (KeyValuePair<string, object> keyValuePair in changeFeedRequestOptions.Properties)
+                        {
+                            if (keyValuePair.Value is string stringValue)
+                            {
+                                additionalHeaders[keyValuePair.Key] = stringValue;
+                            }
+                            else
+                            {
+                                nonStringHeaders[keyValuePair.Key] = keyValuePair.Value;
+                            }
+                        }
+
+                        changeFeedRequestOptions.Properties = nonStringHeaders;
+                    }
+                    else
+                    {
+                        additionalHeaders = null;
+                    }
+
                     CrossPartitionChangeFeedAsyncEnumerator enumerator = CrossPartitionChangeFeedAsyncEnumerator.Create(
                         documentContainer,
-                        changeFeedMode,
-                        changeFeedRequestOptions,
                         new CrossFeedRangeState<ChangeFeedState>(monadicChangeFeedCrossFeedRangeState.Result.FeedRangeStates),
+                        new ChangeFeedPaginationOptions(
+                            changeFeedMode,
+                            changeFeedRequestOptions?.PageSizeHint,
+                            changeFeedRequestOptions?.JsonSerializationFormatOptions?.JsonSerializationFormat,
+                            additionalHeaders),
                         cancellationToken: default);
 
                     TryCatch<CrossPartitionChangeFeedAsyncEnumerator> monadicEnumerator = TryCatch<CrossPartitionChangeFeedAsyncEnumerator>.FromResult(enumerator);

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

@@ -5,6 +5,7 @@
 namespace Microsoft.Azure.Cosmos.ChangeFeed
 {
     using System;
+    using System.Globalization;
     using System.Threading;
     using System.Threading.Tasks;
     using Microsoft.Azure.Cosmos.ChangeFeed.LeaseManagement;
@@ -106,6 +107,17 @@ public override async Task<ResponseMessage> ReadNextAsync(ITrace trace, Cancella
                     // Set time headers if any
                     ChangeFeedStartFromRequestOptionPopulator visitor = new ChangeFeedStartFromRequestOptionPopulator(requestMessage);
                     this.changeFeedStartFrom.Accept(visitor);
+
+                    if (this.changeFeedOptions.PageSizeHint.HasValue)
+                    {
+                        requestMessage.Headers.Add(
+                            HttpConstants.HttpHeaders.PageSize,
+                            this.changeFeedOptions.PageSizeHint.Value.ToString(CultureInfo.InvariantCulture));
+                    }
+
+                    requestMessage.Headers.Add(
+                        HttpConstants.HttpHeaders.A_IM,
+                        HttpConstants.A_IMHeaderValues.IncrementalFeed);
                 },
                 feedRange: this.changeFeedStartFrom.FeedRange,
                 streamPayload: default,

Microsoft.Azure.Cosmos/src/ChangeFeed/Pagination/ChangeFeedNotModifiedPage.cs

@@ -4,14 +4,22 @@
 
 namespace Microsoft.Azure.Cosmos.ChangeFeed.Pagination
 {
+    using System.Collections.Generic;
+    using System.Collections.Immutable;
+
     internal sealed class ChangeFeedNotModifiedPage : ChangeFeedPage
     {
+        private static readonly ImmutableHashSet<string> bannedHeaders = new HashSet<string>().ToImmutableHashSet();
+
         public ChangeFeedNotModifiedPage(
             double requestCharge,
             string activityId,
+            IReadOnlyDictionary<string, string> additionalHeaders,
             ChangeFeedState state)
-            : base(requestCharge, activityId, state)
+            : base(requestCharge, activityId, additionalHeaders, state)
         {
         }
+
+        protected override ImmutableHashSet<string> DerivedClassBannedHeaders => bannedHeaders;
     }
 }

Microsoft.Azure.Cosmos/src/ChangeFeed/Pagination/ChangeFeedPage.cs

@@ -4,23 +4,28 @@
 
 namespace Microsoft.Azure.Cosmos.ChangeFeed.Pagination
 {
-    using System;
+    using System.Collections.Generic;
+    using System.Collections.Immutable;
+    using System.Linq;
     using Microsoft.Azure.Cosmos.Pagination;
+    using Microsoft.Azure.Documents;
 
     internal abstract class ChangeFeedPage : Page<ChangeFeedState>
     {
+        public static readonly ImmutableHashSet<string> BannedHeaders = new HashSet<string>()
+        {
+            HttpConstants.HttpHeaders.ETag,
+        }
+        .Concat(Page<ChangeFeedState>.BannedHeadersBase)
+        .ToImmutableHashSet();
+
         protected ChangeFeedPage(
             double requestCharge,
             string activityId,
+            IReadOnlyDictionary<string, string> additionalHeaders,
             ChangeFeedState state)
-            : base(state)
+            : base(requestCharge, activityId, additionalHeaders, state)
         {
-            this.RequestCharge = requestCharge < 0 ? throw new ArgumentOutOfRangeException(nameof(requestCharge)) : requestCharge;
-            this.ActivityId = activityId ?? throw new ArgumentNullException(nameof(activityId));
         }
-
-        public double RequestCharge { get; }
-
-        public string ActivityId { get; }
     }
 }

Microsoft.Azure.Cosmos/src/ChangeFeed/Pagination/ChangeFeedPaginationOptions.cs

@@ -0,0 +1,43 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+// ------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos.ChangeFeed.Pagination
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Collections.Immutable;
+    using System.Linq;
+    using Microsoft.Azure.Cosmos.Json;
+    using Microsoft.Azure.Cosmos.Pagination;
+    using Microsoft.Azure.Documents;
+
+    internal sealed class ChangeFeedPaginationOptions : PaginationOptions
+    {
+        public static readonly ChangeFeedPaginationOptions Default = new ChangeFeedPaginationOptions(
+            mode: ChangeFeedMode.Incremental);
+
+        public static readonly ImmutableHashSet<string> BannedHeaders = new HashSet<string>()
+        {
+            HttpConstants.HttpHeaders.A_IM,
+            HttpConstants.HttpHeaders.IfModifiedSince,
+            HttpConstants.HttpHeaders.IfNoneMatch,
+        }
+        .Concat(PaginationOptions.bannedAdditionalHeaders)
+        .ToImmutableHashSet();
+
+        public ChangeFeedPaginationOptions(
+            ChangeFeedMode mode,
+            int? pageSizeHint = null,
+            JsonSerializationFormat? jsonSerializationFormat = null,
+            Dictionary<string, string> additionalHeaders = null)
+            : base(pageSizeHint, jsonSerializationFormat, additionalHeaders)
+        {
+            this.Mode = mode ?? throw new ArgumentNullException(nameof(mode));
+        }
+
+        public ChangeFeedMode Mode { get; }
+
+        protected override ImmutableHashSet<string> BannedAdditionalHeaders => BannedHeaders;
+    }
+}

Microsoft.Azure.Cosmos/src/ChangeFeed/Pagination/ChangeFeedPartitionRangePageAsyncEnumerator.cs

@@ -7,45 +7,33 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed.Pagination
     using System;
     using System.Threading;
     using System.Threading.Tasks;
-    using Microsoft.Azure.Cosmos.Json;
     using Microsoft.Azure.Cosmos.Pagination;
     using Microsoft.Azure.Cosmos.Query.Core.Monads;
     using Microsoft.Azure.Cosmos.Tracing;
-    using Microsoft.Azure.Documents;
 
     internal sealed class ChangeFeedPartitionRangePageAsyncEnumerator : PartitionRangePageAsyncEnumerator<ChangeFeedPage, ChangeFeedState>
     {
         private readonly IChangeFeedDataSource changeFeedDataSource;
-        private readonly int pageSize;
-        private readonly ChangeFeedMode changeFeedMode;
-        private readonly JsonSerializationFormat? jsonSerializationFormat;
+        private readonly ChangeFeedPaginationOptions changeFeedPaginationOptions;
 
         public ChangeFeedPartitionRangePageAsyncEnumerator(
             IChangeFeedDataSource changeFeedDataSource,
-            FeedRangeInternal range,
-            int pageSize,
-            ChangeFeedMode changeFeedMode,
-            JsonSerializationFormat? jsonSerializationFormat,
-            ChangeFeedState state,
+            FeedRangeState<ChangeFeedState> feedRangeState,
+            ChangeFeedPaginationOptions changeFeedPaginationOptions,
             CancellationToken cancellationToken)
-            : base(range, cancellationToken, state)
+            : base(feedRangeState, cancellationToken)
         {
             this.changeFeedDataSource = changeFeedDataSource ?? throw new ArgumentNullException(nameof(changeFeedDataSource));
-            this.changeFeedMode = changeFeedMode ?? throw new ArgumentNullException(nameof(changeFeedMode));
-            this.pageSize = pageSize;
-            this.jsonSerializationFormat = jsonSerializationFormat;
+            this.changeFeedPaginationOptions = changeFeedPaginationOptions ?? throw new ArgumentNullException(nameof(changeFeedPaginationOptions));
         }
 
         public override ValueTask DisposeAsync() => default;
 
         protected override Task<TryCatch<ChangeFeedPage>> GetNextPageAsync(
             ITrace trace, 
             CancellationToken cancellationToken) => this.changeFeedDataSource.MonadicChangeFeedAsync(
-            this.State,
-            this.Range,
-            this.pageSize,
-            this.changeFeedMode,
-            this.jsonSerializationFormat,
+            this.FeedRangeState,
+            this.changeFeedPaginationOptions,
             trace,
             cancellationToken);
     }

Microsoft.Azure.Cosmos/src/ChangeFeed/Pagination/ChangeFeedSuccessPage.cs

@@ -5,20 +5,27 @@
 namespace Microsoft.Azure.Cosmos.ChangeFeed.Pagination
 {
     using System;
+    using System.Collections.Generic;
+    using System.Collections.Immutable;
     using System.IO;
 
     internal sealed class ChangeFeedSuccessPage : ChangeFeedPage
     {
+        private static readonly ImmutableHashSet<string> bannedHeaders = new HashSet<string>().ToImmutableHashSet();
+
         public ChangeFeedSuccessPage(
             Stream content,
             double requestCharge,
             string activityId,
+            IReadOnlyDictionary<string, string> additionalHeaders,
             ChangeFeedState state)
-            : base(requestCharge, activityId, state)
+            : base(requestCharge, activityId, additionalHeaders, state)
         {
             this.Content = content ?? throw new ArgumentNullException(nameof(content));
         }
 
         public Stream Content { get; }
+
+        protected override ImmutableHashSet<string> DerivedClassBannedHeaders => bannedHeaders;
     }
 }

Microsoft.Azure.Cosmos/src/ChangeFeed/Pagination/CrossPartitionChangeFeedAsyncEnumerator.cs

@@ -8,7 +8,6 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed.Pagination
     using System.Collections.Generic;
     using System.Threading;
     using System.Threading.Tasks;
-    using Microsoft.Azure.Cosmos.Json;
     using Microsoft.Azure.Cosmos.Pagination;
     using Microsoft.Azure.Cosmos.Query.Core.Monads;
     using Microsoft.Azure.Cosmos.Tracing;
@@ -107,13 +106,15 @@ public async ValueTask<bool> MoveNextAsync(ITrace trace)
                                 changeFeedSuccessPage.Content,
                                 totalRequestCharge,
                                 changeFeedSuccessPage.ActivityId,
+                                changeFeedSuccessPage.AdditionalHeaders,
                                 changeFeedSuccessPage.State);
                         }
                         else
                         {
                             backendPage = new ChangeFeedNotModifiedPage(
                                 totalRequestCharge,
                                 backendPage.ActivityId,
+                                backendPage.AdditionalHeaders,
                                 backendPage.State);
                         }
                     }
@@ -130,30 +131,22 @@ public async ValueTask<bool> MoveNextAsync(ITrace trace)
 
         public static CrossPartitionChangeFeedAsyncEnumerator Create(
             IDocumentContainer documentContainer,
-            ChangeFeedMode changeFeedMode,
-            ChangeFeedRequestOptions changeFeedRequestOptions,
             CrossFeedRangeState<ChangeFeedState> state,
+            ChangeFeedPaginationOptions changeFeedPaginationOptions,
             CancellationToken cancellationToken)
         {
-            changeFeedRequestOptions ??= new ChangeFeedRequestOptions();
+            changeFeedPaginationOptions ??= ChangeFeedPaginationOptions.Default;
 
             if (documentContainer == null)
             {
                 throw new ArgumentNullException(nameof(documentContainer));
             }
 
-            if (changeFeedMode == null)
-            {
-                throw new ArgumentNullException(nameof(changeFeedMode));
-            }
-
             CrossPartitionRangePageAsyncEnumerator<ChangeFeedPage, ChangeFeedState> crossPartitionEnumerator = new CrossPartitionRangePageAsyncEnumerator<ChangeFeedPage, ChangeFeedState>(
                 documentContainer,
                 CrossPartitionChangeFeedAsyncEnumerator.MakeCreateFunction(
                     documentContainer,
-                    changeFeedRequestOptions.PageSizeHint.GetValueOrDefault(int.MaxValue),
-                    changeFeedMode,
-                    changeFeedRequestOptions?.JsonSerializationFormatOptions?.JsonSerializationFormat,
+                    changeFeedPaginationOptions,
                     cancellationToken),
                 comparer: default /* this uses a regular queue instead of prioirty queue */,
                 maxConcurrency: default,
@@ -169,16 +162,11 @@ public static CrossPartitionChangeFeedAsyncEnumerator Create(
 
         private static CreatePartitionRangePageAsyncEnumerator<ChangeFeedPage, ChangeFeedState> MakeCreateFunction(
             IChangeFeedDataSource changeFeedDataSource,
-            int pageSize,
-            ChangeFeedMode changeFeedMode,
-            JsonSerializationFormat? jsonSerializationFormat,
-            CancellationToken cancellationToken) => (FeedRangeInternal range, ChangeFeedState state) => new ChangeFeedPartitionRangePageAsyncEnumerator(
+            ChangeFeedPaginationOptions changeFeedPaginationOptions,
+            CancellationToken cancellationToken) => (FeedRangeState<ChangeFeedState> feedRangeState) => new ChangeFeedPartitionRangePageAsyncEnumerator(
                 changeFeedDataSource,
-                range,
-                pageSize,
-                changeFeedMode,
-                jsonSerializationFormat,
-                state,
+                feedRangeState,
+                changeFeedPaginationOptions,
                 cancellationToken);
     }
 }