ClientModel Prototype: Move response buffering to transport

sdk/core/Azure.Core/src/Pipeline/DisposableHttpPipeline.cs

@@ -24,8 +24,9 @@ public sealed class DisposableHttpPipeline : HttpPipeline, IDisposable
         /// <param name="policies">Policies to be invoked as part of the pipeline in order.</param>
         /// <param name="responseClassifier">The response classifier to be used in invocations.</param>
         /// <param name="isTransportOwnedInternally"> </param>
-        internal DisposableHttpPipeline(HttpPipelineTransport transport, int perCallIndex, int perRetryIndex, HttpPipelinePolicy[] policies, ResponseClassifier responseClassifier, bool isTransportOwnedInternally)
-            : base(transport, perCallIndex, perRetryIndex, policies, responseClassifier)
+        /// <param name="networkTimeout"></param>
+        internal DisposableHttpPipeline(HttpPipelineTransport transport, int perCallIndex, int perRetryIndex, HttpPipelinePolicy[] policies, ResponseClassifier responseClassifier, bool isTransportOwnedInternally, TimeSpan networkTimeout)
+            : base(transport, perCallIndex, perRetryIndex, policies, responseClassifier, networkTimeout)
         {
             this.isTransportOwnedInternally = isTransportOwnedInternally;
         }

sdk/core/Azure.Core/src/Pipeline/HttpPipeline.cs

