Merge branch 'dev'

ZeeLyn · Apr 29, 2019 · f204c29 · f204c29
2 parents 279fa38 + f16b2c3
commit f204c29
Show file tree

Hide file tree

Showing 23 changed files with 76 additions and 76 deletions.
diff --git a/build.cake b/build.cake
@@ -5,7 +5,7 @@
 ///////////////////////////////////////////////////////////////////////////////
 
 var output=Argument<string>("output", "Output");
-var version=Argument<string>("version", "0.0.3");
+var version=Argument<string>("version", "0.0.4");
 var target = Argument<string>("target", "Default");
 var release = Argument<bool>("release", true);
 var nugetApiKey = Argument<string>("nugetApiKey", null);

diff --git a/samples/Sample.Common/CircuitBreakerEvent.cs b/samples/Sample.Common/CircuitBreakerEvent.cs
@@ -16,37 +16,44 @@ public CircuitBreakerEvent(ILogger<CircuitBreakerEvent> logger)
         }
         public async Task OnFallback(string route, MethodInfo methodInfo)
         {
-            Logger.LogTrace("Raise OnFallback");
+            Logger.LogWarning("Raise OnFallback");
+            await Task.CompletedTask;
         }
 
         public async Task OnBreak(string route, MethodInfo methodInfo, Exception exception, TimeSpan time)
         {
-            Logger.LogTrace($"Raise OnBreak;{exception.Message}");
+            Logger.LogError($"Raise OnBreak;{exception.Message}");
+            await Task.CompletedTask;
         }
 
         public async Task OnRest(string route, MethodInfo methodInfo)
         {
-            Logger.LogTrace("Raise OnRest");
+            Logger.LogWarning("Raise OnRest");
+            await Task.CompletedTask;
         }
 
         public async Task OnHalfOpen(string route, MethodInfo methodInfo)
         {
-            Logger.LogTrace("Raise OnHalfOpen");
+            Logger.LogWarning("Raise OnHalfOpen");
+            await Task.CompletedTask;
         }
 
         public async Task OnTimeOut(string route, MethodInfo methodInfo, Exception exception)
         {
-            Logger.LogTrace($"Raise OnTimeOut;{exception.Message}");
+            Logger.LogWarning($"Raise OnTimeOut;{exception.Message}");
+            await Task.CompletedTask;
         }
 
         public async Task OnRetry(string route, MethodInfo methodInfo, Exception exception, int retryTimes)
         {
-            Logger.LogTrace($"Raise OnRetry;{exception.Message};{retryTimes}");
+            Logger.LogWarning($"Raise OnRetry;{exception.Message};{retryTimes}");
+            await Task.CompletedTask;
         }
 
         public async Task OnBulkheadRejected(string route, MethodInfo methodInfo)
         {
-            Logger.LogTrace("Raise OnBulkheadRejected;");
+            Logger.LogWarning("Raise OnBulkheadRejected;");
+            await Task.CompletedTask;
         }
     }
 }
diff --git a/samples/Sample.Server/HelloService.cs b/samples/Sample.Server/HelloService.cs
@@ -15,7 +15,7 @@ public class HelloService : IHelloService
         [ServerMethodInterceptor]
         public async Task<ResultModel> SayHello(string name)
         {
-            // await Task.Delay(2000);
+            //await Task.Delay(5000);
             return await Task.FromResult(new ResultModel { Message = name });
         }
 

diff --git a/samples/Sample.Server/Startup.cs b/samples/Sample.Server/Startup.cs
@@ -44,6 +44,7 @@ public void ConfigureServices(IServiceCollection services)
                  //builder.AddRedisPartitionCaching();
                  //builder.AddRedisCaching();
                  //builder.AddMemoryCaching();
+                 builder.AddCircuitBreaker<CircuitBreakerEvent>();
                  builder.AddOption(UraganoOptions.Remoting_Invoke_CancellationTokenSource_Timeout, TimeSpan.FromSeconds(60));
                  builder.AddOptions();
              });

