Release 2.2.2 (#121)

yang-xiaodong · web-flow · commit c05386016f8c · 2018-04-28T16:03:36.000+08:00
* fixed message enqueue exception in v2.2 * add log to samples * Improved log output. #114 * add default timeout configuration for kafka client. * fixed retry processor bugs. * Fixed kafka producer exception log without logging when publish message. * update version num to 2.2.2 * rename configuration options FailedCallback to FailedThresholdCallback * rename files name. * remove unused files. * modify the error comments. * update samples. * add logs.
diff --git a/build/version.props b/build/version.props
@@ -2,7 +2,7 @@
 	<PropertyGroup>
 		<VersionMajor>2</VersionMajor>
 		<VersionMinor>2</VersionMinor>
-		<VersionPatch>1</VersionPatch>
+		<VersionPatch>2</VersionPatch>
 		<VersionQuality></VersionQuality>
 		<VersionPrefix>$(VersionMajor).$(VersionMinor).$(VersionPatch)</VersionPrefix>
 	</PropertyGroup>
diff --git a/samples/Sample.Kafka.MySql/Controllers/ValuesController.cs b/samples/Sample.Kafka.MySql/Controllers/ValuesController.cs
@@ -37,7 +37,7 @@ public async Task<IActionResult> PublishMessage()
         [CapSubscribe("xxx.xxx.test2")]
         public void Test2(int value)
         {
-            Console.WriteLine(value);
+            Console.WriteLine("Subscriber output message: " + value);
         }
     }
 }
diff --git a/samples/Sample.Kafka.MySql/Program.cs b/samples/Sample.Kafka.MySql/Program.cs
@@ -1,5 +1,6 @@
 ﻿using Microsoft.AspNetCore;
 using Microsoft.AspNetCore.Hosting;
+using NLog.Web;
 
 namespace Sample.Kafka.MySql
 {
@@ -14,7 +15,11 @@ public static void Main(string[] args)
         public static IWebHost BuildWebHost(string[] args) =>
             WebHost.CreateDefaultBuilder(args)
                 .UseStartup<Startup>()
+                .ConfigureLogging((hostingContext, builder) =>
+                {
+                    hostingContext.HostingEnvironment.ConfigureNLog("nlog.config");
+                })
+                .UseNLog()
                 .Build();
-
     }
 }
diff --git a/samples/Sample.Kafka.MySql/Sample.Kafka.MySql.csproj b/samples/Sample.Kafka.MySql/Sample.Kafka.MySql.csproj
@@ -10,6 +10,7 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.AspNetCore.All" Version="2.0.6" />
     <PackageReference Include="MySqlConnector" Version="0.38.0" />
+    <PackageReference Include="NLog.Web.AspNetCore" Version="4.5.2" />
     <PackageReference Include="zipkin4net" Version="1.2.0" />
   </ItemGroup>
   <ItemGroup>
@@ -20,5 +21,10 @@
     <ProjectReference Include="..\..\src\DotNetCore.CAP.MySql\DotNetCore.CAP.MySql.csproj" />
     <ProjectReference Include="..\..\src\DotNetCore.CAP\DotNetCore.CAP.csproj" />
   </ItemGroup>
+  <ItemGroup>
+    <Content Update="nlog.config">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </Content>
+  </ItemGroup>
 
 </Project>
