Add support for SharedFile service registration without the need to run third-party middleware

Author: fanliang11

src/Surging.Core/Surging.Core.CPlatform/Routing/Implementation/SharedFileModule.cs

@@ -0,0 +1,115 @@
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Surging.Core.CPlatform.Cache;
+using Surging.Core.CPlatform.Module;
+using Surging.Core.CPlatform.Mqtt;
+using Surging.Core.CPlatform.Runtime.Client.HealthChecks.Implementation;
+using Surging.Core.CPlatform.Runtime.Client.HealthChecks;
+using Surging.Core.CPlatform.Runtime.Client;
+using Surging.Core.CPlatform.Runtime.Server;
+using Surging.Core.CPlatform.Serialization;
+using Surging.Core.CPlatform.Support;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Surging.Core.CPlatform.Routing.Implementation
+{
+    public class SharedFileModule : EnginePartModule
+    {
+        public override void Initialize(AppModuleContext context)
+        {
+            base.Initialize(context);
+        }
+
+        /// <summary>
+        /// Inject dependent third-party components
+        /// </summary>
+        /// <param name="builder"></param>
+        protected override void RegisterBuilder(ContainerBuilderWrapper builder)
+        {
+            
+            base.RegisterBuilder(builder);
+            var rootpath = AppConfig.ServerOptions.RootPath;
+            if (string.IsNullOrEmpty(rootpath))
+                rootpath = AppContext.BaseDirectory;
+            UseSharedFileRouteManager(builder,System.IO.Path.Combine(rootpath, "reg", $"reg_route.jreg"))
+                              .UseSharedFileCommandManager(builder, System.IO.Path.Combine(rootpath, "reg", $"reg_command.jreg"))
+                              .UseSharedFileCacheManager(builder, System.IO.Path.Combine(rootpath, "reg", $"reg_cahce.jreg"))
+                              .UseSharedFileMqttRouteManager(builder, System.IO.Path.Combine(rootpath, "reg", $"reg_mqtt.jreg"));
+        }
+
+        public SharedFileModule UseSharedFileRouteManager(ContainerBuilderWrapper builder, string filePath)
+        {
+              UseRouteManager(builder, provider =>
+            new SharedFileServiceRouteManager(
+                filePath,
+                provider.GetRequiredService<ISerializer<string>>(),
+                provider.GetRequiredService<IServiceRouteFactory>(),
+                provider.GetRequiredService<ILogger<SharedFileServiceRouteManager>>()));
+            return this;
+        }
+
+        public SharedFileModule UseSharedFileCommandManager(ContainerBuilderWrapper builder, string filePath)
+        {
+              UseCommandManager(builder,provider =>
+            new SharedFileServiceCommandManager(
+                filePath,
+                provider.GetRequiredService<ISerializer<string>>(),
+                provider.GetRequiredService<IServiceEntryManager>(),
+                provider.GetRequiredService<ILogger<SharedFileServiceCommandManager>>()));
+            return this;
+        }
+
+        public SharedFileModule UseSharedFileCacheManager(ContainerBuilderWrapper builder, string filePath)
+        {
+              UseCacheManager(builder,provider =>
+            new SharedFileServiceCacheManager(
+                filePath,
+                provider.GetRequiredService<ISerializer<string>>(),
+                provider.GetRequiredService<IServiceCacheFactory>(),
+                provider.GetRequiredService<ILogger<SharedFileServiceCacheManager>>()));
+            return this;
+        }
+
+        public SharedFileModule UseSharedFileMqttRouteManager(ContainerBuilderWrapper builder, string filePath)
+        {
+              UseMqttRouteManager(builder,provider =>
+            new SharedFileMqttServiceRouteManager(
+                filePath,
+                provider.GetRequiredService<ISerializer<string>>(),
+                provider.GetRequiredService<IMqttServiceFactory>(),
+                provider.GetRequiredService<ILogger<SharedFileMqttServiceRouteManager>>()));
+            return this;
+        }
+
+        public ContainerBuilderWrapper UseCommandManager(ContainerBuilderWrapper builder, Func<IServiceProvider, IServiceCommandManager> factory)
+        {
+            builder.RegisterAdapter(factory).InstancePerLifetimeScope();
+            return builder;
+        }
+
+        public ContainerBuilderWrapper UseCacheManager(ContainerBuilderWrapper builder, Func<IServiceProvider, IServiceCacheManager> factory)
+        {
+            builder.RegisterAdapter(factory).InstancePerLifetimeScope();
+            return builder;
+        }
+
+        public ContainerBuilderWrapper UseRouteManager(ContainerBuilderWrapper builder, Func<IServiceProvider, IServiceRouteManager> factory)
+        {
+            builder.RegisterAdapter(factory).InstancePerLifetimeScope();
+            return builder;
+        }
+
+        public ContainerBuilderWrapper UseMqttRouteManager(ContainerBuilderWrapper builder, Func<IServiceProvider, IMqttServiceRouteManager> factory)
+        {
+            builder.RegisterAdapter(factory).InstancePerLifetimeScope();
+            return builder;
+        }
+
+    }
+}
+

