Copilot commented on code in PR #3978:
URL: https://github.com/apache/thrift/pull/3978#discussion_r4136065524
##########
lib/netstd/Thrift/TBaseClient.cs:
##########
@@ -42,13 +47,235 @@ protected TBaseClient(TProtocol inputProtocol, TProtocol
outputProtocol)
_outputProtocol = outputProtocol ?? throw new
ArgumentNullException(nameof(outputProtocol));
}
- public TProtocol InputProtocol => _inputProtocol;
+ /// <summary>
+ /// Initializes a client with shared protocols and optional per-call
transport support.
+ /// </summary>
+ /// <param name="inputProtocol">The shared input protocol.</param>
+ /// <param name="outputProtocol">The shared output protocol.</param>
+ /// <param name="transport">
+ /// The shared transport. When it implements <see
cref="ITPerCallTransportProvider"/>,
+ /// each client call uses protocols created for a new per-call
transport.
+ /// </param>
+ /// <param name="inputProtocolFactory">The factory used to create
per-call input protocols.</param>
+ /// <param name="outputProtocolFactory">The factory used to create
per-call output protocols.</param>
+ protected TBaseClient(
+ TProtocol inputProtocol,
+ TProtocol outputProtocol,
+ TTransport transport,
+ TProtocolFactory inputProtocolFactory,
+ TProtocolFactory outputProtocolFactory)
+ : this(inputProtocol, outputProtocol)
+ {
+ _perCallTransportProvider = transport as
ITPerCallTransportProvider;
+ if (_perCallTransportProvider != null)
+ {
+ _inputProtocolFactory = inputProtocolFactory ?? throw new
ArgumentNullException(nameof(inputProtocolFactory));
+ _outputProtocolFactory = outputProtocolFactory ?? throw new
ArgumentNullException(nameof(outputProtocolFactory));
+ }
+ }
+
+ /// <summary>
+ /// Initializes a client with protocols created from the supplied
transport and protocol factories.
+ /// </summary>
+ /// <param name="transport">
+ /// The shared transport. When it implements <see
cref="ITPerCallTransportProvider"/>,
+ /// each client call uses protocols created for a new per-call
transport.
+ /// </param>
+ /// <param name="inputProtocolFactory">The factory used to create
input protocols.</param>
+ /// <param name="outputProtocolFactory">The factory used to create
output protocols.</param>
+ /// <exception cref="ArgumentNullException">
+ /// Thrown when <paramref name="transport"/>, <paramref
name="inputProtocolFactory"/>,
+ /// or <paramref name="outputProtocolFactory"/> is <see
langword="null"/>.
+ /// </exception>
+ /// <exception cref="InvalidOperationException">
+ /// Thrown when a protocol factory returns <see langword="null"/>.
+ /// </exception>
+ protected TBaseClient(
+ TTransport transport,
+ TProtocolFactory inputProtocolFactory,
+ TProtocolFactory outputProtocolFactory)
+ {
+ if (transport == null) throw new
ArgumentNullException(nameof(transport));
+ if (inputProtocolFactory == null) throw new
ArgumentNullException(nameof(inputProtocolFactory));
+ if (outputProtocolFactory == null) throw new
ArgumentNullException(nameof(outputProtocolFactory));
+
+ var protocols = CreateProtocols(transport, inputProtocolFactory,
outputProtocolFactory);
+ _inputProtocol = protocols.InputProtocol;
+ _outputProtocol = protocols.OutputProtocol;
+ _perCallTransportProvider = transport as
ITPerCallTransportProvider;
+ if (_perCallTransportProvider != null)
+ {
+ _inputProtocolFactory = inputProtocolFactory;
+ _outputProtocolFactory = outputProtocolFactory;
+ }
+ }
+
+ private static ProtocolPair CreateProtocols(
+ TTransport transport,
+ TProtocolFactory inputProtocolFactory,
+ TProtocolFactory outputProtocolFactory)
+ {
+ TProtocol inputProtocol = null;
+ TProtocol outputProtocol = null;
+ try
+ {
+ inputProtocol = inputProtocolFactory.GetProtocol(transport)
+ ?? throw new InvalidOperationException("The input protocol
factory returned null.");
+ outputProtocol = outputProtocolFactory.GetProtocol(transport)
+ ?? throw new InvalidOperationException("The output
protocol factory returned null.");
+ return new ProtocolPair(inputProtocol, outputProtocol);
+ }
+ catch
+ {
+ outputProtocol?.Dispose();
+ if (inputProtocol != null && !ReferenceEquals(inputProtocol,
outputProtocol))
+ {
+ inputProtocol.Dispose();
+ }
+ if (inputProtocol == null && outputProtocol == null)
+ {
+ transport.Dispose();
+ }
+ throw;
+ }
+ }
- public TProtocol OutputProtocol => _outputProtocol;
+ private sealed class ProtocolPair
+ {
+ public ProtocolPair(TProtocol inputProtocol, TProtocol
outputProtocol)
+ {
+ InputProtocol = inputProtocol;
+ OutputProtocol = outputProtocol;
+ }
+
+ public TProtocol InputProtocol { get; }
+ public TProtocol OutputProtocol { get; }
+ }
+ /// <summary>
+ /// Gets the input protocol for the current call, or the shared input
protocol when no
+ /// per-call transport is active.
+ /// </summary>
+ public TProtocol InputProtocol =>
_perCallProtocols.Value?.InputProtocol ?? _inputProtocol;
+
+ /// <summary>
+ /// Gets the output protocol for the current call, or the shared
output protocol when no
+ /// per-call transport is active.
+ /// </summary>
+ public TProtocol OutputProtocol =>
_perCallProtocols.Value?.OutputProtocol ?? _outputProtocol;
+
+ /// <summary>
+ /// Gets the next client sequence identifier.
+ /// </summary>
public int SeqId
{
- get { return ++_seqId; }
+ get { return Interlocked.Increment(ref _seqId); }
+ }
+
+ /// <summary>
+ /// Executes an operation using protocols scoped to a new per-call
transport when the
+ /// shared transport supports per-call transports.
+ /// </summary>
+ /// <param name="operation">The operation to execute.</param>
+ /// <param name="cancellationToken">The token used to cancel per-call
transport creation.</param>
+ /// <returns>A task that represents the asynchronous
operation.</returns>
+ protected async Task ExecutePerCallAsync(Func<Task> operation,
CancellationToken cancellationToken)
+ {
+ await ExecutePerCallCoreAsync(async () =>
+ {
+ await operation();
+ return true;
+ }, cancellationToken);
+ }
+
+ /// <summary>
+ /// Executes an operation using protocols scoped to a new per-call
transport when the
+ /// shared transport supports per-call transports.
+ /// </summary>
+ /// <typeparam name="TResult">The type of result returned by the
operation.</typeparam>
+ /// <param name="operation">The operation to execute.</param>
+ /// <param name="cancellationToken">The token used to cancel per-call
transport creation.</param>
+ /// <returns>A task that contains the operation result.</returns>
+ protected Task<TResult> ExecutePerCallAsync<TResult>(
+ Func<Task<TResult>> operation,
+ CancellationToken cancellationToken)
+ {
+ return ExecutePerCallCoreAsync(operation, cancellationToken);
+ }
+
+ private async Task<TResult> ExecutePerCallCoreAsync<TResult>(
+ Func<Task<TResult>> operation,
+ CancellationToken cancellationToken)
+ {
+ if (_perCallTransportProvider == null)
+ return await operation();
+
+ var transport = await
_perCallTransportProvider.CreatePerCallTransportAsync(cancellationToken);
Review Comment:
Validate the provider result before invoking either protocol factory.
`ITPerCallTransportProvider` is a public extension point and can return `null`;
in that case the factory failure path reaches `transport.Dispose()` and throws
a secondary `NullReferenceException`, masking the actual contract violation.
This should fail deterministically, just as null protocol-factory results do
below.
--
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]
