Skip to content

C
CAP
Commit 9b67cc4d authored by yangxiaodong's avatar yangxiaodong
Browse files
Options

Refactor.

parent b478b77d
Show whitespace changes
Inline Side-by-side
Showing with 187 additions and 261 deletions
src/DotNetCore.CAP.EntityFrameworkCore/ConsistencyDbContext.cs
View file @ 9b67cc4d
......@@ -24,7 +24,7 @@ namespace DotNetCore.CAP.EntityFrameworkCore
/// <summary>
/// Gets or sets the <see cref="DbSet{ConsistencyMessage}"/> of Messages.
/// </summary>
public DbSet<ConsistencyMessage> Messages { get; set; }
public DbSet<CapMessage> Messages { get; set; }
/// <summary>
/// Configures the schema for the identity framework.
......@@ -34,7 +34,7 @@ namespace DotNetCore.CAP.EntityFrameworkCore
/// </param>
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
modelBuilder.Entity<ConsistencyMessage>(b =>
modelBuilder.Entity<CapMessage>(b =>
{
b.HasKey(m => m.Id);
b.ToTable("ConsistencyMessages");
......
src/DotNetCore.CAP.EntityFrameworkCore/ConsistencyEntityFrameworkBuilderExtensions.cs
View file @ 9b67cc4d
......@@ -18,7 +18,7 @@ namespace Microsoft.Extensions.DependencyInjection
public static CapBuilder AddEntityFrameworkStores<TContext>(this CapBuilder builder)
where TContext : DbContext
{
builder.Services.AddScoped<ICapMessageStore, ConsistencyMessageStore<TContext>>();
builder.Services.AddScoped<ICapMessageStore, CapMessageStore<TContext>>();
return builder;
}
......
src/DotNetCore.CAP.EntityFrameworkCore/ConsistencyMessageStore.cs
View file @ 9b67cc4d
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using DotNetCore.CAP.Infrastructure;
using Microsoft.EntityFrameworkCore;
......@@ -13,186 +11,121 @@ namespace DotNetCore.CAP.EntityFrameworkCore
/// <typeparam name="ConsistencyMessage">The type representing a message.</typeparam>
/// <typeparam name="TContext">The type of the data context class used to access the store.</typeparam>
/// <typeparam name="TKey">The type of the primary key for a message.</typeparam>
public class ConsistencyMessageStore<TContext> : ICapMessageStore where TContext : DbContext
public class CapMessageStore<TContext> : ICapMessageStore where TContext : DbContext
{
private bool _disposed;
/// <summary>
/// Constructs a new instance of <see cref="ConsistencyMessageStore{ConsistencyMessage, TContext, TKey}"/>.
/// </summary>
/// <param name="context">The <see cref="DbContext"/>.</param>
public ConsistencyMessageStore(TContext context)
{
if (context == null)
public CapMessageStore(TContext context)
{
throw new ArgumentNullException(nameof(context));
}
Context = context;
Context = context ?? throw new ArgumentNullException(nameof(context));
}
public TContext Context { get; private set; }
private DbSet<ConsistencyMessage> MessageSet { get { return Context.Set<ConsistencyMessage>(); } }
private DbSet<CapSentMessage> SentMessages { get { return Context.Set<CapSentMessage>(); } }
private DbSet<CapReceivedMessage> ReceivedMessages { get { return Context.Set<CapReceivedMessage>(); } }
/// <summary>
/// Creates the specified <paramref name="message"/> in the consistency message store.
/// Creates the specified <paramref name="message"/> in the cap message store.
/// </summary>
/// <param name="message">The message to create.</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> used to propagate notifications that the operation should be canceled.</param>
/// <returns>The <see cref="Task"/> that represents the asynchronous operation, containing the <see cref="OperateResult"/> of the creation operation.</returns>
public async virtual Task<OperateResult> CreateAsync(ConsistencyMessage message, CancellationToken cancellationToken)
public async Task<OperateResult> StoreSentMessageAsync(CapSentMessage message)
{
cancellationToken.ThrowIfCancellationRequested();
ThrowIfDisposed();
if (message == null)
{
throw new ArgumentNullException(nameof(message));
}
if (message == null) throw new ArgumentNullException(nameof(message));
Context.Add(message);
await SaveChanges(cancellationToken);
await Context.SaveChangesAsync();
return OperateResult.Success;
}
/// <summary>
/// Deletes the specified <paramref name="message"/> from the consistency message store.
/// First Enqueued Message.
/// </summary>
/// <param name="message">The message to delete.</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> used to propagate notifications that the operation should be canceled.</param>
/// <returns>The <see cref="Task"/> that represents the asynchronous operation, containing the <see cref="OperateResult"/> of the update operation.</returns>
public async virtual Task<OperateResult> DeleteAsync(ConsistencyMessage message, CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
ThrowIfDisposed();
if (message == null)
public async Task<CapSentMessage> GetNextSentMessageToBeEnqueuedAsync()
{
throw new ArgumentNullException(nameof(message));
return await SentMessages.FirstOrDefaultAsync(x => x.StateName == StateName.Enqueued);
}
Context.Remove(message);
/// <summary>
/// Updates a message in a store as an asynchronous operation.
/// </summary>
/// <param name="message">The message to update in the store.</param>
public async Task<OperateResult> UpdateSentMessageAsync(CapSentMessage message)
{
if (message == null) throw new ArgumentNullException(nameof(message));
Context.Attach(message);
message.LastRun = DateTime.Now;
Context.Update(message);
try
{
await SaveChanges(cancellationToken);
await Context.SaveChangesAsync();
return OperateResult.Success;
}
catch (DbUpdateConcurrencyException ex)
{
return OperateResult.Failed(new OperateError() { Code = "DbUpdateConcurrencyException", Description = ex.Message });
}
return OperateResult.Success;
}
/// <summary>
/// Finds and returns a message, if any, who has the specified <paramref name="messageId"/>.
/// Deletes the specified <paramref name="message"/> from the consistency message store.
/// </summary>
/// <param name="messageId">The message ID to search for.</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> used to propagate notifications that the operation should be canceled.</param>
/// <returns>
/// The <see cref="Task"/> that represents the asynchronous operation, containing the message matching the specified <paramref name="messageId"/> if it exists.
/// </returns>
public virtual Task<ConsistencyMessage> FindByIdAsync(string messageId, CancellationToken cancellationToken)
/// <param name="message">The message to delete.</param>
public async Task<OperateResult> RemoveSentMessageAsync(CapSentMessage message)
{
if (message == null) throw new ArgumentNullException(nameof(message));
Context.Remove(message);
try
{
await Context.SaveChangesAsync();
return OperateResult.Success;
}
catch (DbUpdateConcurrencyException ex)
{
cancellationToken.ThrowIfCancellationRequested();
ThrowIfDisposed();
return MessageSet.FindAsync(new object[] { messageId }, cancellationToken);
return OperateResult.Failed(new OperateError() { Code = "DbUpdateConcurrencyException", Description = ex.Message });
}
}
/// <summary>
/// Gets the message identifier for the specified <paramref name="message"/>.
/// Creates the specified <paramref name="message"/> in the consistency message store.
/// </summary>
/// <param name="message">The message whose identifier should be retrieved.</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> used to propagate notifications that the operation should be canceled.</param>
/// <returns>The <see cref="Task"/> that represents the asynchronous operation, containing the identifier for the specified <paramref name="message"/>.</returns>
public Task<string> GeConsistencyMessageIdAsync(ConsistencyMessage message, CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
ThrowIfDisposed();
if (message == null)
/// <param name="message">The message to create.</param>
public async Task<OperateResult> StoreReceivedMessageAsync(CapReceivedMessage message)
{
throw new ArgumentNullException(nameof(message));
}
return Task.FromResult(message.Id);
if (message == null) throw new ArgumentNullException(nameof(message));
Context.Add(message);
await Context.SaveChangesAsync();
return OperateResult.Success;
}
/// <summary>
/// Updates the specified <paramref name="message"/> in the message store.
/// </summary>
/// <param name="message">The message to update.</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> used to propagate notifications that the operation should be canceled.</param>
/// <returns>The <see cref="Task"/> that represents the asynchronous operation, containing the <see cref="OperateResult"/> of the update operation.</returns>
public async virtual Task<OperateResult> UpdateAsync(ConsistencyMessage message, CancellationToken cancellationToken)
public async Task<OperateResult> UpdateReceivedMessageAsync(CapReceivedMessage message)
{
cancellationToken.ThrowIfCancellationRequested();
ThrowIfDisposed();
if (message == null)
{
throw new ArgumentNullException(nameof(message));
}
if (message == null) throw new ArgumentNullException(nameof(message));
Context.Attach(message);
message.UpdateTime = DateTime.Now;
message.LastRun = DateTime.Now;
Context.Update(message);
try
{
await SaveChanges(cancellationToken);
await Context.SaveChangesAsync();
return OperateResult.Success;
}
catch (DbUpdateConcurrencyException ex)
{
return OperateResult.Failed(new OperateError() { Code = "DbUpdateConcurrencyException", Description = ex.Message });
}
return OperateResult.Success;
}
public Task<ConsistencyMessage> GetFirstEnqueuedMessageAsync(CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
return MessageSet.AsNoTracking().Where(x => x.Status == MessageStatus.WaitForSend).FirstOrDefaultAsync(cancellationToken);
}
//public void ChangeState(ConsistencyMessage message, MessageStatus status) {
// Context.Attach(message);
// message.Status = status;
// Context.Update(message);
// try {
// await SaveChanges(cancellationToken);
// }
// catch (DbUpdateConcurrencyException ex) {
// return OperateResult.Failed(new OperateError() { Code = "DbUpdateConcurrencyException", Description = ex.Message });
// }
// return OperateResult.Success;
//}
/// <summary>
/// Gets or sets a flag indicating if changes should be persisted after CreateAsync, UpdateAsync and DeleteAsync are called.
/// </summary>
/// <value>
/// True if changes should be automatically persisted, otherwise false.
/// </value>
public bool AutoSaveChanges { get; set; } = true;
/// <summary>Saves the current store.</summary>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> used to propagate notifications that the operation should be canceled.</param>
/// <returns>The <see cref="Task"/> that represents the asynchronous operation.</returns>
protected Task SaveChanges(CancellationToken cancellationToken)
{
return AutoSaveChanges ? Context.SaveChangesAsync(cancellationToken) : Task.CompletedTask;
}
/// <summary>
/// Throws if this class has been disposed.
/// </summary>
protected void ThrowIfDisposed()
{
if (_disposed)
{
throw new ObjectDisposedException(GetType().Name);
}
}
/// <summary>
/// Dispose the store
/// </summary>
public void Dispose()
{
_disposed = true;
}
}
}
\ No newline at end of file
src/DotNetCore.CAP.Kafka/IProcessor.Producer.cs
View file @ 9b67cc4d
......@@ -83,14 +83,14 @@ namespace DotNetCore.CAP.Kafka
var messageStore = provider.GetRequiredService<ICapMessageStore>();
try
{
var message = await messageStore.GetFirstEnqueuedMessageAsync(_cts.Token);
var message = await messageStore.GetNextSentMessageToBeEnqueuedAsync();
if (message != null)
{
var sp = Stopwatch.StartNew();
message.Status = MessageStatus.Processing;
await messageStore.UpdateAsync(message, _cts.Token);
message.StateName = StateName.Processing;
await messageStore.UpdateSentMessageAsync(message);
var jobResult = ExecuteJob(message.Topic, message.Payload);
var jobResult = ExecuteJob(message.KeyName, message.Content);
sp.Stop();
......@@ -100,14 +100,15 @@ namespace DotNetCore.CAP.Kafka
}
else
{
message.Status = MessageStatus.Successed;
await messageStore.UpdateAsync(message, _cts.Token);
//await messageStore.DeleteAsync(message, _cts.Token);
//TODO ： the state will be deleted when release.
message.StateName = StateName.Succeeded;
await messageStore.UpdateSentMessageAsync(message);
_logger.JobExecuted(sp.Elapsed.TotalSeconds);
}
}
}
catch (Exception )
catch (Exception)
{
return false;
}
......
src/DotNetCore.CAP.Kafka/KafkaConsumerClient.cs
View file @ 9b67cc4d
......@@ -14,7 +14,7 @@ namespace DotNetCore.CAP.Kafka
private Consumer<Null, string> _consumerClient;
public event EventHandler<DeliverMessage> MessageReceieved;
public event EventHandler<MessageBase> MessageReceieved;
public IDeserializer<string> StringDeserializer { get; set; }
......@@ -69,11 +69,10 @@ namespace DotNetCore.CAP.Kafka
private void ConsumerClient_OnMessage(object sender, Message<Null, string> e)
{
var message = new DeliverMessage
var message = new MessageBase
{
MessageKey = e.Topic,
Value = e.Value,
Body = Encoding.UTF8.GetBytes(e.Value)
KeyName = e.Topic,
Content = e.Value
};
MessageReceieved?.Invoke(sender, message);
}
......
src/DotNetCore.CAP.Kafka/KafkaTopicAttribute.cs
View file @ 9b67cc4d
......@@ -11,7 +11,8 @@ namespace DotNetCore.CAP.Kafka
: this(topicName, partition, 0) { }
public KafkaTopicAttribute(string topicName, int partition, long offset)
: base(topicName) {
: base(topicName)
{
Offset = offset;
Partition = partition;
}
......@@ -24,7 +25,8 @@ namespace DotNetCore.CAP.Kafka
public bool IsOffset { get { return Offset == 0; } }
public override string ToString() {
public override string ToString()
{
return Name;
}
}
......
src/DotNetCore.CAP.RabbitMQ/RabbitMQConsumerClient.cs
View file @ 9b67cc4d
......@@ -19,7 +19,7 @@ namespace DotNetCore.CAP.RabbitMQ
private string _queueName;
public event EventHandler<DeliverMessage> MessageReceieved;
public event EventHandler<MessageBase> MessageReceieved;
public RabbitMQConsumerClient(string exchange, string hostName)
{
......@@ -65,11 +65,10 @@ namespace DotNetCore.CAP.RabbitMQ
private void OnConsumerReceived(object sender, BasicDeliverEventArgs e)
{
var message = new DeliverMessage
var message = new MessageBase
{
MessageKey = e.RoutingKey,
Body = e.Body,
Value = Encoding.UTF8.GetString(e.Body)
KeyName = e.RoutingKey,
Content = Encoding.UTF8.GetString(e.Body)
};
MessageReceieved?.Invoke(sender, message);
}
......
src/DotNetCore.CAP.RabbitMQ/RabbitMQProducerClient.cs
View file @ 9b67cc4d
......@@ -38,7 +38,8 @@ namespace DotNetCore.CAP.RabbitMQ
}
}
public Task SendAsync<T>(string topic, T contentObj) {
public Task SendAsync<T>(string topic, T contentObj)
{
throw new NotImplementedException();
}
}
......
src/DotNetCore.CAP/Abstractions/ConsumerContext.cs
View file @ 9b67cc4d
......@@ -13,7 +13,7 @@ namespace DotNetCore.CAP.Abstractions
/// </summary>
/// <param name="descriptor">consumer method descriptor. </param>
/// <param name="message"> reveied message.</param>
public ConsumerContext(ConsumerExecutorDescriptor descriptor, DeliverMessage message)
public ConsumerContext(ConsumerExecutorDescriptor descriptor, MessageBase message)
{
ConsumerDescriptor = descriptor ?? throw new ArgumentNullException(nameof(descriptor));
DeliverMessage = message ?? throw new ArgumentNullException(nameof(message));
......@@ -27,6 +27,6 @@ namespace DotNetCore.CAP.Abstractions
/// <summary>
/// consumer reveived message.
/// </summary>
public DeliverMessage DeliverMessage { get; set; }
public MessageBase DeliverMessage { get; set; }
}
}
\ No newline at end of file
src/DotNetCore.CAP/Abstractions/IConsumerServiceSelector.cs
View file @ 9b67cc4d
using System.Collections.Generic;
using DotNetCore.CAP.Abstractions;
namespace DotNetCore.CAP.Abstractions
{
......@@ -18,7 +17,7 @@ namespace DotNetCore.CAP.Abstractions
/// <summary>
/// Selects the best <see cref="ConsumerExecutorDescriptor"/> candidate from <paramref name="candidates"/> for the
/// current message associated with <paramref name="context"/>.
/// current message associated.
/// </summary>
/// <param name="key">topic or exchange router key.</param>
/// <param name="candidates">the set of <see cref="ConsumerExecutorDescriptor"/> candidates.</param>
......
src/DotNetCore.CAP/ICapMessageStore.cs
View file @ 9b67cc4d
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks;
using DotNetCore.CAP.Infrastructure;
namespace DotNetCore.CAP
......@@ -11,49 +10,40 @@ namespace DotNetCore.CAP
public interface ICapMessageStore
{
/// <summary>
/// Finds and returns a message, if any, who has the specified <paramref name="messageId"/>.
/// Creates a new message in a store as an asynchronous operation.
/// </summary>
/// <param name="messageId">The message ID to search for.</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> used to propagate notifications that the operation should be canceled.</param>
/// <returns>
/// The <see cref="Task"/> that represents the asynchronous operation, containing the message matching the specified <paramref name="messageId"/> if it exists.
/// </returns>
Task<ConsistencyMessage> FindByIdAsync(string messageId, CancellationToken cancellationToken);
/// <param name="message">The message to create in the store.</param>
Task<OperateResult> StoreSentMessageAsync(CapSentMessage message);
/// <summary>
/// Creates a new message in a store as an asynchronous operation.
/// Fetches the next message to be executed.
/// </summary>
/// <param name="message">The message to create in the store.</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> used to propagate notifications that the operation should be canceled.</param>
/// <returns>A <see cref="Task{TResult}"/> that represents the <see cref="OperateResult"/> of the asynchronous query.</returns>
Task<OperateResult> CreateAsync(ConsistencyMessage message, CancellationToken cancellationToken);
/// <returns></returns>
Task<CapSentMessage> GetNextSentMessageToBeEnqueuedAsync();
/// <summary>
/// Updates a message in a store as an asynchronous operation.
/// </summary>
/// <param name="message">The message to update in the store.</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> used to propagate notifications that the operation should be canceled.</param>
/// <returns>A <see cref="Task{TResult}"/> that represents the <see cref="OperateResult"/> of the asynchronous query.</returns>
Task<OperateResult> UpdateAsync(ConsistencyMessage message, CancellationToken cancellationToken);
Task<OperateResult> UpdateSentMessageAsync(CapSentMessage message);
/// <summary>
/// Deletes a message from the store as an asynchronous operation.
/// </summary>
/// <param name="message">The message to delete in the store.</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> used to propagate notifications that the operation should be canceled.</param>
/// <returns>A <see cref="Task{TResult}"/> that represents the <see cref="OperateResult"/> of the asynchronous query.</returns>
Task<OperateResult> DeleteAsync(ConsistencyMessage message, CancellationToken cancellationToken);
Task<OperateResult> RemoveSentMessageAsync(CapSentMessage message);
/// <summary>
/// Gets the ID for a message from the store as an asynchronous operation.
/// Creates a new message in a store as an asynchronous operation.
/// </summary>
/// <param name="message">The message whose ID should be returned.</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> used to propagate notifications that the operation should be canceled.</param>
/// <returns>A <see cref="Task{TResult}"/> that contains the ID of the message.</returns>
Task<string> GeConsistencyMessageIdAsync(ConsistencyMessage message, CancellationToken cancellationToken);
/// <param name="message"></param>
/// <returns></returns>
Task<OperateResult> StoreReceivedMessageAsync(CapReceivedMessage message);
Task<ConsistencyMessage> GetFirstEnqueuedMessageAsync(CancellationToken cancellationToken);
// void ChangeState(ConsistencyMessage message, MessageStatus status);
/// <summary>
/// Updates a message in a store as an asynchronous operation.
/// </summary>
/// <param name="message">The message to update in the store.</param>
Task<OperateResult> UpdateReceivedMessageAsync(CapReceivedMessage message);
}
}
\ No newline at end of file
src/DotNetCore.CAP/ICapProducerService.Default.cs
View file @ 9b67cc4d
using System;
using System.Threading;
using System.Threading.Tasks;
using DotNetCore.CAP.Infrastructure;
using Microsoft.Extensions.Logging;
......@@ -13,7 +12,6 @@ namespace DotNetCore.CAP
{
private readonly ICapMessageStore _store;
private readonly ILogger _logger;
private readonly CancellationTokenSource _cts;
public DefaultProducerService(
ICapMessageStore store,
......@@ -21,7 +19,6 @@ namespace DotNetCore.CAP
{
_store = store;
_logger = logger;
_cts = new CancellationTokenSource();
}
public Task SendAsync(string topic, string content)
......@@ -45,20 +42,17 @@ namespace DotNetCore.CAP
private async Task StoreMessage(string topic, string content)
{
var message = new ConsistencyMessage
var message = new CapSentMessage
{
Topic = topic,
Payload = content
KeyName = topic,
Content = content
};
await _store.CreateAsync(message, _cts.Token);
await _store.StoreSentMessageAsync(message);
WaitHandleEx.PulseEvent.Set();
if (_logger.IsEnabled(LogLevel.Debug))
{
_logger.LogDebug("Enqueuing a topic to be store. topic:{topic}, content:{content}", topic, content);
}
_logger.EnqueuingMessage(topic, content);
}
}
}
\ No newline at end of file
src/DotNetCore.CAP/IConsumerClient.cs
View file @ 9b67cc4d
using System;
using System.Collections.Generic;
using System.Text;
using DotNetCore.CAP.Infrastructure;
namespace DotNetCore.CAP
......@@ -16,6 +14,6 @@ namespace DotNetCore.CAP
void Listening(TimeSpan timeout);
event EventHandler<DeliverMessage> MessageReceieved;
event EventHandler<MessageBase> MessageReceieved;
}
}
\ No newline at end of file
src/DotNetCore.CAP/IConsumerClientFactory.cs
View file @ 9b67cc4d
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
namespace DotNetCore.CAP
namespace DotNetCore.CAP
{
/// <summary>
/// Consumer client factory to create consumer client instance.
......
src/DotNetCore.CAP/IConsumerHandler.Default.cs
View file @ 9b67cc4d
using System;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using DotNetCore.CAP.Abstractions;
......@@ -24,7 +23,7 @@ namespace DotNetCore.CAP
private readonly ICapMessageStore _messageStore;
private readonly CancellationTokenSource _cts;
public event EventHandler<ConsistencyMessage> MessageReceieved;
public event EventHandler<CapMessage> MessageReceieved;
private CapStartContext _context;
private Task _compositeTask;
......@@ -37,7 +36,8 @@ namespace DotNetCore.CAP
ILoggerFactory loggerFactory,
ICapMessageStore messageStore,
MethodMatcherCache selector,
IOptions<CapOptions> options) {
IOptions<CapOptions> options)
{
_selector = selector;
_logger = loggerFactory.CreateLogger<ConsumerHandler>();
_loggerFactory = loggerFactory;
......@@ -49,23 +49,29 @@ namespace DotNetCore.CAP
_cts = new CancellationTokenSource();
}
protected virtual void OnMessageReceieved(ConsistencyMessage message) {
protected virtual void OnMessageReceieved(CapMessage message)
{
MessageReceieved?.Invoke(this, message);
}
public void Start() {
public void Start()
{
_context = new CapStartContext(_serviceProvider, _cts.Token);
var matchs = _selector.GetCandidatesMethods(_context);
var groupingMatchs = matchs.GroupBy(x => x.Value.Attribute.GroupOrExchange);
foreach (var matchGroup in groupingMatchs) {
Task.Factory.StartNew(() => {
using (var client = _consumerClientFactory.Create(matchGroup.Key, _options.BrokerUrlList)) {
foreach (var matchGroup in groupingMatchs)
{
Task.Factory.StartNew(() =>
{
using (var client = _consumerClientFactory.Create(matchGroup.Key, _options.BrokerUrlList))
{
client.MessageReceieved += OnMessageReceieved;
foreach (var item in matchGroup) {
foreach (var item in matchGroup)
{
client.Subscribe(item.Key);
}
......@@ -76,19 +82,17 @@ namespace DotNetCore.CAP
_compositeTask = Task.CompletedTask;
}
public virtual void OnMessageReceieved(object sender, DeliverMessage message) {
var consistencyMessage = new ConsistencyMessage() {
Topic = message.MessageKey,
Payload = "Reveived:" + Encoding.UTF8.GetString(message.Body),
Status = MessageStatus.Received
};
public virtual void OnMessageReceieved(object sender, MessageBase message)
{
var capMessage = new CapReceivedMessage(message);
_logger.LogInformation("message receieved message topic name: " + consistencyMessage.Id);
_logger.LogInformation("message receieved message topic name: " + capMessage.Id);
_messageStore.CreateAsync(consistencyMessage, _cts.Token).Wait();
_messageStore.StoreReceivedMessageAsync(capMessage).Wait();
try {
var executeDescriptor = _selector.GetTopicExector(message.MessageKey);
try
{
var executeDescriptor = _selector.GetTopicExector(message.KeyName);
var consumerContext = new ConsumerContext(executeDescriptor, message);
......@@ -96,15 +100,18 @@ namespace DotNetCore.CAP
invoker.InvokeAsync();
_messageStore.UpdateAsync(consistencyMessage, _cts.Token).Wait();
_messageStore.UpdateReceivedMessageAsync(capMessage).Wait();
}
catch (Exception ex) {
catch (Exception ex)
{
_logger.LogError("exception raised when excute method : " + ex.Message);
}
}
public void Dispose() {
if (_disposed) {
public void Dispose()
{
if (_disposed)
{
return;
}
_disposed = true;
......@@ -112,12 +119,15 @@ namespace DotNetCore.CAP
_logger.ServerShuttingDown();
_cts.Cancel();
try {
try
{
_compositeTask.Wait((int)TimeSpan.FromSeconds(60).TotalMilliseconds);
}
catch (AggregateException ex) {
catch (AggregateException ex)
{
var innerEx = ex.InnerExceptions[0];
if (!(innerEx is OperationCanceledException)) {
if (!(innerEx is OperationCanceledException))
{
_logger.ExpectedOperationCanceledException(innerEx);
}
}
......
src/DotNetCore.CAP/Internal/IConsumerInvoker.Default.cs
View file @ 9b67cc4d
using System;
using System.Text;
using System.Threading.Tasks;
using DotNetCore.CAP.Abstractions;
using DotNetCore.CAP.Abstractions.ModelBinding;
......@@ -41,13 +40,13 @@ namespace DotNetCore.CAP.Internal
{
var obj = ActivatorUtilities.GetServiceOrCreateInstance(_serviceProvider, _consumerContext.ConsumerDescriptor.ImplTypeInfo.AsType());
var bodyString = Encoding.UTF8.GetString(_consumerContext.DeliverMessage.Body);
var value = _consumerContext.DeliverMessage.Content;
if (_executor.MethodParameters.Length > 0)
{
var firstParameter = _executor.MethodParameters[0];
var bindingContext = ModelBindingContext.CreateBindingContext(bodyString,
var bindingContext = ModelBindingContext.CreateBindingContext(value,
firstParameter.Name, firstParameter.ParameterType);
_modelBinder.BindModelAsync(bindingContext);
......
src/DotNetCore.CAP/Internal/IConsumerServiceSelector.Default.cs
View file @ 9b67cc4d
......@@ -24,7 +24,8 @@ namespace DotNetCore.CAP.Internal
}
/// <summary>
///
/// Selects the best <see cref="ConsumerExecutorDescriptor"/> candidate from <paramref name="candidates"/> for the
/// current message associated.
/// </summary>
/// <param name="key"></param>
/// <param name="executeDescriptor"></param>
......
src/DotNetCore.CAP/Internal/MethodMatcherCache.cs
View file @ 9b67cc4d
using System;
using System.Collections.Concurrent;
using DotNetCore.CAP.Abstractions;
using DotNetCore.CAP.Infrastructure;
namespace DotNetCore.CAP.Internal
{
......
src/DotNetCore.CAP/LoggerExtensions.cs
View file @ 9b67cc4d
......@@ -17,6 +17,8 @@ namespace DotNetCore.CAP
private static Action<ILogger, string, double, Exception> _cronJobExecuted;
private static Action<ILogger, string, Exception> _cronJobFailed;
private static Action<ILogger, string, string, Exception> _enqueuingMessage;
static LoggerExtensions()
{
_serverStarting = LoggerMessage.Define<int, int>(
......@@ -53,6 +55,16 @@ namespace DotNetCore.CAP
LogLevel.Warning,
4,
"Cron job '{jobName}' failed to execute.");
_enqueuingMessage = LoggerMessage.Define<string, string>(
LogLevel.Debug,
2,
"Enqueuing a topic to the store. NameKey: {NameKey}. Content: {Content}");
}
public static void EnqueuingMessage(this ILogger logger, string nameKey, string content)
{
_enqueuingMessage(logger, nameKey, content, null);
}
public static void ServerStarting(this ILogger logger, int machineProcessorCount, int processorCount)
......
test/DotNetCore.CAP.Test/CAP.BuilderTest.cs
View file @ 9b67cc4d
......@@ -54,6 +54,10 @@ namespace DotNetCore.CAP.Test
{
throw new NotImplementedException();
}
public Task SendAsync<T>(string topic, T contentObj) {
throw new NotImplementedException();
}
}
......@@ -67,32 +71,32 @@ namespace DotNetCore.CAP.Test
private class MyMessageStore : ICapMessageStore
{
public Task<OperateResult> CreateAsync(ConsistencyMessage message, CancellationToken cancellationToken)
public Task<CapSentMessage> GetNextSentMessageToBeEnqueuedAsync()
{
throw new NotImplementedException();
}
public Task<OperateResult> DeleteAsync(ConsistencyMessage message, CancellationToken cancellationToken)
public Task<OperateResult> RemoveSentMessageAsync(CapSentMessage message)
{
throw new NotImplementedException();
}
public Task<ConsistencyMessage> FindByIdAsync(string messageId, CancellationToken cancellationToken)
public Task<OperateResult> StoreReceivedMessageAsync(CapReceivedMessage message)
{
throw new NotImplementedException();
}
public Task<string> GeConsistencyMessageIdAsync(ConsistencyMessage message, CancellationToken cancellationToken)
public Task<OperateResult> StoreSentMessageAsync(CapSentMessage message)
{
throw new NotImplementedException();
}
public Task<ConsistencyMessage> GetFirstEnqueuedMessageAsync(CancellationToken cancellationToken)
public Task<OperateResult> UpdateReceivedMessageAsync(CapReceivedMessage message)
{
throw new NotImplementedException();
}
public Task<OperateResult> UpdateAsync(ConsistencyMessage message, CancellationToken cancellationToken)
public Task<OperateResult> UpdateSentMessageAsync(CapSentMessage message)
{
throw new NotImplementedException();
}
......
test/DotNetCore.CAP.Test/NoopMessageStore.cs
View file @ 9b67cc4d
......@@ -7,42 +7,32 @@ namespace DotNetCore.CAP.Test
{
public class NoopMessageStore : ICapMessageStore
{
public Task<OperateResult> CreateAsync(ConsistencyMessage message, CancellationToken cancellationToken)
public Task<CapSentMessage> GetNextSentMessageToBeEnqueuedAsync()
{
throw new NotImplementedException();
}
public Task<OperateResult> DeleteAsync(ConsistencyMessage message, CancellationToken cancellationToken)
public Task<OperateResult> RemoveSentMessageAsync(CapSentMessage message)
{
throw new NotImplementedException();
}
public void Dispose()
public Task<OperateResult> StoreReceivedMessageAsync(CapReceivedMessage message)
{
throw new NotImplementedException();
}
public Task<ConsistencyMessage> FindByIdAsync(string messageId, CancellationToken cancellationToken)
public Task<OperateResult> StoreSentMessageAsync(CapSentMessage message)
{
throw new NotImplementedException();
}
public Task<string> GeConsistencyMessageIdAsync(ConsistencyMessage message, CancellationToken cancellationToken)
public Task<OperateResult> UpdateReceivedMessageAsync(CapReceivedMessage message)
{
throw new NotImplementedException();
}
public Task<ConsistencyMessage> GetFirstEnqueuedMessageAsync(CancellationToken cancellationToken)
{
throw new NotImplementedException();
}
public Task<string> GetMessageIdAsync(ConsistencyMessage message, CancellationToken cancellationToken)
{
throw new NotImplementedException();
}
public Task<OperateResult> UpdateAsync(ConsistencyMessage message, CancellationToken cancellationToken)
public Task<OperateResult> UpdateSentMessageAsync(CapSentMessage message)
{
throw new NotImplementedException();
}
......
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