Added Stream Method

Microsoft.Azure.Cosmos/src/Resource/Database/Database.cs

@@ -109,6 +109,7 @@ public abstract Task<DatabaseResponse> DeleteAsync(
         /// </value>
         /// <remarks>
         /// Null value indicates a database with no throughput provisioned.
+        /// If requestOptions is set to null on a database with no througput provisioned a 404.0 exception will be thrown
         /// </remarks>
         /// <seealso href="https://docs.microsoft.com/azure/cosmos-db/request-units">Request Units</seealso>
         /// <seealso href="https://docs.microsoft.com/azure/cosmos-db/set-throughput#set-throughput-on-a-database">Set throughput on a database</seealso>
@@ -138,6 +139,19 @@ public abstract Task<ThroughputResponse> ReadThroughputAsync(
             RequestOptions requestOptions,
             CancellationToken cancellationToken = default);
 
+        /// <summary>
+        /// Gets database throughput in measurement of request units per second in the Azure Cosmos service.
+        /// </summary>
+        /// <param name="requestOptions"></param>
+        /// <param name="cancellationToken"></param>
+        /// <returns>A <see cref="Task"/> containing a <see cref="ResponseMessage"/> containing the record of retrieving the provisioned throughput for this database .</returns>
+        /// <exception>https://aka.ms/cosmosdb-dot-net-exceptions</exception>
+        /// <seealso href="https://docs.microsoft.com/azure/cosmos-db/request-units">Request Units</seealso>
+        /// <seealso href="https://docs.microsoft.com/azure/cosmos-db/set-throughput#set-throughput-on-a-database">Set throughput on a database</seealso>
+        public abstract Task<ResponseMessage> ReadThroughputStreamAsync(
+            RequestOptions requestOptions,
+            CancellationToken cancellationToken = default);
+
         /// <summary>
         /// Sets throughput provisioned for a database in measurement of request units per second in the Azure Cosmos service.
         /// </summary>

Microsoft.Azure.Cosmos/src/Resource/Database/DatabaseCore.cs

@@ -93,6 +93,20 @@ public async Task<ThroughputResponse> ReadThroughputAsync(
                 cancellationToken: cancellationToken);
         }
 
+        public async Task<ResponseMessage> ReadThroughputStreamAsync(
+            RequestOptions requestOptions,
+            ITrace trace,
+            CancellationToken cancellationToken)
+        {
+            string rid = await this.GetRIDAsync(cancellationToken);
+            CosmosOffers cosmosOffers = new CosmosOffers(this.ClientContext);
+            return await cosmosOffers.ReadThroughputStreamAsync(
+                targetRID: rid,
+                requestOptions: requestOptions,
+                cancellationToken: cancellationToken,
+                uri: this.LinkUri);
+        }
+
         internal override async Task<ThroughputResponse> ReadThroughputIfExistsAsync(
             RequestOptions requestOptions,
             CancellationToken cancellationToken)

Microsoft.Azure.Cosmos/src/Resource/Database/DatabaseInlineCore.cs

@@ -265,6 +265,16 @@ public override Task<ThroughputResponse> ReadThroughputAsync(
                 (response) => new OpenTelemetryResponse<ThroughputProperties>(response));
         }
 
+        public override Task<ResponseMessage> ReadThroughputStreamAsync(
+            RequestOptions requestOptions,
+            CancellationToken cancellation = default)
+        {
+            return this.ClientContext.OperationHelperAsync(
+                nameof(ReadThroughputStreamAsync),
+                requestOptions,
+                (trace) => base.ReadThroughputStreamAsync(requestOptions, trace, cancellation),
+                (response) => new OpenTelemetryResponse(response));
+        }
         public override Task<ThroughputResponse> ReplaceThroughputAsync(
             int throughput,
             RequestOptions requestOptions = null,

Microsoft.Azure.Cosmos/src/Resource/Offer/CosmosOffers.cs

@@ -30,15 +30,6 @@ internal async Task<ThroughputResponse> ReadThroughputAsync(
             RequestOptions requestOptions,
             CancellationToken cancellationToken = default)
         {
-            if (requestOptions == null)
-            {
-                (OfferV2 nullOfferV2, _) = await this.GetOfferV2Async<OfferV2>(targetRID, failIfNotConfigured: false, cancellationToken: cancellationToken);
-                if (nullOfferV2 == null)
-                {
-                    return new ThroughputResponse(HttpStatusCode.NoContent, null, null, null, null);
-                }
-            }
-
             (OfferV2 offerV2, double requestCharge) = await this.GetOfferV2Async<OfferV2>(targetRID, failIfNotConfigured: true, cancellationToken: cancellationToken);
 
             return await this.GetThroughputResponseAsync(
@@ -52,6 +43,33 @@ internal async Task<ThroughputResponse> ReadThroughputAsync(
 
         }
 
+        internal async Task<ResponseMessage> ReadThroughputStreamAsync(
+            string targetRID,
+            RequestOptions requestOptions,
+            string uri,
+            CancellationToken cancellationToken = default)
+        {
+            (OfferV2 offerV2, double requestCharge) = await this.GetOfferV2Async<OfferV2>(targetRID, failIfNotConfigured: false, cancellationToken: cancellationToken);
+
+            string resourceUri = offerV2 != null ? new Uri(offerV2.SelfLink, UriKind.Relative).OriginalString : uri;
+
+            ResponseMessage responseMessage = await this.ClientContext.ProcessResourceOperationStreamAsync(
+                resourceUri: resourceUri,
+                resourceType: ResourceType.Offer,
+                operationType: OperationType.Read,
+                cosmosContainerCore: null,
+                feedRange: null,
+                streamPayload: null,
+                requestOptions: requestOptions,
+                requestEnricher: null,
+                trace: NoOpTrace.Singleton,
+                cancellationToken: cancellationToken);
+
+            responseMessage.Headers.RequestCharge += requestCharge;
+
+            return responseMessage;
+        }
+
         internal async Task<ThroughputResponse> ReadThroughputIfExistsAsync(
             string targetRID,
             RequestOptions requestOptions,

Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/CosmosThroughputTests.cs

@@ -6,6 +6,7 @@ namespace Microsoft.Azure.Cosmos.SDK.EmulatorTests
 {
     using System;
     using System.Net;
+    using System.Security.Cryptography.X509Certificates;
     using System.Threading;
     using System.Threading.Tasks;
     using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -166,15 +167,21 @@ public async Task ReadThroughutNullRequestOptions()
             Database db1 = await this.cosmosClient.CreateDatabaseAsync(
                 Guid.NewGuid().ToString(),
                 throughput: null);
-            try
-            {
-                await db1.ReadThroughputAsync(requestOptions: null);
-            }
-            catch
-            {
-                await db1.DeleteAsync();
-                Assert.Fail();
-            }
+
+            ResponseMessage responseMessage1 = await db1.ReadThroughputStreamAsync(requestOptions: null);
+
+            Assert.IsNotNull(responseMessage1);
+            Assert.AreEqual(HttpStatusCode.BadRequest, responseMessage1.StatusCode);
+
+            Database db2 = await this.cosmosClient.CreateDatabaseAsync(
+                Guid.NewGuid().ToString(),
+                throughput: 400);
+
+            ResponseMessage responseMessage2 = await db2.ReadThroughputStreamAsync(requestOptions: null);
+
+            Assert.IsNotNull(responseMessage2);
+            Assert.AreEqual(HttpStatusCode.OK, responseMessage2.StatusCode);
+
         }
         private async Task RecreateContainerUsingDifferentClient(
             string databaseId,