diff --git a/samples/Sample.Server/appsettings.json b/samples/Sample.Server/appsettings.json
@@ -50,11 +50,11 @@
     },
     "CircuitBreaker": {
       "Polly": {
-        "timeout": 2000,
-        "retry": 3,
+        "timeout": 1000,
+        "retry": 1,
         "ExceptionsAllowedBeforeBreaking": 10,
         "DurationOfBreak": 60000,
-        "MaxParallelization": 0,
+        "MaxParallelization": 10,
         "MaxQueuingActions": 0
       }
     },

diff --git a/src/Uragano.Abstract/CircuitBreaker/ICircuitBreaker.cs b/src/Uragano.Abstract/CircuitBreaker/ICircuitBreaker.cs
@@ -5,8 +5,6 @@ namespace Uragano.Abstractions.CircuitBreaker
 {
     public interface ICircuitBreaker
     {
-        Task<object> ExecuteAsync(string route, Func<Task<object>> action, Type returnValueType);
-
-        Task ExecuteAsync(string route, Func<Task> action);
+        Task<IServiceResult> ExecuteAsync(string route, Func<Task<IServiceResult>> action, Type returnValueType);
     }
 }
diff --git a/src/Uragano.Abstract/IUraganoBuilder.cs b/src/Uragano.Abstract/IUraganoBuilder.cs
@@ -54,13 +54,15 @@ void AddServiceDiscovery(Type serviceDiscovery, IServiceDiscoveryClientConfigura
         void AddOptions(IConfigurationSection configuration);
 
 
-        void AddCircuitBreaker<TCircuitBreakerEvent>(int timeout = 3000, int retry = 3, int exceptionsAllowedBeforeBreaking = 10, int durationOfBreak = 60 * 1000) where TCircuitBreakerEvent : ICircuitBreakerEvent;
+        void AddCircuitBreaker<TCircuitBreakerEvent>(int timeout = 3000, int retry = 3, int exceptionsAllowedBeforeBreaking = 10, int durationOfBreak = 60 * 1000, int maxParallelization = 0, int maxQueuingActions = 0) where TCircuitBreakerEvent : ICircuitBreakerEvent;
 
         void AddCircuitBreaker(int timeout = 3000, int retry = 3, int exceptionsAllowedBeforeBreaking = 10,
-            int durationOfBreak = 60 * 1000);
+            int durationOfBreak = 60 * 1000, int maxParallelization = 0, int maxQueuingActions = 0);
 
         void AddCircuitBreaker(IConfigurationSection configurationSection);
 
+        void AddCircuitBreaker<TCircuitBreakerEvent>(IConfigurationSection configurationSection) where TCircuitBreakerEvent : ICircuitBreakerEvent;
+
         void AddCodec<TCodec>() where TCodec : ICodec;
 
         void AddCaching<TCaching>(ICachingOptions cachingOptions) where TCaching : class, ICaching;
@@ -90,5 +92,7 @@ public interface IUraganoSampleBuilder : IUraganoBuilder
         void AddOptions();
 
         void AddCircuitBreaker();
+
+        void AddCircuitBreaker<TCircuitBreakerEvent>() where TCircuitBreakerEvent : ICircuitBreakerEvent;
     }
 }
diff --git a/src/Uragano.Abstract/Uragano.Abstractions.csproj b/src/Uragano.Abstract/Uragano.Abstractions.csproj
@@ -4,7 +4,7 @@
     <TargetFramework>netstandard2.0</TargetFramework>
     <LangVersion>latest</LangVersion>
     <GeneratePackageOnBuild>false</GeneratePackageOnBuild>
-    <Version>0.0.3</Version>
+    <Version>0.0.4</Version>
     <Description>A simple, high performance RPC library.</Description>
     <PackageProjectUrl>https://github.com/1100100/Uragano</PackageProjectUrl>
     <PackageTags>Uragano,RPC,DotNetty,Microservice,MessagePack,DynamicProxy,dotnetcore,service-discovery,polly,circuit-breaker,consul,zookeeper,dependency-injection</PackageTags>

diff --git a/src/Uragano.Caching.Memory/Uragano.Caching.Memory.csproj b/src/Uragano.Caching.Memory/Uragano.Caching.Memory.csproj
@@ -4,7 +4,7 @@
     <TargetFramework>netstandard2.0</TargetFramework>
     <LangVersion>latest</LangVersion>
     <GeneratePackageOnBuild>false</GeneratePackageOnBuild>
