简化observer的数据库配置方式

Author: uless

examples/Ray.Client/Program.cs

@@ -18,11 +18,12 @@ static int Main(string[] args)
 
         private static async Task<int> RunMainAsync()
         {
-            try
+
+            using (var client = await StartClientWithRetries())
             {
-                using (var client = await StartClientWithRetries())
+                while (true)
                 {
-                    while (true)
+                    try
                     {
                         Console.WriteLine("start");
                         var times = int.Parse(Console.ReadLine());
@@ -34,13 +35,12 @@ private static async Task<int> RunMainAsync()
                         await Task.Delay(200);
                         Console.WriteLine($"余额为{await client.GetGrain<IAccountRep>(1).GetBalance()}");
                     }
+                    catch (Exception e)
+                    {
+                        Console.WriteLine(e);
+                    }
                 }
             }
-            catch (Exception e)
-            {
-                Console.WriteLine(e);
-                return 1;
-            }
         }
 
         private static async Task<IClusterClient> StartClientWithRetries(int initializeAttemptsBeforeFailing = 5)

examples/Ray.Grain/Account/Account.cs

@@ -1,6 +1,6 @@
 using Orleans;
 using Orleans.Concurrency;
-using Ray.Core.Core.Observer;
+using Ray.Core;
 using Ray.Core.Event;
 using Ray.DistributedTransaction;
 using Ray.EventBus.RabbitMQ;

examples/Ray.Grain/Account/AccountDb.cs

@@ -4,12 +4,11 @@
 using Ray.Core.Event;
 using Ray.IGrains.Actors;
 using Ray.Grain.Events;
