ClientPipeline
Represents an extensible pipeline used by clients that call cloud services
to send and receive HTTP request and responses. Creators of
ClientPipeline can modify how it process a
PipelineMessage by adding PipelinePolicy
instances at various points in the default pipeline.
using System.ClientModel.Internal;
using System.Collections;
using System.Collections.Generic;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading.Tasks;
namespace System.ClientModel.Primitives
{
[System.Runtime.CompilerServices.NullableContext(1)]
[System.Runtime.CompilerServices.Nullable(0)]
public sealed class ClientPipeline
{
[System.Runtime.CompilerServices.Nullable(0)]
private struct PipelineProcessor : IReadOnlyList<PipelinePolicy>, IEnumerable<PipelinePolicy>, IEnumerable, IReadOnlyCollection<PipelinePolicy>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})]
private readonly ReadOnlyMemory<PipelinePolicy> _policies;
[System.Runtime.CompilerServices.Nullable(2)]
private PolicyEnumerator _enumerator;
public PipelinePolicy this[int index] {
get {
return _policies.Span[index];
}
}
public int Count => _policies.Length;
public PipelineProcessor([System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})] ReadOnlyMemory<PipelinePolicy> policies)
{
_enumerator = null;
_policies = policies;
}
public IEnumerator<PipelinePolicy> GetEnumerator()
{
return _enumerator ?? (_enumerator = new PolicyEnumerator(this));
}
IEnumerator IEnumerable.GetEnumerator()
{
return GetEnumerator();
}
}
[System.Runtime.CompilerServices.Nullable(0)]
private class PolicyEnumerator : IEnumerator<PipelinePolicy>, IEnumerator, IDisposable
{
private readonly IReadOnlyList<PipelinePolicy> _policies;
private int _current;
public PipelinePolicy Current {
get {
if (_current >= 0 && _current < _policies.Count)
return _policies[_current];
throw new InvalidOperationException("'Current' is outside the bounds of the policy collection.");
}
}
object IEnumerator.Current {
get {
return Current;
}
}
public PolicyEnumerator(IReadOnlyList<PipelinePolicy> policies)
{
_policies = policies;
_current = -1;
}
public bool MoveNext()
{
return ++_current < _policies.Count;
}
public void Reset()
{
_current = -1;
}
public void Dispose()
{
}
}
[System.Runtime.CompilerServices.Nullable(0)]
internal class RequestOptionsProcessor : IReadOnlyList<PipelinePolicy>, IEnumerable<PipelinePolicy>, IEnumerable, IReadOnlyCollection<PipelinePolicy>
{
private readonly int _perCallIndex;
private readonly int _perTryIndex;
private readonly int _beforeTransportIndex;
private readonly int _length;
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})]
private readonly ReadOnlyMemory<PipelinePolicy> _fixedPolicies;
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})]
private readonly ReadOnlyMemory<PipelinePolicy> _customPerCallPolicies;
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})]
private readonly ReadOnlyMemory<PipelinePolicy> _customPerTryPolicies;
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})]
private readonly ReadOnlyMemory<PipelinePolicy> _customBeforeTransportPolicies;
[System.Runtime.CompilerServices.Nullable(2)]
private PolicyEnumerator _enumerator;
public PipelinePolicy this[int index] {
get {
TryGetPolicy(index, out PipelinePolicy policy);
return policy;
}
}
public int Count => _length;
public RequestOptionsProcessor([System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})] ReadOnlyMemory<PipelinePolicy> fixedPolicies, [System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})] ReadOnlyMemory<PipelinePolicy> perCallPolicies, [System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})] ReadOnlyMemory<PipelinePolicy> perTryPolicies, [System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})] ReadOnlyMemory<PipelinePolicy> beforeTransportPolicies, int perCallIndex, int perTryIndex, int beforeTransportIndex)
{
_fixedPolicies = fixedPolicies;
_customPerCallPolicies = perCallPolicies;
_customPerTryPolicies = perTryPolicies;
_customBeforeTransportPolicies = beforeTransportPolicies;
_perCallIndex = perCallIndex;
_perTryIndex = perTryIndex;
_beforeTransportIndex = beforeTransportIndex;
_length = _fixedPolicies.Length + _customPerCallPolicies.Length + _customPerTryPolicies.Length + _customBeforeTransportPolicies.Length;
}
public IEnumerator<PipelinePolicy> GetEnumerator()
{
return _enumerator ?? (_enumerator = new PolicyEnumerator(this));
}
IEnumerator IEnumerable.GetEnumerator()
{
return GetEnumerator();
}
private bool TryGetPolicy(int index, out PipelinePolicy policy)
{
if (TryGetFixedPerCallPolicy(index, out policy))
return true;
if (TryGetCustomPerCallPolicy(index, out policy))
return true;
if (TryGetFixedPerTryPolicy(index, out policy))
return true;
if (TryGetCustomPerTryPolicy(index, out policy))
return true;
if (TryGetFixedPerTransportPolicy(index, out policy))
return true;
if (TryGetCustomBeforeTransportPolicy(index, out policy))
return true;
if (TryGetFixedTransportPolicy(index, out policy))
return true;
policy = null;
return false;
}
private bool TryGetFixedPerCallPolicy(int index, out PipelinePolicy policy)
{
if (index < _perCallIndex) {
policy = _fixedPolicies.Span[index];
return true;
}
policy = null;
return false;
}
private bool TryGetCustomPerCallPolicy(int index, out PipelinePolicy policy)
{
if (index < _perCallIndex + _customPerCallPolicies.Length) {
policy = _customPerCallPolicies.Span[index - _perCallIndex];
return true;
}
policy = null;
return false;
}
private bool TryGetFixedPerTryPolicy(int index, out PipelinePolicy policy)
{
if (index < _perTryIndex + _customPerCallPolicies.Length) {
policy = _fixedPolicies.Span[index - _customPerCallPolicies.Length];
return true;
}
policy = null;
return false;
}
private bool TryGetCustomPerTryPolicy(int index, out PipelinePolicy policy)
{
if (index < _perTryIndex + _customPerCallPolicies.Length + _customPerTryPolicies.Length) {
policy = _customPerTryPolicies.Span[index - (_perTryIndex + _customPerCallPolicies.Length)];
return true;
}
policy = null;
return false;
}
private bool TryGetFixedPerTransportPolicy(int index, out PipelinePolicy policy)
{
if (index < _beforeTransportIndex + _customPerCallPolicies.Length + _customPerTryPolicies.Length) {
policy = _fixedPolicies.Span[index - (_customPerCallPolicies.Length + _customPerTryPolicies.Length)];
return true;
}
policy = null;
return false;
}
private bool TryGetCustomBeforeTransportPolicy(int index, out PipelinePolicy policy)
{
if (index < _beforeTransportIndex + _customPerCallPolicies.Length + _customPerTryPolicies.Length + _customBeforeTransportPolicies.Length) {
policy = _customBeforeTransportPolicies.Span[index - (_beforeTransportIndex + _customPerCallPolicies.Length + _customPerTryPolicies.Length)];
return true;
}
policy = null;
return false;
}
private bool TryGetFixedTransportPolicy(int index, out PipelinePolicy policy)
{
if (index < _length) {
policy = _fixedPolicies.Span[index - (_customPerCallPolicies.Length + _customPerTryPolicies.Length + _customBeforeTransportPolicies.Length)];
return true;
}
policy = null;
return false;
}
}
private readonly int _perCallIndex;
private readonly int _perTryIndex;
private readonly int _beforeTransportIndex;
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})]
private readonly ReadOnlyMemory<PipelinePolicy> _policies;
private readonly PipelineTransport _transport;
private readonly bool _enableLogging;
private readonly TimeSpan _networkTimeout;
internal static TimeSpan DefaultNetworkTimeout { get; } = TimeSpan.FromSeconds(100);
private ClientPipeline([System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})] ReadOnlyMemory<PipelinePolicy> policies, TimeSpan networkTimeout, int perCallIndex, int perTryIndex, int beforeTransportIndex, bool enableLogging)
{
ReadOnlySpan<PipelinePolicy> span = policies.Span;
if (!(span[policies.Length - 1] is PipelineTransport))
throw new ArgumentException("The last policy must be of type 'PipelineTransport'.", "policies");
span = policies.Span;
_transport = (PipelineTransport)span[policies.Length - 1];
_policies = policies;
_perCallIndex = perCallIndex;
_perTryIndex = perTryIndex;
_beforeTransportIndex = beforeTransportIndex;
_networkTimeout = networkTimeout;
_enableLogging = enableLogging;
}
public static ClientPipeline Create([System.Runtime.CompilerServices.Nullable(2)] ClientPipelineOptions options = null)
{
return Create(options ?? ClientPipelineOptions.Default, ReadOnlySpan<PipelinePolicy>.Empty, ReadOnlySpan<PipelinePolicy>.Empty, ReadOnlySpan<PipelinePolicy>.Empty);
}
public static ClientPipeline Create(ClientPipelineOptions options, [System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})] ReadOnlySpan<PipelinePolicy> perCallPolicies, [System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})] ReadOnlySpan<PipelinePolicy> perTryPolicies, [System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})] ReadOnlySpan<PipelinePolicy> beforeTransportPolicies)
{
Argument.AssertNotNull(options, "options");
options.Freeze();
options.ClientLoggingOptions?.ValidateOptions();
int num = perCallPolicies.Length + perTryPolicies.Length + beforeTransportPolicies.Length;
PipelinePolicy[] perTryPolicies2 = options.PerTryPolicies;
int num2 = num + ((perTryPolicies2 != null) ? perTryPolicies2.Length : 0);
PipelinePolicy[] perCallPolicies2 = options.PerCallPolicies;
int num3 = num2 + ((perCallPolicies2 != null) ? perCallPolicies2.Length : 0);
PipelinePolicy[] beforeTransportPolicies2 = options.BeforeTransportPolicies;
PipelinePolicy[] array = new PipelinePolicy[num3 + ((beforeTransportPolicies2 != null) ? beforeTransportPolicies2.Length : 0) + 1 + (options.AddMessageLoggingPolicy ? 1 : 0) + 1];
int num4 = 0;
perCallPolicies.CopyTo(array.AsSpan(num4));
num4 += perCallPolicies.Length;
if (options.PerCallPolicies != null) {
options.PerCallPolicies.CopyTo(array.AsSpan(num4));
num4 += options.PerCallPolicies.Length;
}
int perCallIndex = num4;
array[num4++] = (options.RetryPolicy ?? options.GetClientRetryPolicy());
perTryPolicies.CopyTo(array.AsSpan(num4));
num4 += perTryPolicies.Length;
if (options.PerTryPolicies != null) {
options.PerTryPolicies.CopyTo(array.AsSpan(num4));
num4 += options.PerTryPolicies.Length;
}
int perTryIndex = num4;
if (options.AddMessageLoggingPolicy)
array[num4++] = (options.MessageLoggingPolicy ?? options.GetMessageLoggingPolicy());
beforeTransportPolicies.CopyTo(array.AsSpan(num4));
num4 += beforeTransportPolicies.Length;
if (options.BeforeTransportPolicies != null) {
options.BeforeTransportPolicies.CopyTo(array.AsSpan(num4));
num4 += options.BeforeTransportPolicies.Length;
}
int beforeTransportIndex = num4;
array[num4++] = (options.Transport ?? options.GetHttpClientPipelineTransport());
bool valueOrDefault = (options.ClientLoggingOptions?.EnableLogging).GetValueOrDefault(true);
return new ClientPipeline(array, options.NetworkTimeout ?? DefaultNetworkTimeout, perCallIndex, perTryIndex, beforeTransportIndex, valueOrDefault);
}
public PipelineMessage CreateMessage()
{
PipelineMessage pipelineMessage = _transport.CreateMessage();
pipelineMessage.NetworkTimeout = _networkTimeout;
return pipelineMessage;
}
public PipelineMessage CreateMessage(Uri uri, string method, [System.Runtime.CompilerServices.Nullable(2)] PipelineMessageClassifier classifier = null)
{
Argument.AssertNotNull(uri, "uri");
Argument.AssertNotNull(method, "method");
PipelineMessage pipelineMessage = CreateMessage();
pipelineMessage.Request.Uri = uri;
pipelineMessage.Request.Method = method;
pipelineMessage.ResponseClassifier = (classifier ?? PipelineMessageClassifier.Default);
return pipelineMessage;
}
public void Send(PipelineMessage message)
{
Argument.AssertNotNull(message, "message");
message.Request.ClientRequestId = (Activity.Current?.Id ?? Guid.NewGuid().ToString());
IReadOnlyList<PipelinePolicy> processor = GetProcessor(message);
processor[0].Process(message, processor, 0);
}
[AsyncStateMachine(typeof(<SendAsync>d__16))]
public ValueTask SendAsync(PipelineMessage message)
{
<SendAsync>d__16 stateMachine = default(<SendAsync>d__16);
stateMachine.<>t__builder = AsyncValueTaskMethodBuilder.Create();
stateMachine.<>4__this = this;
stateMachine.message = message;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
private IReadOnlyList<PipelinePolicy> GetProcessor(PipelineMessage message)
{
if (message.UseCustomRequestPipeline)
return new RequestOptionsProcessor(_policies, message.PerCallPolicies, message.PerTryPolicies, message.BeforeTransportPolicies, _perCallIndex, _perTryIndex, _beforeTransportIndex);
return new PipelineProcessor(_policies);
}
}
}