-    <Version>0.0.3</Version>
+    <Version>0.0.4</Version>
     <Authors>Owen</Authors>
     <Company>Owen</Company>
     <Description>A simple, high performance RPC library.</Description>

diff --git a/src/Uragano.Caching.Redis/Uragano.Caching.Redis.csproj b/src/Uragano.Caching.Redis/Uragano.Caching.Redis.csproj
@@ -4,7 +4,7 @@
     <TargetFramework>netstandard2.0</TargetFramework>
     <LangVersion>latest</LangVersion>
     <GeneratePackageOnBuild>false</GeneratePackageOnBuild>
-    <Version>0.0.3</Version>
+    <Version>0.0.4</Version>
     <Authors>Owen</Authors>
     <Company>Owen</Company>
     <Description>A simple, high performance RPC library.</Description>

diff --git a/src/Uragano.Codec.MessagePack/Uragano.Codec.MessagePack.csproj b/src/Uragano.Codec.MessagePack/Uragano.Codec.MessagePack.csproj
@@ -4,7 +4,7 @@
     <TargetFramework>netstandard2.0</TargetFramework>
     <LangVersion>latest</LangVersion>
     <GeneratePackageOnBuild>false</GeneratePackageOnBuild>
-    <Version>0.0.3</Version>
+    <Version>0.0.4</Version>
     <Description>A simple, high performance RPC library.</Description>
     <PackageProjectUrl>https://github.com/1100100/Uragano</PackageProjectUrl>
     <PackageIconUrl>https://raw.githubusercontent.com/1100100/Uragano/master/icon.png</PackageIconUrl>

diff --git a/src/Uragano.Consul/Uragano.Consul.csproj b/src/Uragano.Consul/Uragano.Consul.csproj
@@ -8,7 +8,7 @@
     <PackageTags>Uragano,RPC,DotNetty,Microservice,MessagePack,DynamicProxy,dotnetcore,service-discovery,polly,circuit-breaker,consul,zookeeper,dependency-injection</PackageTags>
     <PackageProjectUrl>https://github.com/1100100/Uragano</PackageProjectUrl>
     <Description>A simple, high performance RPC library.</Description>
-    <Version>0.0.3</Version>
+    <Version>0.0.4</Version>
     <PackageLicenseUrl></PackageLicenseUrl>
     <Authors>Owen</Authors>
     <Company>Owen</Company>

diff --git a/src/Uragano.Core/PollyCircuitBreaker.cs b/src/Uragano.Core/PollyCircuitBreaker.cs
@@ -4,6 +4,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Threading.Tasks;
+using Uragano.Abstractions;
 using Uragano.Abstractions.CircuitBreaker;
 using Uragano.Abstractions.Service;
 
@@ -17,7 +18,7 @@ public class PollyCircuitBreaker : ICircuitBreaker
 
         private IServiceFactory ServiceFactory { get; }
 
