Commit 7ae7180b authored by yangxiaodong's avatar yangxiaodong

Fix store received message statename bug.

parent 9b67cc4d
...@@ -48,11 +48,12 @@ namespace DotNetCore.CAP ...@@ -48,11 +48,12 @@ namespace DotNetCore.CAP
Content = content Content = content
}; };
message.StateName = StateName.Enqueued;
await _store.StoreSentMessageAsync(message); await _store.StoreSentMessageAsync(message);
WaitHandleEx.PulseEvent.Set(); WaitHandleEx.PulseEvent.Set();
_logger.EnqueuingMessage(topic, content); _logger.EnqueuingSentMessage(topic, content);
} }
} }
} }
\ No newline at end of file
...@@ -84,15 +84,20 @@ namespace DotNetCore.CAP ...@@ -84,15 +84,20 @@ namespace DotNetCore.CAP
public virtual void OnMessageReceieved(object sender, MessageBase message) public virtual void OnMessageReceieved(object sender, MessageBase message)
{ {
var capMessage = new CapReceivedMessage(message); _logger.EnqueuingReceivedMessage(message.KeyName, message.Content);
_logger.LogInformation("message receieved message topic name: " + capMessage.Id);
var capMessage = new CapReceivedMessage(message)
{
StateName = StateName.Enqueued,
Added = DateTime.Now
};
_messageStore.StoreReceivedMessageAsync(capMessage).Wait(); _messageStore.StoreReceivedMessageAsync(capMessage).Wait();
ConsumerExecutorDescriptor executeDescriptor = null;
try try
{ {
var executeDescriptor = _selector.GetTopicExector(message.KeyName); executeDescriptor = _selector.GetTopicExector(message.KeyName);
var consumerContext = new ConsumerContext(executeDescriptor, message); var consumerContext = new ConsumerContext(executeDescriptor, message);
...@@ -104,7 +109,7 @@ namespace DotNetCore.CAP ...@@ -104,7 +109,7 @@ namespace DotNetCore.CAP
} }
catch (Exception ex) catch (Exception ex)
{ {
_logger.LogError("exception raised when excute method : " + ex.Message); _logger.ConsumerMethodExecutingFailed(executeDescriptor.MethodInfo.Name, ex);
} }
} }
......
...@@ -17,7 +17,9 @@ namespace DotNetCore.CAP ...@@ -17,7 +17,9 @@ namespace DotNetCore.CAP
private static Action<ILogger, string, double, Exception> _cronJobExecuted; private static Action<ILogger, string, double, Exception> _cronJobExecuted;
private static Action<ILogger, string, Exception> _cronJobFailed; private static Action<ILogger, string, Exception> _cronJobFailed;
private static Action<ILogger, string, string, Exception> _enqueuingMessage; private static Action<ILogger, string, string, Exception> _enqueuingSentMessage;
private static Action<ILogger, string, string, Exception> _enqueuingReceivdeMessage;
private static Action<ILogger, string, Exception> _executingConsumerMethod;
static LoggerExtensions() static LoggerExtensions()
{ {
...@@ -56,15 +58,35 @@ namespace DotNetCore.CAP ...@@ -56,15 +58,35 @@ namespace DotNetCore.CAP
4, 4,
"Cron job '{jobName}' failed to execute."); "Cron job '{jobName}' failed to execute.");
_enqueuingMessage = LoggerMessage.Define<string, string>( _enqueuingSentMessage = LoggerMessage.Define<string, string>(
LogLevel.Debug, LogLevel.Debug,
2, 2,
"Enqueuing a topic to the store. NameKey: {NameKey}. Content: {Content}"); "Enqueuing a topic to the sent message store. NameKey: {NameKey}. Content: {Content}");
_enqueuingReceivdeMessage = LoggerMessage.Define<string, string>(
LogLevel.Debug,
2,
"Enqueuing a topic to the received message store. NameKey: {NameKey}. Content: {Content}");
_executingConsumerMethod = LoggerMessage.Define<string>(
LogLevel.Error,
5,
"Consumer method '{methodName}' failed to execute.");
}
public static void ConsumerMethodExecutingFailed(this ILogger logger, string methodName, Exception ex)
{
_executingConsumerMethod(logger, methodName, ex);
}
public static void EnqueuingReceivedMessage(this ILogger logger, string nameKey, string content)
{
_enqueuingReceivdeMessage(logger, nameKey, content, null);
} }
public static void EnqueuingMessage(this ILogger logger, string nameKey, string content) public static void EnqueuingSentMessage(this ILogger logger, string nameKey, string content)
{ {
_enqueuingMessage(logger, nameKey, content, null); _enqueuingSentMessage(logger, nameKey, content, null);
} }
public static void ServerStarting(this ILogger logger, int machineProcessorCount, int processorCount) public static void ServerStarting(this ILogger logger, int machineProcessorCount, int processorCount)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment