< Summary

Class:	Microsoft.Azure.ServiceBus.Core.MessageReceiver
Assembly:	Microsoft.Azure.ServiceBus
File(s):	C:\Git\azure-sdk-for-net\sdk\servicebus\Microsoft.Azure.ServiceBus\src\Core\MessageReceiver.cs
Covered lines:	0
Uncovered lines:	672
Coverable lines:	672
Total lines:	1698
Line coverage:	0% (0 of 672)
Covered branches:	0
Total branches:	296
Branch coverage:	0% (0 of 296)

Metrics

Method	Cyclomatic complexity	Line coverage	Branch coverage
.cctor()	-	0%	100%
.ctor(...)	-	0%	0%
.ctor(...)	-	0%	0%
.ctor(...)	-	0%	100%
.ctor(...)	-	0%	100%
.ctor(...)	-	0%	0%
get_RegisteredPlugins()	-	0%	100%
get_ReceiveMode()	-	0%	100%
get_PrefetchCount()	-	0%	100%
set_PrefetchCount(...)	-	0%	0%
get_LastPeekedSequenceNumber()	-	0%	100%
set_LastPeekedSequenceNumber(...)	-	0%	0%
get_Path()	-	0%	100%
get_OperationTimeout()	-	0%	100%
set_OperationTimeout(...)	-	0%	100%
get_ServiceBusConnection()	-	0%	100%
get_LockedUntilUtcInternal()	-	0%	100%
get_SessionIdInternal()	-	0%	100%
get_EntityType()	-	0%	100%
get_LinkException()	-	0%	100%
get_CbsTokenProvider()	-	0%	100%
get_ReceiveLinkManager()	-	0%	100%
get_RequestResponseLinkManager()	-	0%	100%
ReceiveAsync()	-	0%	100%
ReceiveAsync()	-	0%	0%
ReceiveAsync(...)	-	0%	100%
ReceiveAsync()	-	0%	0%
<ReceiveAsync()	-	0%	100%
ReceiveDeferredMessageAsync()	-	0%	0%
ReceiveDeferredMessageAsync()	-	0%	0%
<ReceiveDeferredMessageAsync()	-	0%	100%
CompleteAsync(...)	-	0%	100%
CompleteAsync()	-	0%	0%
AbandonAsync()	-	0%	0%
DeferAsync()	-	0%	0%
DeadLetterAsync()	-	0%	0%
DeadLetterAsync()	-	0%	0%
RenewLockAsync()	-	0%	100%
RenewLockAsync()	-	0%	0%
<RenewLockAsync()	-	0%	100%
PeekAsync()	-	0%	100%
PeekAsync(...)	-	0%	100%
PeekBySequenceNumberAsync()	-	0%	0%
PeekBySequenceNumberAsync()	-	0%	0%
<PeekBySequenceNumberAsync()	-	0%	100%
RegisterMessageHandler(...)	-	0%	100%
RegisterMessageHandler(...)	-	0%	100%
RegisterPlugin(...)	-	0%	0%
UnregisterPlugin(...)	-	0%	0%
GetSessionReceiverLinkAsync()	-	0%	0%
ExecuteRequestResponseAsync()	-	0%	0%
OnClosingAsync()	-	0%	0%
OnReceiveAsync()	-	0%	0%
OnPeekAsync()	-	0%	0%
OnReceiveDeferredMessageAsync()	-	0%	0%
OnCompleteAsync(...)	-	0%	0%
OnAbandonAsync(...)	-	0%	0%
OnDeferAsync(...)	-	0%	0%
OnDeadLetterAsync(...)	-	0%	0%
OnRenewLockAsync()	-	0%	0%
OnMessageHandler(...)	-	0%	0%
CloseSession(...)	-	0%	100%
CloseRequestResponseSession(...)	-	0%	100%
ProcessMessage()	-	0%	0%
ProcessMessages()	-	0%	0%
DisposeMessagesAsync()	-	0%	0%
DisposeMessageRequestResponseAsync()	-	0%	0%
CreateLinkAsync()	-	0%	0%
CreateRequestResponseLinkAsync()	-	0%	100%
OnSessionReceiverLinkClosed(...)	-	0%	0%
ConvertLockTokensToDeliveryTags(...)	-	0%	100%
ThrowIfNotPeekLockMode()	-	0%	0%
ThrowIfSessionLockLost()	-	0%	0%
GetAbandonOutcome(...)	-	0%	100%
GetDeferOutcome(...)	-	0%	100%
GetModifiedOutcome(...)	-	0%	0%
GetRejectedOutcome(...)	-	0%	0%

File(s)