-        private static readonly ConcurrentDictionary<string, AsyncPolicy<object>> Policies = new ConcurrentDictionary<string, AsyncPolicy<object>>();
+        private static readonly ConcurrentDictionary<string, AsyncPolicy<IServiceResult>> Policies = new ConcurrentDictionary<string, AsyncPolicy<IServiceResult>>();
 
         public PollyCircuitBreaker(IServiceProvider serviceProvider, IScriptInjection scriptInjection, IServiceFactory serviceFactory)
         {
@@ -26,39 +27,30 @@ public PollyCircuitBreaker(IServiceProvider serviceProvider, IScriptInjection sc
             ServiceFactory = serviceFactory;
         }
 
-        public async Task<object> ExecuteAsync(string route, Func<Task<object>> action, Type returnValueType)
+        public async Task<IServiceResult> ExecuteAsync(string route, Func<Task<IServiceResult>> action, Type returnValueType)
         {
             var policy = GetPolicy(route, returnValueType);
             return await policy.ExecuteAsync(action);
         }
 
-        public async Task ExecuteAsync(string route, Func<Task> action)
-        {
-            var policy = GetPolicy(route, null);
-            await policy.ExecuteAsync(async () =>
-            {
-                await action();
-                return null;
-            });
-        }
-
-        private AsyncPolicy<object> GetPolicy(string route, Type returnValueType)
+        private AsyncPolicy<IServiceResult> GetPolicy(string route, Type returnValueType)
         {
             return Policies.GetOrAdd(route, key =>
             {
                 var service = ServiceFactory.Get(route);
                 var serviceCircuitBreakerOptions = service.ServiceCircuitBreakerOptions;
                 var circuitBreakerEvent = ServiceProvider.GetService<ICircuitBreakerEvent>();
-                AsyncPolicy<object> policy = Policy<object>.Handle<Exception>().FallbackAsync(
+                AsyncPolicy<IServiceResult> policy = Policy<IServiceResult>.Handle<Exception>().FallbackAsync<IServiceResult>(
                      async ct =>
                      {
+                         //TODO 如果多次降级，根据路由排除此node
                          if (circuitBreakerEvent != null)
                              await circuitBreakerEvent.OnFallback(route, service.ClientMethodInfo);
                          if (returnValueType == null)
-                             return null;
+                             return new ServiceResult(null);
                          if (service.ServiceCircuitBreakerOptions.HasInjection)
-                             return await ScriptInjection.Run(route);
-                         return returnValueType.IsValueType ? Activator.CreateInstance(returnValueType) : null;
+                             return new ServiceResult(await ScriptInjection.Run(route));
+                         return new ServiceResult(returnValueType.IsValueType ? Activator.CreateInstance(returnValueType) : default);
                      });
                 if (serviceCircuitBreakerOptions.ExceptionsAllowedBeforeBreaking > 0)
                 {

diff --git a/src/Uragano.Core/Uragano.Core.csproj b/src/Uragano.Core/Uragano.Core.csproj
@@ -7,7 +7,7 @@
     <PackageIconUrl>https://raw.githubusercontent.com/1100100/Uragano/master/icon.png</PackageIconUrl>
     <PackageProjectUrl>https://github.com/1100100/Uragano</PackageProjectUrl>
     <Description>A simple, high performance RPC library.</Description>
-    <Version>0.0.3</Version>
+    <Version>0.0.4</Version>
     <PackageLicenseUrl></PackageLicenseUrl>
     <GeneratePackageOnBuild>false</GeneratePackageOnBuild>
     <Authors>Owen</Authors>

diff --git a/src/Uragano.Core/UraganoBuilder.cs b/src/Uragano.Core/UraganoBuilder.cs
@@ -230,27 +230,31 @@ public void AddOptions(IConfigurationSection configuration)
 
         #region Circuit breaker
         public void AddCircuitBreaker<TCircuitBreakerEvent>(int timeout = 3000, int retry = 3,
-            int exceptionsAllowedBeforeBreaking = 10, int durationOfBreak = 60000) where TCircuitBreakerEvent : ICircuitBreakerEvent
+            int exceptionsAllowedBeforeBreaking = 10, int durationOfBreak = 60000, int maxParallelization = 0, int maxQueuingActions = 0) where TCircuitBreakerEvent : ICircuitBreakerEvent
         {
             UraganoSettings.CircuitBreakerOptions = new CircuitBreakerOptions
             {
                 Timeout = TimeSpan.FromMilliseconds(timeout),
                 Retry = retry,
                 ExceptionsAllowedBeforeBreaking = exceptionsAllowedBeforeBreaking,
-                DurationOfBreak = TimeSpan.FromMilliseconds(durationOfBreak)
+                DurationOfBreak = TimeSpan.FromMilliseconds(durationOfBreak),
+                MaxParallelization = maxParallelization,
+                MaxQueuingActions = maxQueuingActions
             };
             RegisterSingletonService(typeof(ICircuitBreakerEvent), typeof(TCircuitBreakerEvent));
         }
 
         public void AddCircuitBreaker(int timeout = 3000, int retry = 3, int exceptionsAllowedBeforeBreaking = 10,
-            int durationOfBreak = 60000)
+            int durationOfBreak = 60000, int maxParallelization = 0, int maxQueuingActions = 0)
         {
             UraganoSettings.CircuitBreakerOptions = new CircuitBreakerOptions
             {
                 Timeout = TimeSpan.FromMilliseconds(timeout),
                 Retry = retry,
                 ExceptionsAllowedBeforeBreaking = exceptionsAllowedBeforeBreaking,
-                DurationOfBreak = TimeSpan.FromMilliseconds(durationOfBreak)
+                DurationOfBreak = TimeSpan.FromMilliseconds(durationOfBreak),
+                MaxParallelization = maxParallelization,
+                MaxQueuingActions = maxQueuingActions
             };
         }
 
@@ -261,16 +265,15 @@ public void AddCircuitBreaker(IConfigurationSection configurationSection)
                 Timeout = TimeSpan.FromMilliseconds(configurationSection.GetValue<int>("timeout")),
                 Retry = configurationSection.GetValue<int>("retry"),
                 ExceptionsAllowedBeforeBreaking = configurationSection.GetValue<int>("ExceptionsAllowedBeforeBreaking"),
-                DurationOfBreak = TimeSpan.FromMilliseconds(configurationSection.GetValue<int>("DurationOfBreak"))
+                DurationOfBreak = TimeSpan.FromMilliseconds(configurationSection.GetValue<int>("DurationOfBreak")),
+                MaxParallelization = configurationSection.GetValue<int>("MaxParallelization"),
+                MaxQueuingActions = configurationSection.GetValue<int>("MaxQueuingActions")
             };
-            var eventTypeName = configurationSection.GetValue<string>("EventHandler");
-            if (!string.IsNullOrWhiteSpace(eventTypeName))
-            {
-                var eventType = ReflectHelper.Find(eventTypeName);
-                if (eventType == null)
-                    throw new TypeLoadException($"Cannot load type of {eventTypeName}.");
-                RegisterSingletonService(typeof(ICircuitBreakerEvent), eventType);
-            }
+        }
+
+        public void AddCircuitBreaker<TCircuitBreakerEvent>(IConfigurationSection configurationSection) where TCircuitBreakerEvent : ICircuitBreakerEvent
+        {
+            AddCircuitBreaker<TCircuitBreakerEvent>(configurationSection.GetValue<int>("timeout"), configurationSection.GetValue<int>("retry"), configurationSection.GetValue<int>("ExceptionsAllowedBeforeBreaking"), configurationSection.GetValue<int>("DurationOfBreak"), configurationSection.GetValue<int>("MaxParallelization"), configurationSection.GetValue<int>("MaxQueuingActions"));
         }
 
         #endregion
@@ -494,6 +497,11 @@ public void AddCircuitBreaker()
         {
             AddCircuitBreaker(Configuration.GetSection("Uragano:CircuitBreaker:Polly"));
         }
+
+        public void AddCircuitBreaker<TCircuitBreakerEvent>() where TCircuitBreakerEvent : ICircuitBreakerEvent
+        {
+            AddCircuitBreaker<TCircuitBreakerEvent>(Configuration.GetSection("Uragano:CircuitBreaker:Polly"));
+        }
     }
 }
 