diff --git a/samples/Sample.Kafka.MySql/nlog.config b/samples/Sample.Kafka.MySql/nlog.config
@@ -0,0 +1,26 @@
+﻿<?xml version="1.0" encoding="utf-8"?>
+
+<nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd"
+      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+      autoReload="true"
+      internalLogLevel="Warn"
+      internalLogFile="logs/internal-nlog.txt">
+
+  <variable name="myLogLayout"
+            value="---------------------------------------------------------------------------${newline}Date：${longdate}    Level：${uppercase:${level}}    User：${aspnet-user-identity}(${aspnet-request-ip})    Logger：${logger}  URL：${aspnet-request-method} ${aspnet-request-url:IncludePort=true:IncludeQueryString=true}  Action：${aspnet-mvc-action} ${newline}Message：${message} ${newline}${onexception:Exception：${exception:format=toString}}" />
+  <extensions>
+    <add assembly="NLog.Web.AspNetCore" />
+  </extensions>
+
+  <!-- define various log targets -->
+  <targets>
+    <!-- write logs to file -->
+    <target name="allfile" xsi:type="File" fileName="logs/cap-all-${shortdate}.log"
+            layout="${myLogLayout}" />
+  </targets>
+
+  <rules>
+    <!--All logs, including from Microsoft-->
+    <logger name="*" minlevel="Debug" writeTo="allfile" /> 
+  </rules>
+</nlog>
diff --git a/samples/Sample.RabbitMQ.MySql/AppDbContext.cs b/samples/Sample.RabbitMQ.MySql/AppDbContext.cs
@@ -1,17 +1,12 @@
-﻿using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading.Tasks;
-using Microsoft.EntityFrameworkCore;
+﻿using Microsoft.EntityFrameworkCore;
 
 namespace Sample.RabbitMQ.MySql
 {
     public class AppDbContext : DbContext
     {
         protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
         {
-            optionsBuilder.UseMySql("Server=localhost;Database=Sample.RabbitMQ.MySql;UserId=root;Password=123123;Allow User Variables=True");
-            //optionsBuilder.UseMySql("Server=192.168.2.206;Database=Sample.RabbitMQ.MySql;UserId=root;Password=123123;");
+            optionsBuilder.UseMySql("Server=192.168.10.110;Database=testcap;UserId=root;Password=123123;");
         }
     }
 }
diff --git a/samples/Sample.RabbitMQ.MySql/Controllers/ValuesController.cs b/samples/Sample.RabbitMQ.MySql/Controllers/ValuesController.cs
@@ -21,7 +21,7 @@ public ValuesController(AppDbContext dbContext, ICapPublisher capPublisher)
         public IActionResult PublishMessage()
         {
             _capBus.Publish("sample.rabbitmq.mysql", DateTime.Now);
-            
+
             return Ok();
         }
 
@@ -50,7 +50,7 @@ public async Task<IActionResult> PublishMessageWithTransaction()
         [CapSubscribe("sample.rabbitmq.mysql")]
         public void ReceiveMessage(DateTime time)
         {
-            Console.WriteLine("[sample.rabbitmq.mysql] message received: "+ DateTime.Now.ToString() +" , sent time: " + time.ToString());
+            Console.WriteLine("[sample.rabbitmq.mysql] message received: " + DateTime.Now + ",sent time: " + time);
         }
     }
 }
diff --git a/samples/Sample.RabbitMQ.MySql/Program.cs b/samples/Sample.RabbitMQ.MySql/Program.cs
@@ -1,12 +1,6 @@
-﻿using System;
-using System.Collections.Generic;
-using System.IO;
-using System.Linq;
-using System.Threading.Tasks;
-using Microsoft.AspNetCore;
-using Microsoft.AspNetCore.Builder;
+﻿using Microsoft.AspNetCore;
 using Microsoft.AspNetCore.Hosting;
-using Microsoft.Extensions.Configuration;
+using NLog.Web;
 
 namespace Sample.RabbitMQ.MySql
 {
@@ -20,6 +14,11 @@ public static void Main(string[] args)
         public static IWebHost BuildWebHost(string[] args) =>
             WebHost.CreateDefaultBuilder(args)
                 .UseStartup<Startup>()
+                .ConfigureLogging((hostingContext, builder) =>
+                {
+                    hostingContext.HostingEnvironment.ConfigureNLog("nlog.config");
+                })
+                .UseNLog()
                 .Build();
     }
 }
