Skip to content

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

refactor storage module.

parent 9371e559
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 358 additions and 563 deletions
src/DotNetCore.CAP.EntityFrameworkCore/CapMessageStore.cs deleted 100644 → 0
View file @ 9371e559
using System;
using System.Threading.Tasks;
using DotNetCore.CAP.Models;
using Microsoft.EntityFrameworkCore;
namespace DotNetCore.CAP.EntityFrameworkCore
{
/// <summary>
/// Represents a new instance of a persistence store for the specified message types.
/// </summary>
/// <typeparam name="TContext">The type of the data context class used to access the store.</typeparam>
public class CapMessageStore<TContext> : ICapMessageStore where TContext : DbContext
{
/// <summary>
/// Constructs a new instance of <see cref="TContext"/>.
/// </summary>
/// <param name="context">The <see cref="DbContext"/>.</param>
public CapMessageStore(TContext context)
{
Context = context ?? throw new ArgumentNullException(nameof(context));
}
public TContext Context { get; private set; }
private DbSet<CapSentMessage> SentMessages => Context.Set<CapSentMessage>();
/// <summary>
/// Creates the specified <paramref name="message"/> in the cap message store.
/// </summary>
/// <param name="message">The message to create.</param>
public async Task<OperateResult> StoreSentMessageAsync(CapSentMessage message)
{
if (message == null) throw new ArgumentNullException(nameof(message));
Context.Add(message);
await Context.SaveChangesAsync();
return OperateResult.Success;
}
}
}
\ No newline at end of file
src/DotNetCore.CAP.EntityFrameworkCore/EFStorageConnection.cs
View file @ e1ec8eac
......@@ -56,17 +56,24 @@ OUTPUT DELETED.MessageId,DELETED.[Type];";
var sql = $@"
SELECT TOP (1) *
FROM [{_options.Schema}].[{nameof(CapDbContext.CapSentMessages)}] WITH (readpast)
WHERE StateName = '{StatusName.Enqueued}'";
WHERE StatusName = '{StatusName.Scheduled}'";
var connection = _context.GetDbConnection();
var message = (await connection.QueryAsync<CapSentMessage>(sql)).FirstOrDefault();
try
{
var connection = _context.GetDbConnection();
var message = (await connection.QueryAsync<CapSentMessage>(sql)).FirstOrDefault();
if (message != null)
if (message != null)
{
_context.Attach(message);
}
return message;
}
catch (Exception ex)
{
_context.Attach(message);
throw;
}
return message;
}
// CapReceviedMessage
......
src/DotNetCore.CAP.RabbitMQ/IMessageJobProcessor.RabbitJobProcessor.cs
View file @ e1ec8eac
......@@ -46,7 +46,7 @@ namespace DotNetCore.CAP.RabbitMQ
public Task ProcessAsync(ProcessingContext context)
{
if (context == null) throw new ArgumentNullException(nameof(context));
System.Diagnostics.Debug.WriteLine("RabbitMQ Processor 执行: " + DateTime.Now);
context.ThrowIfStopping();
return ProcessCoreAsync(context);
}
......@@ -64,10 +64,9 @@ namespace DotNetCore.CAP.RabbitMQ
if (!worked)
{
var token = GetTokenToWaitOn(context);
}
await WaitHandleEx.WaitAnyAsync(WaitHandleEx.SentPulseEvent,
context.CancellationToken.WaitHandle, _pollingDelay);
await WaitHandleEx.WaitAnyAsync(WaitHandleEx.SentPulseEvent, token.WaitHandle, _pollingDelay);
}
}
finally
{
......@@ -92,6 +91,7 @@ namespace DotNetCore.CAP.RabbitMQ
{
using (fetched)
{
var message = await connection.GetSentMessageAsync(fetched.MessageId);
try
{
......
src/DotNetCore.CAP.RabbitMQ/RabbitMQConsumerClient.cs
View file @ e1ec8eac
......@@ -56,7 +56,7 @@ namespace DotNetCore.CAP.RabbitMQ
_channel.BasicConsume(_queueName, false, consumer);
while (true)
{
Task.Delay(timeout);
Task.Delay(timeout).Wait();
}
}
......
src/DotNetCore.CAP/CAP.Builder.cs
View file @ e1ec8eac
......@@ -46,27 +46,6 @@ namespace DotNetCore.CAP
return this;
}
/// <summary>
/// Add an <see cref="ICapMessageStore"/> .
/// </summary>
/// <typeparam name="T">The type for the <see cref="ICapMessageStore"/> to add. </typeparam>
/// <returns>The current <see cref="CapBuilder"/> instance.</returns>
public virtual CapBuilder AddMessageStore<T>()
where T : class, ICapMessageStore
{
return AddScoped(typeof(ICapMessageStore), typeof(T));
}
/// <summary>
/// Add an <see cref="IJob"/> for process <see cref="CapJob"/>.
/// </summary>
/// <typeparam name="T">The type of the job.</typeparam>
public virtual CapBuilder AddJobs<T>()
where T : class, IJob
{
return AddSingleton<IJob, T>();
}
/// <summary>
/// Add an <see cref="ICapPublisher"/>.
/// </summary>
......
src/DotNetCore.CAP/ICapMessageStore.cs deleted 100644 → 0
View file @ 9371e559
using System.Threading.Tasks;
using DotNetCore.CAP.Infrastructure;
using DotNetCore.CAP.Models;
namespace DotNetCore.CAP
{
/// <summary>
/// Provides an abstraction for a store which manages CAP message.
/// </summary>
public interface ICapMessageStore
{
/// <summary>
/// Creates a new message in a store as an asynchronous operation.
/// </summary>
/// <param name="message">The message to create in the store.</param>
Task<OperateResult> StoreSentMessageAsync(CapSentMessage message);
}
}
\ No newline at end of file
src/DotNetCore.CAP/ICapPublisher.Default.cs
View file @ e1ec8eac
......@@ -11,14 +11,10 @@ namespace DotNetCore.CAP
/// </summary>
public class DefaultCapPublisher : ICapPublisher
{
private readonly ICapMessageStore _store;
private readonly ILogger _logger;
private readonly ILogger _logger;
public DefaultCapPublisher(
ICapMessageStore store,
ILogger<DefaultCapPublisher> logger)
{
_store = store;
public DefaultCapPublisher( ILogger<DefaultCapPublisher> logger)
{
_logger = logger;
}
......@@ -50,11 +46,11 @@ namespace DotNetCore.CAP
StatusName = StatusName.Enqueued
};
await _store.StoreSentMessageAsync(message);
//await _store.StoreSentMessageAsync(message);
WaitHandleEx.PulseEvent.Set();
// WaitHandleEx.PulseEvent.Set();
_logger.EnqueuingSentMessage(topic, content);
// _logger.EnqueuingSentMessage(topic, content);
}
}
}
\ No newline at end of file
src/DotNetCore.CAP/IConsumerHandler.Default.cs
View file @ e1ec8eac
......@@ -66,7 +66,7 @@ namespace DotNetCore.CAP
client.Listening(_pollingDelay);
}
}, _cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Current);
}, _cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
}
_compositeTask = Task.CompletedTask;
}
......@@ -106,7 +106,7 @@ namespace DotNetCore.CAP
{
var receviedMessage = StoreMessage(scope, message);
client.Commit();
// ProcessMessage(scope, receviedMessage);
// ProcessMessage(scope, receviedMessage);
}
};
}
......
src/DotNetCore.CAP/Infrastructure/StatusName.cs
View file @ e1ec8eac
......@@ -9,6 +9,7 @@ namespace DotNetCore.CAP.Infrastructure
/// </summary>
public struct StatusName
{
public const string Scheduled = nameof(Scheduled);
public const string Enqueued = nameof(Enqueued);
public const string Processing = nameof(Processing);
public const string Succeeded = nameof(Succeeded);
......
src/DotNetCore.CAP/Job/IJob.CapJob.cs deleted 100644 → 0
View file @ 9371e559
using System;
using System.Threading.Tasks;
using DotNetCore.CAP.Internal;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.DependencyInjection;
using DotNetCore.CAP.Abstractions;
using DotNetCore.CAP.Infrastructure;
namespace DotNetCore.CAP.Job
{
public class CapJob : IJob
{
private readonly MethodMatcherCache _selector;
private readonly IConsumerInvokerFactory _consumerInvokerFactory;
private readonly IServiceProvider _serviceProvider;
private readonly ILogger<CapJob> _logger;
private readonly ICapMessageStore _messageStore;
public CapJob(
ILogger<CapJob> logger,
IServiceProvider serviceProvider,
IConsumerInvokerFactory consumerInvokerFactory,
ICapMessageStore messageStore,
MethodMatcherCache selector)
{
_logger = logger;
_serviceProvider = serviceProvider;
_consumerInvokerFactory = consumerInvokerFactory;
_messageStore = messageStore;
_selector = selector;
}
public async Task ExecuteAsync()
{
//var groupedCandidates = _selector.GetCandidatesMethodsOfGroupNameGrouped(_serviceProvider);
//using (var scope = _serviceProvider.CreateScope())
//{
// var provider = scope.ServiceProvider;
// var messageStore = provider.GetService<ICapMessageStore>();
// var nextReceivedMessage = await messageStore.GetNextReceivedMessageToBeExcuted();
// if (nextReceivedMessage != null && groupedCandidates.ContainsKey(nextReceivedMessage.Group))
// {
// try
// {
// await messageStore.ChangeReceivedMessageStateAsync(nextReceivedMessage, StatusName.Processing);
// // If there are multiple consumers in the same group, we will take the first
// var executeDescriptor = groupedCandidates[nextReceivedMessage.Group][0];
// var consumerContext = new ConsumerContext(executeDescriptor, nextReceivedMessage.ToMessageContext());
// var invoker = _consumerInvokerFactory.CreateInvoker(consumerContext);
// await invoker.InvokeAsync();
// await messageStore.ChangeReceivedMessageStateAsync(nextReceivedMessage, StatusName.Succeeded);
// }
// catch (Exception ex)
// {
// _logger.ReceivedMessageRetryExecutingFailed(nextReceivedMessage.KeyName, ex);
// }
// }
//}
}
}
}
\ No newline at end of file
src/DotNetCore.CAP/Job/IJob.cs deleted 100644 → 0
View file @ 9371e559
using System.Threading.Tasks;
namespace DotNetCore.CAP.Job
{
public interface IJob
{
/// <summary>
/// Executes the job.
/// </summary>
Task ExecuteAsync();
}
}
\ No newline at end of file
src/DotNetCore.CAP/Job/IJobProcessor.InfiniteRetry.cs
View file @ e1ec8eac
using System;
using System.Diagnostics;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
......@@ -23,6 +24,7 @@ namespace DotNetCore.CAP.Job
{
while (!context.IsStopping)
{
Debug.WriteLine("InfiniteRetryProcessor在开线程：" + _inner.ToString() + " : " + DateTime.Now);
try
{
await _inner.ProcessAsync(context);
......
src/DotNetCore.CAP/Job/IJobProcessor.JobQueuer.cs
View file @ e1ec8eac
......@@ -46,7 +46,8 @@ namespace DotNetCore.CAP.Job
(sentMessage = await connection.GetNextSentMessageToBeEnqueuedAsync()) != null)
{
var state = new EnqueuedState();
System.Diagnostics.Debug.WriteLine("JobQueuer 执行 内部循环: " + DateTime.Now);
var state = new EnqueuedState();
using (var transaction = connection.CreateTransaction())
{
......@@ -56,7 +57,8 @@ namespace DotNetCore.CAP.Job
}
}
context.ThrowIfStopping();
System.Diagnostics.Debug.WriteLine("JobQueuer 执行: " + DateTime.Now);
context.ThrowIfStopping();
WaitHandleEx.SentPulseEvent.Set();
await WaitHandleEx.WaitAnyAsync(WaitHandleEx.QueuePulseEvent,
......
src/DotNetCore.CAP/Job/IProcessingServer.Job.cs
View file @ e1ec8eac
......@@ -117,7 +117,7 @@ namespace DotNetCore.CAP.Job
}
returnedProcessors.Add(_provider.GetService<JobQueuer>());
returnedProcessors.Add(_provider.GetService<IAdditionalProcessor>());
//returnedProcessors.Add(_provider.GetService<IAdditionalProcessor>());
return returnedProcessors.ToArray();
}
......
test/DotNetCore.CAP.Test/CAP.BuilderTest.cs
View file @ e1ec8eac
......@@ -10,29 +10,29 @@ namespace DotNetCore.CAP.Test
{
public class CapBuilderTest
{
[Fact]
public void CanOverrideMessageStore()
{
var services = new ServiceCollection();
services.AddCap().AddMessageStore<MyMessageStore>();
//[Fact]
//public void CanOverrideMessageStore()
//{
// var services = new ServiceCollection();
// services.AddCap().AddMessageStore<MyMessageStore>();
var thingy = services.BuildServiceProvider()
.GetRequiredService<ICapMessageStore>() as MyMessageStore;
// var thingy = services.BuildServiceProvider()
// .GetRequiredService<ICapMessageStore>() as MyMessageStore;
Assert.NotNull(thingy);
}
// Assert.NotNull(thingy);
////}
[Fact]
public void CanOverrideJobs()
{
var services = new ServiceCollection();
services.AddCap().AddJobs<MyJobTest>();
//[Fact]
//public void CanOverrideJobs()
//{
// var services = new ServiceCollection();
// services.AddCap().AddJobs<MyJobTest>();
var thingy = services.BuildServiceProvider()
.GetRequiredService<IJob>() as MyJobTest;
// var thingy = services.BuildServiceProvider()
// .GetRequiredService<IJob>() as MyJobTest;
Assert.NotNull(thingy);
}
// Assert.NotNull(thingy);
//}
[Fact]
public void CanOverrideProducerService()
......@@ -58,65 +58,6 @@ namespace DotNetCore.CAP.Test
{
throw new NotImplementedException();
}
}
private class MyJobTest : IJob
{
public Task ExecuteAsync()
{
throw new NotImplementedException();
}
}
private class MyMessageStore : ICapMessageStore
{
public Task<OperateResult> ChangeReceivedMessageStateAsync(CapReceivedMessage message, string statusName,
bool autoSaveChanges = true)
{
throw new NotImplementedException();
}
public Task<OperateResult> ChangeSentMessageStateAsync(CapSentMessage message, string statusName,
bool autoSaveChanges = true)
{
throw new NotImplementedException();
}
public Task<CapReceivedMessage> GetNextReceivedMessageToBeExcuted()
{
throw new NotImplementedException();
}
public Task<CapSentMessage> GetNextSentMessageToBeEnqueuedAsync()
{
throw new NotImplementedException();
}
public Task<OperateResult> RemoveSentMessageAsync(CapSentMessage message)
{
throw new NotImplementedException();
}
public Task<OperateResult> StoreReceivedMessageAsync(CapReceivedMessage message)
{
throw new NotImplementedException();
}
public Task<OperateResult> StoreSentMessageAsync(CapSentMessage message)
{
throw new NotImplementedException();
}
public Task<OperateResult> UpdateReceivedMessageAsync(CapReceivedMessage message)
{
throw new NotImplementedException();
}
public Task<OperateResult> UpdateSentMessageAsync(CapSentMessage message)
{
throw new NotImplementedException();
}
}
}
}
}
\ No newline at end of file
test/DotNetCore.CAP.Test/NoopMessageStore.cs
View file @ e1ec8eac
......@@ -5,17 +5,17 @@ using DotNetCore.CAP.Models;
namespace DotNetCore.CAP.Test
{
public class NoopMessageStore : ICapMessageStore
{
public Task<OperateResult> ChangeReceivedMessageStateAsync(CapReceivedMessage message, string statusName,
bool autoSaveChanges = true)
{
throw new NotImplementedException();
}
//public class NoopMessageStore : ICapMessageStore
//{
// public Task<OperateResult> ChangeReceivedMessageStateAsync(CapReceivedMessage message, string statusName,
// bool autoSaveChanges = true)
// {
// throw new NotImplementedException();
// }
public Task<OperateResult> StoreSentMessageAsync(CapSentMessage message)
{
throw new NotImplementedException();
}
}
// public Task<OperateResult> StoreSentMessageAsync(CapSentMessage message)
// {
// throw new NotImplementedException();
// }
//}
}
\ No newline at end of file
test/Shared/MessageManagerTestBase.cs
View file @ e1ec8eac
using System;
using System.Threading.Tasks;
using DotNetCore.CAP.Infrastructure;
using DotNetCore.CAP.Models;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Xunit;
namespace DotNetCore.CAP.Test
{
public abstract class MessageManagerTestBase
{
private const string NullValue = "(null)";
protected virtual bool ShouldSkipDbTests()
{
return false;
}
protected virtual void SetupMessageServices(IServiceCollection services, object context = null)
{
services.AddSingleton<IHttpContextAccessor, HttpContextAccessor>();
services.AddCap();
AddMessageStore(services, context);
services.AddSingleton<ILogger<ICapMessageStore>>(new TestLogger<ICapMessageStore>());
}
protected virtual ICapMessageStore CreateManager(object context = null, IServiceCollection services = null,
Action<IServiceCollection> configureServices = null)
{
if (services == null)
{
services = new ServiceCollection();
}
if (context == null)
{
context = CreateTestContext();
}
SetupMessageServices(services, context);
configureServices?.Invoke(services);
return services.BuildServiceProvider().GetService<ICapMessageStore>();
}
protected abstract object CreateTestContext();
protected abstract CapSentMessage CreateTestSentMessage(string content = "");
protected abstract CapReceivedMessage CreateTestReceivedMessage(string content = "");
protected abstract void AddMessageStore(IServiceCollection services, object context = null);
[Fact]
public async Task CanDeleteSentMessage()
{
if (ShouldSkipDbTests())
{
return;
}
var manager = CreateManager();
var message = CreateTestSentMessage();
var operateResult = await manager.StoreSentMessageAsync(message);
Assert.NotNull(operateResult);
Assert.True(operateResult.Succeeded);
// operateResult = await manager.RemoveSentMessageAsync(message);
// Assert.NotNull(operateResult);
// Assert.True(operateResult.Succeeded);
}
//[Fact]
//public async Task CanUpdateReceivedMessage()
//{
// if (ShouldSkipDbTests())
// {
// return;
// }
// var manager = CreateManager();
// var message = CreateTestReceivedMessage();
// // var operateResult = await manager.StoreReceivedMessageAsync(message);
// // Assert.NotNull(operateResult);
// // Assert.True(operateResult.Succeeded);
// // message.StatusName = StatusName.Processing;
// // operateResult = await manager.UpdateReceivedMessageAsync(message);
// // Assert.NotNull(operateResult);
// // Assert.True(operateResult.Succeeded);
//}
[Fact]
public async Task CanGetNextSendMessage()
{
if (ShouldSkipDbTests())
{
return;
}
var manager = CreateManager();
var message = CreateTestSentMessage();
var operateResult = await manager.StoreSentMessageAsync(message);
Assert.NotNull(operateResult);
Assert.True(operateResult.Succeeded);
// var storeMessage = await manager.GetNextSentMessageToBeEnqueuedAsync();
// Assert.Equal(message, storeMessage);
}
}
}
\ No newline at end of file
//using System;
//using System.Threading.Tasks;
//using DotNetCore.CAP.Infrastructure;
//using DotNetCore.CAP.Models;
//using Microsoft.AspNetCore.Http;
//using Microsoft.Extensions.DependencyInjection;
//using Microsoft.Extensions.Logging;
//using Xunit;
//namespace DotNetCore.CAP.Test
//{
// public abstract class MessageManagerTestBase
// {
// private const string NullValue = "(null)";
// protected virtual bool ShouldSkipDbTests()
// {
// return false;
// }
// protected virtual void SetupMessageServices(IServiceCollection services, object context = null)
// {
// services.AddSingleton<IHttpContextAccessor, HttpContextAccessor>();
// services.AddCap();
// AddMessageStore(services, context);
// services.AddSingleton<ILogger<ICapMessageStore>>(new TestLogger<ICapMessageStore>());
// }
// protected virtual ICapMessageStore CreateManager(object context = null, IServiceCollection services = null,
// Action<IServiceCollection> configureServices = null)
// {
// if (services == null)
// {
// services = new ServiceCollection();
// }
// if (context == null)
// {
// context = CreateTestContext();
// }
// SetupMessageServices(services, context);
// configureServices?.Invoke(services);
// return services.BuildServiceProvider().GetService<ICapMessageStore>();
// }
// protected abstract object CreateTestContext();
// protected abstract CapSentMessage CreateTestSentMessage(string content = "");
// protected abstract CapReceivedMessage CreateTestReceivedMessage(string content = "");
// protected abstract void AddMessageStore(IServiceCollection services, object context = null);
// [Fact]
// public async Task CanDeleteSentMessage()
// {
// if (ShouldSkipDbTests())
// {
// return;
// }
// var manager = CreateManager();
// var message = CreateTestSentMessage();
// var operateResult = await manager.StoreSentMessageAsync(message);
// Assert.NotNull(operateResult);
// Assert.True(operateResult.Succeeded);
// // operateResult = await manager.RemoveSentMessageAsync(message);
// // Assert.NotNull(operateResult);
// // Assert.True(operateResult.Succeeded);
// }
// //[Fact]
// //public async Task CanUpdateReceivedMessage()
// //{
// // if (ShouldSkipDbTests())
// // {
// // return;
// // }
// // var manager = CreateManager();
// // var message = CreateTestReceivedMessage();
// // // var operateResult = await manager.StoreReceivedMessageAsync(message);
// // // Assert.NotNull(operateResult);
// // // Assert.True(operateResult.Succeeded);
// // // message.StatusName = StatusName.Processing;
// // // operateResult = await manager.UpdateReceivedMessageAsync(message);
// // // Assert.NotNull(operateResult);
// // // Assert.True(operateResult.Succeeded);
// //}
// [Fact]
// public async Task CanGetNextSendMessage()
// {
// if (ShouldSkipDbTests())
// {
// return;
// }
// var manager = CreateManager();
// var message = CreateTestSentMessage();
// var operateResult = await manager.StoreSentMessageAsync(message);
// Assert.NotNull(operateResult);
// Assert.True(operateResult.Succeeded);
// // var storeMessage = await manager.GetNextSentMessageToBeEnqueuedAsync();
// // Assert.Equal(message, storeMessage);
// }
// }
//}
\ No newline at end of file
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