@@ -15,6 +15,8 @@ namespace Azure.Core.Pipeline
     /// </summary>
     public class HttpPipeline
     {
+        internal static readonly TimeSpan DefaultNetworkTimeout = TimeSpan.FromSeconds(100);
+
         private static readonly AsyncLocal<HttpMessagePropertiesScope?> CurrentHttpMessagePropertiesScope = new AsyncLocal<HttpMessagePropertiesScope?>();
 
         private protected readonly HttpPipelineTransport _transport;
@@ -40,6 +42,8 @@ public class HttpPipeline
         /// </summary>
         private readonly int _perRetryIndex;
 
+        private readonly TimeSpan _networkTimeout;
+
         /// <summary>
         /// Creates a new instance of <see cref="HttpPipeline"/> with the provided transport, policies and response classifier.
         /// </summary>
@@ -59,14 +63,16 @@ public HttpPipeline(HttpPipelineTransport transport, HttpPipelinePolicy[]? polic
             policies.CopyTo(all, 0);
 
             _pipeline = all;
+            _networkTimeout = DefaultNetworkTimeout;
         }
 
         internal HttpPipeline(
             HttpPipelineTransport transport,
             int perCallIndex,
             int perRetryIndex,
             HttpPipelinePolicy[] pipeline,
-            ResponseClassifier responseClassifier)
+            ResponseClassifier responseClassifier,
+            TimeSpan? networkTimeout)
         {
             ResponseClassifier = responseClassifier ?? throw new ArgumentNullException(nameof(responseClassifier));
 
@@ -77,6 +83,9 @@ internal HttpPipeline(
 
             _perCallIndex = perCallIndex;
             _perRetryIndex = perRetryIndex;
+
+            _networkTimeout = networkTimeout ?? DefaultNetworkTimeout;
+
             _internallyConstructed = true;
         }
 
@@ -85,14 +94,25 @@ internal HttpPipeline(
         /// </summary>
         /// <returns>The request.</returns>
         public Request CreateRequest()
-            => _transport.CreateRequest();
+        {
+            Request request = _transport.CreateRequest();
+            request.NetworkTimeout = _networkTimeout;
+            return request;
+        }
 
         /// <summary>
         /// Creates a new <see cref="HttpMessage"/> instance.
         /// </summary>
         /// <returns>The message.</returns>
         public HttpMessage CreateMessage()
-            => new(CreateRequest(), ResponseClassifier);
+        {
+            Request request = CreateRequest();
+            HttpMessage message = new(request, ResponseClassifier)
+            {
+                NetworkTimeout = request.NetworkTimeout
+            };
+            return message;
+        }
 
         /// <summary>
         /// </summary>
@@ -109,7 +129,11 @@ public HttpMessage CreateMessage(RequestContext? context)
         /// <returns>The message.</returns>
         public HttpMessage CreateMessage(RequestContext? context, ResponseClassifier? classifier = default)
         {
-            HttpMessage message = new HttpMessage(CreateRequest(), classifier ?? ResponseClassifier);
+            Request request = CreateRequest();
+            HttpMessage message = new(request, classifier ?? ResponseClassifier)
+            {
+                NetworkTimeout = request.NetworkTimeout
+            };
 
             if (context != null)
             {
@@ -134,6 +158,7 @@ public ValueTask SendAsync(HttpMessage message, CancellationToken cancellationTo
         {
             message.SetCancellationToken(cancellationToken);
             message.ProcessingStartTime = DateTimeOffset.UtcNow;
+            message.NetworkTimeout ??= _networkTimeout;
             AddHttpMessageProperties(message);
 
             if (message.Policies == null || message.Policies.Count == 0)
@@ -169,6 +194,7 @@ public void Send(HttpMessage message, CancellationToken cancellationToken)
         {
             message.SetCancellationToken(cancellationToken);
             message.ProcessingStartTime = DateTimeOffset.UtcNow;
+            message.NetworkTimeout ??= _networkTimeout;
             AddHttpMessageProperties(message);
 
             if (message.Policies == null || message.Policies.Count == 0)

sdk/core/Azure.Core/src/Pipeline/HttpPipelineBuilder.cs

@@ -43,7 +43,7 @@ public static HttpPipeline Build(
             ((List<HttpPipelinePolicy>)pipelineOptions.PerRetryPolicies).AddRange(perRetryPolicies);
             var result = BuildInternal(pipelineOptions, null);
 
-            return new HttpPipeline(result.Transport, result.PerCallIndex, result.PerRetryIndex, result.Policies, result.Classifier);
+            return new HttpPipeline(result.Transport, result.PerCallIndex, result.PerRetryIndex, result.Policies, result.Classifier, result.NetworkTimeout);
         }
 
         /// <summary>
@@ -63,7 +63,7 @@ public static DisposableHttpPipeline Build(ClientOptions options, HttpPipelinePo
             ((List<HttpPipelinePolicy>)pipelineOptions.PerCallPolicies).AddRange(perCallPolicies);
             ((List<HttpPipelinePolicy>)pipelineOptions.PerRetryPolicies).AddRange(perRetryPolicies);
             var result = BuildInternal(pipelineOptions, transportOptions);
-            return new DisposableHttpPipeline(result.Transport, result.PerCallIndex, result.PerRetryIndex, result.Policies, result.Classifier, result.IsTransportOwned);
+            return new DisposableHttpPipeline(result.Transport, result.PerCallIndex, result.PerRetryIndex, result.Policies, result.Classifier, result.IsTransportOwned, result.NetworkTimeout);
         }
 
         /// <summary>
@@ -74,7 +74,7 @@ public static DisposableHttpPipeline Build(ClientOptions options, HttpPipelinePo
         public static HttpPipeline Build(HttpPipelineOptions options)
         {
             var result = BuildInternal(options, null);
-            return new HttpPipeline(result.Transport, result.PerCallIndex, result.PerRetryIndex, result.Policies, result.Classifier);
+            return new HttpPipeline(result.Transport, result.PerCallIndex, result.PerRetryIndex, result.Policies, result.Classifier, result.NetworkTimeout);
         }
 
         /// <summary>
@@ -87,17 +87,17 @@ public static DisposableHttpPipeline Build(HttpPipelineOptions options, HttpPipe
         {
             Argument.AssertNotNull(transportOptions, nameof(transportOptions));
             var result = BuildInternal(options, transportOptions);
-            return new DisposableHttpPipeline(result.Transport, result.PerCallIndex, result.PerRetryIndex, result.Policies, result.Classifier, result.IsTransportOwned);
+            return new DisposableHttpPipeline(result.Transport, result.PerCallIndex, result.PerRetryIndex, result.Policies, result.Classifier, result.IsTransportOwned, result.NetworkTimeout);
         }
 
-        internal static (ResponseClassifier Classifier, HttpPipelineTransport Transport, int PerCallIndex, int PerRetryIndex, HttpPipelinePolicy[] Policies, bool IsTransportOwned) BuildInternal(
+        internal static (ResponseClassifier Classifier, HttpPipelineTransport Transport, int PerCallIndex, int PerRetryIndex, HttpPipelinePolicy[] Policies, bool IsTransportOwned, TimeSpan NetworkTimeout) BuildInternal(
             HttpPipelineOptions buildOptions,
             HttpPipelineTransportOptions? defaultTransportOptions)
         {
             Argument.AssertNotNull(buildOptions.PerCallPolicies, nameof(buildOptions.PerCallPolicies));
             Argument.AssertNotNull(buildOptions.PerRetryPolicies, nameof(buildOptions.PerRetryPolicies));
 
-            var policies = new List<HttpPipelinePolicy>(8 +
+            var policies = new List<HttpPipelinePolicy>(7 +
                                                         (buildOptions.ClientOptions.Policies?.Count ?? 0) +
                                                         buildOptions.PerCallPolicies.Count +
                                                         buildOptions.PerRetryPolicies.Count);
@@ -181,8 +181,6 @@ void AddNonNullPolicies(HttpPipelinePolicy[] policiesToAdd)
                 policies.Add(new LoggingPolicy(diagnostics.IsLoggingContentEnabled, diagnostics.LoggedContentSizeLimit, sanitizer, assemblyName));
             }
 
-            policies.Add(new ResponseBodyPolicy(buildOptions.ClientOptions.Retry.NetworkTimeout));
-
             policies.Add(new RequestActivityPolicy(isDistributedTracingEnabled, ClientDiagnostics.GetResourceProviderNamespace(buildOptions.ClientOptions.GetType().Assembly), sanitizer));
 
             AddUserPolicies(HttpPipelinePosition.BeforeTransport);
@@ -209,7 +207,7 @@ void AddNonNullPolicies(HttpPipelinePolicy[] policiesToAdd)
 
             buildOptions.ResponseClassifier ??= ResponseClassifier.Shared;
 
-            return (buildOptions.ResponseClassifier, transport, perCallIndex, perRetryIndex, policies.ToArray(), isTransportInternallyCreated);
+            return (buildOptions.ResponseClassifier, transport, perCallIndex, perRetryIndex, policies.ToArray(), isTransportInternallyCreated, buildOptions.ClientOptions.Retry.NetworkTimeout);
         }
 
         // internal for testing

sdk/core/Azure.Core/src/Pipeline/HttpPipelineTransport.Adapter.cs

@@ -0,0 +1,49 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System;
+using System.ClientModel.Primitives;
+using System.Threading.Tasks;
+
+namespace Azure.Core.Pipeline;
+
+public partial class HttpPipelineTransport
+{
+    private class AzureCorePipelineTransport : PipelineTransport
+    {
+        private readonly HttpPipelineTransport _transport;
+
+        public AzureCorePipelineTransport(HttpPipelineTransport transport)
+        {
+            _transport = transport;
+        }
+
+        protected override PipelineMessage CreateMessageCore()
+        {
+            Request request = _transport.CreateRequest();
+            return new HttpMessage(request, ResponseClassifier.Shared);
+        }
+
+        protected override void ProcessCore(PipelineMessage message)
+        {
+            HttpMessage httpMessage = AssertHttpMessage(message);
+            _transport.Process(httpMessage);
+        }
+
+        protected override async ValueTask ProcessCoreAsync(PipelineMessage message)
+        {
+            HttpMessage httpMessage = AssertHttpMessage(message);
+            await _transport.ProcessAsync(httpMessage).ConfigureAwait(false);
+        }
+
+        private static HttpMessage AssertHttpMessage(PipelineMessage message)
+        {
+            if (message is not HttpMessage httpMessage)
+            {
+                throw new InvalidOperationException($"Invalid type for PipelineMessage: '{message?.GetType()}'.");
+            }
+
+            return httpMessage;
+        }
+    }
+}

sdk/core/Azure.Core/src/Pipeline/HttpPipelineTransport.cs

@@ -1,16 +1,36 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
-using System.Threading;
+using System.ClientModel.Primitives;
 using System.Threading.Tasks;
 
 namespace Azure.Core.Pipeline
 {
     /// <summary>
     /// Represents an HTTP pipeline transport used to send HTTP requests and receive responses.
     /// </summary>
-    public abstract class HttpPipelineTransport
+    public abstract partial class HttpPipelineTransport
     {
+        private readonly PipelineTransport _transport;
+
+        /// <summary>
+        /// TBD.
+        /// </summary>
+        protected HttpPipelineTransport()
+        {
+            _transport = new AzureCorePipelineTransport(this);
+        }
+
+        internal void ProcessInternal(HttpMessage message)
+        {
+            _transport.Process(message);
+        }
+
+        internal async ValueTask ProcessInternalAsync(HttpMessage message)
+        {
+            await _transport.ProcessAsync(message).ConfigureAwait(false);
+        }
+
         /// <summary>
         /// Sends the request contained by the <paramref name="message"/> and sets the <see cref="HttpMessage.Response"/> property to received response synchronously.
         /// </summary>

sdk/core/Azure.Core/src/Pipeline/HttpWebRequestTransport.Adapter.cs

@@ -0,0 +1,63 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System;
+using System.ClientModel.Primitives;
+using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
+using System.Globalization;
+using System.IO;
+using System.Net;
+using System.Net.Http.Headers;
+using System.Security.Cryptography.X509Certificates;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Azure.Core.Pipeline
+{
+#if NETFRAMEWORK
+    ///// <summary>
+    ///// The <see cref="HttpWebRequest"/> based <see cref="HttpPipelineTransport"/> implementation.
+    ///// </summary>
+    //internal partial class HttpWebRequestTransport : HttpPipelineTransport
+    //{
+    ////    private class AzureCoreHttpWebRequestTransport : PipelineTransport
+    //    {
+    //        private readonly HttpWebRequestTransport _transport;
+
+    //        public AzureCoreHttpWebRequestTransport(HttpWebRequestTransport transport)
+    //        {
+    //            _transport = transport;
+    //        }
+
+    //        protected override PipelineMessage CreateMessageCore()
+    //        {
+    //            Request request = _transport.CreateRequest();
+    //            return new HttpMessage(request, ResponseClassifier.Shared);
+    //        }
+
+    //        protected override void ProcessCore(PipelineMessage message)
+    //        {
+    //            HttpMessage httpMessage = AssertHttpMessage(message);
+    //            _transport.ProcessSyncOrAsync(httpMessage, async: false).EnsureCompleted();
+    //        }
+
+    //        protected override async ValueTask ProcessCoreAsync(PipelineMessage message)
+    //        {
+    //            HttpMessage httpMessage = AssertHttpMessage(message);
+    //            await _transport.ProcessSyncOrAsync(httpMessage, async: true).ConfigureAwait(false);
+    //        }
+
+    //        private static HttpMessage AssertHttpMessage(PipelineMessage message)
+    //        {
+    //            if (message is not HttpMessage httpMessage)
+    //            {
+    //                throw new InvalidOperationException($"Invalid type for PipelineMessage: '{message?.GetType()}'.");
+    //            }
+
+    //            return httpMessage;
+    //        }
+    //    }
+    //}
+#endif
+}

sdk/core/Azure.Core/src/Pipeline/HttpWebRequestTransport.cs

@@ -18,12 +18,15 @@ namespace Azure.Core.Pipeline
     /// <summary>
     /// The <see cref="HttpWebRequest"/> based <see cref="HttpPipelineTransport"/> implementation.
     /// </summary>
-    internal class HttpWebRequestTransport : HttpPipelineTransport
+    internal partial class HttpWebRequestTransport : HttpPipelineTransport
     {
-        private readonly Action<HttpWebRequest> _configureRequest;
         public static readonly HttpWebRequestTransport Shared = new HttpWebRequestTransport();
+
+        private readonly Action<HttpWebRequest> _configureRequest;
         private readonly IWebProxy? _environmentProxy;
 
+        //private readonly AzureCoreHttpWebRequestTransport _transport;
+
         /// <summary>
         /// Creates a new instance of <see cref="HttpWebRequestTransport"/>
         /// </summary>
@@ -42,19 +45,19 @@ internal HttpWebRequestTransport(Action<HttpWebRequest> configureRequest)
             {
                 _environmentProxy = webProxy;
             }
+
+            //_transport = new(this);
         }
 
         /// <inheritdoc />
         public override void Process(HttpMessage message)
-        {
-            ProcessSyncOrAsync(message, false).EnsureCompleted();
-        }
+            //=> _transport.Process(message);
+            => ProcessSyncOrAsync(message, async: false).EnsureCompleted();
 
         /// <inheritdoc />
         public override async ValueTask ProcessAsync(HttpMessage message)
-        {
-            await ProcessSyncOrAsync(message, true).ConfigureAwait(false);
-        }
+            //=> await _transport.ProcessAsync(message).ConfigureAwait(false);
+            => await ProcessSyncOrAsync(message, async: true).ConfigureAwait(false);
 
         private async ValueTask ProcessSyncOrAsync(HttpMessage message, bool async)
         {