diff --git a/src/Uragano.DynamicProxy/Interceptor/ClientDefaultInterceptor.cs b/src/Uragano.DynamicProxy/Interceptor/ClientDefaultInterceptor.cs
@@ -29,33 +29,21 @@ public override async Task<IServiceResult> Intercept(IInterceptorContext context
             //No circuit breaker
             if (UraganoSettings.CircuitBreakerOptions == null)
             {
-                if (ctx.ReturnType != null)
-                    return new ServiceResult(await Exec(ctx.ServiceName, ctx.ServiceRoute, ctx.Args, ctx.Meta,
-                        ctx.ReturnType));
-                await Exec(ctx.ServiceName, ctx.ServiceRoute, ctx.Args, ctx.Meta, null);
-                return new ServiceResult(null);
+                return await Exec(ctx.ServiceName, ctx.ServiceRoute, ctx.Args, ctx.Meta);
             }
             //Circuit breaker
-            if (ctx.ReturnType != null)
-                return new ServiceResult(await CircuitBreaker.ExecuteAsync(ctx.ServiceRoute,
-                    async () => await Exec(ctx.ServiceName, ctx.ServiceRoute, ctx.Args, ctx.Meta,
-                        ctx.ReturnType), ctx.ReturnType));
-
-            await CircuitBreaker.ExecuteAsync(ctx.ServiceRoute,
-                async () => { await Exec(ctx.ServiceName, ctx.ServiceRoute, ctx.Args, ctx.Meta, null); });
-            return new ServiceResult(null);
+            return await CircuitBreaker.ExecuteAsync(ctx.ServiceRoute,
+                async () => await Exec(ctx.ServiceName, ctx.ServiceRoute, ctx.Args, ctx.Meta), ctx.ReturnType);
         }
 
