Validate that EnableCrossEntityTransactions is configured when SendsAtomicWithReceive is enabled

src/ManualTests.HostV4/SomeEvent.cs

@@ -0,0 +1,8 @@
+using NServiceBus;
+
+public class SomeEvent : IEvent
+{
+    public SomeEvent()
+    {
+    }
+}

src/ManualTests.HostV4/SomeEventMessageHandler.cs

@@ -0,0 +1,15 @@
+using System.Threading.Tasks;
+using NServiceBus;
+using NServiceBus.Logging;
+
+public class SomeEventMessageHandler : IHandleMessages<SomeEvent>
+{
+    static readonly ILog Log = LogManager.GetLogger<SomeEventMessageHandler>();
+
+    public Task Handle(SomeEvent message, IMessageHandlerContext context)
+    {
+        Log.Warn($"Handling {nameof(SomeEvent)} in {nameof(SomeEventMessageHandler)}");
+
+        return Task.CompletedTask;
+    }
+}

src/ManualTests.HostV4/Startup.cs

@@ -3,7 +3,6 @@
 
 [assembly: FunctionsStartup(typeof(Startup))]
 [assembly: NServiceBusTriggerFunction("InProcess-HostV4", SendsAtomicWithReceive = true)]
-
 public class Startup : FunctionsStartup
 {
     public override void Configure(IFunctionsHostBuilder builder)

src/ManualTests.HostV4/TriggerMessageHandler.cs

@@ -6,10 +6,11 @@ public class TriggerMessageHandler : IHandleMessages<TriggerMessage>
 {
     static readonly ILog Log = LogManager.GetLogger<TriggerMessageHandler>();
 
-    public Task Handle(TriggerMessage message, IMessageHandlerContext context)
+    public async Task Handle(TriggerMessage message, IMessageHandlerContext context)
     {
         Log.Warn($"Handling {nameof(TriggerMessage)} in {nameof(TriggerMessageHandler)}");
 
-        return context.SendLocal(new SomeOtherMessage());
+        await context.SendLocal(new SomeOtherMessage()).ConfigureAwait(false);
+        await context.Publish(new SomeEvent()).ConfigureAwait(false);
     }
 }

src/ManualTests.HostV4/host.json

@@ -1,3 +1,8 @@
 {
-    "version": "2.0"
+  "version": "2.0",
+  "extensions": {
+    "ServiceBus": {
+      "EnableCrossEntityTransactions": true
+    }
+  }
 }

src/NServiceBus.AzureFunctions.InProcess.ServiceBus/FunctionsHostBuilderExtensions.cs

@@ -45,7 +45,8 @@ public static void UseNServiceBus(
             Action<ServiceBusTriggeredEndpointConfiguration> configurationFactory = null)
         {
             var config = functionsHostBuilder.GetContext().Configuration;
-            var serviceBusConfiguration = new ServiceBusTriggeredEndpointConfiguration(endpointName, config, null);
+
+            var serviceBusConfiguration = new ServiceBusTriggeredEndpointConfiguration(endpointName, config, GetTransportTransactionMode());
             configurationFactory?.Invoke(serviceBusConfiguration);
             RegisterEndpointFactory(functionsHostBuilder, serviceBusConfiguration);
         }
@@ -60,7 +61,7 @@ public static void UseNServiceBus(
             Action<ServiceBusTriggeredEndpointConfiguration> configurationFactory = null)
         {
             var config = functionsHostBuilder.GetContext().Configuration;
-            var serviceBusConfiguration = new ServiceBusTriggeredEndpointConfiguration(endpointName, config, connectionString);
+            var serviceBusConfiguration = new ServiceBusTriggeredEndpointConfiguration(endpointName, config, GetTransportTransactionMode(), connectionString);
             configurationFactory?.Invoke(serviceBusConfiguration);
             RegisterEndpointFactory(functionsHostBuilder, serviceBusConfiguration);
         }
@@ -109,5 +110,19 @@ internal static IStartableEndpointWithExternallyManagedContainer Configure(
                     endpointConfiguration,
                     serviceCollection);
         }
+
+        static TransportTransactionMode GetTransportTransactionMode()
+        {
+            foreach (var assembly in AppDomain.CurrentDomain.GetAssemblies())
+            {
+                var attribute = assembly.GetCustomAttribute<NServiceBusTriggerFunctionAttribute>();
+
+                if (attribute != null)
+                {
+                    return attribute.SendsAtomicWithReceive ? TransportTransactionMode.SendsAtomicWithReceive : TransportTransactionMode.ReceiveOnly;
+                }
+            }
+            return TransportTransactionMode.SendsAtomicWithReceive;
+        }
     }
 }

