Add overrideable OnProcess* methods (Azure#19772)

* Add overrideable OnProcess* methods

* PR FB

Author: JoshLove-msft

sdk/servicebus/Azure.Messaging.ServiceBus/api/Azure.Messaging.ServiceBus.netstandard2.0.cs

@@ -203,7 +203,6 @@ protected ServiceBusProcessor() { }
         public virtual int MaxConcurrentCalls { get { throw null; } }
         public virtual int PrefetchCount { get { throw null; } }
         public virtual Azure.Messaging.ServiceBus.ServiceBusReceiveMode ReceiveMode { get { throw null; } }
-        public virtual string TransactionGroup { get { throw null; } }
         public event System.Func<Azure.Messaging.ServiceBus.ProcessErrorEventArgs, System.Threading.Tasks.Task> ProcessErrorAsync { add { } remove { } }
         public event System.Func<Azure.Messaging.ServiceBus.ProcessMessageEventArgs, System.Threading.Tasks.Task> ProcessMessageAsync { add { } remove { } }
         public virtual System.Threading.Tasks.Task CloseAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
@@ -212,6 +211,8 @@ protected ServiceBusProcessor() { }
         public override bool Equals(object obj) { throw null; }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
         public override int GetHashCode() { throw null; }
+        protected internal virtual System.Threading.Tasks.Task OnProcessErrorAsync(Azure.Messaging.ServiceBus.ProcessErrorEventArgs args) { throw null; }
+        protected internal virtual System.Threading.Tasks.Task OnProcessMessageAsync(Azure.Messaging.ServiceBus.ProcessMessageEventArgs args) { throw null; }
         public virtual System.Threading.Tasks.Task StartProcessingAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public virtual System.Threading.Tasks.Task StopProcessingAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
@@ -368,6 +369,7 @@ protected ServiceBusSessionProcessor() { }
         public virtual bool AutoCompleteMessages { get { throw null; } }
         public virtual string EntityPath { get { throw null; } }
         public virtual string FullyQualifiedNamespace { get { throw null; } }
+        protected virtual Azure.Messaging.ServiceBus.ServiceBusProcessor InnerProcessor { get { throw null; } }
         public virtual bool IsClosed { get { throw null; } }
         public virtual bool IsProcessing { get { throw null; } }
         public virtual System.TimeSpan MaxAutoLockRenewalDuration { get { throw null; } }
@@ -376,7 +378,6 @@ protected ServiceBusSessionProcessor() { }
         public virtual int PrefetchCount { get { throw null; } }
         public virtual Azure.Messaging.ServiceBus.ServiceBusReceiveMode ReceiveMode { get { throw null; } }
         public virtual System.TimeSpan? SessionIdleTimeout { get { throw null; } }
-        public virtual string TransactionGroup { get { throw null; } }
         public event System.Func<Azure.Messaging.ServiceBus.ProcessErrorEventArgs, System.Threading.Tasks.Task> ProcessErrorAsync { add { } remove { } }
         public event System.Func<Azure.Messaging.ServiceBus.ProcessSessionMessageEventArgs, System.Threading.Tasks.Task> ProcessMessageAsync { add { } remove { } }
         public event System.Func<Azure.Messaging.ServiceBus.ProcessSessionEventArgs, System.Threading.Tasks.Task> SessionClosingAsync { add { } remove { } }
@@ -387,6 +388,10 @@ protected ServiceBusSessionProcessor() { }
         public override bool Equals(object obj) { throw null; }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
         public override int GetHashCode() { throw null; }
+        protected internal virtual System.Threading.Tasks.Task OnProcessErrorAsync(Azure.Messaging.ServiceBus.ProcessErrorEventArgs args) { throw null; }
+        protected internal virtual System.Threading.Tasks.Task OnProcessSessionMessageAsync(Azure.Messaging.ServiceBus.ProcessSessionMessageEventArgs args) { throw null; }
+        protected internal virtual System.Threading.Tasks.Task OnSessionClosingAsync(Azure.Messaging.ServiceBus.ProcessSessionEventArgs args) { throw null; }
+        protected internal virtual System.Threading.Tasks.Task OnSessionInitializingAsync(Azure.Messaging.ServiceBus.ProcessSessionEventArgs args) { throw null; }
         public virtual System.Threading.Tasks.Task StartProcessingAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public virtual System.Threading.Tasks.Task StopProcessingAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]

sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/ReceiverManager.cs

