Add MySql support (#221)

src/Akka.Persistence.Sql.Tests.Common/Containers/MySqlContainer.cs

@@ -26,20 +26,22 @@ public sealed class MySqlContainer : DockerContainer
 
         private readonly DbConnectionStringBuilder _connectionStringBuilder;
 
-        public MySqlContainer() : base("mysql", "latest", $"mysql-{Guid.NewGuid():N}")
+        public MySqlContainer() : base("mysql", "8", $"mysql-{Guid.NewGuid():N}")
         {
             _connectionStringBuilder = new DbConnectionStringBuilder
             {
                 ["Server"] = "localhost",
                 ["Port"] = Port.ToString(),
                 ["User Id"] = User,
-                ["Password"] = Password
+                ["Password"] = Password,
+                ["allowPublicKeyRetrieval"] = "true",
+                ["Allow User Variables"] = "true"
             };
         }
 
         public override string ConnectionString => _connectionStringBuilder.ToString();
 
-        public override string ProviderName => LinqToDB.ProviderName.MySql;
+        public override string ProviderName => LinqToDB.ProviderName.MySqlOfficial;
 
         private int Port { get; } = ThreadLocalRandom.Current.Next(9000, 10000);
 
@@ -70,23 +72,56 @@ protected override void ConfigureContainer(CreateContainerParameters parameters)
             parameters.Env = new[]
             {
                 $"MYSQL_ROOT_PASSWORD={Password}",
-                $"MYSQL_DATABASE={DatabaseName}"
             };
         }
 
+        protected override async Task AfterContainerStartedAsync()
+        {
+            await using var connection = new MySqlConnection(ConnectionString);
+            await connection.OpenAsync();
+
+            await using var command = new MySqlCommand
+            {
+                CommandText = "SET GLOBAL max_connections = 999;",
+                Connection = connection
+            };
+            await command.ExecuteNonQueryAsync();
+            await connection.CloseAsync();
+
+            await base.AfterContainerStartedAsync();
+        }
+
         public override async Task InitializeDbAsync()
         {
-            GenerateDatabaseName();
+            _connectionStringBuilder["Database"] = "sys";
 
             await using var connection = new MySqlConnection(ConnectionString);
             await connection.OpenAsync();
 
+            if (!string.IsNullOrWhiteSpace(DatabaseName))
+            {
+                try
+                {
+                    await using var dropCommand = new MySqlCommand
+                    {
+                        CommandText = @$"DROP DATABASE IF EXISTS `{DatabaseName}`;",
+                        Connection = connection
+                    };
+                    await dropCommand.ExecuteNonQueryAsync();
+                }
+                catch
+                {
+                    // no-op
+                }
+            }
+
+            GenerateDatabaseName();
+
             await using var command = new MySqlCommand
             {
-                CommandText = $"CREATE DATABASE {DatabaseName}",
+                CommandText = $"CREATE DATABASE `{DatabaseName}`;",
                 Connection = connection
             };
-
             await command.ExecuteNonQueryAsync();
             await connection.CloseAsync();
         }

src/Akka.Persistence.Sql.Tests/Akka.Persistence.Sql.Tests.csproj

@@ -13,6 +13,7 @@
     <PackageReference Include="Akka.Persistence.Sqlite" />
     <PackageReference Include="Akka.Persistence.SqlServer" />
     <PackageReference Include="Akka.Persistence.PostgreSql" />
+    <PackageReference Include="Akka.Persistence.MySql" />
     <PackageReference Include="Akka.Serialization.Hyperion" />
 
     <PackageReference Include="Docker.DotNet" />
@@ -22,6 +23,8 @@
 
     <PackageReference Include="System.Data.SQLite.Core" />
 
+    <PackageReference Include="MySql.Data" />
+
     <PackageReference Include="xunit" />
     <PackageReference Include="xunit.runner.visualstudio">
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>

src/Akka.Persistence.Sql.Tests/MySql/Compatibility/MySqlCommonJournalCompatibilitySpec.cs

