Internal ReadFeed: Adds pagination library adoption

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

@@ -13,7 +13,6 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed
     using Microsoft.Azure.Cosmos.Pagination;
     using Microsoft.Azure.Cosmos.Query.Core;
     using Microsoft.Azure.Cosmos.Query.Core.Monads;
-    using Microsoft.Azure.Cosmos.Query.Core.Pipeline.CrossPartition.Parallel;
     using Microsoft.Azure.Cosmos.Routing;
 
     internal sealed class ChangeFeedIteratorCore : FeedIteratorInternal

Microsoft.Azure.Cosmos/src/Pagination/DocumentContainer.cs

@@ -13,6 +13,7 @@ namespace Microsoft.Azure.Cosmos.Pagination
     using Microsoft.Azure.Cosmos.Query.Core;
     using Microsoft.Azure.Cosmos.Query.Core.Monads;
     using Microsoft.Azure.Cosmos.Query.Core.Pipeline;
+    using Microsoft.Azure.Cosmos.ReadFeed.Pagination;
     using Microsoft.Azure.Documents;
 
     /// <summary>
@@ -83,24 +84,28 @@ public Task<Record> ReadItemAsync(
                     cancellationToken),
                 cancellationToken);
 
-        public Task<TryCatch<DocumentContainerPage>> MonadicReadFeedAsync(
+        public Task<TryCatch<ReadFeedPage>> MonadicReadFeedAsync(
+            ReadFeedState readFeedState,
             FeedRangeInternal feedRange,
-            ResourceId resourceIdentifer,
+            QueryRequestOptions queryRequestOptions,
             int pageSize,
             CancellationToken cancellationToken) => this.monadicDocumentContainer.MonadicReadFeedAsync(
+                readFeedState,
                 feedRange,
-                resourceIdentifer,
+                queryRequestOptions,
                 pageSize,
                 cancellationToken);
 
-        public Task<DocumentContainerPage> ReadFeedAsync(
+        public Task<ReadFeedPage> ReadFeedAsync(
+            ReadFeedState readFeedState,
             FeedRangeInternal feedRange,
-            ResourceId resourceIdentifier,
+            QueryRequestOptions queryRequestOptions,
             int pageSize,
-            CancellationToken cancellationToken) => TryCatch<DocumentContainerPage>.UnsafeGetResultAsync(
+            CancellationToken cancellationToken) => TryCatch<ReadFeedPage>.UnsafeGetResultAsync(
                 this.MonadicReadFeedAsync(
+                    readFeedState,
                     feedRange,
-                    resourceIdentifier,
+                    queryRequestOptions,
                     pageSize,
                     cancellationToken),
                 cancellationToken);

Microsoft.Azure.Cosmos/src/Pagination/IDocumentContainer.cs

@@ -9,9 +9,9 @@ namespace Microsoft.Azure.Cosmos.Pagination
     using Microsoft.Azure.Cosmos.ChangeFeed.Pagination;
     using Microsoft.Azure.Cosmos.CosmosElements;
     using Microsoft.Azure.Cosmos.Query.Core.Pipeline.CrossPartition;
-    using Microsoft.Azure.Documents;
+    using Microsoft.Azure.Cosmos.ReadFeed.Pagination;
 
-    internal interface IDocumentContainer : IMonadicDocumentContainer, IFeedRangeProvider, IQueryDataSource, IChangeFeedDataSource
+    internal interface IDocumentContainer : IMonadicDocumentContainer, IFeedRangeProvider, IQueryDataSource, IReadFeedDataSource, IChangeFeedDataSource 
     {
         Task<Record> CreateItemAsync(
             CosmosObject payload,
@@ -22,12 +22,6 @@ Task<Record> ReadItemAsync(
             string identifier,
             CancellationToken cancellationToken);
 
-        Task<DocumentContainerPage> ReadFeedAsync(
-            FeedRangeInternal feedRange,
-            ResourceId resourceIdentifier,
-            int pageSize,
-            CancellationToken cancellationToken);
-
         Task SplitAsync(
             FeedRangeInternal feedRange,
             CancellationToken cancellationToken);

Microsoft.Azure.Cosmos/src/Pagination/IMonadicDocumentContainer.cs

@@ -10,9 +10,9 @@ namespace Microsoft.Azure.Cosmos.Pagination
     using Microsoft.Azure.Cosmos.CosmosElements;
     using Microsoft.Azure.Cosmos.Query.Core.Monads;
     using Microsoft.Azure.Cosmos.Query.Core.Pipeline.CrossPartition;
-    using Microsoft.Azure.Documents;
+    using Microsoft.Azure.Cosmos.ReadFeed.Pagination;
 
-    internal interface IMonadicDocumentContainer : IMonadicFeedRangeProvider, IMonadicQueryDataSource, IMonadicChangeFeedDataSource
+    internal interface IMonadicDocumentContainer : IMonadicFeedRangeProvider, IMonadicQueryDataSource, IMonadicReadFeedDataSource, IMonadicChangeFeedDataSource
     {
         Task<TryCatch<Record>> MonadicCreateItemAsync(
             CosmosObject payload,
@@ -23,12 +23,6 @@ Task<TryCatch<Record>> MonadicReadItemAsync(
             string identifer,
             CancellationToken cancellationToken);
 
-        Task<TryCatch<DocumentContainerPage>> MonadicReadFeedAsync(
-            FeedRangeInternal feedRange,
-            ResourceId resourceIdentifer,
-            int pageSize,
-            CancellationToken cancellationToken);
-
         Task<TryCatch> MonadicSplitAsync(
             FeedRangeInternal feedRange,
             CancellationToken cancellationToken);

Microsoft.Azure.Cosmos/src/Pagination/NetworkAttachedDocumentContainer.cs

@@ -18,28 +18,32 @@ namespace Microsoft.Azure.Cosmos.Pagination
     using Microsoft.Azure.Cosmos.Query.Core.Monads;
     using Microsoft.Azure.Cosmos.Query.Core.Pipeline;
     using Microsoft.Azure.Cosmos.Query.Core.QueryClient;
+    using Microsoft.Azure.Cosmos.ReadFeed.Pagination;
     using Microsoft.Azure.Documents;
 
     internal sealed class NetworkAttachedDocumentContainer : IMonadicDocumentContainer
     {
-        private readonly ContainerCore container;
+        private readonly ContainerInternal container;
         private readonly CosmosQueryClient cosmosQueryClient;
-        private readonly CosmosClientContext cosmosClientContext;
         private readonly QueryRequestOptions queryRequestOptions;
         private readonly CosmosDiagnosticsContext diagnosticsContext;
+        private readonly string resourceLink;
+        private readonly ResourceType resourceType;
 
         public NetworkAttachedDocumentContainer(
-            ContainerCore container,
+            ContainerInternal container,
             CosmosQueryClient cosmosQueryClient,
-            CosmosClientContext cosmosClientContext,
             CosmosDiagnosticsContext diagnosticsContext,
-            QueryRequestOptions queryRequestOptions = null)
+            QueryRequestOptions queryRequestOptions = null,
+            string resourceLink = null,
+            ResourceType resourceType = ResourceType.Document)
         {
             this.container = container ?? throw new ArgumentNullException(nameof(container));
             this.cosmosQueryClient = cosmosQueryClient ?? throw new ArgumentNullException(nameof(cosmosQueryClient));
-            this.cosmosClientContext = cosmosClientContext ?? throw new ArgumentNullException(nameof(cosmosClientContext));
             this.diagnosticsContext = diagnosticsContext;
             this.queryRequestOptions = queryRequestOptions;
+            this.resourceLink = resourceLink ?? this.container.LinkUri;
+            this.resourceType = resourceType;
         }
 
         public Task<TryCatch> MonadicSplitAsync(
@@ -91,7 +95,7 @@ public async Task<TryCatch<List<FeedRangeEpk>>> MonadicGetChildRangeAsync(
         {
             try
             {
-                ContainerProperties containerProperties = await this.cosmosClientContext.GetCachedContainerPropertiesAsync(
+                ContainerProperties containerProperties = await this.container.ClientContext.GetCachedContainerPropertiesAsync(
                     this.container.LinkUri,
                     cancellationToken);
                 List<PartitionKeyRange> overlappingRanges = await this.cosmosQueryClient.GetTargetPartitionKeyRangeByFeedRangeAsync(
@@ -113,13 +117,90 @@ await this.container.GetRIDAsync(cancellationToken),
             }
         }
 
-        public Task<TryCatch<DocumentContainerPage>> MonadicReadFeedAsync(
+        public async Task<TryCatch<ReadFeedPage>> MonadicReadFeedAsync(
+            ReadFeedState readFeedState,
             FeedRangeInternal feedRange,
-            ResourceId resourceIdentifer,
+            QueryRequestOptions queryRequestOptions,
             int pageSize,
             CancellationToken cancellationToken)
         {
-            throw new NotImplementedException();
+            cancellationToken.ThrowIfCancellationRequested();
+
+            if (feedRange is FeedRangeEpk feedRangeEpk)
+            {
+                ContainerProperties containerProperties = await this.container.ClientContext.GetCachedContainerPropertiesAsync(
+                    this.container.LinkUri,
+                    cancellationToken);
+                List<PartitionKeyRange> overlappingRanges = await this.cosmosQueryClient.GetTargetPartitionKeyRangeByFeedRangeAsync(
+                    this.container.LinkUri,
+                    await this.container.GetRIDAsync(cancellationToken),
+                    containerProperties.PartitionKey,
+                    feedRange);
+
+                if ((overlappingRanges == null) || (overlappingRanges.Count != 1))
+                {
+                    // Simulate a split exception, since we don't have a partition key range id to route to.
+                    CosmosException goneException = new CosmosException(
+                        message: $"Epk Range: {feedRangeEpk.Range} is gone.",
+                        statusCode: System.Net.HttpStatusCode.Gone,
+                        subStatusCode: (int)SubStatusCodes.PartitionKeyRangeGone,
+                        activityId: Guid.NewGuid().ToString(),
+                        requestCharge: default);
+
+                    return TryCatch<ReadFeedPage>.FromException(goneException);
+                }
+            }
+
+            if (queryRequestOptions != null)
+            {
+                queryRequestOptions.MaxItemCount = pageSize;
+            }
+
+            ResponseMessage responseMessage = await this.container.ClientContext.ProcessResourceOperationStreamAsync(
+               resourceUri: this.resourceLink,
+               resourceType: this.resourceType,
+               operationType: OperationType.ReadFeed,
+               requestOptions: queryRequestOptions,
+               cosmosContainerCore: this.container,
+               requestEnricher: request =>
+               {
+                   if (!(readFeedState.ContinuationToken is CosmosNull))
+                   {
+                       request.Headers.ContinuationToken = (readFeedState.ContinuationToken as CosmosString).Value;
+                   }
+
+                   feedRange.Accept(FeedRangeRequestMessagePopulatorVisitor.Singleton, request);
+               },
+               partitionKey: queryRequestOptions?.PartitionKey,
+               streamPayload: default,
+               diagnosticsContext: this.diagnosticsContext,
+               cancellationToken: cancellationToken);
+
+            TryCatch<ReadFeedPage> monadicReadFeedPage;
+            if (responseMessage.StatusCode == HttpStatusCode.OK)
+            {
+                ReadFeedPage readFeedPage = new ReadFeedPage(
+                    responseMessage.Content,
+                    responseMessage.Headers.RequestCharge,
+                    responseMessage.Headers.ActivityId,
+                    responseMessage.Headers.ContinuationToken != null ? new ReadFeedState(CosmosString.Create(responseMessage.Headers.ContinuationToken)) : null);
+
+                monadicReadFeedPage = TryCatch<ReadFeedPage>.FromResult(readFeedPage);
+            }
+            else
+            {
+                CosmosException cosmosException = new CosmosException(
+                    responseMessage.ErrorMessage,
+                    statusCode: responseMessage.StatusCode,
+                    (int)responseMessage.Headers.SubStatusCode,
+                    responseMessage.Headers.ActivityId,
+                    responseMessage.Headers.RequestCharge);
+                cosmosException.Headers.ContinuationToken = responseMessage.Headers.ContinuationToken;
+
+                monadicReadFeedPage = TryCatch<ReadFeedPage>.FromException(cosmosException);
+            }
+
+            return monadicReadFeedPage;
         }
 
         public async Task<TryCatch<QueryPage>> MonadicQueryAsync(
@@ -135,7 +216,7 @@ public async Task<TryCatch<QueryPage>> MonadicQueryAsync(
             {
                 case FeedRangePartitionKey feedRangePartitionKey:
                     {
-                        ContainerProperties containerProperties = await this.cosmosClientContext.GetCachedContainerPropertiesAsync(
+                        ContainerProperties containerProperties = await this.container.ClientContext.GetCachedContainerPropertiesAsync(
                             this.container.LinkUri,
                             cancellationToken);
                         PartitionKeyDefinition partitionKeyDefinition = await this.container.GetPartitionKeyDefinitionAsync(cancellationToken);
@@ -163,8 +244,8 @@ await this.container.GetRIDAsync(cancellationToken),
                         queryRequestOptions.PartitionKey = feedRangePartitionKey.PartitionKey;
 
                         monadicQueryPage = await this.cosmosQueryClient.ExecuteItemQueryAsync(
-                            this.container.LinkUri,
-                            Documents.ResourceType.Document,
+                            this.resourceLink,
+                            this.resourceType,
                             Documents.OperationType.Query,
                             Guid.NewGuid(),
                             queryRequestOptions,
@@ -183,8 +264,8 @@ await this.container.GetRIDAsync(cancellationToken),
                 case FeedRangePartitionKeyRange feedRangePartitionKeyRange:
                     {
                         monadicQueryPage = await this.cosmosQueryClient.ExecuteItemQueryAsync(
-                            this.container.LinkUri,
-                            Documents.ResourceType.Document,
+                            this.resourceLink,
+                            this.resourceType,
                             Documents.OperationType.Query,
                             Guid.NewGuid(),
                             requestOptions: queryRequestOptions,
@@ -202,7 +283,7 @@ await this.container.GetRIDAsync(cancellationToken),
 
                 case FeedRangeEpk feedRangeEpk:
                     {
-                        ContainerProperties containerProperties = await this.cosmosClientContext.GetCachedContainerPropertiesAsync(
+                        ContainerProperties containerProperties = await this.container.ClientContext.GetCachedContainerPropertiesAsync(
                             this.container.LinkUri,
                             cancellationToken);
                         List<PartitionKeyRange> overlappingRanges = await this.cosmosQueryClient.GetTargetPartitionKeyRangeByFeedRangeAsync(
@@ -225,8 +306,8 @@ await this.container.GetRIDAsync(cancellationToken),
                         }
 
                         monadicQueryPage = await this.cosmosQueryClient.ExecuteItemQueryAsync(
-                            this.container.LinkUri,
-                            Documents.ResourceType.Document,
+                            this.resourceLink,
+                            this.resourceType,
                             Documents.OperationType.Query,
                             Guid.NewGuid(),
                             requestOptions: queryRequestOptions,
@@ -260,7 +341,7 @@ public async Task<TryCatch<ChangeFeedPage>> MonadicChangeFeedAsync(
             if (feedRange is FeedRangeEpk feedRangeEpk)
             {
                 // convert into physical range or throw a split exception
-                ContainerProperties containerProperties = await this.cosmosClientContext.GetCachedContainerPropertiesAsync(
+                ContainerProperties containerProperties = await this.container.ClientContext.GetCachedContainerPropertiesAsync(
                             this.container.LinkUri,
                             cancellationToken);
                 List<PartitionKeyRange> overlappingRanges = await this.cosmosQueryClient.GetTargetPartitionKeyRangeByFeedRangeAsync(
@@ -285,7 +366,7 @@ await this.container.GetRIDAsync(cancellationToken),
                 feedRange = new FeedRangePartitionKeyRange(overlappingRanges[0].Id);
             }
 
-            ResponseMessage responseMessage = await this.cosmosClientContext.ProcessResourceOperationStreamAsync(
+            ResponseMessage responseMessage = await this.container.ClientContext.ProcessResourceOperationStreamAsync(
                 resourceUri: this.container.LinkUri,
                 resourceType: ResourceType.Document,
                 operationType: OperationType.ReadFeed,
@@ -303,7 +384,7 @@ await this.container.GetRIDAsync(cancellationToken),
                 },
                 partitionKey: default,
                 streamPayload: default,
-                diagnosticsContext: default,
+                diagnosticsContext: this.diagnosticsContext,
                 cancellationToken: cancellationToken);
 
             TryCatch<ChangeFeedPage> monadicChangeFeedPage;

Microsoft.Azure.Cosmos/src/Query/v3Query/QueryIterator.cs

@@ -81,7 +81,6 @@ public static QueryIterator Create(
             NetworkAttachedDocumentContainer networkAttachedDocumentContainer = new NetworkAttachedDocumentContainer(
                 containerCore,
                 client,
-                clientContext,
                 queryPipelineCreationDiagnostics,
                 queryRequestOptions);
             DocumentContainer documentContainer = new DocumentContainer(networkAttachedDocumentContainer);