@@ -12,59 +12,44 @@
 namespace Azure.Messaging.ServiceBus
 {
     /// <summary>
-    /// Represents a single receiver instance that multiple threads spawned by the
+    /// Represents a single receiver instance that multiple tasks spawned by the
     /// <see cref="ServiceBusProcessor"/> may be using to receive and process messages.
     /// The manager will delegate to the user provided callbacks and handle automatic
     /// locking of messages.
     /// </summary>
     internal class ReceiverManager
     {
         protected virtual ServiceBusReceiver Receiver { get; set; }
-        protected readonly ServiceBusConnection _connection;
-        protected readonly string _fullyQualifiedNamespace;
-        protected readonly string _entityPath;
-        protected readonly string _identifier;
+
+        protected readonly ServiceBusProcessor Processor;
         protected readonly TimeSpan? _maxReceiveWaitTime;
         private readonly ServiceBusReceiverOptions _receiverOptions;
-        protected readonly ServiceBusProcessorOptions _processorOptions;
-        private readonly Func<ProcessErrorEventArgs, Task> _errorHandler;
-        private readonly Func<ProcessMessageEventArgs, Task> _messageHandler;
+        protected readonly ServiceBusProcessorOptions ProcessorOptions;
         protected readonly EntityScopeFactory _scopeFactory;
         protected readonly IList<ServiceBusPlugin> _plugins;
 
-        protected bool AutoRenewLock => _processorOptions.MaxAutoLockRenewalDuration > TimeSpan.Zero;
+        protected bool AutoRenewLock => ProcessorOptions.MaxAutoLockRenewalDuration > TimeSpan.Zero;
 
         public ReceiverManager(
-            ServiceBusConnection connection,
-            string fullyQualifiedNamespace,
-            string entityPath,
-            string identifier,
-            ServiceBusProcessorOptions processorOptions,
-            Func<ProcessMessageEventArgs, Task> messageHandler,
-            Func<ProcessErrorEventArgs, Task> errorHandler,
+            ServiceBusProcessor processor,
             EntityScopeFactory scopeFactory,
             IList<ServiceBusPlugin> plugins)
         {
-            _connection = connection;
-            _fullyQualifiedNamespace = fullyQualifiedNamespace;
-            _entityPath = entityPath;
-            _processorOptions = processorOptions;
+            Processor = processor;
+            ProcessorOptions = processor.Options;
             _receiverOptions = new ServiceBusReceiverOptions
             {
-                ReceiveMode = _processorOptions.ReceiveMode,
-                PrefetchCount = _processorOptions.PrefetchCount,
+                ReceiveMode = ProcessorOptions.ReceiveMode,
+                PrefetchCount = ProcessorOptions.PrefetchCount,
             };
-            _maxReceiveWaitTime = _processorOptions.MaxReceiveWaitTime;
-            _identifier = identifier;
+            _maxReceiveWaitTime = ProcessorOptions.MaxReceiveWaitTime;
             _plugins = plugins;
             Receiver = new ServiceBusReceiver(
-                connection: _connection,
-                entityPath: _entityPath,
+                connection: Processor.Connection,
+                entityPath: Processor.EntityPath,
                 isSessionEntity: false,
                 plugins: _plugins,
                 options: _receiverOptions);
-            _errorHandler = errorHandler;
-            _messageHandler = messageHandler;
             _scopeFactory = scopeFactory;
         }
 
@@ -121,8 +106,8 @@ await RaiseExceptionReceived(
                     new ProcessErrorEventArgs(
                         ex,
                         errorSource,
-                        _fullyQualifiedNamespace,
-                        _entityPath,
+                        Processor.FullyQualifiedNamespace,
+                        Processor.EntityPath,
                         cancellationToken))
                     .ConfigureAwait(false);
             }
@@ -147,12 +132,6 @@ await ProcessOneMessage(
             }
         }
 
-        /// <summary>
-        ///
-        /// </summary>
-        /// <param name="message"></param>
-        /// <param name="cancellationToken"></param>
-        /// <returns></returns>
         private async Task ProcessOneMessage(
             ServiceBusReceivedMessage message,
             CancellationToken cancellationToken)