C:\Git\azure-sdk-for-net\sdk\servicebus\Microsoft.Azure.ServiceBus\src\Core\MessageReceiver.cs

	#	Line		Line coverage
		1		// Copyright (c) Microsoft. All rights reserved.
		2		// Licensed under the MIT license. See LICENSE file in the project root for full license information.
		3
		4		namespace Microsoft.Azure.ServiceBus.Core
		5		{
		6		using System;
		7		using System.Collections.Generic;
		8		using System.Diagnostics;
		9		using System.Linq;
		10		using System.Threading;
		11		using System.Threading.Tasks;
		12		using System.Transactions;
		13		using Microsoft.Azure.Amqp;
		14		using Microsoft.Azure.Amqp.Encoding;
		15		using Microsoft.Azure.Amqp.Framing;
		16		using Microsoft.Azure.ServiceBus.Amqp;
		17		using Microsoft.Azure.ServiceBus.Primitives;
		18
		19		/// <summary>
		20		/// The MessageReceiver can be used to receive messages from Queues and Subscriptions and acknowledge them.
		21		/// </summary>
		22		/// <example>
		23		/// Create a new MessageReceiver to receive a message from a Subscription
		24		/// <code>
		25		/// IMessageReceiver messageReceiver = new MessageReceiver(
		26		///     namespaceConnectionString,
		27		///     EntityNameHelper.FormatSubscriptionPath(topicName, subscriptionName),
		28		///     ReceiveMode.PeekLock);
		29		/// </code>
		30		///
		31		/// Receive a message from the Subscription.
		32		/// <code>
		33		/// var message = await messageReceiver.ReceiveAsync();
		34		/// await messageReceiver.CompleteAsync(message.SystemProperties.LockToken);
		35		/// </code>
		36		/// </example>
		37		/// <remarks>
		38		/// The MessageReceiver provides advanced functionality that is not found in the
		39		/// <see cref="QueueClient" /> or <see cref="SubscriptionClient" />. For instance,
		40		/// <see cref="ReceiveAsync()"/>, which allows you to receive messages on demand, but also requires
		41		/// you to manually renew locks using <see cref="RenewLockAsync(Message)"/>.
		42		/// It uses AMQP protocol to communicate with service.
		43		/// </remarks>
		44		public class MessageReceiver : ClientEntity, IMessageReceiver
		45		{
	0	46		private static readonly TimeSpan DefaultBatchFlushInterval = TimeSpan.FromMilliseconds(20);
		47
		48		readonly ConcurrentExpiringSet<Guid> requestResponseLockedMessages;
		49		readonly bool isSessionReceiver;
		50		readonly object messageReceivePumpSyncLock;
		51		readonly ActiveClientLinkManager clientLinkManager;
		52		readonly ServiceBusDiagnosticSource diagnosticSource;
		53
		54		int prefetchCount;
		55		long lastPeekedSequenceNumber;
		56		MessageReceivePump receivePump;
		57		CancellationTokenSource receivePumpCancellationTokenSource;
		58
		59		/// <summary>
		60		/// Creates a new MessageReceiver from a <see cref="ServiceBusConnectionStringBuilder"/>.
		61		/// </summary>
		62		/// <param name="connectionStringBuilder">The <see cref="ServiceBusConnectionStringBuilder"/> having entity leve
		63		/// <param name="receiveMode">The <see cref="ServiceBus.ReceiveMode"/> used to specify how messages are received
		64		/// <param name="retryPolicy">The <see cref="RetryPolicy"/> that will be used when communicating with Service Bu
		65		/// <param name="prefetchCount">The <see cref="PrefetchCount"/> that specifies the upper limit of messages this
		66		/// will actively receive regardless of whether a receive operation is pending. Defaults to 0.</param>
		67		/// <remarks>Creates a new connection to the entity, which is opened during the first operation.</remarks>
		68		public MessageReceiver(
		69		ServiceBusConnectionStringBuilder connectionStringBuilder,
		70		ReceiveMode receiveMode = ReceiveMode.PeekLock,
		71		RetryPolicy retryPolicy = null,
		72		int prefetchCount = Constants.DefaultClientPrefetchCount)
	0	73		: this(connectionStringBuilder?.GetNamespaceConnectionString(), connectionStringBuilder?.EntityPath, receive
		74		{
	0	75		}
		76
		77		/// <summary>
		78		/// Creates a new MessageReceiver from a specified connection string and entity path.
		79		/// </summary>
		80		/// <param name="connectionString">Namespace connection string used to communicate with Service Bus. Must not co
		81		/// <param name="entityPath">The path of the entity for this receiver. For Queues this will be the name, but for
		82		/// You can use <see cref="EntityNameHelper.FormatSubscriptionPath(string, string)"/>, to help create this path.
		83		/// <param name="receiveMode">The <see cref="ServiceBus.ReceiveMode"/> used to specify how messages are received
		84		/// <param name="retryPolicy">The <see cref="RetryPolicy"/> that will be used when communicating with Service Bu
		85		/// <param name="prefetchCount">The <see cref="PrefetchCount"/> that specifies the upper limit of messages this
		86		/// will actively receive regardless of whether a receive operation is pending. Defaults to 0.</param>
		87		/// <remarks>Creates a new connection to the entity, which is opened during the first operation.</remarks>
		88		public MessageReceiver(
		89		string connectionString,
		90		string entityPath,
		91		ReceiveMode receiveMode = ReceiveMode.PeekLock,
		92		RetryPolicy retryPolicy = null,
		93		int prefetchCount = Constants.DefaultClientPrefetchCount)
	0	94		: this(entityPath, null, receiveMode, new ServiceBusConnection(connectionString), null, retryPolicy, prefetc
		95		{
	0	96		if (string.IsNullOrWhiteSpace(connectionString))
		97		{
	0	98		throw Fx.Exception.ArgumentNullOrWhiteSpace(connectionString);
		99		}
		100
	0	101		this.OwnsConnection = true;
	0	102		}
		103
		104		/// <summary>
		105		/// Creates a new MessageReceiver from a specified endpoint, entity path, and token provider.
		106		/// </summary>
		107		/// <param name="endpoint">Fully qualified domain name for Service Bus. Most likely, {yournamespace}.servicebus.
		108		/// <param name="entityPath">Queue path.</param>
		109		/// <param name="tokenProvider">Token provider which will generate security tokens for authorization.</param>
		110		/// <param name="transportType">Transport type.</param>
		111		/// <param name="receiveMode">Mode of receive of messages. Defaults to <see cref="ReceiveMode"/>.PeekLock.</para
		112		/// <param name="retryPolicy">Retry policy for queue operations. Defaults to <see cref="RetryPolicy.Default"/></
		113		/// <param name="prefetchCount">The <see cref="PrefetchCount"/> that specifies the upper limit of messages this
		114		/// will actively receive regardless of whether a receive operation is pending. Defaults to 0.</param>
		115		/// <remarks>Creates a new connection to the entity, which is opened during the first operation.</remarks>
		116		public MessageReceiver(
		117		string endpoint,
		118		string entityPath,
		119		ITokenProvider tokenProvider,
		120		TransportType transportType = TransportType.Amqp,
		121		ReceiveMode receiveMode = ReceiveMode.PeekLock,
		122		RetryPolicy retryPolicy = null,
		123		int prefetchCount = Constants.DefaultClientPrefetchCount)
	0	124		: this(entityPath, null, receiveMode, new ServiceBusConnection(endpoint, transportType, retryPolicy) {TokenP
		125		{
	0	126		this.OwnsConnection = true;
	0	127		}
		128
		129		/// <summary>
		130		/// Creates a new AMQP MessageReceiver on a given <see cref="ServiceBusConnection"/>
		131		/// </summary>
		132		/// <param name="serviceBusConnection">Connection object to the service bus namespace.</param>
		133		/// <param name="entityPath">The path of the entity for this receiver. For Queues this will be the name, but for
		134		/// You can use <see cref="EntityNameHelper.FormatSubscriptionPath(string, string)"/>, to help create this path.
		135		/// <param name="receiveMode">The <see cref="ServiceBus.ReceiveMode"/> used to specify how messages are received
		136		/// <param name="retryPolicy">The <see cref="RetryPolicy"/> that will be used when communicating with Service Bu
		137		/// <param name="prefetchCount">The <see cref="PrefetchCount"/> that specifies the upper limit of messages this
		138		/// will actively receive regardless of whether a receive operation is pending. Defaults to 0.</param>
		139		public MessageReceiver(
		140		ServiceBusConnection serviceBusConnection,
		141		string entityPath,
		142		ReceiveMode receiveMode = ReceiveMode.PeekLock,
		143		RetryPolicy retryPolicy = null,
		144		int prefetchCount = Constants.DefaultClientPrefetchCount)
	0	145		: this(entityPath, null, receiveMode, serviceBusConnection, null, retryPolicy, prefetchCount)
		146		{
	0	147		this.OwnsConnection = false;
	0	148		}
		149
		150		internal MessageReceiver(
		151		string entityPath,
		152		MessagingEntityType? entityType,
		153		ReceiveMode receiveMode,
		154		ServiceBusConnection serviceBusConnection,
		155		ICbsTokenProvider cbsTokenProvider,
		156		RetryPolicy retryPolicy,
		157		int prefetchCount = Constants.DefaultClientPrefetchCount,
		158		string sessionId = null,
		159		bool isSessionReceiver = false)
	0	160		: base(nameof(MessageReceiver), entityPath, retryPolicy ?? RetryPolicy.Default)
		161		{
	0	162		MessagingEventSource.Log.MessageReceiverCreateStart(serviceBusConnection?.Endpoint.Authority, entityPath, re
		163
	0	164		if (string.IsNullOrWhiteSpace(entityPath))
		165		{
	0	166		throw Fx.Exception.ArgumentNullOrWhiteSpace(entityPath);
		167		}
		168
	0	169		this.ServiceBusConnection = serviceBusConnection ?? throw new ArgumentNullException(nameof(serviceBusConnect
	0	170		this.ReceiveMode = receiveMode;
	0	171		this.Path = entityPath;
	0	172		this.EntityType = entityType;
	0	173		this.ServiceBusConnection.ThrowIfClosed();
		174
	0	175		if (cbsTokenProvider != null)
		176		{
	0	177		this.CbsTokenProvider = cbsTokenProvider;
		178		}
	0	179		else if (this.ServiceBusConnection.TokenProvider != null)
		180		{
	0	181		this.CbsTokenProvider = new TokenProviderAdapter(this.ServiceBusConnection.TokenProvider, this.ServiceBu
		182		}
		183		else
		184		{
	0	185		throw new ArgumentNullException($"{nameof(ServiceBusConnection)} doesn't have a valid token provider");
		186		}
		187
	0	188		this.SessionIdInternal = sessionId;
	0	189		this.isSessionReceiver = isSessionReceiver;
	0	190		this.ReceiveLinkManager = new FaultTolerantAmqpObject<ReceivingAmqpLink>(this.CreateLinkAsync, CloseSession)
	0	191		this.RequestResponseLinkManager = new FaultTolerantAmqpObject<RequestResponseAmqpLink>(this.CreateRequestRes
	0	192		this.requestResponseLockedMessages = new ConcurrentExpiringSet<Guid>();
	0	193		this.PrefetchCount = prefetchCount;
	0	194		this.messageReceivePumpSyncLock = new object();
	0	195		this.clientLinkManager = new ActiveClientLinkManager(this, this.CbsTokenProvider);
	0	196		this.diagnosticSource = new ServiceBusDiagnosticSource(entityPath, serviceBusConnection.Endpoint);
	0	197		MessagingEventSource.Log.MessageReceiverCreateStop(serviceBusConnection.Endpoint.Authority, entityPath, this
	0	198		}
		199
		200		/// <summary>
		201		/// Gets a list of currently registered plugins.
		202		/// </summary>
	0	203		public override IList<ServiceBusPlugin> RegisteredPlugins { get; } = new List<ServiceBusPlugin>();
		204
		205		/// <summary>
		206		/// Gets the <see cref="ServiceBus.ReceiveMode"/> of the current receiver.
		207		/// </summary>
	0	208		public ReceiveMode ReceiveMode { get; protected set; }
		209
		210		/// <summary>
		211		/// Prefetch speeds up the message flow by aiming to have a message readily available for local retrieval when a
		212		/// Setting a non-zero value prefetches PrefetchCount number of messages.
		213		/// Setting the value to zero turns prefetch off.
		214		/// Defaults to 0.
		215		/// </summary>
		216		/// <remarks>
		217		/// <para>
		218		/// When Prefetch is enabled, the receiver will quietly acquire more messages, up to the PrefetchCount limit, th
		219		/// immediately asks for. A single initial Receive/ReceiveAsync call will therefore acquire a message for immedi
		220		/// that will be returned as soon as available, and the client will proceed to acquire further messages to fill
		221		/// </para>
		222		/// <para>
		223		/// While messages are available in the prefetch buffer, any subsequent ReceiveAsync calls will be immediately s
		224		/// replenished in the background as space becomes available.If there are no messages available for delivery, th
		225		/// buffer and then wait or block as expected.
		226		/// </para>
		227		/// <para>Prefetch also works equivalently with the <see cref="RegisterMessageHandler(Func{Message,CancellationT
		228		/// <para>Updates to this value take effect on the next receive call to the service.</para>
		229		/// </remarks>
		230		public int PrefetchCount
		231		{
	0	232		get => this.prefetchCount;
		233
		234		set
		235		{
	0	236		if (value < 0)
		237		{
	0	238		throw Fx.Exception.ArgumentOutOfRange(nameof(this.PrefetchCount), value, "Value cannot be less than
		239		}
	0	240		this.prefetchCount = value;
	0	241		if (this.ReceiveLinkManager.TryGetOpenedObject(out var link))
		242		{
	0	243		link.SetTotalLinkCredit((uint)value, true, true);
		244		}
	0	245		}
		246		}
		247
		248		/// <summary>Gets the sequence number of the last peeked message.</summary>
		249		/// <seealso cref="PeekAsync()"/>
		250		public long LastPeekedSequenceNumber
		251		{
	0	252		get => this.lastPeekedSequenceNumber;
		253
		254		internal set
		255		{
	0	256		if (value < 0)
		257		{
	0	258		throw new ArgumentOutOfRangeException(nameof(this.LastPeekedSequenceNumber), value.ToString());
		259		}
		260
	0	261		this.lastPeekedSequenceNumber = value;
	0	262		}
		263		}
		264
		265		/// <summary>The path of the entity for this receiver. For Queues this will be the name, but for Subscriptions t
	0	266		public override string Path { get; }
		267
		268		/// <summary>
		269		/// Duration after which individual operations will timeout.
		270		/// </summary>
		271		public override TimeSpan OperationTimeout {
	0	272		get => this.ServiceBusConnection.OperationTimeout;
	0	273		set => this.ServiceBusConnection.OperationTimeout = value;
		274		}
		275
		276		/// <summary>
		277		/// Connection object to the service bus namespace.
		278		/// </summary>
	0	279		public override ServiceBusConnection ServiceBusConnection { get; }
		280
		281		/// <summary>
		282		/// Gets the DateTime that the current receiver is locked until. This is only applicable when Sessions are used.
		283		/// </summary>
	0	284		internal DateTime LockedUntilUtcInternal { get; set; }
		285
		286		/// <summary>
		287		/// Gets the SessionId of the current receiver. This is only applicable when Sessions are used.
		288		/// </summary>
	0	289		internal string SessionIdInternal { get; set; }
		290
	0	291		internal MessagingEntityType? EntityType { get; }
		292
	0	293		Exception LinkException { get; set; }
		294
	0	295		ICbsTokenProvider CbsTokenProvider { get; }
		296
	0	297		internal FaultTolerantAmqpObject<ReceivingAmqpLink> ReceiveLinkManager { get; }
		298
	0	299		FaultTolerantAmqpObject<RequestResponseAmqpLink> RequestResponseLinkManager { get; }
		300
		301		/// <summary>
		302		/// Receive a message from the entity defined by <see cref="Path"/> using <see cref="ReceiveMode"/> mode.
		303		/// </summary>
		304		/// <returns>The message received. Returns null if no message is found.</returns>
		305		/// <remarks>Operation will time out after duration of <see cref="ClientEntity.OperationTimeout"/></remarks>
		306		public Task<Message> ReceiveAsync()
		307		{
	0	308		return this.ReceiveAsync(this.OperationTimeout);
		309		}
		310
		311		/// <summary>
		312		/// Receive a message from the entity defined by <see cref="Path"/> using <see cref="ReceiveMode"/> mode.
		313		/// </summary>
		314		/// <param name="operationTimeout">The time span the client waits for receiving a message before it times out.</
		315		/// <returns>The message received. Returns null if no message is found.</returns>
		316		/// <remarks>
		317		/// The parameter <paramref name="operationTimeout"/> includes the time taken by the receiver to establish a con
		318		/// (either during the first receive or when connection needs to be re-established). If establishing the connect
		319		/// times out, this will throw <see cref="ServiceBusTimeoutException"/>.
		320		/// </remarks>
		321		public async Task<Message> ReceiveAsync(TimeSpan operationTimeout)
		322		{
	0	323		var messages = await this.ReceiveAsync(1, operationTimeout).ConfigureAwait(false);
	0	324		if (messages != null && messages.Count > 0)
		325		{
	0	326		return messages[0];
		327		}
		328
	0	329		return null;
	0	330		}
		331
		332		/// <summary>
		333		/// Receives a maximum of <paramref name="maxMessageCount"/> messages from the entity defined by <see cref="Path
		334		/// </summary>
		335		/// <param name="maxMessageCount">The maximum number of messages that will be received.</param>
		336		/// <returns>List of messages received. Returns null if no message is found.</returns>
		337		/// <remarks>Receiving less than <paramref name="maxMessageCount"/> messages is not an indication of empty entit
		338		public Task<IList<Message>> ReceiveAsync(int maxMessageCount)
		339		{
	0	340		return this.ReceiveAsync(maxMessageCount, this.OperationTimeout);
		341		}
		342
		343		/// <summary>
		344		/// Receives a maximum of <paramref name="maxMessageCount"/> messages from the entity defined by <see cref="Path
		345		/// </summary>
		346		/// <param name="maxMessageCount">The maximum number of messages that will be received.</param>
		347		/// <param name="operationTimeout">The time span the client waits for receiving a message before it times out.</
		348		/// <returns>List of messages received. Returns null if no message is found.</returns>
		349		/// <remarks>Receiving less than <paramref name="maxMessageCount"/> messages is not an indication of empty entit
		350		/// The parameter <paramref name="operationTimeout"/> includes the time taken by the receiver to establish a con
		351		/// (either during the first receive or when connection needs to be re-established). If establishing the connect
		352		/// times out, this will throw <see cref="ServiceBusTimeoutException"/>.
		353		/// </remarks>
		354		public async Task<IList<Message>> ReceiveAsync(int maxMessageCount, TimeSpan operationTimeout)
		355		{
	0	356		this.ThrowIfClosed();
		357
	0	358		if (operationTimeout <= TimeSpan.Zero)
		359		{
	0	360		throw Fx.Exception.ArgumentOutOfRange(nameof(operationTimeout), operationTimeout, Resources.TimeoutMustB
		361		}
		362
	0	363		MessagingEventSource.Log.MessageReceiveStart(this.ClientId, maxMessageCount);
		364
	0	365		bool isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
	0	366		Activity activity = isDiagnosticSourceEnabled ? this.diagnosticSource.ReceiveStart(maxMessageCount) : null;
	0	367		Task receiveTask = null;
		368
	0	369		IList<Message> unprocessedMessageList = null;
		370		try
		371		{
	0	372		receiveTask = this.RetryPolicy.RunOperation(
	0	373		async () =>
	0	374		{
	0	375		unprocessedMessageList = await this.OnReceiveAsync(maxMessageCount, operationTimeout)
	0	376		.ConfigureAwait(false);
	0	377		}, operationTimeout);
	0	378		await receiveTask.ConfigureAwait(false);
		379
	0	380		}
	0	381		catch (Exception exception)
		382		{
	0	383		if (isDiagnosticSourceEnabled)
		384		{
	0	385		this.diagnosticSource.ReportException(exception);
		386		}
		387
	0	388		MessagingEventSource.Log.MessageReceiveException(this.ClientId, exception);
	0	389		throw;
		390		}
		391		finally
		392		{
	0	393		this.diagnosticSource.ReceiveStop(activity, maxMessageCount, receiveTask?.Status, unprocessedMessageList
		394		}
		395
	0	396		MessagingEventSource.Log.MessageReceiveStop(this.ClientId, unprocessedMessageList?.Count ?? 0);
		397
	0	398		if (unprocessedMessageList == null)
		399		{
	0	400		return unprocessedMessageList;
		401		}
		402
	0	403		return await this.ProcessMessages(unprocessedMessageList).ConfigureAwait(false);
	0	404		}
		405
		406		/// <summary>
		407		/// Receives a specific deferred message identified by <paramref name="sequenceNumber"/>.
		408		/// </summary>
		409		/// <param name="sequenceNumber">The sequence number of the message that will be received.</param>
		410		/// <returns>Message identified by sequence number <paramref name="sequenceNumber"/>. Returns null if no such me
		411		/// Throws if the message has not been deferred.</returns>
		412		/// <seealso cref="DeferAsync"/>
		413		public async Task<Message> ReceiveDeferredMessageAsync(long sequenceNumber)
		414		{
	0	415		var messages = await this.ReceiveDeferredMessageAsync(new[] { sequenceNumber }).ConfigureAwait(false);
	0	416		if (messages != null && messages.Count > 0)
		417		{
	0	418		return messages[0];
		419		}
		420
	0	421		return null;
	0	422		}
		423
		424		/// <summary>
		425		/// Receives a <see cref="IList{Message}"/> of deferred messages identified by <paramref name="sequenceNumbers"/
		426		/// </summary>
		427		/// <param name="sequenceNumbers">An <see cref="IEnumerable{T}"/> containing the sequence numbers to receive.</p
		428		/// <returns>Messages identified by sequence number are returned. Returns null if no messages are found.
		429		/// Throws if the messages have not been deferred.</returns>
		430		/// <seealso cref="DeferAsync"/>
		431		public async Task<IList<Message>> ReceiveDeferredMessageAsync(IEnumerable<long> sequenceNumbers)
		432		{
	0	433		this.ThrowIfClosed();
	0	434		this.ThrowIfNotPeekLockMode();
		435
	0	436		if (sequenceNumbers == null)
		437		{
	0	438		throw Fx.Exception.ArgumentNull(nameof(sequenceNumbers));
		439		}
	0	440		var sequenceNumberList = sequenceNumbers.ToArray();
	0	441		if (sequenceNumberList.Length == 0)
		442		{
	0	443		throw Fx.Exception.ArgumentNull(nameof(sequenceNumbers));
		444		}
		445
	0	446		MessagingEventSource.Log.MessageReceiveDeferredMessageStart(this.ClientId, sequenceNumberList.Length, sequen
		447
	0	448		bool isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
	0	449		Activity activity = isDiagnosticSourceEnabled ? this.diagnosticSource.ReceiveDeferredStart(sequenceNumberLis
	0	450		Task receiveTask = null;
		451
	0	452		IList<Message> messages = null;
		453		try
		454		{
	0	455		receiveTask = this.RetryPolicy.RunOperation(
	0	456		async () =>
	0	457		{
	0	458		messages = await this.OnReceiveDeferredMessageAsync(sequenceNumberList).ConfigureAwait(false);
	0	459		}, this.OperationTimeout);
	0	460		await receiveTask.ConfigureAwait(false);
	0	461		}
	0	462		catch (Exception exception)
		463		{
	0	464		if (isDiagnosticSourceEnabled)
		465		{
	0	466		this.diagnosticSource.ReportException(exception);
		467		}
		468
	0	469		MessagingEventSource.Log.MessageReceiveDeferredMessageException(this.ClientId, exception);
	0	470		throw;
		471		}
		472		finally
		473		{
	0	474		this.diagnosticSource.ReceiveDeferredStop(activity, sequenceNumberList, receiveTask?.Status, messages);
		475		}
	0	476		MessagingEventSource.Log.MessageReceiveDeferredMessageStop(this.ClientId, messages?.Count ?? 0);
		477
	0	478		return messages;
	0	479		}
		480
		481		/// <summary>
		482		/// Completes a <see cref="Message"/> using its lock token. This will delete the message from the service.
		483		/// </summary>
		484		/// <param name="lockToken">The lock token of the corresponding message to complete.</param>
		485		/// <remarks>
		486		/// A lock token can be found in <see cref="Message.SystemPropertiesCollection.LockToken"/>,
		487		/// only when <see cref="ReceiveMode"/> is set to <see cref="ServiceBus.ReceiveMode.PeekLock"/>.
		488		/// This operation can only be performed on messages that were received by this receiver.
		489		/// </remarks>
		490		public Task CompleteAsync(string lockToken)
		491		{
	0	492		return this.CompleteAsync(new[] { lockToken });
		493		}
		494
		495		/// <summary>
		496		/// Completes a series of <see cref="Message"/> using a list of lock tokens. This will delete the message from t
		497		/// </summary>
		498		/// <remarks>
		499		/// A lock token can be found in <see cref="Message.SystemPropertiesCollection.LockToken"/>,
		500		/// only when <see cref="ReceiveMode"/> is set to <see cref="ServiceBus.ReceiveMode.PeekLock"/>.
		501		/// This operation can only be performed on messages that were received by this receiver.
		502		/// </remarks>
		503		/// <param name="lockTokens">An <see cref="IEnumerable{T}"/> containing the lock tokens of the corresponding mes
		504		public async Task CompleteAsync(IEnumerable<string> lockTokens)
		505		{
	0	506		this.ThrowIfClosed();
	0	507		this.ThrowIfNotPeekLockMode();
	0	508		if (lockTokens == null)
		509		{
	0	510		throw Fx.Exception.ArgumentNull(nameof(lockTokens));
		511		}
	0	512		var lockTokenList = lockTokens.ToList();
	0	513		if (lockTokenList.Count == 0)
		514		{
	0	515		throw Fx.Exception.Argument(nameof(lockTokens), Resources.ListOfLockTokensCannotBeEmpty);
		516		}
		517
	0	518		MessagingEventSource.Log.MessageCompleteStart(this.ClientId, lockTokenList.Count, lockTokenList);
	0	519		bool isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
	0	520		Activity activity = isDiagnosticSourceEnabled ? this.diagnosticSource.CompleteStart(lockTokenList) : null;
	0	521		Task completeTask = null;
		522
		523		try
		524		{
	0	525		completeTask =
	0	526		this.RetryPolicy.RunOperation(() => this.OnCompleteAsync(lockTokenList), this.OperationTimeout);
	0	527		await completeTask.ConfigureAwait(false);
	0	528		}
	0	529		catch (Exception exception)
		530		{
	0	531		if (isDiagnosticSourceEnabled)
		532		{
	0	533		this.diagnosticSource.ReportException(exception);
		534		}
		535
	0	536		MessagingEventSource.Log.MessageCompleteException(this.ClientId, exception);
		537
	0	538		throw;
		539		}
		540		finally
		541		{
	0	542		this.diagnosticSource.CompleteStop(activity, lockTokenList, completeTask?.Status);
		543		}
		544
	0	545		MessagingEventSource.Log.MessageCompleteStop(this.ClientId);
	0	546		}
		547
		548		/// <summary>
		549		/// Abandons a <see cref="Message"/> using a lock token. This will make the message available again for processi
		550		/// </summary>
		551		/// <param name="lockToken">The lock token of the corresponding message to abandon.</param>
		552		/// <param name="propertiesToModify">The properties of the message to modify while abandoning the message.</para
		553		/// <remarks>A lock token can be found in <see cref="Message.SystemPropertiesCollection.LockToken"/>,
		554		/// only when <see cref="ReceiveMode"/> is set to <see cref="ServiceBus.ReceiveMode.PeekLock"/>.
		555		/// Abandoning a message will increase the delivery count on the message.
		556		/// This operation can only be performed on messages that were received by this receiver.
		557		/// </remarks>
		558		public async Task AbandonAsync(string lockToken, IDictionary<string, object> propertiesToModify = null)
		559		{
	0	560		this.ThrowIfClosed();
	0	561		this.ThrowIfNotPeekLockMode();
		562
	0	563		MessagingEventSource.Log.MessageAbandonStart(this.ClientId, 1, lockToken);
	0	564		bool isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
	0	565		Activity activity = isDiagnosticSourceEnabled ? this.diagnosticSource.DisposeStart("Abandon", lockToken) : n
	0	566		Task abandonTask = null;
		567
		568		try
		569		{
	0	570		abandonTask = this.RetryPolicy.RunOperation(() => this.OnAbandonAsync(lockToken, propertiesToModify),
	0	571		this.OperationTimeout);
	0	572		await abandonTask.ConfigureAwait(false);
	0	573		}
	0	574		catch (Exception exception)
		575		{
	0	576		if (isDiagnosticSourceEnabled)
		577		{
	0	578		this.diagnosticSource.ReportException(exception);
		579		}
		580
	0	581		MessagingEventSource.Log.MessageAbandonException(this.ClientId, exception);
	0	582		throw;
		583		}
		584		finally
		585		{
	0	586		this.diagnosticSource.DisposeStop(activity, lockToken, abandonTask?.Status);
		587		}
		588
		589
	0	590		MessagingEventSource.Log.MessageAbandonStop(this.ClientId);
	0	591		}
		592
		593		/// <summary>Indicates that the receiver wants to defer the processing for the message.</summary>
		594		/// <param name="lockToken">The lock token of the <see cref="Message" />.</param>
		595		/// <param name="propertiesToModify">The properties of the message to modify while deferring the message.</param
		596		/// <remarks>
		597		/// A lock token can be found in <see cref="Message.SystemPropertiesCollection.LockToken"/>,
		598		/// only when <see cref="ReceiveMode"/> is set to <see cref="ServiceBus.ReceiveMode.PeekLock"/>.
		599		/// In order to receive this message again in the future, you will need to save the <see cref="Message.SystemPro
		600		/// and receive it using <see cref="ReceiveDeferredMessageAsync(long)"/>.
		601		/// Deferring messages does not impact message's expiration, meaning that deferred messages can still expire.
		602		/// This operation can only be performed on messages that were received by this receiver.
		603		/// </remarks>
		604		public async Task DeferAsync(string lockToken, IDictionary<string, object> propertiesToModify = null)
		605		{
	0	606		this.ThrowIfClosed();
	0	607		this.ThrowIfNotPeekLockMode();
		608
	0	609		MessagingEventSource.Log.MessageDeferStart(this.ClientId, 1, lockToken);
	0	610		bool isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
	0	611		Activity activity = isDiagnosticSourceEnabled ? this.diagnosticSource.DisposeStart("Defer", lockToken) : nul
	0	612		Task deferTask = null;
		613
		614		try
		615		{
	0	616		deferTask = this.RetryPolicy.RunOperation(() => this.OnDeferAsync(lockToken, propertiesToModify),
	0	617		this.OperationTimeout);
	0	618		await deferTask.ConfigureAwait(false);
	0	619		}
	0	620		catch (Exception exception)
		621		{
	0	622		if (isDiagnosticSourceEnabled)
		623		{
	0	624		this.diagnosticSource.ReportException(exception);
		625		}
		626
	0	627		MessagingEventSource.Log.MessageDeferException(this.ClientId, exception);
	0	628		throw;
		629		}
		630		finally
		631		{
	0	632		this.diagnosticSource.DisposeStop(activity, lockToken, deferTask?.Status);
		633		}
	0	634		MessagingEventSource.Log.MessageDeferStop(this.ClientId);
	0	635		}
		636
		637		/// <summary>
		638		/// Moves a message to the deadletter sub-queue.
		639		/// </summary>
		640		/// <param name="lockToken">The lock token of the corresponding message to deadletter.</param>
		641		/// <param name="propertiesToModify">The properties of the message to modify while moving to sub-queue.</param>
		642		/// <remarks>
		643		/// A lock token can be found in <see cref="Message.SystemPropertiesCollection.LockToken"/>,
		644		/// only when <see cref="ReceiveMode"/> is set to <see cref="ServiceBus.ReceiveMode.PeekLock"/>.
		645		/// In order to receive a message from the deadletter queue, you will need a new <see cref="IMessageReceiver"/>,
		646		/// You can use <see cref="EntityNameHelper.FormatDeadLetterPath(string)"/> to help with this.
		647		/// This operation can only be performed on messages that were received by this receiver.
		648		/// </remarks>
		649		public async Task DeadLetterAsync(string lockToken, IDictionary<string, object> propertiesToModify = null)
		650		{
	0	651		this.ThrowIfClosed();
	0	652		this.ThrowIfNotPeekLockMode();
		653
	0	654		MessagingEventSource.Log.MessageDeadLetterStart(this.ClientId, 1, lockToken);
	0	655		bool isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
	0	656		Activity activity = isDiagnosticSourceEnabled ? this.diagnosticSource.DisposeStart("DeadLetter", lockToken)
	0	657		Task deadLetterTask = null;
		658
		659		try
		660		{
	0	661		deadLetterTask = this.RetryPolicy.RunOperation(() => this.OnDeadLetterAsync(lockToken, propertiesToModif
	0	662		this.OperationTimeout);
	0	663		await deadLetterTask.ConfigureAwait(false);
	0	664		}
	0	665		catch (Exception exception)
		666		{
	0	667		if (isDiagnosticSourceEnabled)
		668		{
	0	669		this.diagnosticSource.ReportException(exception);
		670		}
		671
	0	672		MessagingEventSource.Log.MessageDeadLetterException(this.ClientId, exception);
	0	673		throw;
		674		}
		675		finally
		676		{
	0	677		this.diagnosticSource.DisposeStop(activity, lockToken, deadLetterTask?.Status);
		678		}
	0	679		MessagingEventSource.Log.MessageDeadLetterStop(this.ClientId);
	0	680		}
		681
		682		/// <summary>
		683		/// Moves a message to the deadletter sub-queue.
		684		/// </summary>
		685		/// <param name="lockToken">The lock token of the corresponding message to deadletter.</param>
		686		/// <param name="deadLetterReason">The reason for deadlettering the message.</param>
		687		/// <param name="deadLetterErrorDescription">The error description for deadlettering the message.</param>
		688		/// <remarks>
		689		/// A lock token can be found in <see cref="Message.SystemPropertiesCollection.LockToken"/>,
		690		/// only when <see cref="ReceiveMode"/> is set to <see cref="ServiceBus.ReceiveMode.PeekLock"/>.
		691		/// In order to receive a message from the deadletter queue, you will need a new <see cref="IMessageReceiver"/>,
		692		/// You can use <see cref="EntityNameHelper.FormatDeadLetterPath(string)"/> to help with this.
		693		/// This operation can only be performed on messages that were received by this receiver.
		694		/// </remarks>
		695		public async Task DeadLetterAsync(string lockToken, string deadLetterReason, string deadLetterErrorDescription =
		696		{
	0	697		this.ThrowIfClosed();
	0	698		this.ThrowIfNotPeekLockMode();
		699
	0	700		MessagingEventSource.Log.MessageDeadLetterStart(this.ClientId, 1, lockToken);
	0	701		bool isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
	0	702		Activity activity = isDiagnosticSourceEnabled ? this.diagnosticSource.DisposeStart("DeadLetter", lockToken)
	0	703		Task deadLetterTask = null;
		704
		705		try
		706		{
	0	707		deadLetterTask =
	0	708		this.RetryPolicy.RunOperation(
	0	709		() => this.OnDeadLetterAsync(lockToken, null, deadLetterReason, deadLetterErrorDescription),
	0	710		this.OperationTimeout);
	0	711		await deadLetterTask.ConfigureAwait(false);
	0	712		}
	0	713		catch (Exception exception)
		714		{
	0	715		if (isDiagnosticSourceEnabled)
		716		{
	0	717		this.diagnosticSource.ReportException(exception);
		718		}
		719
	0	720		MessagingEventSource.Log.MessageDeadLetterException(this.ClientId, exception);
	0	721		throw;
		722		}
		723		finally
		724		{
	0	725		this.diagnosticSource.DisposeStop(activity, lockToken, deadLetterTask?.Status);
		726		}
		727
	0	728		MessagingEventSource.Log.MessageDeadLetterStop(this.ClientId);
	0	729		}
		730
		731		/// <summary>
		732		/// Renews the lock on the message specified by the lock token. The lock will be renewed based on the setting sp
		733		/// </summary>
		734		/// <remarks>
		735		/// When a message is received in <see cref="ServiceBus.ReceiveMode.PeekLock"/> mode, the message is locked on t
		736		/// receiver instance for a duration as specified during the Queue/Subscription creation (LockDuration).
		737		/// If processing of the message requires longer than this duration, the lock needs to be renewed.
		738		/// For each renewal, it resets the time the message is locked by the LockDuration set on the Entity.
		739		/// </remarks>
		740		public async Task RenewLockAsync(Message message)
		741		{
	0	742		message.SystemProperties.LockedUntilUtc = await RenewLockAsync(message.SystemProperties.LockToken).Configure
	0	743		}
		744
		745		/// <summary>
		746		/// Renews the lock on the message. The lock will be renewed based on the setting specified on the queue.
		747		/// <returns>New lock token expiry date and time in UTC format.</returns>
		748		/// </summary>
		749		/// <param name="lockToken">Lock token associated with the message.</param>
		750		/// <remarks>
		751		/// When a message is received in <see cref="ServiceBus.ReceiveMode.PeekLock"/> mode, the message is locked on t
		752		/// receiver instance for a duration as specified during the Queue/Subscription creation (LockDuration).
		753		/// If processing of the message requires longer than this duration, the lock needs to be renewed.
		754		/// For each renewal, it resets the time the message is locked by the LockDuration set on the Entity.
		755		/// </remarks>
		756		public async Task<DateTime> RenewLockAsync(string lockToken)
		757		{
	0	758		this.ThrowIfClosed();
	0	759		this.ThrowIfNotPeekLockMode();
		760
	0	761		MessagingEventSource.Log.MessageRenewLockStart(this.ClientId, 1, lockToken);
	0	762		bool isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
	0	763		Activity activity = isDiagnosticSourceEnabled ? this.diagnosticSource.RenewLockStart(lockToken) : null;
	0	764		Task renewTask = null;
		765
	0	766		var lockedUntilUtc = DateTime.MinValue;
		767
		768		try
		769		{
	0	770		renewTask = this.RetryPolicy.RunOperation(
	0	771		async () => lockedUntilUtc = await this.OnRenewLockAsync(lockToken).ConfigureAwait(false),
	0	772		this.OperationTimeout);
	0	773		await renewTask.ConfigureAwait(false);
	0	774		}
	0	775		catch (Exception exception)
		776		{
	0	777		if (isDiagnosticSourceEnabled)
		778		{
	0	779		this.diagnosticSource.ReportException(exception);
		780		}
		781
	0	782		MessagingEventSource.Log.MessageRenewLockException(this.ClientId, exception);
	0	783		throw;
		784		}
		785		finally
		786		{
	0	787		this.diagnosticSource.RenewLockStop(activity, lockToken, renewTask?.Status, lockedUntilUtc);
		788		}
	0	789		MessagingEventSource.Log.MessageRenewLockStop(this.ClientId);
		790
	0	791		return lockedUntilUtc;
	0	792		}
		793
		794		/// <summary>
		795		/// Fetches the next active message without changing the state of the receiver or the message source.
		796		/// </summary>
		797		/// <remarks>
		798		/// The first call to <see cref="PeekAsync()"/> fetches the first active message for this receiver. Each subsequ
		799		/// fetches the subsequent message in the entity.
		800		/// Unlike a received message, peeked message will not have lock token associated with it, and hence it cannot b
		801		/// Also, unlike <see cref="ReceiveAsync()"/>, this method will fetch even Deferred messages (but not Deadletter
		802		/// </remarks>
		803		/// <returns>The <see cref="Message" /> that represents the next message to be read. Returns null when nothing t
		804		public Task<Message> PeekAsync()
		805		{
	0	806		return this.PeekBySequenceNumberAsync(this.lastPeekedSequenceNumber + 1);
		807		}
		808
		809		/// <summary>
		810		/// Fetches the next batch of active messages without changing the state of the receiver or the message source.
		811		/// </summary>
		812		/// <remarks>
		813		/// The first call to <see cref="PeekAsync()"/> fetches the first active message for this receiver. Each subsequ
		814		/// fetches the subsequent message in the entity.
		815		/// Unlike a received message, peeked message will not have lock token associated with it, and hence it cannot b
		816		/// Also, unlike <see cref="ReceiveAsync()"/>, this method will fetch even Deferred messages (but not Deadletter
		817		/// </remarks>
		818		/// <returns>List of <see cref="Message" /> that represents the next message to be read. Returns null when nothi
		819		public Task<IList<Message>> PeekAsync(int maxMessageCount)
		820		{
	0	821		return this.PeekBySequenceNumberAsync(this.lastPeekedSequenceNumber + 1, maxMessageCount);
		822		}
		823
		824		/// <summary>
		825		/// Asynchronously reads the next message without changing the state of the receiver or the message source.
		826		/// </summary>
		827		/// <param name="fromSequenceNumber">The sequence number from where to read the message.</param>
		828		/// <returns>The asynchronous operation that returns the <see cref="Message" /> that represents the next message
		829		public async Task<Message> PeekBySequenceNumberAsync(long fromSequenceNumber)
		830		{
	0	831		var messages = await this.PeekBySequenceNumberAsync(fromSequenceNumber, 1).ConfigureAwait(false);
	0	832		return messages?.FirstOrDefault();
	0	833		}
		834
		835		/// <summary>Peeks a batch of messages.</summary>
		836		/// <param name="fromSequenceNumber">The starting point from which to browse a batch of messages.</param>
		837		/// <param name="messageCount">The number of messages to retrieve.</param>
		838		/// <returns>A batch of messages peeked.</returns>
		839		public async Task<IList<Message>> PeekBySequenceNumberAsync(long fromSequenceNumber, int messageCount)
		840		{
	0	841		this.ThrowIfClosed();
	0	842		IList<Message> messages = null;
		843
	0	844		MessagingEventSource.Log.MessagePeekStart(this.ClientId, fromSequenceNumber, messageCount);
	0	845		bool isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
	0	846		Activity activity = isDiagnosticSourceEnabled ? this.diagnosticSource.PeekStart(fromSequenceNumber, messageC
	0	847		Task peekTask = null;
		848
		849		try
		850		{
	0	851		peekTask = this.RetryPolicy.RunOperation(
	0	852		async () =>
	0	853		{
	0	854		messages = await this.OnPeekAsync(fromSequenceNumber, messageCount).ConfigureAwait(false);
	0	855		}, this.OperationTimeout);
		856
	0	857		await peekTask.ConfigureAwait(false);
	0	858		}
	0	859		catch (Exception exception)
		860		{
	0	861		if (isDiagnosticSourceEnabled)
		862		{
	0	863		this.diagnosticSource.ReportException(exception);
		864		}
		865
	0	866		MessagingEventSource.Log.MessagePeekException(this.ClientId, exception);
	0	867		throw;
		868		}
		869		finally
		870		{
	0	871		this.diagnosticSource.PeekStop(activity, fromSequenceNumber, messageCount, peekTask?.Status, messages);
		872		}
		873
	0	874		MessagingEventSource.Log.MessagePeekStop(this.ClientId, messages?.Count ?? 0);
	0	875		return messages;
	0	876		}
		877
		878		/// <summary>
		879		/// Receive messages continuously from the entity. Registers a message handler and begins a new thread to receiv
		880		/// This handler(<see cref="Func{Message, CancellationToken, Task}"/>) is awaited on every time a new message is
		881		/// </summary>
		882		/// <param name="handler">A <see cref="Func{T1, T2, TResult}"/> that processes messages.</param>
		883		/// <param name="exceptionReceivedHandler">A <see cref="Func{T1, TResult}"/> that is used to notify exceptions.<
		884		public void RegisterMessageHandler(Func<Message, CancellationToken, Task> handler, Func<ExceptionReceivedEventAr
		885		{
	0	886		this.RegisterMessageHandler(handler, new MessageHandlerOptions(exceptionReceivedHandler));
	0	887		}
		888
		889		/// <summary>
		890		/// Receive messages continuously from the entity. Registers a message handler and begins a new thread to receiv
		891		/// This handler(<see cref="Func{Message, CancellationToken, Task}"/>) is awaited on every time a new message is
		892		/// </summary>
		893		/// <param name="handler">A <see cref="Func{Message, CancellationToken, Task}"/> that processes messages.</param
		894		/// <param name="messageHandlerOptions">The <see cref="MessageHandlerOptions"/> options used to configure the se
		895		/// <remarks>Enable prefetch to speed up the receive rate.</remarks>
		896		public void RegisterMessageHandler(Func<Message, CancellationToken, Task> handler, MessageHandlerOptions message
		897		{
	0	898		this.ThrowIfClosed();
	0	899		this.OnMessageHandler(messageHandlerOptions, handler);
	0	900		}
		901
		902		/// <summary>
		903		/// Registers a <see cref="ServiceBusPlugin"/> to be used with this receiver.
		904		/// </summary>
		905		public override void RegisterPlugin(ServiceBusPlugin serviceBusPlugin)
		906		{
	0	907		this.ThrowIfClosed();
	0	908		if (serviceBusPlugin == null)
		909		{
	0	910		throw new ArgumentNullException(nameof(serviceBusPlugin), Resources.ArgumentNullOrWhiteSpace.FormatForUs
		911		}
	0	912		if (this.RegisteredPlugins.Any(p => p.Name == serviceBusPlugin.Name))
		913		{
	0	914		throw new ArgumentException(nameof(serviceBusPlugin), Resources.PluginAlreadyRegistered.FormatForUser(se
		915		}
	0	916		this.RegisteredPlugins.Add(serviceBusPlugin);
	0	917		}
		918
		919		/// <summary>
		920		/// Unregisters a <see cref="ServiceBusPlugin"/>.
		921		/// </summary>
		922		/// <param name="serviceBusPluginName">The <see cref="ServiceBusPlugin.Name"/> of the plugin to be unregistered.
		923		public override void UnregisterPlugin(string serviceBusPluginName)
		924		{
	0	925		this.ThrowIfClosed();
	0	926		if (this.RegisteredPlugins == null)
		927		{
	0	928		return;
		929		}
	0	930		if (string.IsNullOrWhiteSpace(serviceBusPluginName))
		931		{
	0	932		throw new ArgumentNullException(nameof(serviceBusPluginName), Resources.ArgumentNullOrWhiteSpace.FormatF
		933		}
	0	934		if (this.RegisteredPlugins.Any(p => p.Name == serviceBusPluginName))
		935		{
	0	936		var plugin = this.RegisteredPlugins.First(p => p.Name == serviceBusPluginName);
	0	937		this.RegisteredPlugins.Remove(plugin);
		938		}
	0	939		}
		940
		941		internal async Task GetSessionReceiverLinkAsync(TimeSpan serverWaitTime)
		942		{
	0	943		var timeoutHelper = new TimeoutHelper(serverWaitTime, true);
	0	944		var receivingAmqpLink = await this.ReceiveLinkManager.GetOrCreateAsync(timeoutHelper.RemainingTime()).Config
		945
	0	946		var source = (Source)receivingAmqpLink.Settings.Source;
	0	947		if (!source.FilterSet.TryGetValue<string>(AmqpClientConstants.SessionFilterName, out var tempSessionId))
		948		{
	0	949		receivingAmqpLink.Session.SafeClose();
	0	950		throw new ServiceBusException(true, Resources.SessionFilterMissing);
		951		}
		952
	0	953		if (string.IsNullOrWhiteSpace(tempSessionId))
		954		{
	0	955		receivingAmqpLink.Session.SafeClose();
	0	956		throw new ServiceBusException(true, Resources.AmqpFieldSessionId);
		957		}
		958
	0	959		receivingAmqpLink.Closed += this.OnSessionReceiverLinkClosed;
	0	960		this.SessionIdInternal = tempSessionId;
	0	961		this.LockedUntilUtcInternal = receivingAmqpLink.Settings.Properties.TryGetValue<long>(AmqpClientConstants.Lo
	0	962		? new DateTime(lockedUntilUtcTicks, DateTimeKind.Utc)
	0	963		: DateTime.MinValue;
	0	964		}
		965
		966		internal async Task<AmqpResponseMessage> ExecuteRequestResponseAsync(AmqpRequestMessage amqpRequestMessage)
		967		{
	0	968		var amqpMessage = amqpRequestMessage.AmqpMessage;
	0	969		if (this.isSessionReceiver)
		970		{
	0	971		this.ThrowIfSessionLockLost();
		972		}
		973
	0	974		var timeoutHelper = new TimeoutHelper(this.OperationTimeout, true);
		975
	0	976		ArraySegment<byte> transactionId = AmqpConstants.NullBinary;
	0	977		var ambientTransaction = Transaction.Current;
	0	978		if (ambientTransaction != null)
		979		{
	0	980		transactionId = await AmqpTransactionManager.Instance.EnlistAsync(ambientTransaction, this.ServiceBusCon
		981		}
		982
	0	983		if (!this.RequestResponseLinkManager.TryGetOpenedObject(out var requestResponseAmqpLink))
		984		{
	0	985		MessagingEventSource.Log.CreatingNewLink(this.ClientId, this.isSessionReceiver, this.SessionIdInternal,
	0	986		requestResponseAmqpLink = await this.RequestResponseLinkManager.GetOrCreateAsync(timeoutHelper.Remaining
		987		}
		988
	0	989		var responseAmqpMessage = await Task.Factory.FromAsync(
	0	990		(c, s) => requestResponseAmqpLink.BeginRequest(amqpMessage, transactionId, timeoutHelper.RemainingTime()
	0	991		(a) => requestResponseAmqpLink.EndRequest(a),
	0	992		this).ConfigureAwait(false);
		993
	0	994		return AmqpResponseMessage.CreateResponse(responseAmqpMessage);
	0	995		}
		996
		997		protected override async Task OnClosingAsync()
		998		{
	0	999		this.clientLinkManager.Close();
	0	1000		lock (this.messageReceivePumpSyncLock)
		1001		{
	0	1002		if (this.receivePump != null)
		1003		{
	0	1004		this.receivePumpCancellationTokenSource.Cancel();
	0	1005		this.receivePumpCancellationTokenSource.Dispose();
	0	1006		this.receivePump = null;
		1007		}
	0	1008		}
	0	1009		await this.ReceiveLinkManager.CloseAsync().ConfigureAwait(false);
	0	1010		await this.RequestResponseLinkManager.CloseAsync().ConfigureAwait(false);
	0	1011		this.requestResponseLockedMessages.Close();
	0	1012		}
		1013
		1014		protected virtual async Task<IList<Message>> OnReceiveAsync(int maxMessageCount, TimeSpan serverWaitTime)
		1015		{
	0	1016		ReceivingAmqpLink receiveLink = null;
		1017
	0	1018		if (this.isSessionReceiver)
		1019		{
	0	1020		this.ThrowIfSessionLockLost();
		1021		}
		1022
		1023		try
		1024		{
	0	1025		var timeoutHelper = new TimeoutHelper(serverWaitTime, true);
	0	1026		if(!this.ReceiveLinkManager.TryGetOpenedObject(out receiveLink))
		1027		{
	0	1028		MessagingEventSource.Log.CreatingNewLink(this.ClientId, this.isSessionReceiver, this.SessionIdIntern
	0	1029		receiveLink = await this.ReceiveLinkManager.GetOrCreateAsync(timeoutHelper.RemainingTime()).Configur
		1030		}
		1031
	0	1032		IList<Message> brokeredMessages = null;
	0	1033		this.ThrowIfClosed();
		1034
	0	1035		IEnumerable<AmqpMessage> amqpMessages = null;
	0	1036		var hasMessages = await Task.Factory.FromAsync(
	0	1037		(c, s) => receiveLink.BeginReceiveRemoteMessages(maxMessageCount, DefaultBatchFlushInterval, timeout
	0	1038		a => receiveLink.EndReceiveMessages(a, out amqpMessages),
	0	1039		this).ConfigureAwait(false);
		1040		Exception exception;
	0	1041		if ((exception = receiveLink.GetInnerException()) != null)
		1042		{
	0	1043		throw exception;
		1044		}
		1045
	0	1046		if (hasMessages && amqpMessages != null)
		1047		{
	0	1048		foreach (var amqpMessage in amqpMessages)
		1049		{
	0	1050		if (this.ReceiveMode == ReceiveMode.ReceiveAndDelete)
		1051		{
	0	1052		receiveLink.DisposeDelivery(amqpMessage, true, AmqpConstants.AcceptedOutcome);
		1053		}
		1054
	0	1055		var message = AmqpMessageConverter.AmqpMessageToSBMessage(amqpMessage);
	0	1056		if (brokeredMessages == null)
		1057		{
	0	1058		brokeredMessages = new List<Message>();
		1059		}
		1060
	0	1061		brokeredMessages.Add(message);
		1062		}
		1063		}
		1064
	0	1065		return brokeredMessages;
		1066		}
		1067		catch (Exception exception)
		1068		{
	0	1069		throw AmqpExceptionHelper.GetClientException(exception, receiveLink?.GetTrackingId(), null, receiveLink?
		1070		}
	0	1071		}
		1072
		1073		protected virtual async Task<IList<Message>> OnPeekAsync(long fromSequenceNumber, int messageCount = 1)
		1074		{
		1075		try
		1076		{
	0	1077		var amqpRequestMessage = AmqpRequestMessage.CreateRequest(
	0	1078		ManagementConstants.Operations.PeekMessageOperation,
	0	1079		this.OperationTimeout,
	0	1080		null);
		1081
	0	1082		if (this.ReceiveLinkManager.TryGetOpenedObject(out var receiveLink))
		1083		{
	0	1084		amqpRequestMessage.AmqpMessage.ApplicationProperties.Map[ManagementConstants.Request.AssociatedLinkN
		1085		}
		1086
	0	1087		amqpRequestMessage.Map[ManagementConstants.Properties.FromSequenceNumber] = fromSequenceNumber;
	0	1088		amqpRequestMessage.Map[ManagementConstants.Properties.MessageCount] = messageCount;
		1089
	0	1090		if (!string.IsNullOrWhiteSpace(this.SessionIdInternal))
		1091		{
	0	1092		amqpRequestMessage.Map[ManagementConstants.Properties.SessionId] = this.SessionIdInternal;
		1093		}
		1094
	0	1095		var messages = new List<Message>();
		1096
	0	1097		var amqpResponseMessage = await this.ExecuteRequestResponseAsync(amqpRequestMessage).ConfigureAwait(fals
	0	1098		if (amqpResponseMessage.StatusCode == AmqpResponseStatusCode.OK)
		1099		{
	0	1100		Message message = null;
	0	1101		var messageList = amqpResponseMessage.GetListValue<AmqpMap>(ManagementConstants.Properties.Messages)
	0	1102		foreach (AmqpMap entry in messageList)
		1103		{
	0	1104		var payload = (ArraySegment<byte>)entry[ManagementConstants.Properties.Message];
	0	1105		var amqpMessage = AmqpMessage.CreateAmqpStreamMessage(new BufferListStream(new[] { payload }), t
	0	1106		message = AmqpMessageConverter.AmqpMessageToSBMessage(amqpMessage, true);
	0	1107		messages.Add(message);
		1108		}
		1109
	0	1110		if (message != null)
		1111		{
	0	1112		this.LastPeekedSequenceNumber = message.SystemProperties.SequenceNumber;
		1113		}
		1114
	0	1115		return messages;
		1116		}
		1117
	0	1118		if (amqpResponseMessage.StatusCode == AmqpResponseStatusCode.NoContent ||
	0	1119		(amqpResponseMessage.StatusCode == AmqpResponseStatusCode.NotFound && Equals(AmqpClientConstants.Mes
		1120		{
	0	1121		return messages;
		1122		}
		1123
	0	1124		throw amqpResponseMessage.ToMessagingContractException();
		1125		}
		1126		catch (Exception exception)
		1127		{
	0	1128		throw AmqpExceptionHelper.GetClientException(exception);
		1129		}
	0	1130		}
		1131
		1132		protected virtual async Task<IList<Message>> OnReceiveDeferredMessageAsync(long[] sequenceNumbers)
		1133		{
	0	1134		var messages = new List<Message>();
		1135		try
		1136		{
	0	1137		var amqpRequestMessage = AmqpRequestMessage.CreateRequest(ManagementConstants.Operations.ReceiveBySequen
		1138
	0	1139		if (this.ReceiveLinkManager.TryGetOpenedObject(out var receiveLink))
		1140		{
	0	1141		amqpRequestMessage.AmqpMessage.ApplicationProperties.Map[ManagementConstants.Request.AssociatedLinkN
		1142		}
	0	1143		amqpRequestMessage.Map[ManagementConstants.Properties.SequenceNumbers] = sequenceNumbers;
	0	1144		amqpRequestMessage.Map[ManagementConstants.Properties.ReceiverSettleMode] = (uint)(this.ReceiveMode == R
	0	1145		if (!string.IsNullOrWhiteSpace(this.SessionIdInternal))
		1146		{
	0	1147		amqpRequestMessage.Map[ManagementConstants.Properties.SessionId] = this.SessionIdInternal;
		1148		}
		1149
	0	1150		var response = await this.ExecuteRequestResponseAsync(amqpRequestMessage).ConfigureAwait(false);
		1151
	0	1152		if (response.StatusCode == AmqpResponseStatusCode.OK)
		1153		{
	0	1154		var amqpMapList = response.GetListValue<AmqpMap>(ManagementConstants.Properties.Messages);
	0	1155		foreach (var entry in amqpMapList)
		1156		{
	0	1157		var payload = (ArraySegment<byte>)entry[ManagementConstants.Properties.Message];
	0	1158		var amqpMessage = AmqpMessage.CreateAmqpStreamMessage(new BufferListStream(new[] { payload }), t
	0	1159		var message = AmqpMessageConverter.AmqpMessageToSBMessage(amqpMessage);
	0	1160		if (entry.TryGetValue<Guid>(ManagementConstants.Properties.LockToken, out var lockToken))
		1161		{
	0	1162		message.SystemProperties.LockTokenGuid = lockToken;
	0	1163		this.requestResponseLockedMessages.AddOrUpdate(lockToken, message.SystemProperties.LockedUnt
		1164		}
		1165
	0	1166		messages.Add(message);
		1167		}
		1168		}
		1169		else
		1170		{
	0	1171		throw response.ToMessagingContractException();
		1172		}
	0	1173		}
		1174		catch (Exception exception)
		1175		{
	0	1176		throw AmqpExceptionHelper.GetClientException(exception);
		1177		}
		1178
	0	1179		return messages;
	0	1180		}
		1181
		1182		protected virtual Task OnCompleteAsync(IEnumerable<string> lockTokens)
		1183		{
	0	1184		var lockTokenGuids = lockTokens.Select(lt => new Guid(lt)).ToArray();
	0	1185		if (lockTokenGuids.Any(lt => this.requestResponseLockedMessages.Contains(lt)))
		1186		{
	0	1187		return this.DisposeMessageRequestResponseAsync(lockTokenGuids, DispositionStatus.Completed);
		1188		}
	0	1189		return this.DisposeMessagesAsync(lockTokenGuids, AmqpConstants.AcceptedOutcome);
		1190		}
		1191
		1192		protected virtual Task OnAbandonAsync(string lockToken, IDictionary<string, object> propertiesToModify = null)
		1193		{
	0	1194		var lockTokens = new[] { new Guid(lockToken) };
	0	1195		if (lockTokens.Any(lt => this.requestResponseLockedMessages.Contains(lt)))
		1196		{
	0	1197		return this.DisposeMessageRequestResponseAsync(lockTokens, DispositionStatus.Abandoned, propertiesToModi
		1198		}
	0	1199		return this.DisposeMessagesAsync(lockTokens, GetAbandonOutcome(propertiesToModify));
		1200		}
		1201
		1202		protected virtual Task OnDeferAsync(string lockToken, IDictionary<string, object> propertiesToModify = null)
		1203		{
	0	1204		var lockTokens = new[] { new Guid(lockToken) };
	0	1205		if (lockTokens.Any(lt => this.requestResponseLockedMessages.Contains(lt)))
		1206		{
	0	1207		return this.DisposeMessageRequestResponseAsync(lockTokens, DispositionStatus.Defered, propertiesToModify
		1208		}
	0	1209		return this.DisposeMessagesAsync(lockTokens, GetDeferOutcome(propertiesToModify));
		1210		}
		1211
		1212		protected virtual Task OnDeadLetterAsync(string lockToken, IDictionary<string, object> propertiesToModify = null
		1213		{
	0	1214		if (deadLetterReason != null && deadLetterReason.Length > Constants.MaxDeadLetterReasonLength)
		1215		{
	0	1216		throw new ArgumentOutOfRangeException(nameof(deadLetterReason), $"Maximum permitted length is {Constants
		1217		}
		1218
	0	1219		if (deadLetterErrorDescription != null && deadLetterErrorDescription.Length > Constants.MaxDeadLetterReasonL
		1220		{
	0	1221		throw new ArgumentOutOfRangeException(nameof(deadLetterErrorDescription), $"Maximum permitted length is
		1222		}
		1223
	0	1224		var lockTokens = new[] { new Guid(lockToken) };
	0	1225		if (lockTokens.Any(lt => this.requestResponseLockedMessages.Contains(lt)))
		1226		{
	0	1227		return this.DisposeMessageRequestResponseAsync(lockTokens, DispositionStatus.Suspended, propertiesToModi
		1228		}
		1229
	0	1230		return this.DisposeMessagesAsync(lockTokens, GetRejectedOutcome(propertiesToModify, deadLetterReason, deadLe
		1231		}
		1232
		1233		protected virtual async Task<DateTime> OnRenewLockAsync(string lockToken)
		1234		{
		1235		DateTime lockedUntilUtc;
		1236		try
		1237		{
		1238		// Create an AmqpRequest Message to renew  lock
	0	1239		var amqpRequestMessage = AmqpRequestMessage.CreateRequest(ManagementConstants.Operations.RenewLockOperat
		1240
	0	1241		if (this.ReceiveLinkManager.TryGetOpenedObject(out var receiveLink))
		1242		{
	0	1243		amqpRequestMessage.AmqpMessage.ApplicationProperties.Map[ManagementConstants.Request.AssociatedLinkN
		1244		}
	0	1245		amqpRequestMessage.Map[ManagementConstants.Properties.LockTokens] = new[] { new Guid(lockToken) };
		1246
	0	1247		var amqpResponseMessage = await this.ExecuteRequestResponseAsync(amqpRequestMessage).ConfigureAwait(fals
		1248
	0	1249		if (amqpResponseMessage.StatusCode == AmqpResponseStatusCode.OK)
		1250		{
	0	1251		var lockedUntilUtcTimes = amqpResponseMessage.GetValue<IEnumerable<DateTime>>(ManagementConstants.Pr
	0	1252		lockedUntilUtc = lockedUntilUtcTimes.First();
		1253		}
		1254		else
		1255		{
	0	1256		throw amqpResponseMessage.ToMessagingContractException();
		1257		}
	0	1258		}
		1259		catch (Exception exception)
		1260		{
	0	1261		throw AmqpExceptionHelper.GetClientException(exception);
		1262		}
		1263
	0	1264		return lockedUntilUtc;
	0	1265		}
		1266
		1267		/// <summary> </summary>
		1268		protected virtual void OnMessageHandler(
		1269		MessageHandlerOptions registerHandlerOptions,
		1270		Func<Message, CancellationToken, Task> callback)
		1271		{
	0	1272		MessagingEventSource.Log.RegisterOnMessageHandlerStart(this.ClientId, registerHandlerOptions);
		1273
	0	1274		lock (this.messageReceivePumpSyncLock)
		1275		{
	0	1276		if (this.receivePump != null)
		1277		{
	0	1278		throw new InvalidOperationException(Resources.MessageHandlerAlreadyRegistered);
		1279		}
		1280
	0	1281		this.receivePumpCancellationTokenSource = new CancellationTokenSource();
	0	1282		this.receivePump = new MessageReceivePump(this, registerHandlerOptions, callback, this.ServiceBusConnect
	0	1283		}
		1284
		1285		try
		1286		{
	0	1287		this.receivePump.StartPump();
	0	1288		}
	0	1289		catch (Exception exception)
		1290		{
	0	1291		MessagingEventSource.Log.RegisterOnMessageHandlerException(this.ClientId, exception);
	0	1292		lock (this.messageReceivePumpSyncLock)
		1293		{
	0	1294		if (this.receivePump != null)
		1295		{
	0	1296		this.receivePumpCancellationTokenSource.Cancel();
	0	1297		this.receivePumpCancellationTokenSource.Dispose();
	0	1298		this.receivePump = null;
		1299		}
	0	1300		}
		1301
	0	1302		throw;
		1303		}
		1304
	0	1305		MessagingEventSource.Log.RegisterOnMessageHandlerStop(this.ClientId);
	0	1306		}
		1307
		1308		static void CloseSession(ReceivingAmqpLink link)
		1309		{
	0	1310		link.Session.SafeClose();
	0	1311		}
		1312
		1313		static void CloseRequestResponseSession(RequestResponseAmqpLink requestResponseAmqpLink)
		1314		{
	0	1315		requestResponseAmqpLink.Session.SafeClose();
	0	1316		}
		1317
		1318		async Task<Message> ProcessMessage(Message message)
		1319		{
	0	1320		var processedMessage = message;
	0	1321		foreach (var plugin in this.RegisteredPlugins)
		1322		{
		1323		try
		1324		{
	0	1325		MessagingEventSource.Log.PluginCallStarted(plugin.Name, message.MessageId);
	0	1326		processedMessage = await plugin.AfterMessageReceive(message).ConfigureAwait(false);
	0	1327		MessagingEventSource.Log.PluginCallCompleted(plugin.Name, message.MessageId);
	0	1328		}
	0	1329		catch (Exception ex)
		1330		{
	0	1331		MessagingEventSource.Log.PluginCallFailed(plugin.Name, message.MessageId, ex);
	0	1332		if (!plugin.ShouldContinueOnException)
		1333		{
	0	1334		throw;
		1335		}
	0	1336		}
	0	1337		}
	0	1338		return processedMessage;
	0	1339		}
		1340
		1341		async Task<IList<Message>> ProcessMessages(IList<Message> messageList)
		1342		{
	0	1343		if (this.RegisteredPlugins.Count < 1)
		1344		{
	0	1345		return messageList;
		1346		}
		1347
	0	1348		var processedMessageList = new List<Message>();
	0	1349		foreach (var message in messageList)
		1350		{
	0	1351		var processedMessage = await this.ProcessMessage(message).ConfigureAwait(false);
	0	1352		processedMessageList.Add(processedMessage);
		1353		}
		1354
	0	1355		return processedMessageList;
	0	1356		}
		1357
		1358		async Task DisposeMessagesAsync(IEnumerable<Guid> lockTokens, Outcome outcome)
		1359		{
	0	1360		if(this.isSessionReceiver)
		1361		{
	0	1362		this.ThrowIfSessionLockLost();
		1363		}
		1364
	0	1365		var timeoutHelper = new TimeoutHelper(this.OperationTimeout, true);
	0	1366		List<ArraySegment<byte>> deliveryTags = this.ConvertLockTokensToDeliveryTags(lockTokens);
		1367
	0	1368		ReceivingAmqpLink receiveLink = null;
		1369		try
		1370		{
	0	1371		ArraySegment<byte> transactionId = AmqpConstants.NullBinary;
	0	1372		var ambientTransaction = Transaction.Current;
	0	1373		if (ambientTransaction != null)
		1374		{
	0	1375		transactionId = await AmqpTransactionManager.Instance.EnlistAsync(ambientTransaction, this.ServiceBu
		1376		}
		1377
	0	1378		if (!this.ReceiveLinkManager.TryGetOpenedObject(out receiveLink))
		1379		{
	0	1380		MessagingEventSource.Log.CreatingNewLink(this.ClientId, this.isSessionReceiver, this.SessionIdIntern
	0	1381		receiveLink = await this.ReceiveLinkManager.GetOrCreateAsync(timeoutHelper.RemainingTime()).Configur
		1382		}
		1383
	0	1384		var disposeMessageTasks = new Task<Outcome>[deliveryTags.Count];
	0	1385		var i = 0;
	0	1386		foreach (ArraySegment<byte> deliveryTag in deliveryTags)
		1387		{
	0	1388		disposeMessageTasks[i++] = Task.Factory.FromAsync(
	0	1389		(c, s) => receiveLink.BeginDisposeMessage(deliveryTag, transactionId, outcome, true, timeoutHelp
	0	1390		a => receiveLink.EndDisposeMessage(a),
	0	1391		this);
		1392		}
		1393
	0	1394		var outcomes = await Task.WhenAll(disposeMessageTasks).ConfigureAwait(false);
	0	1395		Error error = null;
	0	1396		foreach (var item in outcomes)
		1397		{
	0	1398		var disposedOutcome = item.DescriptorCode == Rejected.Code && ((error = ((Rejected)item).Error) != n
	0	1399		if (disposedOutcome != null)
		1400		{
	0	1401		if (error.Condition.Equals(AmqpErrorCode.NotFound))
		1402		{
	0	1403		if (this.isSessionReceiver)
		1404		{
	0	1405		throw new SessionLockLostException(Resources.SessionLockExpiredOnMessageSession);
		1406		}
		1407
	0	1408		throw new MessageLockLostException(Resources.MessageLockLost);
		1409		}
		1410
	0	1411		throw error.ToMessagingContractException();
		1412		}
		1413		}
	0	1414		}
	0	1415		catch (Exception exception)
		1416		{
	0	1417		if (exception is OperationCanceledException &&
	0	1418		receiveLink != null && receiveLink.State != AmqpObjectState.Opened)
		1419		{
		1420		// The link state is lost, We need to return a non-retriable error.
	0	1421		MessagingEventSource.Log.LinkStateLost(this.ClientId, receiveLink.Name, receiveLink.State, this.isSe
	0	1422		if (this.isSessionReceiver)
		1423		{
	0	1424		throw new SessionLockLostException(Resources.SessionLockExpiredOnMessageSession);
		1425		}
		1426
	0	1427		throw new MessageLockLostException(Resources.MessageLockLost);
		1428		}
		1429
	0	1430		throw AmqpExceptionHelper.GetClientException(exception);
		1431		}
	0	1432		}
		1433
		1434		async Task DisposeMessageRequestResponseAsync(Guid[] lockTokens, DispositionStatus dispositionStatus, IDictionar
		1435		{
		1436		try
		1437		{
		1438		// Create an AmqpRequest Message to update disposition
	0	1439		var amqpRequestMessage = AmqpRequestMessage.CreateRequest(ManagementConstants.Operations.UpdateDispositi
		1440
	0	1441		if (this.ReceiveLinkManager.TryGetOpenedObject(out var receiveLink))
		1442		{
	0	1443		amqpRequestMessage.AmqpMessage.ApplicationProperties.Map[ManagementConstants.Request.AssociatedLinkN
		1444		}
	0	1445		amqpRequestMessage.Map[ManagementConstants.Properties.LockTokens] = lockTokens;
	0	1446		amqpRequestMessage.Map[ManagementConstants.Properties.DispositionStatus] = dispositionStatus.ToString().
		1447
	0	1448		if (deadLetterReason != null)
		1449		{
	0	1450		amqpRequestMessage.Map[ManagementConstants.Properties.DeadLetterReason] = deadLetterReason;
		1451		}
		1452
	0	1453		if (deadLetterDescription != null)
		1454		{
	0	1455		amqpRequestMessage.Map[ManagementConstants.Properties.DeadLetterDescription] = deadLetterDescription
		1456		}
		1457
	0	1458		if (propertiesToModify != null)
		1459		{
	0	1460		var amqpPropertiesToModify = new AmqpMap();
	0	1461		foreach (var pair in propertiesToModify)
		1462		{
	0	1463		if (AmqpMessageConverter.TryGetAmqpObjectFromNetObject(pair.Value, MappingType.ApplicationProper
		1464		{
	0	1465		amqpPropertiesToModify[new MapKey(pair.Key)] = amqpObject;
		1466		}
		1467		else
		1468		{
	0	1469		throw new NotSupportedException(
	0	1470		Resources.InvalidAmqpMessageProperty.FormatForUser(pair.Key.GetType()));
		1471		}
		1472		}
		1473
	0	1474		if (amqpPropertiesToModify.Count > 0)
		1475		{
	0	1476		amqpRequestMessage.Map[ManagementConstants.Properties.PropertiesToModify] = amqpPropertiesToModi
		1477		}
		1478		}
		1479
	0	1480		if (!string.IsNullOrWhiteSpace(this.SessionIdInternal))
		1481		{
	0	1482		amqpRequestMessage.Map[ManagementConstants.Properties.SessionId] = this.SessionIdInternal;
		1483		}
		1484
	0	1485		var amqpResponseMessage = await this.ExecuteRequestResponseAsync(amqpRequestMessage).ConfigureAwait(fals
	0	1486		if (amqpResponseMessage.StatusCode != AmqpResponseStatusCode.OK)
		1487		{
	0	1488		throw amqpResponseMessage.ToMessagingContractException();
		1489		}
	0	1490		}
		1491		catch (Exception exception)
		1492		{
	0	1493		throw AmqpExceptionHelper.GetClientException(exception);
		1494		}
	0	1495		}
		1496
		1497		async Task<ReceivingAmqpLink> CreateLinkAsync(TimeSpan timeout)
		1498		{
	0	1499		FilterSet filterMap = null;
		1500
	0	1501		MessagingEventSource.Log.AmqpReceiveLinkCreateStart(this.ClientId, false, this.EntityType, this.Path);
		1502
	0	1503		if (this.isSessionReceiver)
		1504		{
	0	1505		filterMap = new FilterSet { { AmqpClientConstants.SessionFilterName, this.SessionIdInternal } };
		1506		}
		1507
	0	1508		var amqpLinkSettings = new AmqpLinkSettings
	0	1509		{
	0	1510		Role = true,
	0	1511		TotalLinkCredit = (uint)this.PrefetchCount,
	0	1512		AutoSendFlow = this.PrefetchCount > 0,
	0	1513		Source = new Source { Address = this.Path, FilterSet = filterMap },
	0	1514		SettleType = (this.ReceiveMode == ReceiveMode.PeekLock) ? SettleMode.SettleOnDispose : SettleMode.Settle
	0	1515		};
		1516
	0	1517		if (this.EntityType != null)
		1518		{
	0	1519		amqpLinkSettings.AddProperty(AmqpClientConstants.EntityTypeName, (int)this.EntityType);
		1520		}
		1521
	0	1522		amqpLinkSettings.AddProperty(AmqpClientConstants.TimeoutName, (uint)timeout.TotalMilliseconds);
		1523
	0	1524		var endpointUri = new Uri(this.ServiceBusConnection.Endpoint, this.Path);
	0	1525		var claims = new[] { ClaimConstants.Listen };
	0	1526		var amqpSendReceiveLinkCreator = new AmqpSendReceiveLinkCreator(
	0	1527		this.Path,
	0	1528		this.ServiceBusConnection,
	0	1529		endpointUri,
	0	1530		new string[] { endpointUri.AbsoluteUri },
	0	1531		claims,
	0	1532		this.CbsTokenProvider,
	0	1533		amqpLinkSettings,
	0	1534		this.ClientId);
		1535
	0	1536		Tuple<AmqpObject, DateTime> linkDetails = await amqpSendReceiveLinkCreator.CreateAndOpenAmqpLinkAsync().Conf
		1537
	0	1538		var receivingAmqpLink = (ReceivingAmqpLink) linkDetails.Item1;
	0	1539		var activeSendReceiveClientLink = new ActiveSendReceiveClientLink(
	0	1540		receivingAmqpLink,
	0	1541		endpointUri,
	0	1542		new string[] { endpointUri.AbsoluteUri },
	0	1543		claims,
	0	1544		linkDetails.Item2);
		1545
	0	1546		this.clientLinkManager.SetActiveSendReceiveLink(activeSendReceiveClientLink);
		1547
	0	1548		MessagingEventSource.Log.AmqpReceiveLinkCreateStop(this.ClientId);
		1549
	0	1550		return receivingAmqpLink;
	0	1551		}
		1552
		1553		// TODO: Consolidate the link creation paths
		1554		async Task<RequestResponseAmqpLink> CreateRequestResponseLinkAsync(TimeSpan timeout)
		1555		{
	0	1556		var entityPath = this.Path + '/' + AmqpClientConstants.ManagementAddress;
		1557
	0	1558		MessagingEventSource.Log.AmqpReceiveLinkCreateStart(this.ClientId, true, this.EntityType, entityPath);
	0	1559		var amqpLinkSettings = new AmqpLinkSettings();
	0	1560		amqpLinkSettings.AddProperty(AmqpClientConstants.EntityTypeName, AmqpClientConstants.EntityTypeManagement);
		1561
	0	1562		var endpointUri = new Uri(this.ServiceBusConnection.Endpoint, entityPath);
	0	1563		string[] claims = { ClaimConstants.Manage, ClaimConstants.Listen };
	0	1564		var amqpRequestResponseLinkCreator = new AmqpRequestResponseLinkCreator(
	0	1565		entityPath,
	0	1566		this.ServiceBusConnection,
	0	1567		endpointUri,
	0	1568		new string[] { endpointUri.AbsoluteUri },
	0	1569		claims,
	0	1570		this.CbsTokenProvider,
	0	1571		amqpLinkSettings,
	0	1572		this.ClientId);
		1573
	0	1574		var linkDetails = await amqpRequestResponseLinkCreator.CreateAndOpenAmqpLinkAsync().ConfigureAwait(false);
		1575
	0	1576		var requestResponseAmqpLink = (RequestResponseAmqpLink)linkDetails.Item1;
	0	1577		var activeRequestResponseClientLink = new ActiveRequestResponseLink(
	0	1578		requestResponseAmqpLink,
	0	1579		endpointUri,
	0	1580		new string[] { endpointUri.AbsoluteUri },
	0	1581		claims,
	0	1582		linkDetails.Item2);
	0	1583		this.clientLinkManager.SetActiveRequestResponseLink(activeRequestResponseClientLink);
		1584
	0	1585		MessagingEventSource.Log.AmqpReceiveLinkCreateStop(this.ClientId);
	0	1586		return requestResponseAmqpLink;
	0	1587		}
		1588
		1589		void OnSessionReceiverLinkClosed(object sender, EventArgs e)
		1590		{
	0	1591		var receivingAmqpLink = (ReceivingAmqpLink)sender;
	0	1592		if (receivingAmqpLink != null)
		1593		{
	0	1594		var exception = receivingAmqpLink.GetInnerException();
	0	1595		if (!(exception is SessionLockLostException))
		1596		{
	0	1597		exception = new SessionLockLostException("Session lock lost. Accept a new session", exception);
		1598		}
		1599
	0	1600		this.LinkException = exception;
	0	1601		MessagingEventSource.Log.SessionReceiverLinkClosed(this.ClientId, this.SessionIdInternal, this.LinkExcep
		1602		}
	0	1603		}
		1604
		1605		List<ArraySegment<byte>> ConvertLockTokensToDeliveryTags(IEnumerable<Guid> lockTokens)
		1606		{
	0	1607		return lockTokens.Select(lockToken => new ArraySegment<byte>(lockToken.ToByteArray())).ToList();
		1608		}
		1609
		1610		void ThrowIfNotPeekLockMode()
		1611		{
	0	1612		if (this.ReceiveMode != ReceiveMode.PeekLock)
		1613		{
	0	1614		throw Fx.Exception.AsError(new InvalidOperationException("The operation is only supported in 'PeekLock'
		1615		}
	0	1616		}
		1617
		1618		void ThrowIfSessionLockLost()
		1619		{
	0	1620		if (this.LinkException != null)
		1621		{
	0	1622		throw this.LinkException;
		1623		}
	0	1624		}
		1625
		1626		Outcome GetAbandonOutcome(IDictionary<string, object> propertiesToModify)
		1627		{
	0	1628		return this.GetModifiedOutcome(propertiesToModify, false);
		1629		}
		1630
		1631		Outcome GetDeferOutcome(IDictionary<string, object> propertiesToModify)
		1632		{
	0	1633		return this.GetModifiedOutcome(propertiesToModify, true);
		1634		}
		1635
		1636		Outcome GetModifiedOutcome(IDictionary<string, object> propertiesToModify, bool undeliverableHere)
		1637		{
	0	1638		Modified modified = new Modified();
	0	1639		if (undeliverableHere)
		1640		{
	0	1641		modified.UndeliverableHere = true;
		1642		}
		1643
	0	1644		if (propertiesToModify != null)
		1645		{
	0	1646		modified.MessageAnnotations = new Fields();
	0	1647		foreach (var pair in propertiesToModify)
		1648		{
	0	1649		if (AmqpMessageConverter.TryGetAmqpObjectFromNetObject(pair.Value, MappingType.ApplicationProperty,
		1650		{
	0	1651		modified.MessageAnnotations.Add(pair.Key, amqpObject);
		1652		}
		1653		else
		1654		{
	0	1655		throw new NotSupportedException(Resources.InvalidAmqpMessageProperty.FormatForUser(pair.Key.GetT
		1656		}
		1657		}
		1658		}
		1659
	0	1660		return modified;
		1661		}
		1662
		1663		Rejected GetRejectedOutcome(IDictionary<string, object> propertiesToModify, string deadLetterReason, string dead
		1664		{
	0	1665		var rejected = AmqpConstants.RejectedOutcome;
	0	1666		if (deadLetterReason != null || deadLetterErrorDescription != null || propertiesToModify != null)
		1667		{
	0	1668		rejected = new Rejected { Error = new Error { Condition = AmqpClientConstants.DeadLetterName, Info = new
	0	1669		if (deadLetterReason != null)
		1670		{
	0	1671		rejected.Error.Info.Add(Message.DeadLetterReasonHeader, deadLetterReason);
		1672		}
		1673
	0	1674		if (deadLetterErrorDescription != null)
		1675		{
	0	1676		rejected.Error.Info.Add(Message.DeadLetterErrorDescriptionHeader, deadLetterErrorDescription);
		1677		}
		1678
	0	1679		if (propertiesToModify != null)
		1680		{
	0	1681		foreach (var pair in propertiesToModify)
		1682		{
	0	1683		if (AmqpMessageConverter.TryGetAmqpObjectFromNetObject(pair.Value, MappingType.ApplicationProper
		1684		{
	0	1685		rejected.Error.Info.Add(pair.Key, amqpObject);
		1686		}
		1687		else
		1688		{
	0	1689		throw new NotSupportedException(Resources.InvalidAmqpMessageProperty.FormatForUser(pair.Key.
		1690		}
		1691		}
		1692		}
		1693		}
		1694
	0	1695		return rejected;
		1696		}
		1697		}
		1698		}