-        private async Task<object> Exec(string serviceName, string route, object[] args, Dictionary<string, string> meta, Type returnValueType)
+        private async Task<IServiceResult> Exec(string serviceName, string route, object[] args, Dictionary<string, string> meta)
         {
             var node = await LoadBalancing.GetNextNode(serviceName, route, args, meta);
             var client = await ClientFactory.CreateClientAsync(serviceName, node);
             var result = await client.SendAsync(new InvokeMessage(route, args, meta));
             if (result.Status != RemotingStatus.Ok)
                 throw new RemoteInvokeException(route, result.Result?.ToString(), result.Status);
-            if (returnValueType == null)
-                return null;
-            return result.Result;
+            return result;
         }
     }
 }
diff --git a/src/Uragano.DynamicProxy/ServiceFactory.cs b/src/Uragano.DynamicProxy/ServiceFactory.cs
@@ -58,6 +58,8 @@ public void Create(string route, MethodInfo serverMethodInfo, MethodInfo clientM
                             breaker.ExceptionsAllowedBeforeBreaking =
                                 globalCircuitBreaker.ExceptionsAllowedBeforeBreaking;
                             breaker.DurationOfBreak = globalCircuitBreaker.DurationOfBreak;
+                            breaker.MaxParallelization = globalCircuitBreaker.MaxParallelization;
+                            breaker.MaxQueuingActions = globalCircuitBreaker.MaxQueuingActions;
                         }
 
                         if (circuitBreakerAttr != null)

diff --git a/src/Uragano.DynamicProxy/Uragano.DynamicProxy.csproj b/src/Uragano.DynamicProxy/Uragano.DynamicProxy.csproj
@@ -8,7 +8,7 @@
     <PackageIconUrl>https://raw.githubusercontent.com/1100100/Uragano/master/icon.png</PackageIconUrl>
     <PackageProjectUrl>https://github.com/1100100/Uragano</PackageProjectUrl>
     <Description>A simple, high performance RPC library.</Description>
-    <Version>0.0.3</Version>
+    <Version>0.0.4</Version>
     <PackageLicenseUrl></PackageLicenseUrl>
     <Authors>Owen</Authors>
     <Company>Owen</Company>

diff --git a/src/Uragano.Logging.Exceptionless/Uragano.Logging.Exceptionless.csproj b/src/Uragano.Logging.Exceptionless/Uragano.Logging.Exceptionless.csproj
@@ -3,7 +3,7 @@
   <PropertyGroup>
     <TargetFramework>netstandard2.0</TargetFramework>
     <GeneratePackageOnBuild>false</GeneratePackageOnBuild>
-    <Version>0.0.3</Version>
+    <Version>0.0.4</Version>
     <Authors>Owen</Authors>
     <Company>Owen</Company>
     <Description>A simple, high performance RPC library.</Description>

diff --git a/src/Uragano.Logging.Log4net/Uragano.Logging.Log4Net.csproj b/src/Uragano.Logging.Log4net/Uragano.Logging.Log4Net.csproj
@@ -3,7 +3,7 @@
   <PropertyGroup>
     <TargetFramework>netcoreapp2.2</TargetFramework>
     <GeneratePackageOnBuild>false</GeneratePackageOnBuild>
-    <Version>0.0.3</Version>
+    <Version>0.0.4</Version>
     <Authors>Owen</Authors>
     <Company>Owen</Company>
     <Description>A simple, high performance RPC library.</Description>