src/Surging.Core/Surging.Core.CPlatform/Routing/Implementation/SharedFileMqttServiceRouteManager.cs

@@ -0,0 +1,278 @@
+using Microsoft.Extensions.Logging;
+using Surging.Core.CPlatform.Address;
+using Surging.Core.CPlatform.Mqtt;
+using Surging.Core.CPlatform.Mqtt.Implementation;
+using Surging.Core.CPlatform.Serialization;
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Surging.Core.CPlatform.Routing.Implementation
+{
+    public class SharedFileMqttServiceRouteManager : MqttServiceRouteManagerBase, IDisposable
+    {
+        #region Field
+
+        private readonly string _filePath;
+        private readonly ISerializer<string> _serializer;
+        private readonly IMqttServiceFactory _mqttServiceFactory;
+        private readonly ILogger<SharedFileMqttServiceRouteManager> _logger;
+        private MqttServiceRoute[] _routes;
+        private readonly FileSystemWatcher _fileSystemWatcher;
+
+        #endregion Field
+
+        #region Constructor
+
+        public SharedFileMqttServiceRouteManager(string filePath, ISerializer<string> serializer,
+            IMqttServiceFactory mqttServiceFactory, ILogger<SharedFileMqttServiceRouteManager> logger) : base(serializer)
+        {
+            _filePath = filePath;
+            _serializer = serializer;
+            _mqttServiceFactory = mqttServiceFactory;
+            _logger = logger;
+
+            var directoryName = Path.GetDirectoryName(filePath);
+            if (!Directory.Exists(directoryName))
+                Directory.CreateDirectory(directoryName);
+            if (!File.Exists(filePath)) File.Create(filePath).Close();
+            _fileSystemWatcher = new FileSystemWatcher(directoryName, Path.GetFileName(filePath));
+
+            _fileSystemWatcher.Changed += _fileSystemWatcher_Changed;
+            _fileSystemWatcher.Created += _fileSystemWatcher_Changed;
+            _fileSystemWatcher.Deleted += _fileSystemWatcher_Changed;
+            _fileSystemWatcher.Renamed += _fileSystemWatcher_Changed;
+            _fileSystemWatcher.IncludeSubdirectories = false;
+            _fileSystemWatcher.EnableRaisingEvents = true;
+        }
+
+        #endregion Constructor
+
+        #region Implementation of IDisposable
+
+        /// <summary>
+        /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
+        /// </summary>
+        public void Dispose()
+        {
+            _fileSystemWatcher?.Dispose();
+        }
+
+        #endregion Implementation of IDisposable
+
+        #region Overrides of ServiceRouteManagerBase
+
+        /// <summary>
+        ///     获取所有可用的服务路由信息。
+        /// </summary>
+        /// <returns>服务路由集合。</returns>
+        public override async Task<IEnumerable<MqttServiceRoute>> GetRoutesAsync()
+        {
+            if (_routes == null)
+                await EntryRoutes(_filePath);
+            return _routes;
+        }
+
+   
+        /// <summary>
+        ///     清空所有的服务路由。
+        /// </summary>
+        /// <returns>一个任务。</returns>
+        public override Task ClearAsync()
+        {
+            if (File.Exists(_filePath))
+                File.Delete(_filePath);
+            return Task.FromResult(0);
+        }
+
+        /// <summary>
+        ///     设置服务路由。
+        /// </summary>
+        /// <param name="routes">服务路由集合。</param>
+        /// <returns>一个任务。</returns>
+        protected override async Task SetRoutesAsync(IEnumerable<MqttServiceDescriptor> routes)
+        {
+            using (var fileStream = new FileStream(_filePath, FileMode.OpenOrCreate, FileAccess.Write, FileShare.None))
+            {
+                fileStream.SetLength(0);
+                using (var writer = new StreamWriter(fileStream, Encoding.UTF8))
+                {
+                    await writer.WriteAsync(_serializer.Serialize(routes));
+                }
+            }
+        }
+
+        public override async Task RemveAddressAsync(IEnumerable<AddressModel> Address)
+        {
+            var routes = await GetRoutesAsync();
+            foreach (var route in routes)
+            {
+                route.MqttEndpoint = route.MqttEndpoint.Except(Address);
+            }
+            await base.SetRoutesAsync(routes);
+        }
+
+        #endregion Overrides of ServiceRouteManagerBase
+
+        #region Private Method
+
+        private async Task<IEnumerable<MqttServiceRoute>> GetRoutes(string file)
+        {
+            MqttServiceRoute[] routes;
+            if (File.Exists(file))
+            {
+                if (_logger.IsEnabled(LogLevel.Debug))
+                    _logger.LogDebug($"准备从文件：{file}中获取服务路由。");
+                string content;
+                while (true)
+                {
+                    try
+                    {
+                        using (
+                            var fileStream = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
+                        {
+                            var reader = new StreamReader(fileStream, Encoding.UTF8);
+                            content = await reader.ReadToEndAsync();
+                        }
+                        break;
+                    }
+                    catch (IOException)
+                    {
+                    }
+                }
+                try
+                {
+                    var serializer = _serializer;
+                    routes =
+                    (await
+                        _mqttServiceFactory.CreateMqttServiceRoutesAsync(
+                            serializer.Deserialize<string, MqttServiceDescriptor[]>(content))).ToArray();
+                    if (_logger.IsEnabled(LogLevel.Information))
+                        _logger.LogInformation(
+                            $"成功获取到以下路由信息：{string.Join(",", routes.Select(i => i.MqttDescriptor.Topic))}。");
+                }
+                catch (Exception exception)
+                {
+                    if (_logger.IsEnabled(LogLevel.Error))
+                        _logger.LogError(exception, "获取路由信息时发生了错误。");
+                    routes = new MqttServiceRoute[0];
+                }
+            }
+            else
+            {
+                if (_logger.IsEnabled(LogLevel.Warning))
+                    _logger.LogWarning($"无法获取路由信息，因为文件：{file}不存在。");
+                routes = new MqttServiceRoute[0];
+            }
+            return routes;
+        }
+
+        private async Task EntryRoutes(string file)
+        {
+            var oldRoutes = _routes?.ToArray();
+            var newRoutes = (await GetRoutes(file)).ToArray();
+            _routes = newRoutes;
+            if (oldRoutes == null)
+            {
+                //触发服务路由创建事件。
+                OnCreated(newRoutes.Select(route => new MqttServiceRouteEventArgs(route)).ToArray());
+            }
+            else
+            {
+                //旧的服务Id集合。
+                var oldServiceIds = oldRoutes.Select(i => i.MqttDescriptor.Topic).ToArray();
+                //新的服务Id集合。
+                var newServiceIds = newRoutes.Select(i => i.MqttDescriptor.Topic).ToArray();
+
+                //被删除的服务Id集合
+                var removeServiceIds = oldServiceIds.Except(newServiceIds).ToArray();
+                //新增的服务Id集合。
+                var addServiceIds = newServiceIds.Except(oldServiceIds).ToArray();
+                //可能被修改的服务Id集合。
+                var mayModifyServiceIds = newServiceIds.Except(removeServiceIds).ToArray();
+
+                //触发服务路由创建事件。
+                OnCreated(
+                    newRoutes.Where(i => addServiceIds.Contains(i.MqttDescriptor.Topic))
+                        .Select(route => new MqttServiceRouteEventArgs(route))
+                        .ToArray());
+
+                //触发服务路由删除事件。
+                OnRemoved(
+                    oldRoutes.Where(i => removeServiceIds.Contains(i.MqttDescriptor.Topic))
+                        .Select(route => new MqttServiceRouteEventArgs(route))
+                        .ToArray());
+
+                //触发服务路由变更事件。
+                var currentMayModifyRoutes =
+                    newRoutes.Where(i => mayModifyServiceIds.Contains(i.MqttDescriptor.Topic)).ToArray();
+                var oldMayModifyRoutes =
+                    oldRoutes.Where(i => mayModifyServiceIds.Contains(i.MqttDescriptor.Topic)).ToArray();
+
+                foreach (var oldMayModifyRoute in oldMayModifyRoutes)
+                {
+                    if (!currentMayModifyRoutes.Contains(oldMayModifyRoute))
+                        OnChanged(
+                            new MqttServiceRouteChangedEventArgs(
+                                currentMayModifyRoutes.First(
+                                    i => i.MqttDescriptor.Topic == oldMayModifyRoute.MqttDescriptor.Topic),
+                                oldMayModifyRoute));
+                }
+            }
+        }
+
+        private async void _fileSystemWatcher_Changed(object sender, FileSystemEventArgs e)
+        {
+            if (_logger.IsEnabled(LogLevel.Information))
+                _logger.LogInformation($"文件{_filePath}发生了变更，将重新获取路由信息。");
+
+            if (e.ChangeType == WatcherChangeTypes.Changed)
+            {
+                string content;
+                try
+                {
+                    content = File.ReadAllText(_filePath, Encoding.UTF8);
+                }
+                catch (IOException) //还没有操作完，忽略本次修改
+                {
+                    return;
+                }
+                if (!string.IsNullOrWhiteSpace(content))
+                {
+                    await EntryRoutes(_filePath);
+                }
+                else
+                {
+                    return;
+                }
+            }
+
+            await EntryRoutes(_filePath);
+        }
+
+        public override async Task RemoveByTopicAsync(string topic, IEnumerable<AddressModel> endpoint)
+        {
+            var routes = await GetRoutesAsync();
+            try
+            {
+                var route = routes.Where(p => p.MqttDescriptor.Topic == topic).SingleOrDefault();
+                if (route != null)
+                {
+                    route.MqttEndpoint = route.MqttEndpoint.Except(endpoint);
+                    await base.SetRoutesAsync(new MqttServiceRoute[] { route });
+                }
+            }
+            catch (Exception ex)
+            {
+                throw ex;
+            }
+        }
+
+
+
+        #endregion Private Method
+    }
+}