@@ -0,0 +1,35 @@
+// -----------------------------------------------------------------------
+//  <copyright file="PostgreSqlCommonJournalCompatibilitySpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using System;
+using Akka.Persistence.Sql.Tests.Common.Containers;
+using Xunit;
+using Xunit.Abstractions;
+#if !DEBUG
+using Akka.Persistence.Sql.Tests.Common.Internal.Xunit;
+#endif
+
+namespace Akka.Persistence.Sql.Tests.MySql.Compatibility
+{
+#if !DEBUG
+    [SkipWindows]
+#endif
+    [Collection(nameof(MySqlPersistenceSpec))]
+    public class MySqlCommonJournalCompatibilitySpec : SqlCommonJournalCompatibilitySpec<MySqlContainer>
+    {
+        public MySqlCommonJournalCompatibilitySpec(ITestOutputHelper output, MySqlContainer fixture)
+            : base(fixture, output)
+        {
+        }
+
+        protected override Func<MySqlContainer, Configuration.Config> Config => fixture 
+            => MySqlCompatibilitySpecConfig.InitJournalConfig(fixture, "event_journal", "metadata");
+
+        protected override string OldJournal => "akka.persistence.journal.mysql";
+
+        protected override string NewJournal => "akka.persistence.journal.sql";
+    }
+}

src/Akka.Persistence.Sql.Tests/MySql/Compatibility/MySqlCommonSnapshotCompatibilitySpec.cs

@@ -0,0 +1,35 @@
+// -----------------------------------------------------------------------
+//  <copyright file="PostgreSqlCommonSnapshotCompatibilitySpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using System;
+using Akka.Persistence.Sql.Tests.Common.Containers;
+using Xunit;
+using Xunit.Abstractions;
+#if !DEBUG
+using Akka.Persistence.Sql.Tests.Common.Internal.Xunit;
+#endif
+
+namespace Akka.Persistence.Sql.Tests.MySql.Compatibility
+{
+#if !DEBUG
+    [SkipWindows]
+#endif
+    [Collection(nameof(MySqlPersistenceSpec))]
+    public class MySqlCommonSnapshotCompatibilitySpec : SqlCommonSnapshotCompatibilitySpec<MySqlContainer>
+    {
+        public MySqlCommonSnapshotCompatibilitySpec(ITestOutputHelper output, MySqlContainer fixture)
+            : base(fixture, output)
+        {
+        }
+
+        protected override string OldSnapshot => "akka.persistence.snapshot-store.mysql";
+
+        protected override string NewSnapshot => "akka.persistence.snapshot-store.sql";
+
+        protected override Func<MySqlContainer, Configuration.Config> Config => fixture
+            => MySqlCompatibilitySpecConfig.InitSnapshotConfig(fixture, "snapshot_store");
+    }
+}

src/Akka.Persistence.Sql.Tests/MySql/Compatibility/MySqlCompatibilitySpecConfig.cs

@@ -0,0 +1,90 @@
+// -----------------------------------------------------------------------
+//  <copyright file="PostgreSqlCompatibilitySpecConfig.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using Akka.Persistence.MySql.Journal;
+using Akka.Persistence.MySql.Snapshot;
+using Akka.Persistence.Sql.Journal;
+using Akka.Persistence.Sql.Snapshot;
+using Akka.Persistence.Sql.Tests.Common.Containers;
+
+namespace Akka.Persistence.Sql.Tests.MySql.Compatibility
+{
+    public static class MySqlCompatibilitySpecConfig
+    {
+        public static Configuration.Config InitSnapshotConfig(
+            MySqlContainer fixture,
+            string tableName)
+            => $@"
+                akka.persistence {{
+                    publish-plugin-commands = on
+                    snapshot-store {{
+		                mysql {{
+			                class = ""{typeof(MySqlSnapshotStore).AssemblyQualifiedName}""
+			                plugin-dispatcher = ""akka.actor.default-dispatcher""
+			                connection-string = ""{fixture.ConnectionString}""
+			                connection-timeout = 30s
+			                table-name = {tableName}
+			                auto-initialize = on
+			                sequential-access = on
+		                }}
+
+                        sql {{
+                            class = ""{typeof(SqlSnapshotStore).AssemblyQualifiedName}""
+                            plugin-dispatcher = ""akka.actor.default-dispatcher""
+                            connection-string = ""{fixture.ConnectionString}""
+                            provider-name = {fixture.ProviderName}
+                            table-mapping = mysql
+                            auto-initialize = true
+                            postgresql {{
+                                snapshot {{
+                                    table-name = ""{tableName}""
+                                }}
+                            }}
+                        }}
+                    }}
+                }}";
+
+        public static Configuration.Config InitJournalConfig(
+            MySqlContainer fixture,
+            string tableName,
+            string metadataTableName)
+            => $@"
+                akka.persistence {{
+                    publish-plugin-commands = on
+                    journal {{
+                        mysql {{
+                            class = ""{typeof(MySqlJournal).AssemblyQualifiedName}""
+                            plugin-dispatcher = ""akka.actor.default-dispatcher""
+                            connection-string = ""{fixture.ConnectionString}""
+                            connection-timeout = 30s
+                            table-name = ""{tableName}""
+                            metadata-table-name = ""{metadataTableName}""
+                            auto-initialize = on
+                        }}
+
+                        sql {{
+                            class = ""{typeof(SqlWriteJournal).AssemblyQualifiedName}""
+                            plugin-dispatcher = ""akka.persistence.dispatchers.default-plugin-dispatcher""
+                            connection-string = ""{fixture.ConnectionString}""
+                            provider-name = ""{fixture.ProviderName}""
+                            parallelism = 3
+                            table-mapping = mysql
+                            auto-initialize = true
+                            tag-write-mode = Csv
+                            delete-compatibility-mode = true
+                            postgresql {{
+                                journal {{
+                                    table-name = ""{tableName}""
+                                }}
+                                metadata {{
+                                    table-name = ""{metadataTableName}""
+                                }}
+                            }}
+                        }}
+                    }}
+                }}";
+    }
+}

