Copilot commented on code in PR #3930:
URL: https://github.com/apache/thrift/pull/3930#discussion_r4076302408
##########
lib/netstd/Thrift/Transport/Client/THttpTransport.cs:
##########
@@ -298,5 +298,78 @@ protected override void Dispose(bool disposing)
}
_isDisposed = true;
}
+
+ public ValueTask<TTransport>
CreatePerCallTransportAsync(CancellationToken cancellationToken = default)
+ {
+ return new ValueTask<TTransport>(new THttpPerCallTransport(this));
+ }
+
+ /// <summary>
+ /// Sends the given request stream to the server and returns the HTTP
response.
+ /// </summary>
+ /// <param name="request">The request stream to send.</param>
+ /// <param name="cancellationToken">The cancellation token to cancel
the operation.</param>
+ /// <returns>The HTTP response message.</returns>
+ /// <exception cref="TTransportException">Thrown if an error occurs
during the request.</exception>
+ internal async Task<HttpResponseMessage> SendAsync(Stream request,
CancellationToken cancellationToken)
+ {
+ // Snapshot the field locally: Close()/Dispose() running
concurrently on the parent
+ // transport can null out or dispose _httpClient while a per-call
transport is in flight.
+ var httpClient = _httpClient ?? throw new
TTransportException(TTransportException.ExceptionType.NotOpen,
+ "The transport has been closed.");
+ var content = new StreamContent(request);
+ content.Headers.ContentType = ContentType ?? new
MediaTypeHeaderValue("application/x-thrift");
+
+ var message = new HttpRequestMessage(HttpMethod.Post, _uri)
+ {
+ Content = content
+ };
+
+ // Note: message/content are intentionally not disposed here.
Disposing StreamContent
+ // would also dispose the caller-owned "request" stream, which is
owned and disposed by
+ // THttpPerCallTransport (or the parent's own _outputStream)
instead.
+ HttpResponseMessage response = null;
+ try
+ {
+ response = await httpClient.SendAsync(
+ message,
+ HttpCompletionOption.ResponseHeadersRead,
+ cancellationToken);
+
+ response.EnsureSuccessStatusCode();
+ return response;
+ }
+ catch (IOException iox)
+ {
+ response?.Dispose();
+ throw new
TTransportException(TTransportException.ExceptionType.Unknown, iox.ToString(),
iox);
+ }
+ catch (HttpRequestException wx)
+ {
+ response?.Dispose();
+ throw new
TTransportException(TTransportException.ExceptionType.Unknown,
+ "Couldn't connect to server: " + wx, wx);
+ }
+ catch (ObjectDisposedException odx)
+ {
+ response?.Dispose();
+ throw new
TTransportException(TTransportException.ExceptionType.NotOpen,
+ "The transport has been closed.", odx);
+ }
+ catch (OperationCanceledException ocx)
+ {
+ response?.Dispose();
+ throw new
TTransportException(TTransportException.ExceptionType.Interrupted, ocx.Message,
ocx);
+ }
+ catch (Exception ex)
+ {
+ response?.Dispose();
+ throw new
TTransportException(TTransportException.ExceptionType.Unknown, ex.Message, ex);
+ }
+ finally
+ {
+ ResetMessageSizeAndConsumedBytes();
Review Comment:
`SendAsync` is operating on the shared parent transport, so this reset
mutates the parent's `KnownMessageSize`/`RemainingMessageSize` while
independent per-call transports are in flight. If the parent is also being used
to read a response (or has a size established by a layered transport), a
per-call completion can reset that state to `MaxMessageSize` and bypass the
parent's message-size accounting. The per-call transport owns its own counters,
and the parent `FlushAsync` already resets the parent's counters; remove this
reset from the shared send path.
##########
lib/netstd/Thrift/Transport/Client/THttpPerCallTransport.cs:
##########
@@ -0,0 +1,134 @@
+// Licensed to the Apache Software Foundation(ASF) under one
+// or more contributor license agreements.See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+using System;
+using System.IO;
+using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Thrift.Transport.Client
+{
+ /// <summary>
+ /// Implementation of a per-call transport for THttpTransport.
+ /// Each call to the server will create a new HTTP request and response.
+ /// </summary>
+ internal class THttpPerCallTransport : TEndpointTransport
+ {
+ private readonly THttpTransport _parent;
+ private readonly MemoryStream _request = new MemoryStream();
+
+ private HttpResponseMessage _response;
+ private Stream _responseStream;
+ private bool _sent;
+ private bool _disposed;
+
+ public THttpPerCallTransport(THttpTransport parent) :
base(parent.Configuration)
+ {
+ this._parent = parent;
+ }
+
+ public override bool IsOpen => !_disposed;
+
+ public override Task WriteAsync(
+ byte[] buffer,
+ int offset,
+ int length,
+ CancellationToken cancellationToken)
+ {
+ if (_sent)
+ {
+ throw new InvalidOperationException("The request has already
been sent.");
+ }
+
+ return _request.WriteAsync(
+ buffer,
+ offset,
+ length,
+ cancellationToken);
+ }
+
+ public override async Task FlushAsync(
+ CancellationToken cancellationToken)
+ {
+ if (_sent)
+ {
+ throw new InvalidOperationException("The request has already
been sent.");
+ }
+
+ _request.Position = 0;
+ _response = await _parent.SendAsync(_request,cancellationToken);
+ _sent = true;
+
+#if NET5_0_OR_GREATER
+ _responseStream = await
_response.Content.ReadAsStreamAsync(cancellationToken);
Review Comment:
`_sent` is set before `ReadAsStreamAsync` has succeeded. If obtaining the
content stream is canceled or throws, the transport is left with `_sent ==
true` and `_responseStream == null`; the next `ReadAsync` then fails with a
`NullReferenceException`, while the response is only released if the caller
later disposes the transport. Acquire the stream first, assign the fields only
after success, and dispose the response on this failure path.
##########
lib/netstd/Thrift/Transport/Client/THttpTransport.cs:
##########
@@ -298,5 +298,78 @@ protected override void Dispose(bool disposing)
}
_isDisposed = true;
}
+
+ public ValueTask<TTransport>
CreatePerCallTransportAsync(CancellationToken cancellationToken = default)
+ {
+ return new ValueTask<TTransport>(new THttpPerCallTransport(this));
+ }
+
+ /// <summary>
+ /// Sends the given request stream to the server and returns the HTTP
response.
+ /// </summary>
+ /// <param name="request">The request stream to send.</param>
+ /// <param name="cancellationToken">The cancellation token to cancel
the operation.</param>
+ /// <returns>The HTTP response message.</returns>
+ /// <exception cref="TTransportException">Thrown if an error occurs
during the request.</exception>
+ internal async Task<HttpResponseMessage> SendAsync(Stream request,
CancellationToken cancellationToken)
+ {
+ // Snapshot the field locally: Close()/Dispose() running
concurrently on the parent
+ // transport can null out or dispose _httpClient while a per-call
transport is in flight.
+ var httpClient = _httpClient ?? throw new
TTransportException(TTransportException.ExceptionType.NotOpen,
+ "The transport has been closed.");
+ var content = new StreamContent(request);
+ content.Headers.ContentType = ContentType ?? new
MediaTypeHeaderValue("application/x-thrift");
+
+ var message = new HttpRequestMessage(HttpMethod.Post, _uri)
+ {
+ Content = content
+ };
+
+ // Note: message/content are intentionally not disposed here.
Disposing StreamContent
+ // would also dispose the caller-owned "request" stream, which is
owned and disposed by
+ // THttpPerCallTransport (or the parent's own _outputStream)
instead.
+ HttpResponseMessage response = null;
+ try
+ {
+ response = await httpClient.SendAsync(
+ message,
+ HttpCompletionOption.ResponseHeadersRead,
+ cancellationToken);
Review Comment:
Using `ResponseHeadersRead` changes the timeout boundary:
`HttpClient.Timeout` no longer covers the response body, so a server that sends
headers and then stalls can leave `THttpPerCallTransport.ReadAsync` blocked
indefinitely when callers use the default cancellation token, despite
`ConnectTimeout` being configured. Preserve an equivalent per-call deadline for
body reads or document this changed behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