-using Ray.Core.Core.Observer;
 using Ray.Core;
 
 namespace Ray.Grain
 {
-    [Observer(DefaultObserverGroup.secondary, typeof(Account))]
+    [Observer(DefaultObserverGroup.secondary, "db", typeof(Account))]
     public sealed class AccountDb : DbGrain<Account, long>, IAccountDb
     {
         public AccountDb(ILogger<AccountDb> logger) : base(logger)

examples/Ray.Grain/Account/AccountFlow.cs

@@ -5,11 +5,10 @@
 using Ray.Core.Event;
 using Ray.IGrains.Actors;
 using Ray.Grain.Events;
-using Ray.Core.Core.Observer;
 
 namespace Ray.Grain
 {
-    [Observer(DefaultObserverGroup.primary, typeof(Account))]
+    [Observer(DefaultObserverGroup.primary,"flow", typeof(Account))]
     public sealed class AccountFlow : ConcurrentObserverGrain<Account, long>, IAccountFlow
     {
         public AccountFlow(ILogger<AccountFlow> logger) : base(logger)

examples/Ray.Grain/Account/AccountRep.cs

@@ -3,13 +3,12 @@
 using Orleans;
 using Ray.Core;
 using Ray.Core.Core.Grains;
-using Ray.Core.Core.Observer;
 using Ray.IGrains.Actors;
 using Ray.IGrains.States;
 
 namespace Ray.Grain
 {
-    [Observer(DefaultObserverGroup.primary, typeof(Account))]
+    [Observer(DefaultObserverGroup.primary, null, typeof(Account))]
     public sealed class AccountRep : TxShadowGrain<Account, long, AccountState>, IAccountRep
     {
         public AccountRep(ILogger<AccountRep> logger) : base(logger)

examples/Ray.Grain/SQLStorageConfig.cs

@@ -11,14 +11,14 @@ public static class SQLStorageConfig
         public static IServiceCollection PSQLConfigure(this IServiceCollection serviceCollection)
         {
             serviceCollection.AddSingleton<IConfigureBuilder<long, Account>>(new PSQLConfigureBuilder<long, Account>((provider, id, parameter) =>
-            new IntegerKeyOptions(provider, "core_event", "account")).Observe<AccountRep>().Observe<AccountDb>("db").Observe<AccountFlow>("observer"));
+            new IntegerKeyOptions(provider, "core_event", "account")).AutoRegistrationObserver());
 
             return serviceCollection;
         }
         public static IServiceCollection MySQLConfigure(this IServiceCollection serviceCollection)
         {
             serviceCollection.AddSingleton<IConfigureBuilder<long, Account>>(new MySQLConfigureBuilder<long, Account>((provider, id, parameter) =>
-            new IntegerKeyOptions(provider, "core_event", "account")).Observe<AccountRep>().Observe<AccountDb>("db").Observe<AccountFlow>("observer"));
+            new IntegerKeyOptions(provider, "core_event", "account")).AutoRegistrationObserver());
 
             return serviceCollection;
         }

src/Ray.Core/Core/Observer/Attributes/ObservableAttribute.cs

@@ -1,6 +1,6 @@
 using System;
 
-namespace Ray.Core.Core.Observer
+namespace Ray.Core
 {
     /// <summary>
     /// 标记为可观察

src/Ray.Core/Core/Observer/Attributes/ObserverAttribute.cs

@@ -1,20 +1,29 @@
 using System;
 
-namespace Ray.Core.Core.Observer
+namespace Ray.Core
 {
     /// <summary>
     /// 标记为观察者
     /// </summary>
     [AttributeUsage(AttributeTargets.Class, AllowMultiple = false)]
     public class ObserverAttribute : Attribute
     {
-        public ObserverAttribute(string group, Type observable, Type observer = default)
+        /// <summary>
+        /// 事件监听者标记
+        /// </summary>
+        /// <param name="group">监听者分组</param>
+        /// <param name="name">监听者名称(如果是shadow请设置为null)</param>
+        /// <param name="observable">被监听的Type</param>
+        /// <param name="observer">监听者的Type</param>
+        public ObserverAttribute(string group, string name, Type observable, Type observer = default)
         {
             Group = group;
+            Name = name;
             Observable = observable;
             Observer = observer;
         }
         public string Group { get; set; }
+        public string Name { get; set; }
         public Type Observable { get; set; }
         public Type Observer { get; set; }
     }

src/Ray.Core/Core/Observer/ObserverUnitContainer.cs

@@ -4,7 +4,6 @@
 using System.Collections.Concurrent;
 using Ray.Core.Abstractions;
 using Ray.Core.Exceptions;
-using Ray.Core.Core.Observer;
 using Orleans;
 
 namespace Ray.Core
@@ -94,11 +93,11 @@ public object GetUnit(Type grainType)
                 throw new UnfindObserverUnitException(grainType.FullName);
         }
 
-        public void Register(IGrainID followUnit)
+        public void Register(IGrainID observerUnit)
         {
-            if (!unitDict.TryAdd(followUnit.GrainType, followUnit))
+            if (!unitDict.TryAdd(observerUnit.GrainType, observerUnit))
             {
-                throw new ObserverUnitRepeatedException(followUnit.GrainType.FullName);
+                throw new ObserverUnitRepeatedException(observerUnit.GrainType.FullName);
             }
         }
     }

src/Ray.Core/Storage/Configuration/ConfigureBuilder.cs

@@ -27,7 +27,11 @@ public ConfigureBuilder(
 
         protected void Observe<Follow>(Func<IServiceProvider, PrimaryKey, Parameter, FollowConfig> generator)
         {
-            followConfigGeneratorDict.Add(typeof(Follow), generator);
+            Observe(typeof(Follow), generator);
+        }
+        protected void Observe(Type type, Func<IServiceProvider, PrimaryKey, Parameter, FollowConfig> generator)
+        {
+            followConfigGeneratorDict.Add(type, generator);
         }
         readonly SemaphoreSlim seamphore = new SemaphoreSlim(1, 1);
         public async ValueTask<IStorageOptions> GetConfig(IServiceProvider serviceProvider, PrimaryKey primaryKey)