add mysql project unit tests.

test/DotNetCore.CAP.MySql.Test/ConnectionUtil.cs

+using System;
+using MySql.Data.MySqlClient;
+
+namespace DotNetCore.CAP.MySql.Test
+{
+    public static class ConnectionUtil
+    {
+        private const string DatabaseVariable = "Cap_MySql_DatabaseName";
+        private const string ConnectionStringTemplateVariable = "Cap_MySql_ConnectionStringTemplate";
+
+        private const string MasterDatabaseName = "information_schema";
+        private const string DefaultDatabaseName = @"DotNetCore.CAP.MySql.Test";
+
+        private const string DefaultConnectionStringTemplate =
+            @"Server=localhost;Database={0};Uid=root;Pwd=123123;";
+
+        public static string GetDatabaseName()
+        {
+            return Environment.GetEnvironmentVariable(DatabaseVariable) ?? DefaultDatabaseName;
+        }
+
+        public static string GetMasterConnectionString()
+        {
+            return string.Format(GetConnectionStringTemplate(), MasterDatabaseName);
+        }
+
+        public static string GetConnectionString()
+        {
+            return string.Format(GetConnectionStringTemplate(), GetDatabaseName());
+        }
+
+        private static string GetConnectionStringTemplate()
+        {
+            return
+                Environment.GetEnvironmentVariable(ConnectionStringTemplateVariable) ??
+                DefaultConnectionStringTemplate;
+        }
+
+        public static MySqlConnection CreateConnection(string connectionString = null)
+        {
+            connectionString = connectionString ?? GetConnectionString();
+            var connection = new MySqlConnection(connectionString);
+            connection.Open();
+            return connection;
+        }
+    }
+}
\ No newline at end of file

test/DotNetCore.CAP.MySql.Test/DatabaseTestHost.cs

