Opt-out from reusing SQL Server transport's connection

src/SqlPersistence.Tests/APIApprovals.Approve.approved.txt

@@ -132,7 +132,11 @@ namespace NServiceBus
             public Oracle() { }
         }
     }
-    public class SqlDialectSettings<T>
+    public abstract class SqlDialectSettings
+    {
+        protected SqlDialectSettings(NServiceBus.SqlDialect dialect) { }
+    }
+    public class SqlDialectSettings<T> : NServiceBus.SqlDialectSettings
         where T : NServiceBus.SqlDialect, new ()
     {
         public SqlDialectSettings() { }
@@ -141,6 +145,7 @@ namespace NServiceBus
     {
         public static void ConnectionBuilder(this NServiceBus.PersistenceExtensions<NServiceBus.Persistence.Sql.SqlPersistence> configuration, System.Func<System.Data.Common.DbConnection> connectionBuilder) { }
         public static void DisableInstaller(this NServiceBus.PersistenceExtensions<NServiceBus.Persistence.Sql.SqlPersistence> configuration) { }
+        public static void DoNotUseSqlServerTransportConnection(this NServiceBus.SqlDialectSettings<NServiceBus.SqlDialect.MsSqlServer> dialectSettings) { }
         public static NServiceBus.Persistence.Sql.SagaSettings SagaSettings(this NServiceBus.PersistenceExtensions<NServiceBus.Persistence.Sql.SqlPersistence> configuration) { }
         [System.ObsoleteAttribute("Use `persistence.SqlDialect<SqlDialect.DialectType>().Schema(\"schema_name\")` inst" +
             "ead. The member currently throws a NotImplementedException. Will be removed in v" +

src/SqlPersistence.Tests/Integration/SqlTransportIntegrationTests.cs

@@ -1,4 +1,5 @@
 //TODO: re enable when sql transport is updated
+//TODO: add tests for the new option to not reuse the connection
 
 /**
 using System;

src/SqlPersistence/Config/SqlDialectSettings.cs

@@ -3,16 +3,32 @@ namespace NServiceBus
     /// <summary>
     /// Exposes settings options available for the selected database engine.
     /// </summary>
-    public class SqlDialectSettings<T> where T : SqlDialect, new()
+    public abstract class SqlDialectSettings
+    {
+        internal SqlDialect Dialect;
+
+        /// <summary>
+        /// Exposes settings options available for the selected database engine.
+        /// </summary>
+        protected SqlDialectSettings(SqlDialect dialect)
+        {
+            this.Dialect = dialect;
+        }
+    }
+
+    /// <summary>
+    /// Exposes settings options available for the selected database engine.
+    /// </summary>
+    public class SqlDialectSettings<T> : SqlDialectSettings
+        where T : SqlDialect, new()
     {
         /// <summary>
         /// Exposes settings options available for the selected database engine.
         /// </summary>
-        public SqlDialectSettings()
+        public SqlDialectSettings() : base(new T())
         {
-            Settings = new T();
         }
 
-        internal T Settings { get; }
+        internal T TypedDialect => (T)Dialect;
     }
 }

src/SqlPersistence/Config/SqlPersistenceConfig_SqlDialect.cs

@@ -13,9 +13,9 @@ public static partial class SqlPersistenceConfig
 
         internal static SqlDialect GetSqlDialect(this ReadOnlySettings settings)
         {
-            if (settings.TryGet("SqlPersistence.SqlDialect", out SqlDialect value))
+            if (settings.TryGet("SqlPersistence.SqlDialect", out SqlDialectSettings value))
             {
-                return value;
+                return value.Dialect;
             }
             throw new Exception("Must specify SQL dialect using persistence.SqlDialect<T>() method.");
         }
@@ -27,11 +27,17 @@ internal static SqlDialect GetSqlDialect(this ReadOnlySettings settings)
         public static SqlDialectSettings<T> SqlDialect<T>(this PersistenceExtensions<SqlPersistence> configuration) where T : SqlDialect, new()
         {
             var settings = configuration.GetSettings();
-
+
+            SqlDialectSettings<T> dialectSettings;
+            if (settings.TryGet("SqlPersistence.SqlDialect", out dialectSettings))
+            {
+                return dialectSettings;
+            }
+
             var type = typeof(SqlDialectSettings<>).MakeGenericType(typeof(T));
-            var instance = (SqlDialectSettings<T>)Activator.CreateInstance(type);
-            settings.Set("SqlPersistence.SqlDialect", instance.Settings);
-            return instance;
+            dialectSettings = (SqlDialectSettings<T>)Activator.CreateInstance(type);
+            settings.Set("SqlPersistence.SqlDialect", dialectSettings);
+            return dialectSettings;
         }
     }
 }

src/SqlPersistence/Config/SqlPersistenceConfig_Schema.cs → src/SqlPersistence/Config/SqlPersistenceConfig_SqlDialect_MsSqlServer.cs

@@ -11,8 +11,16 @@ public static void Schema(this SqlDialectSettings<SqlDialect.MsSqlServer> dialec
             Guard.AgainstNull(nameof(dialectSettings), dialectSettings);
             Guard.AgainstNullAndEmpty(nameof(schema), schema);
             Guard.AgainstSqlDelimiters(nameof(schema), schema);
-            dialectSettings.Settings.Schema = schema;
+            dialectSettings.TypedDialect.Schema = schema;
         }
 
+        /// <summary>
+        /// Instructs the persistence to not use the connection established by the SQL Server transport.
+        /// </summary>
+        public static void DoNotUseSqlServerTransportConnection(this SqlDialectSettings<SqlDialect.MsSqlServer> dialectSettings)
+        {
+            Guard.AgainstNull(nameof(dialectSettings), dialectSettings);
+            dialectSettings.TypedDialect.DoNotUseTransportConnection = true;
+        }
     }
 }

src/SqlPersistence/Installer.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Generic;
 using System.Data.Common;
 using System.IO;
 using System.Linq;

src/SqlPersistence/SynchronizedStorage/SqlDialect.cs

@@ -0,0 +1,14 @@
+namespace NServiceBus
+{
+    using System;
+    using System.Data.Common;
+    using System.Threading.Tasks;
+    using Extensibility;
+    using Persistence;
+    using Transport;
+
+    public partial class SqlDialect
+    {
+        internal abstract Task<CompletableSynchronizedStorageSession> TryAdaptTransportConnection(TransportTransaction transportTransaction, ContextBag context, Func<DbConnection> connectionBuilder, Func<DbConnection, DbTransaction, bool, StorageSession> storageSessionFactory);
+    }
+}

src/SqlPersistence/SynchronizedStorage/SqlDialect_MsSqlServer.cs

@@ -0,0 +1,54 @@
+namespace NServiceBus
+{
+    using System;
+    using System.Data.Common;
+    using System.Threading.Tasks;
+    using System.Transactions;
+    using Extensibility;
+    using Persistence;
+    using Transport;
+
+    public partial class SqlDialect
+    {
+        public partial class MsSqlServer
+        {
+            internal override async Task<CompletableSynchronizedStorageSession> TryAdaptTransportConnection(TransportTransaction transportTransaction, ContextBag context, Func<DbConnection> connectionBuilder, Func<DbConnection, DbTransaction, bool, StorageSession> storageSessionFactory)
+            {
+                if (DoNotUseTransportConnection)
+                {
+                    return null;
+                }
+
+                //SQL server transport in native TX mode
+                if (transportTransaction.TryGet("System.Data.SqlClient.SqlConnection", out DbConnection existingSqlConnection) &&
+                    transportTransaction.TryGet("System.Data.SqlClient.SqlTransaction", out DbTransaction existingSqlTransaction))
+                {
+                    return storageSessionFactory(existingSqlConnection, existingSqlTransaction, false);
+                }
+
+                // Transport supports DTC and uses TxScope owned by the transport
+                var scopeTx = Transaction.Current;
+                if (transportTransaction.TryGet(out Transaction transportTx) &&
+                    scopeTx != null &&
+                    transportTx != scopeTx)
+                {
+                    throw new Exception("A TransactionScope has been opened in the current context overriding the one created by the transport. "
+                                        + "This setup can result in inconsistent data because operations done via connections enlisted in the context scope won't be committed "
+                                        + "atomically with the receive transaction. To manually control the TransactionScope in the pipeline switch the transport transaction mode "
+                                        + $"to values lower than '{nameof(TransportTransactionMode.TransactionScope)}'.");
+                }
+                var ambientTransaction = transportTx ?? scopeTx;
+                if (ambientTransaction == null)
+                {
+                    //Other modes handled by creating a new session.
+                    return null;
+                }
+                var connection = await connectionBuilder.OpenConnection().ConfigureAwait(false);
+                connection.EnlistTransaction(ambientTransaction);
+                return storageSessionFactory(connection, null, true);
+            }
+
+            internal bool DoNotUseTransportConnection { get; set; }
+        }
+    }
+}

src/SqlPersistence/SynchronizedStorage/SqlDialect_MySql.cs

@@ -0,0 +1,23 @@
+namespace NServiceBus
+{
+    using System;
+    using System.Data.Common;
+    using System.Threading.Tasks;
+    using Extensibility;
+    using Persistence;
+    using Transport;
+
+    public partial class SqlDialect
+    {
+        public partial class MySql
+        {
+            static Task<CompletableSynchronizedStorageSession> result = Task.FromResult((CompletableSynchronizedStorageSession)null);
+
+            internal override Task<CompletableSynchronizedStorageSession> TryAdaptTransportConnection(TransportTransaction transportTransaction, ContextBag context, Func<DbConnection> connectionBuilder, Func<DbConnection, DbTransaction, bool, StorageSession> storageSessionFactory)
+            {
+                // MySQL does not support DTC so we should not enlist if transport has such a transaction.
+                return result;
+            }
+        }
+    }
+}

src/SqlPersistence/SynchronizedStorage/SqlDialect_Oracle.cs

@@ -0,0 +1,40 @@
+namespace NServiceBus
+{
+    using System;
+    using System.Data.Common;
+    using System.Threading.Tasks;
+    using System.Transactions;
+    using Extensibility;
+    using Persistence;
+    using Transport;
+
+    public partial class SqlDialect
+    {
+        public partial class Oracle
+        {
+            internal override async Task<CompletableSynchronizedStorageSession> TryAdaptTransportConnection(TransportTransaction transportTransaction, ContextBag context, Func<DbConnection> connectionBuilder, Func<DbConnection, DbTransaction, bool, StorageSession> storageSessionFactory)
+            {
+                // Oracle supports DTC so we should enlist in the transport's TransactionScope if present
+                var scopeTx = Transaction.Current;
+                if (transportTransaction.TryGet(out Transaction transportTx) &&
+                    scopeTx != null &&
+                    transportTx != scopeTx)
+                {
+                    throw new Exception("A TransactionScope has been opened in the current context overriding the one created by the transport. "
+                                        + "This setup can result in inconsistent data because operations done via connections enlisted in the context scope won't be committed "
+                                        + "atomically with the receive transaction. To manually control the TransactionScope in the pipeline switch the transport transaction mode "
+                                        + $"to values lower than '{nameof(TransportTransactionMode.TransactionScope)}'.");
+                }
+                var ambientTransaction = transportTx ?? scopeTx;
+                if (ambientTransaction == null)
+                {
+                    //Other modes handled by creating a new session.
+                    return null;
+                }
+                var connection = await connectionBuilder.OpenConnection().ConfigureAwait(false);
+                connection.EnlistTransaction(ambientTransaction);
+                return storageSessionFactory(connection, null, true);
+            }
+        }
+    }
+}

src/SqlPersistence/SynchronizedStorage/StorageAdapter.cs

@@ -1,7 +1,6 @@
 using System;
 using System.Data.Common;
 using System.Threading.Tasks;
-using System.Transactions;
 using NServiceBus;
 using NServiceBus.Extensibility;
 using NServiceBus.Outbox;
@@ -13,12 +12,14 @@ class StorageAdapter : ISynchronizedStorageAdapter
     static Task<CompletableSynchronizedStorageSession> EmptyResultTask = Task.FromResult(default(CompletableSynchronizedStorageSession));
 
     SagaInfoCache infoCache;
+    SqlDialect dialect;
     Func<DbConnection> connectionBuilder;
 
-    public StorageAdapter(Func<DbConnection> connectionBuilder, SagaInfoCache infoCache)
+    public StorageAdapter(Func<DbConnection> connectionBuilder, SagaInfoCache infoCache, SqlDialect dialect)
     {
         this.connectionBuilder = connectionBuilder;
         this.infoCache = infoCache;
+        this.dialect = dialect;
     }
 
     public Task<CompletableSynchronizedStorageSession> TryAdapt(OutboxTransaction transaction, ContextBag context)
@@ -32,35 +33,9 @@ public Task<CompletableSynchronizedStorageSession> TryAdapt(OutboxTransaction tr
         return Task.FromResult(session);
     }
 
-    public async Task<CompletableSynchronizedStorageSession> TryAdapt(TransportTransaction transportTransaction, ContextBag context)
+    public Task<CompletableSynchronizedStorageSession> TryAdapt(TransportTransaction transportTransaction, ContextBag context)
     {
-        //SQL server transport in native TX mode
-        if (transportTransaction.TryGet("System.Data.SqlClient.SqlConnection", out DbConnection existingSqlConnection) &&
-            transportTransaction.TryGet("System.Data.SqlClient.SqlTransaction", out DbTransaction existingSqlTransaction))
-        {
-            return new StorageSession(existingSqlConnection, existingSqlTransaction, false, infoCache);
-        }
-
-        // Transport supports DTC and uses TxScope owned by the transport
-        var scopeTx = Transaction.Current;
-        if (transportTransaction.TryGet(out Transaction transportTx) &&
-            scopeTx != null &&
-            transportTx != scopeTx)
-        {
-            throw new Exception("A TransactionScope has been opened in the current context overriding the one created by the transport. "
-                + "This setup can result in inconsistent data because operations done via connections enlisted in the context scope won't be committed "
-                + "atomically with the receive transaction. To manually control the TransactionScope in the pipeline switch the transport transaction mode "
-                + $"to values lower than '{nameof(TransportTransactionMode.TransactionScope)}'.");
-        }
-        var ambientTransaction = transportTx ?? scopeTx;
-        if (ambientTransaction == null)
-        {
-            //Other modes handled by creating a new session.
-            return null;
-        }
-        var connection = await connectionBuilder.OpenConnection().ConfigureAwait(false);
-        connection.EnlistTransaction(ambientTransaction);
-        return new StorageSession(connection, null, true, infoCache);
-
+        return dialect.TryAdaptTransportConnection(transportTransaction, context, connectionBuilder, 
+            (conn, trans, ownsTx) => new StorageSession(conn, trans, ownsTx, infoCache));
     }
 }

src/SqlPersistence/SynchronizedStorage/StorageSessionFeature.cs

@@ -27,7 +27,7 @@ protected override void Setup(FeatureConfigurationContext context)
         {
             infoCache = BuildSagaInfoCache(sqlDialect, settings);
             container.ConfigureComponent(() => new SynchronizedStorage(connectionBuilder, infoCache), DependencyLifecycle.SingleInstance);
-            container.ConfigureComponent(() => new StorageAdapter(connectionBuilder, infoCache), DependencyLifecycle.SingleInstance);
+            container.ConfigureComponent(() => new StorageAdapter(connectionBuilder, infoCache, sqlDialect), DependencyLifecycle.SingleInstance);
         }
         if (isSagasEnabledForSqlPersistence)
         {