src/Akka.Persistence.Sql.Tests/MySql/MySqlJournalSpec.cs

@@ -0,0 +1,60 @@
+// -----------------------------------------------------------------------
+//  <copyright file="PostgreSqlJournalSpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using System;
+using Akka.Configuration;
+using Akka.Persistence.Sql.Tests.Common.Containers;
+using Akka.Persistence.TCK.Journal;
+using FluentAssertions.Extensions;
+using Xunit;
+using Xunit.Abstractions;
+#if !DEBUG
+using Akka.Persistence.Sql.Tests.Common.Internal.Xunit;
+#endif
+
+namespace Akka.Persistence.Sql.Tests.MySql
+{
+#if !DEBUG
+    [SkipWindows]
+#endif
+    [Collection(nameof(MySqlPersistenceSpec))]
+    public class MySqlJournalSpec : JournalSpec
+    {
+        public MySqlJournalSpec(ITestOutputHelper output, MySqlContainer fixture)
+            : base(Configuration(fixture), nameof(MySqlJournalSpec), output)
+        {
+            Initialize();
+        }
+
+        protected override bool SupportsSerialization => false;
+
+        public static Configuration.Config Configuration(MySqlContainer fixture)
+        {
+            if (!fixture.InitializeDbAsync().Wait(10.Seconds()))
+                throw new Exception("Failed to clean up database in 10 seconds");
+
+            return ConfigurationFactory.ParseString(@$"
+akka.persistence {{
+    publish-plugin-commands = on
+    journal {{
+        plugin = ""akka.persistence.journal.sql""
+        sql {{
+            connection-string = ""{fixture.ConnectionString}""
+            provider-name = ""{fixture.ProviderName}""
+        }}
+    }}
+    snapshot-store {{
+        plugin = ""akka.persistence.snapshot-store.sql""
+        sql {{
+            connection-string = ""{fixture.ConnectionString}""
+            provider-name = ""{fixture.ProviderName}""
+        }}
+    }}
+}}")
+                .WithFallback(SqlPersistence.DefaultConfiguration);
+        }
+    }
+}

src/Akka.Persistence.Sql.Tests/MySql/MySqlPersistenceSpec.cs

@@ -0,0 +1,14 @@
+// -----------------------------------------------------------------------
+//  <copyright file="PostgreSqlPersistenceSpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using Akka.Persistence.Sql.Tests.Common.Containers;
+using Xunit;
+
+namespace Akka.Persistence.Sql.Tests.MySql
+{
+    [CollectionDefinition(nameof(MySqlPersistenceSpec), DisableParallelization = true)]
+    public sealed class MySqlPersistenceSpec : ICollectionFixture<MySqlContainer> { }
+}