+using System.Data;
+using System.Threading;
+using Dapper;
+using Microsoft.EntityFrameworkCore;
+
+namespace DotNetCore.CAP.MySql.Test
+{
+    public abstract class DatabaseTestHost : TestHost
+    {
+        private static bool _sqlObjectInstalled;
+        public static object _lock = new object();
+
+        protected override void PostBuildServices()
+        {
+            base.PostBuildServices();
+            lock (_lock)
+            {
+                if (!_sqlObjectInstalled)
+                {
+                    InitializeDatabase();
+                }
+            }
+        }
+
+        public override void Dispose()
+        {
+            DeleteAllData();
+            base.Dispose();
+        }
+
+        private void InitializeDatabase()
+        {
+            using (CreateScope())
+            {
+                var storage = GetService<MySqlStorage>();
+                var token = new CancellationTokenSource().Token;
+                CreateDatabase();
+                storage.InitializeAsync(token).Wait();
+                _sqlObjectInstalled = true;
+            }
+        }
+
+        private void CreateDatabase()
+        {
+            var masterConn = ConnectionUtil.GetMasterConnectionString();
+            var databaseName = ConnectionUtil.GetDatabaseName();
+            using (var connection = ConnectionUtil.CreateConnection(masterConn))
+            {
+                connection.Execute($@"
+DROP DATABASE IF EXISTS `{databaseName}`;
+CREATE DATABASE `{databaseName}`;");
+            }
+        }
+
+        private void DeleteAllData()
+        {
+            var conn = ConnectionUtil.GetConnectionString();
+
+            using (var connection = ConnectionUtil.CreateConnection(conn))
+            {
+                connection.Execute($@"
+TRUNCATE TABLE `cap.published`;
+TRUNCATE TABLE `cap.received`;
+TRUNCATE TABLE `cap.queue`;");
+            }
+        }
+    }
+}
\ No newline at end of file

test/DotNetCore.CAP.MySql.Test/DotNetCore.CAP.MySql.Test.csproj

@@ -25,6 +25,7 @@
   <ItemGroup>
     <PackageReference Include="Dapper" Version="1.50.2" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.0.0" />
+    <PackageReference Include="MySqlConnector" Version="0.24.0" />
     <PackageReference Include="xunit.runner.visualstudio" Version="2.2.0" />
     <PackageReference Include="xunit" Version="2.2.0" />
     <PackageReference Include="Microsoft.AspNetCore.Http" Version="1.1.2" />
@@ -37,5 +38,9 @@
     <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="1.1.2" />
     <PackageReference Include="Microsoft.Extensions.Configuration.EnvironmentVariables" Version="1.1.2" />
   </ItemGroup>
+  
+  <ItemGroup>
+    <Service Include="{82a7f48d-3b50-4b1e-b82e-3ada8210c358}" />
+  </ItemGroup>
 
 </Project>
\ No newline at end of file

test/DotNetCore.CAP.MySql.Test/MySqlStorageConnectionTest.cs

+//using System;
+//using System.Threading.Tasks;
+//using Dapper;
+//using DotNetCore.CAP.Infrastructure;
+//using DotNetCore.CAP.Models;
+//using Xunit;
+
+//namespace DotNetCore.CAP.MySql.Test
+//{
+//    [Collection("MySql")]
+//    public class MySqlStorageConnectionTest : DatabaseTestHost
+//    {
+//        private MySqlStorageConnection _storage;
+
+//        public MySqlStorageConnectionTest()
+//        {
+//            var options = GetService<MySqlOptions>();
+//            _storage = new MySqlStorageConnection(options);
+//        }
+
+//        [Fact]
+//        public async Task GetPublishedMessageAsync_Test()
+//        {
+//            var sql = "INSERT INTO [Cap].[Published]([Name],[Content],[Retries],[Added],[ExpiresAt],[StatusName]) OUTPUT INSERTED.Id VALUES(@Name,@Content,@Retries,@Added,@ExpiresAt,@StatusName);";
+//            var publishMessage = new CapPublishedMessage
+//            {
+//                Name = "MySqlStorageConnectionTest",
+//                Content = "",
+//                StatusName = StatusName.Scheduled
+//            };
+//            var insertedId = default(int);
+//            using (var connection = ConnectionUtil.CreateConnection())
+//            {
+//                insertedId = connection.QueryFirst<int>(sql, publishMessage);
+//            }
+//            var message = await _storage.GetPublishedMessageAsync(insertedId);
+//            Assert.NotNull(message);
+//            Assert.Equal("MySqlStorageConnectionTest", message.Name);
+//            Assert.Equal(StatusName.Scheduled, message.StatusName);
+//        }
+
+//        [Fact]
+//        public async Task FetchNextMessageAsync_Test()
+//        {
+//            var sql = "INSERT INTO [Cap].[Queue]([MessageId],[MessageType]) VALUES(@MessageId,@MessageType);";
+//            var queue = new CapQueue
+//            {
+//                MessageId = 3333,
+//                MessageType = MessageType.Publish
+//            };
+//            using (var connection = ConnectionUtil.CreateConnection())
+//            {
+//                connection.Execute(sql, queue);
+//            }
+//            var fetchedMessage = await _storage.FetchNextMessageAsync();
+//            fetchedMessage.Dispose();
+//            Assert.NotNull(fetchedMessage);
+//            Assert.Equal(MessageType.Publish, fetchedMessage.MessageType);
+//            Assert.Equal(3333, fetchedMessage.MessageId);
+//        }
+
+//        [Fact]
+//        public async Task StoreReceivedMessageAsync_Test()
+//        {
+//            var receivedMessage = new CapReceivedMessage
+//            {
+//                Name = "MySqlStorageConnectionTest",
+//                Content = "",
+//                Group = "mygroup",
+//                StatusName = StatusName.Scheduled
+//            };
+
+//            Exception exception = null;
+//            try
+//            {
+//                await _storage.StoreReceivedMessageAsync(receivedMessage);
+//            }
+//            catch (Exception ex)
+//            {
+//                exception = ex;
+//            }
+//            Assert.Null(exception);
+//        }
+
+//        [Fact]
+//        public async Task GetReceivedMessageAsync_Test()
+//        {
+
+//            var sql = $@"
+//        INSERT INTO [Cap].[Received]([Name],[Group],[Content],[Retries],[Added],[ExpiresAt],[StatusName]) OUTPUT INSERTED.Id
+//        VALUES(@Name,@Group,@Content,@Retries,@Added,@ExpiresAt,@StatusName);";
+//            var receivedMessage = new CapReceivedMessage
+//            {
+//                Name = "MySqlStorageConnectionTest",
+//                Content = "",
+//                Group = "mygroup",
+//                StatusName = StatusName.Scheduled
+//            };
+//            var insertedId = default(int);
+//            using (var connection = ConnectionUtil.CreateConnection())
+//            {
+//                insertedId = connection.QueryFirst<int>(sql, receivedMessage);
+//            }
+
+//            var message = await _storage.GetReceivedMessageAsync(insertedId);
+
+//            Assert.NotNull(message);
+//            Assert.Equal(StatusName.Scheduled, message.StatusName);
+//            Assert.Equal("MySqlStorageConnectionTest", message.Name);
+//            Assert.Equal("mygroup", message.Group);
+//        }
+
+//        [Fact]
+//        public async Task GetNextReceviedMessageToBeEnqueuedAsync_Test()
+//        {
+//            var receivedMessage = new CapReceivedMessage
+//            {
+//                Name = "MySqlStorageConnectionTest",
+//                Content = "",
+//                Group = "mygroup",
+//                StatusName = StatusName.Scheduled
+//            };
+//            await _storage.StoreReceivedMessageAsync(receivedMessage);
+
+//            var message = await _storage.GetNextReceviedMessageToBeEnqueuedAsync();
+
+//            Assert.NotNull(message);
+//            Assert.Equal(StatusName.Scheduled, message.StatusName);
+//            Assert.Equal("MySqlStorageConnectionTest", message.Name);
+//            Assert.Equal("mygroup", message.Group);
+//        }
+
+//    }
+//}

test/DotNetCore.CAP.MySql.Test/MySqlStorageTest.cs

+using Xunit;
+using Dapper;
+
+namespace DotNetCore.CAP.MySql.Test
+{
+    [Collection("MySql")]
+    public class MySqlStorageTest : DatabaseTestHost
+    {
+        private readonly string _dbName;
+        private readonly string _masterDbConnectionString;
+
+
+        public MySqlStorageTest()
+        {
+            _dbName = ConnectionUtil.GetDatabaseName();
+            _masterDbConnectionString = ConnectionUtil.GetMasterConnectionString();
+        }
+
+        [Fact]
+        public void Database_IsExists()
+        {
+            using (var connection = ConnectionUtil.CreateConnection(_masterDbConnectionString))
+            {
+                var databaseName = ConnectionUtil.GetDatabaseName();
+                var sql = $@"SELECT SCHEMA_NAME FROM SCHEMATA WHERE SCHEMA_NAME = '{databaseName}'";
+                var result = connection.QueryFirstOrDefault<string>(sql);
+                Assert.NotNull(result);
+                Assert.True(databaseName.Equals(result, System.StringComparison.CurrentCultureIgnoreCase));
+            }
+        }
+
+        [Fact]
+        public void DatabaseTable_Published_IsExists()
+        {
+            var tableName = "cap.published";
+            using (var connection = ConnectionUtil.CreateConnection(_masterDbConnectionString))
+            {
+                var sql = $"SELECT TABLE_NAME FROM `TABLES` WHERE TABLE_SCHEMA='{_dbName}' AND TABLE_NAME = '{tableName}'";
+                var result = connection.QueryFirstOrDefault<string>(sql);
+                Assert.NotNull(result);
+                Assert.Equal(tableName, result);
+            }
+        }
+
+        [Fact]
+        public void DatabaseTable_Queue_IsExists()
+        {
+            var tableName = "cap.queue";
+            using (var connection = ConnectionUtil.CreateConnection(_masterDbConnectionString))
+            {
+                var sql = $"SELECT TABLE_NAME FROM `TABLES` WHERE TABLE_SCHEMA='{_dbName}' AND TABLE_NAME = '{tableName}'";
+                var result = connection.QueryFirstOrDefault<string>(sql);
+                Assert.NotNull(result);
+                Assert.Equal(tableName, result);
+            }
+        }
+
+        [Fact]
+        public void DatabaseTable_Received_IsExists()
+        {
+            var tableName = "cap.received";
+            using (var connection = ConnectionUtil.CreateConnection(_masterDbConnectionString))
+            {
+                var sql = $"SELECT TABLE_NAME FROM `TABLES` WHERE TABLE_SCHEMA='{_dbName}' AND TABLE_NAME = '{tableName}'";
+                var result = connection.QueryFirstOrDefault<string>(sql);
+                Assert.NotNull(result);
+                Assert.Equal(tableName, result);
+            }
+        }
+    }
+}

test/DotNetCore.CAP.MySql.Test/TestHost.cs

+using System;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace DotNetCore.CAP.MySql.Test
+{
+    public abstract class TestHost : IDisposable
+    {
+        protected IServiceCollection _services;
+        protected string _connectionString;
+        private IServiceProvider _provider;
+        private IServiceProvider _scopedProvider;
+
+        public TestHost()
+        {
+            CreateServiceCollection();
+            PreBuildServices();
+            BuildServices();
+            PostBuildServices();
+        }
+
+        protected IServiceProvider Provider => _scopedProvider ?? _provider;
+
+        private void CreateServiceCollection()
+        {
+            var services = new ServiceCollection();
+
+            services.AddOptions();
+            services.AddLogging();
+
+            _connectionString = ConnectionUtil.GetConnectionString();
+            services.AddSingleton(new MySqlOptions { ConnectionString = _connectionString });
+            services.AddSingleton<MySqlStorage>();
+
+            _services = services;
+        }
+
+        protected virtual void PreBuildServices()
+        {
+        }
+
+        private void BuildServices()
+        {
+            _provider = _services.BuildServiceProvider();
+        }
+
+        protected virtual void PostBuildServices()
+        {
+        }
+
+        public IDisposable CreateScope()
+        {
+            var scope = CreateScope(_provider);
+            var loc = scope.ServiceProvider;
+            _scopedProvider = loc;
+            return new DelegateDisposable(() =>
+            {
+                if (_scopedProvider == loc)
+                {
+                    _scopedProvider = null;
+                }
+                scope.Dispose();
+            });
+        }
+
+        public IServiceScope CreateScope(IServiceProvider provider)
+        {
+            var scope = provider.GetService<IServiceScopeFactory>().CreateScope();
+            return scope;
+        }
+
+        public T GetService<T>() => Provider.GetService<T>();
+
+        public T Ensure<T>(ref T service)
+            where T : class
+            => service ?? (service = GetService<T>());
+
+        public virtual void Dispose()
+        {
+            (_provider as IDisposable)?.Dispose();
+        }
+
+        private class DelegateDisposable : IDisposable
+        {
+            private Action _dispose;
+
+            public DelegateDisposable(Action dispose)
+            {
+                _dispose = dispose;
+            }
+
+            public void Dispose()
+            {
+                _dispose();
+            }
+        }
+    }
+}
\ No newline at end of file