src/NServiceBus.AzureFunctions.InProcess.ServiceBus/ServiceBusTriggeredEndpointConfiguration.cs

@@ -6,6 +6,7 @@
     using AzureFunctions.InProcess.ServiceBus;
     using Configuration.AdvancedExtensibility;
     using Logging;
+    using Microsoft.Azure.WebJobs.ServiceBus;
     using Microsoft.Extensions.Configuration;
     using Serialization;
 
@@ -40,7 +41,7 @@ static ServiceBusTriggeredEndpointConfiguration()
         /// <summary>
         /// Creates a serverless NServiceBus endpoint.
         /// </summary>
-        internal ServiceBusTriggeredEndpointConfiguration(string endpointName, IConfiguration configuration, string connectionString = default)
+        internal ServiceBusTriggeredEndpointConfiguration(string endpointName, IConfiguration configuration, TransportTransactionMode transportTransactionMode, string connectionString = default)
         {
             var endpointConfiguration = new EndpointConfiguration(endpointName);
 
@@ -74,9 +75,23 @@ internal ServiceBusTriggeredEndpointConfiguration(string endpointName, IConfigur
                 }
             }
 
+            if (transportTransactionMode == TransportTransactionMode.SendsAtomicWithReceive)
+            {
+                var serviceBusOptions = configuration.GetSection("AzureFunctionsJobHost:extensions:ServiceBus")
+                    .Get<ServiceBusOptions>();
+
+                if (serviceBusOptions == null || serviceBusOptions.EnableCrossEntityTransactions)
+                {
+                    throw new Exception("SendsAtomicWithReceive mode requires EnableCrossEntityTransactions needs to be enabled on the ServiceBusOptions.");
+                }
+            }
+
             Transport = new AzureServiceBusTransport(connectionString);
 
-            serverlessTransport = new ServerlessTransport(Transport);
+            serverlessTransport = new ServerlessTransport(Transport)
+            {
+                TransportTransactionMode = transportTransactionMode
+            };
             var serverlessRouting = endpointConfiguration.UseTransport(serverlessTransport);
             Routing = new RoutingSettings<AzureServiceBusTransport>(serverlessRouting.GetSettings());
 

src/ServiceBus.AcceptanceTests/FunctionEndpointComponent.cs

@@ -83,7 +83,7 @@ public FunctionRunner(IList<object> messages,
 
             public override Task Start(CancellationToken token)
             {
-                var functionEndpointConfiguration = new ServiceBusTriggeredEndpointConfiguration(Name, default, null);
+                var functionEndpointConfiguration = new ServiceBusTriggeredEndpointConfiguration(Name, default, TransportTransactionMode.ReceiveOnly);
                 configurationCustomization(functionEndpointConfiguration);
                 var endpointConfiguration = functionEndpointConfiguration.AdvancedConfiguration;
 

src/ServiceBus.Tests/When_no_connection_string_is_provided.cs

@@ -18,7 +18,7 @@ public void Should_guide_user_towards_success()
                 Environment.SetEnvironmentVariable(defaultConnectionStringKey, null, EnvironmentVariableTarget.Process);
 
                 var exception = Assert.Throws<Exception>(
-                    () => new ServiceBusTriggeredEndpointConfiguration("SampleEndpoint", default, null),
+                    () => new ServiceBusTriggeredEndpointConfiguration("SampleEndpoint", default, TransportTransactionMode.ReceiveOnly),
                     "Exception should be thrown at endpoint creation so that the error will be found during functions startup"
                 );
 

src/ServiceBus.Tests/When_shipping_handlers_in_dedicated_assembly.cs

@@ -24,7 +24,7 @@ public async Task Should_load_handlers_from_assembly_when_using_FunctionsHostBui
 
             var serviceCollection = new ServiceCollection();
 
-            var configuration = new ServiceBusTriggeredEndpointConfiguration("assemblyTest", default, null);
+            var configuration = new ServiceBusTriggeredEndpointConfiguration("assemblyTest", default, TransportTransactionMode.ReceiveOnly);
             configuration.UseSerialization<XmlSerializer>();
 
             var endpointConfiguration = configuration.AdvancedConfiguration;