Add MessageActions/SessionMessageActions (Azure#19074)

* Add MessageActions/SessionMessageActions

* PR FB

* Export API

Author: JoshLove-msft

sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/api/Microsoft.Azure.WebJobs.Extensions.ServiceBus.netstandard2.0.cs

@@ -52,8 +52,8 @@ public enum EntityType
     public partial class MessageProcessor
     {
         public MessageProcessor(Azure.Messaging.ServiceBus.ServiceBusProcessor processor) { }
-        public virtual System.Threading.Tasks.Task<bool> BeginProcessingMessageAsync(Azure.Messaging.ServiceBus.ServiceBusReceiver receiver, Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, System.Threading.CancellationToken cancellationToken) { throw null; }
-        public virtual System.Threading.Tasks.Task CompleteProcessingMessageAsync(Azure.Messaging.ServiceBus.ServiceBusReceiver receiver, Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, Microsoft.Azure.WebJobs.Host.Executors.FunctionResult result, System.Threading.CancellationToken cancellationToken) { throw null; }
+        public virtual System.Threading.Tasks.Task<bool> BeginProcessingMessageAsync(Microsoft.Azure.WebJobs.ServiceBus.ServiceBusMessageActions messageActions, Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, System.Threading.CancellationToken cancellationToken) { throw null; }
+        public virtual System.Threading.Tasks.Task CompleteProcessingMessageAsync(Microsoft.Azure.WebJobs.ServiceBus.ServiceBusMessageActions messageActions, Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, Microsoft.Azure.WebJobs.Host.Executors.FunctionResult result, System.Threading.CancellationToken cancellationToken) { throw null; }
     }
     public partial class MessagingProvider
     {
@@ -65,6 +65,15 @@ public MessagingProvider(Microsoft.Extensions.Options.IOptions<Microsoft.Azure.W
         public virtual Azure.Messaging.ServiceBus.ServiceBusProcessor CreateProcessor(string entityPath, string connectionString) { throw null; }
         public virtual Microsoft.Azure.WebJobs.ServiceBus.SessionMessageProcessor CreateSessionMessageProcessor(string entityPath, string connectionString) { throw null; }
     }
+    public partial class ServiceBusMessageActions
+    {
+        internal ServiceBusMessageActions() { }
+        public virtual System.Threading.Tasks.Task AbandonMessageAsync(Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, System.Collections.Generic.IDictionary<string, object> propertiesToModify = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+        public virtual System.Threading.Tasks.Task CompleteMessageAsync(Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+        public virtual System.Threading.Tasks.Task DeadLetterMessageAsync(Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, System.Collections.Generic.IDictionary<string, object> propertiesToModify = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+        public virtual System.Threading.Tasks.Task DeadLetterMessageAsync(Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, string deadLetterReason, string deadLetterErrorDescription = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+        public virtual System.Threading.Tasks.Task DeferMessageAsync(Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, System.Collections.Generic.IDictionary<string, object> propertiesToModify = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+    }
     public partial class ServiceBusOptions : Microsoft.Azure.WebJobs.Hosting.IOptionsFormatter
     {
         public ServiceBusOptions() { }
@@ -75,13 +84,19 @@ public ServiceBusOptions() { }
         public int MaxConcurrentCalls { get { throw null; } set { } }
         public int MaxConcurrentSessions { get { throw null; } set { } }
         public int MaxMessages { get { throw null; } set { } }
-        public System.TimeSpan? MaxWaitTime { get { throw null; } set { } }
         public int PrefetchCount { get { throw null; } set { } }
         public Azure.Messaging.ServiceBus.ServiceBusRetryOptions RetryOptions { get { throw null; } set { } }
+        public System.TimeSpan? SessionIdleTimeout { get { throw null; } set { } }
         public Azure.Messaging.ServiceBus.ServiceBusTransportType TransportType { get { throw null; } set { } }
         public System.Net.IWebProxy WebProxy { get { throw null; } set { } }
         public string Format() { throw null; }
     }
+    public partial class ServiceBusSessionMessageActions : Microsoft.Azure.WebJobs.ServiceBus.ServiceBusMessageActions
+    {
+        internal ServiceBusSessionMessageActions() { }
+        public virtual System.Threading.Tasks.Task<System.BinaryData> GetSessionStateAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+        public virtual System.Threading.Tasks.Task SetSessionStateAsync(System.BinaryData sessionState, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+    }
     public partial class ServiceBusWebJobsStartup : Microsoft.Azure.WebJobs.Hosting.IWebJobsStartup
     {
         public ServiceBusWebJobsStartup() { }
@@ -90,8 +105,8 @@ public void Configure(Microsoft.Azure.WebJobs.IWebJobsBuilder builder) { }
     public partial class SessionMessageProcessor
     {
         public SessionMessageProcessor(Azure.Messaging.ServiceBus.ServiceBusSessionProcessor processor) { }
-        public virtual System.Threading.Tasks.Task<bool> BeginProcessingMessageAsync(Azure.Messaging.ServiceBus.ServiceBusSessionReceiver receiver, Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, System.Threading.CancellationToken cancellationToken) { throw null; }
-        public virtual System.Threading.Tasks.Task CompleteProcessingMessageAsync(Azure.Messaging.ServiceBus.ServiceBusSessionReceiver receiver, Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, Microsoft.Azure.WebJobs.Host.Executors.FunctionResult result, System.Threading.CancellationToken cancellationToken) { throw null; }
+        public virtual System.Threading.Tasks.Task<bool> BeginProcessingMessageAsync(Microsoft.Azure.WebJobs.ServiceBus.ServiceBusSessionMessageActions sessionActions, Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, System.Threading.CancellationToken cancellationToken) { throw null; }
+        public virtual System.Threading.Tasks.Task CompleteProcessingMessageAsync(Microsoft.Azure.WebJobs.ServiceBus.ServiceBusSessionMessageActions sessionActions, Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, Microsoft.Azure.WebJobs.Host.Executors.FunctionResult result, System.Threading.CancellationToken cancellationToken) { throw null; }
     }
 }
 namespace Microsoft.Extensions.Hosting

sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Config/ServiceBusHostBuilderExtensions.cs

@@ -75,6 +75,8 @@ public static IWebJobsBuilder AddServiceBus(this IWebJobsBuilder builder, Action
                         "SessionHandlerOptions:MaxConcurrentSessions",
                         options.MaxConcurrentSessions);
 
+                    options.SessionIdleTimeout = section.GetValue("SessionHandlerOptions:MessageWaitTime", options.SessionIdleTimeout);
+
                     section.Bind(options);
 
                     configure(options);

sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Config/ServiceBusOptions.cs

@@ -131,19 +131,20 @@ public int MaxConcurrentSessions
         /// </summary>
         public Func<ProcessErrorEventArgs, Task> ExceptionHandler { get; set; }
 
-        /// <summary>
-        /// Gets or sets an optional <see cref="TimeSpan"/> specifying the maximum time to wait when attempting to receive messages.
-        /// If not specified, the <see cref="ServiceBusRetryOptions.TryTimeout"/> will be used. This only applies for functions that receive
-        /// a batch of messages.
-        /// </summary>
-        public TimeSpan? MaxWaitTime { get; set; }  // TODO this should probably be exposed in the Processor as well and would then also apply for functions that receive a single message.
-
         /// <summary>
         /// Gets or sets the maximum number of messages that will be passed to each function call. This only applies for functions that receive
         /// a batch of messages. The default value is 1000.
         /// </summary>
         public int MaxMessages { get; set; } = 1000;
 
+        /// <summary>
+        /// Gets or sets the maximum amount of time to wait for a message to be received for the
+        /// currently active session. After this time has elapsed, the processor will close the session
+        /// and attempt to process another session.
+        /// If not specified, the <see cref="ServiceBusRetryOptions.TryTimeout"/> will be used.
+        /// </summary>
+        public TimeSpan? SessionIdleTimeout { get; set; }
+
         /// <summary>
         /// Formats the options as JSON objects for display.
         /// </summary>
@@ -171,7 +172,7 @@ public string Format()
                 { nameof(MaxConcurrentCalls), MaxConcurrentCalls },
                 { nameof(MaxConcurrentSessions), MaxConcurrentSessions },
                 { nameof(MaxMessages), MaxMessages },
-                { nameof(MaxWaitTime), MaxWaitTime.ToString() ?? string.Empty }
+                { nameof(SessionIdleTimeout), SessionIdleTimeout.ToString() ?? string.Empty }
             };
 
             return options.ToString(Formatting.Indented);
@@ -199,7 +200,8 @@ internal ServiceBusSessionProcessorOptions ToSessionProcessorOptions() =>
                 AutoCompleteMessages = AutoCompleteMessages,
                 PrefetchCount = PrefetchCount,
                 MaxAutoLockRenewalDuration = MaxAutoLockRenewalDuration,
-                MaxConcurrentSessions = MaxConcurrentSessions
+                MaxConcurrentSessions = MaxConcurrentSessions,
+                SessionIdleTimeout = SessionIdleTimeout
             };
 
         internal ServiceBusClientOptions ToClientOptions() =>

sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Listeners/ServiceBusListener.cs

@@ -5,7 +5,6 @@
 using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
 using System.Linq;
-using System.Reflection;
 using System.Threading;
 using System.Threading.Tasks;
 using Azure.Core.Pipeline;
@@ -194,40 +193,37 @@ internal async Task ProcessMessageAsync(ProcessMessageEventArgs args)
         {
             using (CancellationTokenSource linkedCts = CancellationTokenSource.CreateLinkedTokenSource(args.CancellationToken, _cancellationTokenSource.Token))
             {
-                //TODO consider using internals visible or exposing the Receiver property instead of reflection
-                ServiceBusReceiver receiver = (ServiceBusReceiver) typeof(ProcessMessageEventArgs).GetField("_receiver", BindingFlags.NonPublic | BindingFlags.Instance).GetValue(args);
-
-                if (!await _messageProcessor.BeginProcessingMessageAsync(receiver, args.Message, linkedCts.Token).ConfigureAwait(false))
+                var actions = new ServiceBusMessageActions(args);
+                if (!await _messageProcessor.BeginProcessingMessageAsync(actions, args.Message, linkedCts.Token).ConfigureAwait(false))
                 {
                     return;
                 }
 
                 ServiceBusTriggerInput input = ServiceBusTriggerInput.CreateSingle(args.Message);
-                input.Receiver = receiver;
+                input.MessageActions = actions;
 
                 TriggeredFunctionData data = input.GetTriggerFunctionData();
                 FunctionResult result = await _triggerExecutor.TryExecuteAsync(data, linkedCts.Token).ConfigureAwait(false);
-                await _messageProcessor.CompleteProcessingMessageAsync(receiver, args.Message, result, linkedCts.Token).ConfigureAwait(false);
+                await _messageProcessor.CompleteProcessingMessageAsync(actions, args.Message, result, linkedCts.Token).ConfigureAwait(false);
             }
         }
 
         internal async Task ProcessSessionMessageAsync(ProcessSessionMessageEventArgs args)
         {
             using (CancellationTokenSource linkedCts = CancellationTokenSource.CreateLinkedTokenSource(args.CancellationToken, _cancellationTokenSource.Token))
             {
-                ServiceBusSessionReceiver receiver = (ServiceBusSessionReceiver)typeof(ProcessSessionMessageEventArgs).GetField("_sessionReceiver", BindingFlags.NonPublic | BindingFlags.Instance).GetValue(args);
-
-                if (!await _sessionMessageProcessor.BeginProcessingMessageAsync(receiver, args.Message, linkedCts.Token).ConfigureAwait(false))
+                var actions = new ServiceBusSessionMessageActions(args);
+                if (!await _sessionMessageProcessor.BeginProcessingMessageAsync(actions, args.Message, linkedCts.Token).ConfigureAwait(false))
                 {
                     return;
                 }
 
                 ServiceBusTriggerInput input = ServiceBusTriggerInput.CreateSingle(args.Message);
-                input.SessionReceiver = receiver;
+                input.MessageActions = actions;
 
                 TriggeredFunctionData data = input.GetTriggerFunctionData();
                 FunctionResult result = await _triggerExecutor.TryExecuteAsync(data, linkedCts.Token).ConfigureAwait(false);
-                await _sessionMessageProcessor.CompleteProcessingMessageAsync(receiver, args.Message, result, linkedCts.Token).ConfigureAwait(false);
+                await _sessionMessageProcessor.CompleteProcessingMessageAsync(actions, args.Message, result, linkedCts.Token).ConfigureAwait(false);
             }
         }
 
@@ -273,21 +269,20 @@ internal void StartMessageBatchReceiver(CancellationToken cancellationToken)
                             }
                         }
 
-                        IReadOnlyList<ServiceBusReceivedMessage> messages = await receiver.ReceiveMessagesAsync(
-                            _serviceBusOptions.MaxMessages,
-                            _serviceBusOptions.MaxWaitTime).ConfigureAwait(false);
+                        IReadOnlyList<ServiceBusReceivedMessage> messages =
+                            await receiver.ReceiveMessagesAsync(_serviceBusOptions.MaxMessages).ConfigureAwait(false);
 
                         if (messages != null)
                         {
                             ServiceBusReceivedMessage[] messagesArray = messages.ToArray();
                             ServiceBusTriggerInput input = ServiceBusTriggerInput.CreateBatch(messagesArray);
                             if (_isSessionsEnabled)
                             {
-                                input.SessionReceiver = (ServiceBusSessionReceiver) receiver;
+                                input.MessageActions = new ServiceBusSessionMessageActions((ServiceBusSessionReceiver) receiver);
                             }
                             else
                             {
-                                input.Receiver = receiver;
+                                input.MessageActions = new ServiceBusMessageActions(receiver);
                             }
                             FunctionResult result = await _triggerExecutor.TryExecuteAsync(input.GetTriggerFunctionData(), cancellationToken).ConfigureAwait(false);
 

…sions.ServiceBus/src/MessageProcessor.cs …ceBus/src/Primitives/MessageProcessor.cssdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/MessageProcessor.cs renamed to sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Primitives/MessageProcessor.cs sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/MessageProcessor.cs renamed to sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Primitives/MessageProcessor.cs

@@ -10,37 +10,33 @@
 namespace Microsoft.Azure.WebJobs.ServiceBus
 {
     /// <summary>
-    /// This class defines a strategy used for processing ServiceBus messages.
+    /// This class defines a strategy used for processing Service Bus messages.
     /// </summary>
-    /// <remarks>
-    /// Custom <see cref="ServiceBus.MessageProcessor"/> implementations can be specified by implementing
-    /// a custom <see cref="MessagingProvider"/> and setting it via ServiceBusOptions.MessagingProvider.
-    /// </remarks>
     public class MessageProcessor
     {
         /// <summary>
-        /// Constructs a new instance.
+        /// Initializes a new instance of <see cref="MessageProcessor"/>.
         /// </summary>
-        /// <param name="processor">The <see cref="Processor"/>.</param>
+        /// <param name="processor">The <see cref="ServiceBusProcessor"/> to use.</param>
         public MessageProcessor(ServiceBusProcessor processor)
         {
             Processor = processor ?? throw new ArgumentNullException(nameof(processor));
         }
 
         /// <summary>
-        /// Gets or sets the <see cref="Processor"/> that will be used by the <see cref="Processor"/>.
+        /// Gets or sets the <see cref="ServiceBusProcessor"/> that will be used by the <see cref="Processor"/>.
         /// </summary>
         internal ServiceBusProcessor Processor { get; set; }
 
         /// <summary>
         /// This method is called when there is a new message to process, before the job function is invoked.
         /// This allows any preprocessing to take place on the message before processing begins.
         /// </summary>
-        /// <param name="receiver"></param>
-        /// <param name="message"></param>
-        /// <param name="cancellationToken"></param>
+        /// <param name="messageActions">The set of actions that can be performed on a <see cref="ServiceBusReceivedMessage"/>.</param>
+        /// <param name="message">The <see cref="ServiceBusReceivedMessage"/> to process.</param>
+        /// <param name="cancellationToken">A cancellation token that will be cancelled when the processor is shutting down.</param>
         /// <returns>A <see cref="Task"/> that returns true if the message processing should continue, false otherwise.</returns>
-        public virtual Task<bool> BeginProcessingMessageAsync(ServiceBusReceiver receiver, ServiceBusReceivedMessage message, CancellationToken cancellationToken)
+        public virtual Task<bool> BeginProcessingMessageAsync(ServiceBusMessageActions messageActions, ServiceBusReceivedMessage message, CancellationToken cancellationToken)
         {
             return Task.FromResult<bool>(true);
         }
@@ -53,12 +49,12 @@ public virtual Task<bool> BeginProcessingMessageAsync(ServiceBusReceiver receive
         /// is configured. E.g. if <see cref="ServiceBusProcessorOptions.AutoCompleteMessages"/> is false, it is up to the job function to complete
         /// the message.
         /// </remarks>
-        /// <param name="receiver"></param>
-        /// <param name="message"></param>
+        /// <param name="messageActions">The set of actions that can be performed on a <see cref="ServiceBusReceivedMessage"/>.</param>
+        /// <param name="message">The <see cref="ServiceBusReceivedMessage"/> to process.</param>
         /// <param name="result">The <see cref="FunctionResult"/> from the job invocation.</param>
-        /// <param name="cancellationToken">The <see cref="CancellationToken"/> to use</param>
+        /// <param name="cancellationToken">A cancellation token that will be cancelled when the processor is shutting down.</param>
         /// <returns>A <see cref="Task"/> that will complete the message processing.</returns>
-        public virtual Task CompleteProcessingMessageAsync(ServiceBusReceiver receiver, ServiceBusReceivedMessage message, FunctionResult result, CancellationToken cancellationToken)
+        public virtual Task CompleteProcessingMessageAsync(ServiceBusMessageActions messageActions, ServiceBusReceivedMessage message, FunctionResult result, CancellationToken cancellationToken)
         {
             if (message is null)
             {