diff --git a/samples/Sample.RabbitMQ.MySql/Sample.RabbitMQ.MySql.csproj b/samples/Sample.RabbitMQ.MySql/Sample.RabbitMQ.MySql.csproj
@@ -12,6 +12,7 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.AspNetCore.All" Version="2.0.6" />
     <PackageReference Include="Pomelo.EntityFrameworkCore.MySql" Version="2.0.1" />
+    <PackageReference Include="NLog.Web.AspNetCore" Version="4.5.2" />
   </ItemGroup>
   <ItemGroup>
     <DotNetCliToolReference Include="Microsoft.EntityFrameworkCore.Tools.DotNet" Version="2.0.0" />
@@ -21,5 +22,10 @@
     <ProjectReference Include="..\..\src\DotNetCore.CAP.RabbitMQ\DotNetCore.CAP.RabbitMQ.csproj" />
     <ProjectReference Include="..\..\src\DotNetCore.CAP\DotNetCore.CAP.csproj" />
   </ItemGroup>
+  <ItemGroup>
+    <Content Update="nlog.config">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </Content>
+  </ItemGroup>
 
 </Project>
diff --git a/samples/Sample.RabbitMQ.MySql/nlog.config b/samples/Sample.RabbitMQ.MySql/nlog.config
@@ -0,0 +1,26 @@
+﻿<?xml version="1.0" encoding="utf-8"?>
+
+<nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd"
+      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+      autoReload="true"
+      internalLogLevel="Warn"
+      internalLogFile="logs/internal-nlog.txt">
+
+  <variable name="myLogLayout"
+            value="---------------------------------------------------------------------------${newline}Date：${longdate}    Level：${uppercase:${level}}    User：${aspnet-user-identity}(${aspnet-request-ip})    Logger：${logger}  URL：${aspnet-request-method} ${aspnet-request-url:IncludePort=true:IncludeQueryString=true}  Action：${aspnet-mvc-action} ${newline}Message：${message} ${newline}${onexception:Exception：${exception:format=toString}}" />
+  <extensions>
+    <add assembly="NLog.Web.AspNetCore" />
+  </extensions>
+
+  <!-- define various log targets -->
+  <targets>
+    <!-- write logs to file -->
+    <target name="allfile" xsi:type="File" fileName="logs/cap-all-${shortdate}.log"
+            layout="${myLogLayout}" />
+  </targets>
+
+  <rules>
+    <!--All logs, including from Microsoft-->
+    <logger name="*" minlevel="Debug" writeTo="allfile" />
+  </rules>
+</nlog>
diff --git a/src/DotNetCore.CAP.Kafka/CAP.KafkaOptions.cs b/src/DotNetCore.CAP.Kafka/CAP.KafkaOptions.cs
@@ -57,6 +57,8 @@ internal IEnumerable<KeyValuePair<string, object>> AsKafkaConfig()
                 MainConfig["socket.blocking.max.ms"] = "10";
                 MainConfig["enable.auto.commit"] = "false";
                 MainConfig["log.connection.close"] = "false";
+                MainConfig["request.timeout.ms"] = "3000";
+                MainConfig["message.timeout.ms"] = "5000";
 
                 _kafkaConfig = MainConfig.AsEnumerable();
             }
diff --git a/src/DotNetCore.CAP.Kafka/IConnectionPool.Default.cs b/src/DotNetCore.CAP.Kafka/IConnectionPool.Default.cs
diff --git a/src/DotNetCore.CAP.Kafka/IPublishMessageSender.Kafka.cs b/src/DotNetCore.CAP.Kafka/IPublishMessageSender.Kafka.cs
@@ -36,11 +36,7 @@ public override async Task<OperateResult> PublishAsync(string keyName, string co
 
                 if (message.Error.HasError)
                 {
-                    return OperateResult.Failed(new OperateError
-                    {
-                        Code = message.Error.Code.ToString(),
-                        Description = message.Error.Reason
-                    });
+                    throw new PublisherSentFailedException(message.Error.ToString());
                 }
 
                 _logger.LogDebug($"kafka topic message [{keyName}] has been published.");
@@ -61,6 +57,6 @@ public override async Task<OperateResult> PublishAsync(string keyName, string co
                     producer.Dispose();
                 }
             }