@@ -177,18 +156,18 @@ private async Task ProcessOneMessage(
 
                 try
                 {
-                    ServiceBusEventSource.Log.ProcessorMessageHandlerStart(_identifier, message.SequenceNumber);
+                    ServiceBusEventSource.Log.ProcessorMessageHandlerStart(Processor.Identifier, message.SequenceNumber);
                     await OnMessageHandler(message, cancellationToken).ConfigureAwait(false);
-                    ServiceBusEventSource.Log.ProcessorMessageHandlerComplete(_identifier, message.SequenceNumber);
+                    ServiceBusEventSource.Log.ProcessorMessageHandlerComplete(Processor.Identifier, message.SequenceNumber);
                 }
                 catch (Exception ex)
                 {
-                    ServiceBusEventSource.Log.ProcessorMessageHandlerException(_identifier, message.SequenceNumber, ex.ToString());
+                    ServiceBusEventSource.Log.ProcessorMessageHandlerException(Processor.Identifier, message.SequenceNumber, ex.ToString());
                     throw;
                 }
 
                 if (Receiver.ReceiveMode == ServiceBusReceiveMode.PeekLock &&
-                    _processorOptions.AutoCompleteMessages &&
+                    ProcessorOptions.AutoCompleteMessages &&
                     !message.IsSettled)
                 {
                     errorSource = ServiceBusErrorSource.Complete;
@@ -214,8 +193,8 @@ await RaiseExceptionReceived(
                     new ProcessErrorEventArgs(
                         ex,
                         errorSource,
-                        _fullyQualifiedNamespace,
-                        _entityPath,
+                        Processor.FullyQualifiedNamespace,
+                        Processor.EntityPath,
                         cancellationToken))
                     .ConfigureAwait(false);
 
@@ -244,8 +223,8 @@ await RaiseExceptionReceived(
                             new ProcessErrorEventArgs(
                                 exception,
                                 ServiceBusErrorSource.Abandon,
-                                _fullyQualifiedNamespace,
-                                _entityPath,
+                                Processor.FullyQualifiedNamespace,
+                                Processor.EntityPath,
                                 cancellationToken))
                         .ConfigureAwait(false);
                     }
@@ -264,26 +243,20 @@ protected virtual async Task OnMessageHandler(ServiceBusReceivedMessage message,
                 message,
                 Receiver,
                 processorCancellationToken);
-            await _messageHandler(args).ConfigureAwait(false);
+            await Processor.OnProcessMessageAsync(args).ConfigureAwait(false);
         }
 
-        /// <summary>
-        ///
-        /// </summary>
-        /// <param name="message"></param>
-        /// <param name="cancellationTokenSource"></param>
-        /// <returns></returns>
         private async Task RenewMessageLock(
             ServiceBusReceivedMessage message,
             CancellationTokenSource cancellationTokenSource)
         {
-            cancellationTokenSource.CancelAfter(_processorOptions.MaxAutoLockRenewalDuration);
+            cancellationTokenSource.CancelAfter(ProcessorOptions.MaxAutoLockRenewalDuration);
             CancellationToken cancellationToken = cancellationTokenSource.Token;
             while (!cancellationToken.IsCancellationRequested)
             {
                 try
                 {
-                    ServiceBusEventSource.Log.ProcessorRenewMessageLockStart(_identifier, 1, message.LockToken);
+                    ServiceBusEventSource.Log.ProcessorRenewMessageLockStart(Processor.Identifier, 1, message.LockToken);
                     TimeSpan delay = CalculateRenewDelay(message.LockedUntil);
 
                     // We're awaiting the task created by 'ContinueWith' to avoid awaiting the Delay task which may be canceled
@@ -300,11 +273,11 @@ private async Task RenewMessageLock(
                     }
 
                     await Receiver.RenewMessageLockAsync(message, cancellationToken).ConfigureAwait(false);
-                    ServiceBusEventSource.Log.ProcessorRenewMessageLockComplete(_identifier);
+                    ServiceBusEventSource.Log.ProcessorRenewMessageLockComplete(Processor.Identifier);
                 }
                 catch (Exception ex) when (!(ex is TaskCanceledException))
                 {
-                    ServiceBusEventSource.Log.ProcessorRenewMessageLockException(_identifier, ex.ToString());
+                    ServiceBusEventSource.Log.ProcessorRenewMessageLockException(Processor.Identifier, ex.ToString());
                     await HandleRenewLockException(ex, cancellationToken).ConfigureAwait(false);
 
                     // if the error was not transient, break out of the loop
@@ -363,22 +336,17 @@ await RaiseExceptionReceived(
                     new ProcessErrorEventArgs(
                         ex,
                         ServiceBusErrorSource.RenewLock,
-                        _fullyQualifiedNamespace,
-                        _entityPath,
+                        Processor.FullyQualifiedNamespace,
+                        Processor.EntityPath,
                         cancellationToken)).ConfigureAwait(false);
             }
         }
 
-        /// <summary>
-        ///
-        /// </summary>
-        /// <param name="eventArgs"></param>
-        /// <returns></returns>
         protected async Task RaiseExceptionReceived(ProcessErrorEventArgs eventArgs)
         {
             try
             {
-                await _errorHandler(eventArgs).ConfigureAwait(false);
+                await Processor.OnProcessErrorAsync(eventArgs).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -387,11 +355,6 @@ protected async Task RaiseExceptionReceived(ProcessErrorEventArgs eventArgs)
             }
         }
 
-        /// <summary>
-        ///
-        /// </summary>
-        /// <param name="lockedUntil"></param>
-        /// <returns></returns>
         protected static TimeSpan CalculateRenewDelay(DateTimeOffset lockedUntil)
         {
             var remainingTime = lockedUntil - DateTimeOffset.UtcNow;