ServiceBusProcessor.cs

// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.

usingSystem;
usingSystem.Collections.Generic;
usingSystem.ComponentModel;
usingSystem.Diagnostics.CodeAnalysis;
usingSystem.Globalization;
usingSystem.Linq;
usingSystem.Threading;
usingSystem.Threading.Tasks;
usingAzure.Core;
usingAzure.Core.Shared;
usingAzure.Messaging.ServiceBus.Core;
usingAzure.Messaging.ServiceBus.Diagnostics;

namespaceAzure.Messaging.ServiceBus
{
/// <summary>
/// The <see cref="ServiceBusProcessor"/> provides an abstraction around a set of <see cref="ServiceBusReceiver"/>
/// that allows using an event based model for processing received <see cref="ServiceBusReceivedMessage" />.
/// It is constructed by calling <see cref="ServiceBusClient.CreateProcessor(string, ServiceBusProcessorOptions)"/>.
/// The message handler is specified with the <see cref="ProcessMessageAsync"/>
/// property. The error handler is specified with the <see cref="ProcessErrorAsync"/> property.
/// To start processing after the handlers have been specified, call <see cref="StartProcessingAsync"/>.
/// </summary>
/// <remarks>
/// The <see cref="ServiceBusProcessor" /> is safe to cache and use for the lifetime of an application
/// or until the <see cref="ServiceBusClient" /> that it was created by is disposed. Caching the processor
/// is recommended when the application is processing messages regularly. The sender is responsible for
/// ensuring efficient network, CPU, and memory use. Calling <see cref="DisposeAsync" /> on the
/// associated <see cref="ServiceBusClient" /> as the application is shutting down will ensure that
/// network resources and other unmanaged objects used by the processor are properly cleaned up.
/// </remarks>
#pragma warning disable CA1001// Types that own disposable fields should be disposable
publicclassServiceBusProcessor:IAsyncDisposable
#pragma warning restore CA1001// Types that own disposable fields should be disposable
{
privateFunc<ProcessMessageEventArgs,Task>_processMessageAsync;

privateFunc<ProcessSessionMessageEventArgs,Task>_processSessionMessageAsync;

privateFunc<ProcessErrorEventArgs,Task>_processErrorAsync;

internalFunc<ProcessSessionEventArgs,Task>_sessionInitializingAsync;

internalFunc<ProcessSessionEventArgs,Task>_sessionClosingAsync;

// we use int.MaxValue as the maxCount since the concurrency can be changed dynamically by the user
privatereadonlySemaphoreSlim_messageHandlerSemaphore=newSemaphoreSlim(0,int.MaxValue);

privatereadonlyobject_optionsLock=new();

/// <summary>
/// The primitive for ensuring that the service is not overloaded with
/// accept session requests.
/// </summary>
privatereadonlySemaphoreSlim_maxConcurrentAcceptSessionsSemaphore=new(0,int.MaxValue);

/// <summary>The primitive for synchronizing access during start and close operations.</summary>
privatereadonlySemaphoreSlim_processingStartStopSemaphore=new(1,1);

privateCancellationTokenSourceRunningTaskTokenSource{get;set;}

privateTaskActiveReceiveTask{get;set;}

/// <summary>
/// Gets the fully qualified Service Bus namespace that the receiver is associated with. This is likely
/// to be similar to <c>{yournamespace}.servicebus.windows.net</c>.
/// </summary>
publicvirtualstringFullyQualifiedNamespace=>Connection.FullyQualifiedNamespace;

/// <summary>
/// Gets the path of the Service Bus entity that the processor is connected to, specific to the
/// Service Bus namespace that contains it.
/// </summary>
publicvirtualstringEntityPath{get;}

/// <summary>
/// Gets the ID used to identify this processor. This can be used to correlate logs and exceptions.
/// </summary>
publicvirtualstringIdentifier=>Options.Identifier;

/// <summary>
/// Gets the <see cref="ReceiveMode"/> used to specify how messages are received. Defaults to PeekLock mode.
/// </summary>
/// <value>
/// The receive mode is specified using <see cref="ServiceBusProcessorOptions.ReceiveMode"/>
/// and has a default mode of <see cref="ServiceBusReceiveMode.PeekLock"/>.
/// </value>
publicvirtualServiceBusReceiveModeReceiveMode=>Options.ReceiveMode;

/// <summary>
/// Gets whether the processor is configured to process session entities.
/// </summary>
internalboolIsSessionProcessor{get;}

/// <summary>
/// Gets the number of messages that will be eagerly requested from Queues or Subscriptions
/// during processing. This is intended to help maximize throughput by allowing the
/// processor to receive from a local cache rather than waiting on a service request.
/// </summary>
/// <value>
/// The prefetch count is specified using <see cref="ServiceBusProcessorOptions.PrefetchCount"/>
/// and has a default value of 0.
/// </value>
publicvirtualintPrefetchCount=>Options.PrefetchCount;

/// <summary>
/// Gets whether or not this processor is currently processing messages.
/// </summary>
///
/// <value>
/// <c>true</c> if the processor is processing messages; otherwise, <c>false</c>.
/// </value>
publicvirtualboolIsProcessing=>ActiveReceiveTask!=null;

internalServiceBusProcessorOptionsOptions{get;}

/// <summary>
/// The active connection to the Azure Service Bus service, enabling client communications for metadata
/// about the associated Service Bus entity and access to transport-aware consumers.
/// </summary>
internalreadonlyServiceBusConnectionConnection;

/// <summary>Gets the maximum number of concurrent calls to the
/// <see cref="ProcessMessageAsync"/> message handler the processor should initiate.
/// </summary>
/// <value>
/// The number of maximum concurrent calls is specified using <see cref="ServiceBusProcessorOptions.MaxConcurrentCalls"/>
/// and has a default value of 1.
/// </value>
publicvirtualintMaxConcurrentCalls=>Options.MaxConcurrentCalls;
privateint_currentConcurrentCalls;

internalintMaxConcurrentSessions=>_maxConcurrentSessions;
privatevolatileint_maxConcurrentSessions;
privateint_currentConcurrentSessions;

internalintMaxConcurrentCallsPerSession=>_maxConcurrentCallsPerSession;
privatevolatileint_maxConcurrentCallsPerSession;

privateint_currentAcceptSessions;

internalTimeSpan?MaxReceiveWaitTime=>Options.MaxReceiveWaitTime;

/// <summary>
/// Gets a value that indicates whether the processor should automatically
/// complete messages after the message handler has completed processing. If the
/// message handler triggers an exception, the message will not be automatically
/// completed.
/// </summary>
/// <remarks>
/// If the message handler triggers an exception and did not settle the message,
/// then the message will be automatically abandoned, irrespective of <see cref= "AutoCompleteMessages" />.
/// </remarks>
/// <value>
/// The option to auto complete messages is specified using <see cref="ServiceBusProcessorOptions.AutoCompleteMessages"/>
/// and has a default value of <c>true</c>.
/// </value>
publicvirtualboolAutoCompleteMessages{get;}

/// <summary>
/// Gets the maximum duration within which the lock will be renewed automatically. This
/// value should be greater than the longest message lock duration; for example, the LockDuration Property.
/// </summary>
/// <remarks>The message renew can continue for sometime in the background
/// after completion of message and result in a few false MessageLockLostExceptions temporarily.</remarks>
/// <value>
/// The maximum duration for lock renewal is specified using <see cref="ServiceBusProcessorOptions.MaxAutoLockRenewalDuration"/>
/// and has a default value of 5 minutes.
/// </value>
publicvirtualTimeSpanMaxAutoLockRenewalDuration=>Options.MaxAutoLockRenewalDuration;

/// <summary>
/// The instance of <see cref="ServiceBusEventSource" /> which can be mocked for testing.
/// </summary>
internalServiceBusEventSourceLogger{get;set;}=ServiceBusEventSource.Log;

/// <summary>
/// Indicates whether or not this <see cref="ServiceBusProcessor"/> has been closed.
/// </summary>
///
/// <value>
/// <c>true</c> if the processor is closed; otherwise, <c>false</c>.
/// </value>
publicvirtualboolIsClosed
{
get=>_closed;
privateset=>_closed=value;
}

// If the user has listed named sessions, and they
// have MaxConcurrentSessions greater or equal to the number
// of sessions, we can leave the sessions open at all times
// instead of cycling through them as receive calls time out.
internalboolKeepOpenOnReceiveTimeout=>_sessionIds.Length>0&&_maxConcurrentSessions>=_sessionIds.Length;

/// <summary>Indicates whether or not this instance has been closed.</summary>
privatevolatilebool_closed;

privatereadonlystring[]_sessionIds;

privatereadonlyMessagingClientDiagnostics_clientDiagnostics;

// deliberate usage of List instead of IList for faster enumeration and less allocations
privatereadonlyList<ReceiverManager>_receiverManagers=newList<ReceiverManager>();
privatereadonlyServiceBusSessionProcessor_sessionProcessor;
internalList<(TaskTask,CancellationTokenSourceCts)>TaskTuples{get;privateset;}=new();

privatereadonlyList<ReceiverManager>_orphanedReceiverManagers=new();
privateCancellationTokenSource_handlerCts=new();
privatereadonlyint_processorCount=Environment.ProcessorCount;

/// <summary>
/// Initializes a new instance of the <see cref="ServiceBusProcessor"/> class.
/// </summary>
///
/// <param name="connection">The <see cref="ServiceBusConnection" /> connection to use for communication with the Service Bus service.</param>
/// <param name="entityPath">The queue name or subscription path to process messages from.</param>
/// <param name="isSessionEntity">Whether or not the processor is associated with a session entity.</param>
/// <param name="options">The set of options to use when configuring the processor.</param>
/// <param name="sessionIds">An optional set of session Ids to limit processing to.
/// Only applies if isSessionEntity is true.</param>
/// <param name="maxConcurrentSessions">The max number of sessions that can be processed concurrently.
/// Only applies if isSessionEntity is true.</param>
/// <param name="maxConcurrentCallsPerSession">The max number of concurrent calls per session.
/// Only applies if isSessionEntity is true.</param>
/// <param name="sessionProcessor">If this is for a session processor, will contain the session processor instance.</param>
internalServiceBusProcessor(
ServiceBusConnectionconnection,
stringentityPath,
boolisSessionEntity,
ServiceBusProcessorOptionsoptions,
string[]sessionIds=default,
intmaxConcurrentSessions=default,
intmaxConcurrentCallsPerSession=default,
ServiceBusSessionProcessorsessionProcessor=default)
{
Argument.AssertNotNullOrWhiteSpace(entityPath,nameof(entityPath));
Argument.AssertNotNull(connection,nameof(connection));
Argument.AssertNotNull(connection.RetryOptions,nameof(connection.RetryOptions));
connection.ThrowIfClosed();

Options=options?.Clone()??newServiceBusProcessorOptions();
Connection=connection;
EntityPath=EntityNameFormatter.FormatEntityPath(entityPath,Options.SubQueue);
Options.Identifier=string.IsNullOrEmpty(Options.Identifier)?DiagnosticUtilities.GenerateIdentifier(EntityPath):Options.Identifier;

_maxConcurrentSessions=maxConcurrentSessions;
_maxConcurrentCallsPerSession=maxConcurrentCallsPerSession;
_sessionIds=sessionIds??Array.Empty<string>();
_sessionProcessor=sessionProcessor;

if(isSessionEntity)
{
Options.MaxConcurrentCalls=(_sessionIds.Length>0
?Math.Min(_sessionIds.Length,_maxConcurrentSessions)
:_maxConcurrentSessions)*_maxConcurrentCallsPerSession;
}

AutoCompleteMessages=Options.AutoCompleteMessages;

IsSessionProcessor=isSessionEntity;
_clientDiagnostics=newMessagingClientDiagnostics(
DiagnosticProperty.DiagnosticNamespace,
DiagnosticProperty.ResourceProviderNamespace,
DiagnosticProperty.ServiceBusServiceContext,
FullyQualifiedNamespace,
EntityPath);
}

/// <summary>
/// Initializes a new instance of the <see cref="ServiceBusProcessor"/> class for mocking.
/// </summary>
protectedServiceBusProcessor()
{
// assign default options since some of the properties reach into the options
Options=newServiceBusProcessorOptions();
_sessionIds=Array.Empty<string>();
}

/// <summary>
/// Initializes a new instance of the <see cref="ServiceBusProcessor"/> class for use with derived types.
/// </summary>
/// <param name="client">The client instance to use for the processor.</param>
/// <param name="queueName">The name of the queue to processor from.</param>
/// <param name="options">The set of options to use when configuring the processor.</param>
protectedServiceBusProcessor(ServiceBusClientclient,stringqueueName,ServiceBusProcessorOptionsoptions):
this(client?.Connection,queueName,false,options)
{
}

/// <summary>
/// Initializes a new instance of the <see cref="ServiceBusProcessor"/> class for use with derived types.
/// </summary>
/// <param name="client">The client instance to use for the processor.</param>
/// <param name="topicName">The topic to create a processor for.</param>
/// <param name="subscriptionName">The subscription to create a processor for.</param>
/// <param name="options">The set of options to use when configuring the processor.</param>
protectedServiceBusProcessor(ServiceBusClientclient,stringtopicName,stringsubscriptionName,
ServiceBusProcessorOptionsoptions):
this(client?.Connection,EntityNameFormatter.FormatSubscriptionPath(topicName,subscriptionName),false,options)
{
}

/// <summary>
/// Determines whether the specified <see cref="System.Object" /> is equal to this instance.
/// </summary>
///
/// <param name="obj">The <see cref="System.Object" /> to compare with this instance.</param>
///
/// <returns><c>true</c> if the specified <see cref="System.Object" /> is equal to this instance; otherwise, <c>false</c>.</returns>
[EditorBrowsable(EditorBrowsableState.Never)]
publicoverrideboolEquals(objectobj)=>base.Equals(obj);

/// <summary>
/// Returns a hash code for this instance.
/// </summary>
///
/// <returns>A hash code for this instance, suitable for use in hashing algorithms and data structures like a hash table.</returns>
///
[EditorBrowsable(EditorBrowsableState.Never)]
publicoverrideintGetHashCode()=>base.GetHashCode();

/// <summary>
/// Converts the instance to string representation.
/// </summary>
///
/// <returns>A <see cref="System.String" /> that represents this instance.</returns>
[EditorBrowsable(EditorBrowsableState.Never)]
publicoverridestringToString()=>base.ToString();

/// <summary>
/// The handler responsible for processing messages received from the Queue
/// or Subscription.
/// Implementation is mandatory.
/// </summary>
/// <remarks>
/// It is not recommended that the state of the processor be managed directly from within this handler; requesting to start or stop the processor may result in
/// a deadlock scenario.
/// </remarks>
[SuppressMessage("Usage","AZC0003:DO make service methods virtual.",
Justification="This member follows the standard .NET event pattern; override via the associated On<<EVENT>> method.")]
publiceventFunc<ProcessMessageEventArgs,Task>ProcessMessageAsync
{
add
{
Argument.AssertNotNull(value,nameof(ProcessMessageAsync));

if(_processMessageAsync!=default)
{
#pragma warning disable CA1065// Do not raise exceptions in unexpected locations
thrownewNotSupportedException(Resources.HandlerHasAlreadyBeenAssigned);
#pragma warning restore CA1065// Do not raise exceptions in unexpected locations
}

EnsureNotRunningAndInvoke(()=>_processMessageAsync=value);
}

remove
{
Argument.AssertNotNull(value,nameof(ProcessMessageAsync));

if(_processMessageAsync!=value)
{
#pragma warning disable CA1065// Do not raise exceptions in unexpected locations
thrownewArgumentException(Resources.HandlerHasNotBeenAssigned);
#pragma warning restore CA1065// Do not raise exceptions in unexpected locations
}

EnsureNotRunningAndInvoke(()=>_processMessageAsync=default);
}
}

/// <summary>
/// The handler responsible for processing messages received from the Queue
/// or Subscription. Implementation is mandatory.
/// </summary>
[SuppressMessage("Usage","AZC0003:DO make service methods virtual.",
Justification="This member follows the standard .NET event pattern; override via the associated On<<EVENT>> method.")]
internaleventFunc<ProcessSessionMessageEventArgs,Task>ProcessSessionMessageAsync
{
add
{
Argument.AssertNotNull(value,nameof(ProcessSessionMessageAsync));

if(_processSessionMessageAsync!=default)
{
thrownewNotSupportedException(Resources.HandlerHasAlreadyBeenAssigned);
}

EnsureNotRunningAndInvoke(()=>_processSessionMessageAsync=value);
}

remove
{
Argument.AssertNotNull(value,nameof(ProcessSessionMessageAsync));

if(_processSessionMessageAsync!=value)
{
thrownewArgumentException(Resources.HandlerHasNotBeenAssigned);
}

EnsureNotRunningAndInvoke(()=>_processSessionMessageAsync=default);
}
}

/// <summary>
/// The handler responsible for processing unhandled exceptions thrown while
/// this processor is running.
/// Implementation is mandatory.
/// </summary>
/// <remarks>
/// It is not recommended that the state of the processor be managed directly from within this handler; requesting to start or stop the processor may result in
/// a deadlock scenario.
/// </remarks>
[SuppressMessage("Usage","AZC0003:DO make service methods virtual.",
Justification="This member follows the standard .NET event pattern; override via the associated On<<EVENT>> method.")]
publiceventFunc<ProcessErrorEventArgs,Task>ProcessErrorAsync
{
add
{
Argument.AssertNotNull(value,nameof(ProcessErrorAsync));

if(_processErrorAsync!=default)
{
#pragma warning disable CA1065// Do not raise exceptions in unexpected locations
thrownewNotSupportedException(Resources.HandlerHasAlreadyBeenAssigned);
#pragma warning restore CA1065// Do not raise exceptions in unexpected locations
}

EnsureNotRunningAndInvoke(()=>_processErrorAsync=value);
}

remove
{
Argument.AssertNotNull(value,nameof(ProcessErrorAsync));

if(_processErrorAsync!=value)
{
#pragma warning disable CA1065// Do not raise exceptions in unexpected locations
thrownewArgumentException(Resources.HandlerHasNotBeenAssigned);
#pragma warning restore CA1065// Do not raise exceptions in unexpected locations
}

EnsureNotRunningAndInvoke(()=>_processErrorAsync=default);
}
}

/// <summary>
/// Optional handler that can be set to be notified when a new session is about to be processed.
/// </summary>
[SuppressMessage("Usage","AZC0003:DO make service methods virtual.",
Justification="This member follows the standard .NET event pattern; override via the associated On<<EVENT>> method.")]
internaleventFunc<ProcessSessionEventArgs,Task>SessionInitializingAsync
{
add
{
Argument.AssertNotNull(value,nameof(SessionInitializingAsync));

if(_sessionInitializingAsync!=default)
{
thrownewNotSupportedException(Resources.HandlerHasAlreadyBeenAssigned);
}

EnsureNotRunningAndInvoke(()=>_sessionInitializingAsync=value);
}

remove
{
Argument.AssertNotNull(value,nameof(SessionInitializingAsync));
if(_sessionInitializingAsync!=value)
{
thrownewArgumentException(Resources.HandlerHasNotBeenAssigned);
}

EnsureNotRunningAndInvoke(()=>_sessionInitializingAsync=default);
}
}

/// <summary>
/// Optional handler that can be set to be notified when a session is about to be closed for processing.
/// This means that the most recent <see cref="ServiceBusReceiver.ReceiveMessageAsync"/> call timed out so there are currently no messages
/// available to be received for the session.
/// </summary>
[SuppressMessage("Usage","AZC0003:DO make service methods virtual.",
Justification="This member follows the standard .NET event pattern; override via the associated On<<EVENT>> method.")]
internaleventFunc<ProcessSessionEventArgs,Task>SessionClosingAsync
{
add
{
Argument.AssertNotNull(value,nameof(SessionClosingAsync));

if(_sessionClosingAsync!=default)
{
thrownewNotSupportedException(Resources.HandlerHasAlreadyBeenAssigned);
}

EnsureNotRunningAndInvoke(()=>_sessionClosingAsync=value);
}

remove
{
Argument.AssertNotNull(value,nameof(SessionClosingAsync));
if(_sessionClosingAsync!=value)
{
thrownewArgumentException(Resources.HandlerHasNotBeenAssigned);
}

EnsureNotRunningAndInvoke(()=>_sessionClosingAsync=default);
}
}

/// <summary>
/// Invokes the process message event handler after a message has been received.
/// This method can be overridden to raise an event manually for testing purposes.
/// </summary>
/// <param name="args">The event args containing information related to the message.</param>
protectedinternalvirtualasyncTaskOnProcessMessageAsync(ProcessMessageEventArgsargs)
{
try
{
await_processMessageAsync(args).ConfigureAwait(false);
}
finally
{
args.EndExecutionScope();
}
}

/// <summary>
/// Invokes the error event handler when an error has occurred during processing.
/// This method can be overridden to raise an event manually for testing purposes.
/// </summary>
/// <param name="args">The event args containing information related to the error.</param>
protectedinternalvirtualasyncTaskOnProcessErrorAsync(ProcessErrorEventArgsargs)
{
await_processErrorAsync(args).ConfigureAwait(false);
}

internalasyncTaskOnProcessSessionMessageAsync(ProcessSessionMessageEventArgsargs)
{
try
{
await_processSessionMessageAsync(args).ConfigureAwait(false);
}
finally
{
args.EndExecutionScope();
}
}

internalasyncTaskOnSessionInitializingAsync(ProcessSessionEventArgsargs)
{
await_sessionInitializingAsync(args).ConfigureAwait(false);
}

internalasyncTaskOnSessionClosingAsync(ProcessSessionEventArgsargs)
{
await_sessionClosingAsync(args).ConfigureAwait(false);
}

/// <summary>
/// Signals the processor to begin processing messages. Should this method be
/// called while the processor is already running, an
/// <see cref="InvalidOperationException"/> is thrown.
/// </summary>
///
/// <param name="cancellationToken">A <see cref="CancellationToken"/> instance to
/// signal the request to cancel the start operation. This won't affect the
/// processor once it starts running.
/// </param>
/// <exception cref="InvalidOperationException">
/// This can occur if the processor is already running. This can be checked via the <see cref="IsProcessing"/> property.
/// This can also occur if event handlers have not been specified for the <see cref="ProcessMessageAsync"/> or
/// the <see cref="ProcessErrorAsync"/> events.
/// </exception>
publicvirtualasyncTaskStartProcessingAsync(
CancellationTokencancellationToken=default)
{
Argument.AssertNotDisposed(IsClosed,nameof(ServiceBusProcessor));
Connection.ThrowIfClosed();
cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
boolreleaseGuard=false;
try
{
await_processingStartStopSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
releaseGuard=true;

if(ActiveReceiveTask==null)
{
Logger.StartProcessingStart(Identifier);

try
{
ValidateMessageHandler();
ValidateErrorHandler();
cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();

// We expect the token source to be null, but we are playing safe.

RunningTaskTokenSource?.Cancel();
RunningTaskTokenSource?.Dispose();
RunningTaskTokenSource=newCancellationTokenSource();

// Start the main running task.
ActiveReceiveTask=RunReceiveTaskAsync(RunningTaskTokenSource.Token);
}
catch(Exceptionexception)
{
Logger.StartProcessingException(Identifier,exception.ToString());
throw;
}

Logger.StartProcessingComplete(Identifier);
}
else
{
thrownewInvalidOperationException(Resources.RunningMessageProcessorCannotPerformOperation);
}
}
finally
{
if(releaseGuard)
{
_processingStartStopSemaphore.Release();
}
}
}

privatevoidReconcileReceiverManagers(intmaxConcurrentSessions,intprefetchCount)
{
if(_receiverManagers.Count==0)
{
if(IsSessionProcessor)
{
varnumReceivers=_sessionIds.Length>0?_sessionIds.Length:_maxConcurrentSessions;
for(inti=0;i<numReceivers;i++)
{
varsessionId=_sessionIds.Length>0?_sessionIds[i]:null;

_receiverManagers.Add(
newSessionReceiverManager(
_sessionProcessor,
sessionId,
_maxConcurrentAcceptSessionsSemaphore,
_clientDiagnostics,
KeepOpenOnReceiveTimeout));
}
}
else
{
_receiverManagers.Add(
newReceiverManager(
this,
_clientDiagnostics,
false));
}
}
else
{
if(IsSessionProcessor&&_sessionIds.Length==0)
{
vardiffSessions=maxConcurrentSessions-_currentConcurrentSessions;

if(diffSessions>0)
{
for(inti=0;i<diffSessions;i++)
{
_receiverManagers.Add(
newSessionReceiverManager(
_sessionProcessor,
null,
_maxConcurrentAcceptSessionsSemaphore,
_clientDiagnostics,
KeepOpenOnReceiveTimeout));
}
}
else
{
intdiffSessionsLimit=Math.Abs(diffSessions);
for(inti=0;i<diffSessionsLimit;i++)
{
// These should generally be closed as part of the normal bookkeeping in SessionReceiverManager,
// but we will track them so that they can be explicitly closed when stopping, just like we do with
// _receiverManagers.
_orphanedReceiverManagers.Add(_receiverManagers[0]);

// these tasks will be awaited when closing the orphaned receivers as part of CloseAsync
_=_receiverManagers[0].CancelAsync();
_receiverManagers.RemoveAt(0);
}
}
}

intreceiverManagers=_receiverManagers.Count;
for(inti=0;i<receiverManagers;i++)
{
varreceiverManager=_receiverManagers[i];
receiverManager.UpdatePrefetchCount(prefetchCount);
}
}
}

privatevoidValidateErrorHandler()
{
if(_processErrorAsync==null)
{
thrownewInvalidOperationException(string.Format(CultureInfo.CurrentCulture,
Resources.CannotStartMessageProcessorWithoutHandler,nameof(ProcessErrorAsync)));
}
}

privatevoidValidateMessageHandler()
{
if(IsSessionProcessor)
{
if(_processSessionMessageAsync==null)
{
thrownewInvalidOperationException(string.Format(CultureInfo.CurrentCulture,
Resources.CannotStartMessageProcessorWithoutHandler,nameof(ProcessMessageAsync)));
}
}
elseif(_processMessageAsync==null)
{
thrownewInvalidOperationException(string.Format(CultureInfo.CurrentCulture,
Resources.CannotStartMessageProcessorWithoutHandler,nameof(ProcessMessageAsync)));
}
}

/// <summary>
/// Signals the processor to stop processing messaging. Should this method be
/// called while the processor is not running, no action is taken. This method
/// will not close the underlying receivers, but will cause the receivers to stop
/// receiving. Any in-flight message handlers will be awaited, and this method will not return
/// until all in-flight message handlers have returned. To close the underlying receivers, <see cref="CloseAsync"/>
/// should be called. If <see cref="CloseAsync"/> is called, the processor cannot be restarted.
/// If you wish to resume processing at some point after calling this method, you can call
/// <see cref="StartProcessingAsync"/>.
/// </summary>
///
/// <param name="cancellationToken">A <see cref="CancellationToken"/> instance to
/// signal the request to cancel the stop operation. If the operation is successfully
/// canceled, the processor will keep running.</param>
publicvirtualasyncTaskStopProcessingAsync(CancellationTokencancellationToken=default)
{
boolreleaseGuard=false;
try
{
await_processingStartStopSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
releaseGuard=true;

if(ActiveReceiveTask!=null)
{
Logger.StopProcessingStart(Identifier);
cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();

// Cancel the current running task. Catch any exception that are triggered due to customer token registrations, and
// log these as warnings as we don't want to forego stopping due to these exceptions.
try
{
RunningTaskTokenSource.Cancel();
}
catch(Exceptionexception)
{
Logger.ProcessorStoppingCancellationWarning(Identifier,exception.ToString());
}

RunningTaskTokenSource.Dispose();
RunningTaskTokenSource=null;

// Now that a cancellation request has been issued, wait for the running task to finish. In case something
// unexpected happened and it stopped working midway, this is the moment we expect to catch an exception.
try
{
awaitActiveReceiveTask.ConfigureAwait(false);
}
catch(OperationCanceledException)
{
// Nothing to do here. These exceptions are expected.
}
finally
{
// If an unexpected exception occurred while awaiting the receive task, we still want to dispose and set to null
// as the task is complete and there is no use in awaiting it again if StopProcessingAsync is called again.
ActiveReceiveTask.Dispose();
ActiveReceiveTask=null;
}
}
}
catch(Exceptionexception)
{
Logger.StopProcessingException(Identifier,exception.ToString());
throw;
}
finally
{
if(releaseGuard)
{
_processingStartStopSemaphore.Release();
}
}

Logger.StopProcessingComplete(Identifier);
}

/// <summary>
/// Runs the Receive task in as many threads as are
/// specified in the <see cref="MaxConcurrentCalls"/> property.
/// </summary>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> instance to signal the request to cancel the operation.</param>
privateasyncTaskRunReceiveTaskAsync(CancellationTokencancellationToken)
{
CancellationTokenSourcelinkedHandlerTcs=CancellationTokenSource.CreateLinkedTokenSource(cancellationToken,_handlerCts.Token);

try
{
while(!cancellationToken.IsCancellationRequested&&!Connection.IsClosed)
{
awaitReconcileConcurrencyAsync().ConfigureAwait(false);

foreach(ReceiverManagerreceiverManagerin_receiverManagers)
{
// reset the linkedHandlerTcs if it was already cancelled due to user updating the concurrency
// do this before the synchronous Wait call as that does not respect the TCS.
if(linkedHandlerTcs.IsCancellationRequested)
{
linkedHandlerTcs.Dispose();
linkedHandlerTcs=CancellationTokenSource.CreateLinkedTokenSource(cancellationToken,_handlerCts.Token);
break;
}

// Do a quick synchronous check before we resort to async/await with the state-machine overhead.
if(!_messageHandlerSemaphore.Wait(0,CancellationToken.None))
{
try
{
await_messageHandlerSemaphore.WaitAsync(linkedHandlerTcs.Token).ConfigureAwait(false);
}
catch(OperationCanceledException)
{
linkedHandlerTcs.Dispose();
// reset the linkedHandlerTcs if it was already cancelled due to user updating the concurrency
linkedHandlerTcs=CancellationTokenSource.CreateLinkedTokenSource(cancellationToken,_handlerCts.Token);
// allow the loop to wake up when tcs is signaled
break;
}
}

// hold onto all the tasks that we are starting so that when cancellation is requested,
// we can await them to make sure we surface any unexpected exceptions, i.e. exceptions
// other than TaskCanceledExceptions
// Instead of using the array overload which allocates an array, we use the overload that has two parameters
// and pass in CancellationToken.None. This should be safe since CanBeCanceled will return false.
varlinkedCts=CancellationTokenSource.CreateLinkedTokenSource(cancellationToken,CancellationToken.None);

TaskTuples.Add(
(
ReceiveAndProcessMessagesAsync(receiverManager,linkedCts.Token),
linkedCts)
);

if(TaskTuples.Count>Options.MaxConcurrentCalls)
{
List<(TaskTask,CancellationTokenSourceCts)>remaining=new();
foreach(vartupleinTaskTuples)
{
if(tuple.Task.IsCompleted)
{
tuple.Cts.Dispose();
}
else
{
remaining.Add(tuple);
}
}

TaskTuples=remaining;
}
}
}

// If the main loop is aborting due to the connection being canceled, then
// force the processor to stop.
if(!cancellationToken.IsCancellationRequested&&Connection.IsClosed)
{
Logger.ProcessorClientClosedException(Identifier);

// Because this is a highly unusual situation
// and goes against the goal of resiliency for the processor, surface
// a fatal exception with an explicit message detailing that processing
// will be stopped.
try
{
awaitOnProcessErrorAsync(
newProcessErrorEventArgs(
newObjectDisposedException(nameof(ServiceBusConnection),
Resources.DisposedConnectionMessageProcessorMustStop),
ServiceBusErrorSource.Receive,
FullyQualifiedNamespace,
EntityPath,
Identifier,
cancellationToken))
.ConfigureAwait(false);
}
catch(Exceptionex)
{
// Don't bubble up exceptions raised from customer exception handler.
Logger.ProcessorErrorHandlerThrewException(ex.ToString(),Identifier);
}

// This call will deadlock if awaited, as StopProcessingAsync awaits
// completion of this task. The processor is already known to be in a bad
// state and exceptions in StopProcessingAsync are likely and will be logged
// in that call. There is little value in surfacing those expected exceptions
// to the customer error handler as well; allow StopProcessingAsync to run
// in a fire-and-forget manner.
_=StopProcessingAsync(CancellationToken.None);
}
}
finally
{
try
{
// await all tasks using WhenAll so that we ensure every task finishes even if there are exceptions
// (rather than try/catch each await)
awaitTask.WhenAll(TaskTuples.Select(t =>t.Task)).ConfigureAwait(false);
}
finally
{
foreach(var(_,cts)inTaskTuples)
{
cts.Dispose();
}

TaskTuples.Clear();

linkedHandlerTcs.Dispose();
}
}
}

privateasyncTaskReceiveAndProcessMessagesAsync(
ReceiverManagerreceiverManager,
CancellationTokencancellationToken)
{
try
{
awaitreceiverManager.ReceiveAndProcessMessagesAsync(cancellationToken).ConfigureAwait(false);
}
finally
{
_messageHandlerSemaphore.Release();
}
}

/// <summary>
/// Invokes a specified action only if this <see cref="ServiceBusProcessor" /> instance is not running.
/// </summary>
/// <param name="action">The action to invoke.</param>
/// <exception cref="InvalidOperationException">Method is invoked while the event processor is running.</exception>
internalvoidEnsureNotRunningAndInvoke(Actionaction)
{
if(ActiveReceiveTask==null)
{
try
{
_processingStartStopSemaphore.Wait();
if(ActiveReceiveTask==null)
{
action?.Invoke();
}
else
{
thrownewInvalidOperationException(Resources.RunningMessageProcessorCannotPerformOperation);
}
}
finally
{
_processingStartStopSemaphore.Release();
}
}
else
{
thrownewInvalidOperationException(Resources.RunningMessageProcessorCannotPerformOperation);
}
}

/// <summary>
/// Performs the task needed to clean up resources used by the <see cref="ServiceBusProcessor" />. Any in-flight message handlers will
/// be awaited. Once all message handling has completed, the underlying links will be closed. After this point, the method will return.
/// This is equivalent to calling <see cref="DisposeAsync"/>.
/// </summary>
/// <param name="cancellationToken"> An optional<see cref="CancellationToken"/> instance to signal the
/// request to cancel the operation.</param>
publicvirtualasyncTaskCloseAsync(
CancellationTokencancellationToken=default)
{
IsClosed=true;