-        } 
+        }
     }
 }
diff --git a/src/DotNetCore.CAP.RabbitMQ/IConnectionChannelPool.Default.cs b/src/DotNetCore.CAP.RabbitMQ/IConnectionChannelPool.Default.cs
diff --git a/src/DotNetCore.CAP/Abstractions/CapPublisherBase.cs b/src/DotNetCore.CAP/Abstractions/CapPublisherBase.cs
@@ -132,7 +132,7 @@ private void CheckIsUsingEF(string name)
             {
                 throw new InvalidOperationException(
                     "If you are using the EntityFramework, you need to configure the DbContextType first." +
-                    " otherwise you need to use overloaded method with IDbConnection and IDbTransaction.");
+                    " otherwise you need to use overloaded method with IDbTransaction.");
             }
         }
 
diff --git a/src/DotNetCore.CAP/CAP.Options.cs b/src/DotNetCore.CAP/CAP.Options.cs
@@ -58,9 +58,9 @@ public CapOptions()
         public int FailedRetryInterval { get; set; }
 
         /// <summary>
-        /// We’ll invoke this call-back with message type,name,content when requeue failed message.
+        /// We’ll invoke this call-back with message type,name,content when retry failed (send or executed) messages equals <see cref="FailedRetryCount"/> times.
         /// </summary>
-        public Action<MessageType, string, string> FailedCallback { get; set; }
+        public Action<MessageType, string, string> FailedThresholdCallback { get; set; }
 
         /// <summary>
         /// The number of message retries, the retry will stop when the threshold is reached.
diff --git a/src/DotNetCore.CAP/ICapPublisher.cs b/src/DotNetCore.CAP/ICapPublisher.cs
@@ -15,7 +15,7 @@ public interface ICapPublisher
         /// (EntityFramework) Asynchronous publish a object message.
         /// <para>
         /// If you are using the EntityFramework, you need to configure the DbContextType first.
-        /// otherwise you need to use overloaded method with IDbConnection and IDbTransaction.
+        /// otherwise you need to use overloaded method with  IDbTransaction.
         /// </para>
         /// </summary>
         /// <typeparam name="T">The type of content object.</typeparam>
@@ -28,7 +28,7 @@ public interface ICapPublisher
         /// (EntityFramework) Publish a object message.
         /// <para>
         /// If you are using the EntityFramework, you need to configure the DbContextType first.
-        /// otherwise you need to use overloaded method with IDbConnection and IDbTransaction.
+        /// otherwise you need to use overloaded method with IDbTransaction.
         /// </para>
         /// </summary>
         /// <typeparam name="T">The type of content object.</typeparam>
diff --git a/src/DotNetCore.CAP/ICapSubscribe.cs b/src/DotNetCore.CAP/ICapSubscribe.cs
@@ -4,7 +4,7 @@
 namespace DotNetCore.CAP
 {
     /// <summary>
-    /// An empty interface, which is used to mark the current class have a CAP methods.
+    /// An empty interface, which is used to mark the current class have a CAP subscriber methods.
     /// </summary>
     public interface ICapSubscribe
     {
diff --git a/src/DotNetCore.CAP/IConsumerHandler.Default.cs b/src/DotNetCore.CAP/IConsumerHandler.Default.cs
@@ -178,6 +178,8 @@ private void StoreMessage(CapReceivedMessage receivedMessage)
 
         private (Guid, string) TracingBefore(string topic, string values)
         {
+            _logger.LogDebug("CAP received topic message:" + topic);
+
             Guid operationId = Guid.NewGuid();
 
             var eventData = new BrokerConsumeEventData(
diff --git a/src/DotNetCore.CAP/IFetchedMessage.cs b/src/DotNetCore.CAP/IFetchedMessage.cs
diff --git a/src/DotNetCore.CAP/IPublishMessageSender.Base.cs b/src/DotNetCore.CAP/IPublishMessageSender.Base.cs
@@ -67,15 +67,13 @@ public async Task<OperateResult> SendAsync(CapPublishedMessage message)
             }
             else
             {
-                TracingError(operationId, message.Name, sendValues, result.Exception, startTime, stopwatch.Elapsed);
-
-                _logger.MessagePublishException(message.Id, result.Exception);
+                TracingError(operationId, message, result, startTime, stopwatch.Elapsed);
 
                 await SetFailedState(message, result.Exception, out bool stillRetry);
 
                 if (stillRetry)
                 {
-                    _logger.SenderRetrying(3);
+                    _logger.SenderRetrying(message.Id, message.Retries);
 
                     await SendAsync(message);
                 }
@@ -109,20 +107,11 @@ private Task SetSuccessfulState(CapPublishedMessage message)
         private Task SetFailedState(CapPublishedMessage message, Exception ex, out bool stillRetry)
         {
             IState newState = new FailedState();
-
-            if (ex is PublisherSentFailedException)
-            {
-                stillRetry = false;
-                message.Retries = _options.FailedRetryCount; // not retry if PublisherSentFailedException
-            }
-            else
+            stillRetry = UpdateMessageForRetryAsync(message);
+            if (stillRetry)
             {
-                stillRetry = UpdateMessageForRetryAsync(message);
-                if (stillRetry)
-                {
-                    _logger.ConsumerExecutionFailedWillRetry(ex);
-                    return Task.CompletedTask;
-                }
+                _logger.ConsumerExecutionFailedWillRetry(ex);
+                return Task.CompletedTask;
             }
 
             AddErrorReasonToContent(message, ex);
@@ -166,14 +155,18 @@ private void TracingAfter(Guid operationId, string topic, string values, DateTim
             _logger.MessageHasBeenSent(du.TotalSeconds);
         }
 
-        private void TracingError(Guid operationId, string topic, string values, Exception ex, DateTimeOffset startTime, TimeSpan du)
+        private void TracingError(Guid operationId, CapPublishedMessage message, OperateResult result, DateTimeOffset startTime, TimeSpan du)
         {
+            var ex = new PublisherSentFailedException(result.ToString(), result.Exception);
+
+            _logger.MessagePublishException(message.Id, result.ToString(), ex);
+
             var eventData = new BrokerPublishErrorEventData(
                 operationId,
                 "",
                 ServersAddress,
-                topic,
-                values,
+                message.Name,
+                message.Content,
                 ex,
                 startTime,
                 du);
diff --git a/src/DotNetCore.CAP/Infrastructure/Helper.cs b/src/DotNetCore.CAP/Infrastructure/Helper.cs
@@ -86,7 +86,6 @@ public static bool IsComplexType(Type type)
             return !CanConvertFromString(type);
         }
 
-
         public static string AddExceptionProperty(string json, Exception exception)
         {
             var jObject = ToJObject(exception);
diff --git a/src/DotNetCore.CAP/Internal/PublisherSentFailedException.cs b/src/DotNetCore.CAP/Internal/PublisherSentFailedException.cs
@@ -7,6 +7,10 @@ namespace DotNetCore.CAP.Internal
 {
     public class PublisherSentFailedException : Exception
     {
+        public PublisherSentFailedException(string message) : base(message)
+        {
+        }
+
         public PublisherSentFailedException(string message, Exception ex) : base(message, ex)
         {
         }
diff --git a/src/DotNetCore.CAP/LoggerExtensions.cs b/src/DotNetCore.CAP/LoggerExtensions.cs
diff --git a/src/DotNetCore.CAP/Processor/IProcessor.NeedRetry.cs b/src/DotNetCore.CAP/Processor/IProcessor.NeedRetry.cs

Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,7 @@ public async Task<IActionResult> PublishMessage()`
`37`	`37`	`[CapSubscribe("xxx.xxx.test2")]`
`38`	`38`	`public void Test2(int value)`
`39`	`39`	`{`
`40`		`- Console.WriteLine(value);`
	`40`	`+ Console.WriteLine("Subscriber output message: " + value);`
`41`	`41`	`}`
`42`	`42`	`}`
`43`	`43`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,6 @@`
`1`	`1`	`using Microsoft.AspNetCore;`
`2`	`2`	`using Microsoft.AspNetCore.Hosting;`
	`3`	`+using NLog.Web;`
`3`	`4`
`4`	`5`	`namespace Sample.Kafka.MySql`
`5`	`6`	`{`
`@@ -14,7 +15,11 @@ public static void Main(string[] args)`
`14`	`15`	`public static IWebHost BuildWebHost(string[] args) =>`
`15`	`16`	`WebHost.CreateDefaultBuilder(args)`
`16`	`17`	`.UseStartup<Startup>()`
	`18`	`+ .ConfigureLogging((hostingContext, builder) =>`
	`19`	`+ {`
	`20`	`+ hostingContext.HostingEnvironment.ConfigureNLog("nlog.config");`
	`21`	`+ })`
	`22`	`+ .UseNLog()`
`17`	`23`	`.Build();`
`18`		`-`
`19`	`24`	`}`
`20`	`25`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,17 +1,12 @@`
`1`		`-using System;`
`2`		`-using System.Collections.Generic;`
`3`		`-using System.Linq;`
`4`		`-using System.Threading.Tasks;`
`5`		`-using Microsoft.EntityFrameworkCore;`
	`1`	`+using Microsoft.EntityFrameworkCore;`
`6`	`2`
`7`	`3`	`namespace Sample.RabbitMQ.MySql`
`8`	`4`	`{`
`9`	`5`	`public class AppDbContext : DbContext`
`10`	`6`	`{`
`11`	`7`	`protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)`
`12`	`8`	`{`
`13`		`- optionsBuilder.UseMySql("Server=localhost;Database=Sample.RabbitMQ.MySql;UserId=root;Password=123123;Allow User Variables=True");`
`14`		`- //optionsBuilder.UseMySql("Server=192.168.2.206;Database=Sample.RabbitMQ.MySql;UserId=root;Password=123123;");`
	`9`	`+ optionsBuilder.UseMySql("Server=192.168.10.110;Database=testcap;UserId=root;Password=123123;");`
`15`	`10`	`}`
`16`	`11`	`}`
`17`	`12`	`}`
Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@ public ValuesController(AppDbContext dbContext, ICapPublisher capPublisher)`
`21`	`21`	`public IActionResult PublishMessage()`
`22`	`22`	`{`
`23`	`23`	`_capBus.Publish("sample.rabbitmq.mysql", DateTime.Now);`
`24`		`-`
	`24`	`+`
`25`	`25`	`return Ok();`
`26`	`26`	`}`
`27`	`27`
`@@ -50,7 +50,7 @@ public async Task<IActionResult> PublishMessageWithTransaction()`
`50`	`50`	`[CapSubscribe("sample.rabbitmq.mysql")]`
`51`	`51`	`public void ReceiveMessage(DateTime time)`
`52`	`52`	`{`
`53`		`- Console.WriteLine("[sample.rabbitmq.mysql] message received: "+ DateTime.Now.ToString() +" , sent time: " + time.ToString());`
	`53`	`+ Console.WriteLine("[sample.rabbitmq.mysql] message received: " + DateTime.Now + ",sent time: " + time);`
`54`	`54`	`}`
`55`	`55`	`}`
`56`	`56`	`}`