refactor

samples/Sample.Kafka/AppDbContext.cs

@@ -13,7 +13,8 @@ namespace Sample.Kafka
     {
         protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
         {
-            optionsBuilder.UseSqlServer("Server=192.168.2.206;Initial Catalog=Test;User Id=cmswuliu;Password=h7xY81agBn*Veiu3;MultipleActiveResultSets=True");
+            //optionsBuilder.UseSqlServer("Server=192.168.2.206;Initial Catalog=Test;User Id=cmswuliu;Password=h7xY81agBn*Veiu3;MultipleActiveResultSets=True");
+            optionsBuilder.UseSqlServer("Server=DESKTOP-M9R8T31;Initial Catalog=Test;User Id=sa;Password=P@ssw0rd;MultipleActiveResultSets=True");
         }
     }
 }

samples/Sample.Kafka/Controllers/ValuesController.cs

@@ -1,9 +1,10 @@
 using System;
 using System.Threading.Tasks;
 using DotNetCore.CAP;
-using DotNetCore.CAP.Kafka;
+using DotNetCore.CAP.RabbitMQ;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.AspNetCore.Mvc;
+using Dapper;
 
 namespace Sample.Kafka.Controllers
 {
@@ -11,10 +12,12 @@ namespace Sample.Kafka.Controllers
     public class ValuesController : Controller, ICapSubscribe
     {
         private readonly ICapPublisher _producer;
+        private readonly AppDbContext _dbContext ;
 
-        public ValuesController(ICapPublisher producer)
+        public ValuesController(ICapPublisher producer, AppDbContext dbContext)
         {
             _producer = producer;
+            _dbContext = dbContext;
         }
 
         [Route("/")]
@@ -33,11 +36,11 @@ namespace Sample.Kafka.Controllers
         }
 
         [Route("~/send")]
-        public async Task<IActionResult> SendTopic([FromServices] AppDbContext dbContext)
+        public async Task<IActionResult> SendTopic()
         {
-            using (var trans = dbContext.Database.BeginTransaction())
+            using (var trans = _dbContext.Database.BeginTransaction())
             {
-                await _producer.PublishAsync("zzwl.topic.finace.callBack", new Person { Name = "Test", Age = 11 });
+                await _producer.PublishAsync("zzwl.topic.finace.callBack","");
 
                 trans.Commit();
             }

samples/Sample.Kafka/Sample.Kafka.csproj

@@ -25,7 +25,6 @@
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\..\src\DotNetCore.CAP.EntityFrameworkCore\DotNetCore.CAP.EntityFrameworkCore.csproj" />
-    <ProjectReference Include="..\..\src\DotNetCore.CAP.Kafka\DotNetCore.CAP.Kafka.csproj" />
     <ProjectReference Include="..\..\src\DotNetCore.CAP.RabbitMQ\DotNetCore.CAP.RabbitMQ.csproj" />
     <ProjectReference Include="..\..\src\DotNetCore.CAP\DotNetCore.CAP.csproj" />
   </ItemGroup>

samples/Sample.Kafka/Startup.cs

@@ -28,13 +28,14 @@ namespace Sample.Kafka
 
             services.AddCap()
                     .AddEntityFrameworkStores<AppDbContext>(x=> {
-                        x.ConnectionString = "Server=192.168.2.206;Initial Catalog=Test;User Id=cmswuliu;Password=h7xY81agBn*Veiu3;MultipleActiveResultSets=True";
+                        //x.ConnectionString = "Server=192.168.2.206;Initial Catalog=Test;User Id=cmswuliu;Password=h7xY81agBn*Veiu3;MultipleActiveResultSets=True";
+                        x.ConnectionString = "Server=DESKTOP-M9R8T31;Initial Catalog=Test;User Id=sa;Password=P@ssw0rd;MultipleActiveResultSets=True";
                     })
                     .AddRabbitMQ(x =>
                     {
-                        x.HostName = "192.168.2.206";
-                        x.UserName = "admin";
-                        x.Password = "123123";
+                        x.HostName = "localhost";
+                       // x.UserName = "admin";
+                       // x.Password = "123123";
                     });
             //.AddKafka(x => x.Servers = "");
             

src/DotNetCore.CAP.EntityFrameworkCore/CAP.BuilderExtensions.cs

@@ -23,8 +23,10 @@ namespace Microsoft.Extensions.DependencyInjection
             builder.Services.AddSingleton<IStorage, EFStorage>();
             builder.Services.AddScoped<IStorageConnection, EFStorageConnection>();
 
-            builder.Services.AddTransient<IAdditionalProcessor, DefaultAdditionalProcessor>();
+            builder.Services.AddScoped<ICapPublisher, CapPublisher>();
 
+            builder.Services.AddTransient<IAdditionalProcessor, DefaultAdditionalProcessor>();
+            
             builder.Services.Configure(actionOptions);
             
             var sqlServerOptions = new SqlServerOptions();

src/DotNetCore.CAP.EntityFrameworkCore/CAP.EFOptions.cs

@@ -12,7 +12,7 @@ namespace DotNetCore.CAP.EntityFrameworkCore
 
         public EFOptions()
         {
-            ConnectionString = "Server=DESKTOP-M9R8T31;Initial Catalog=WebApp1;User Id=sa;Password=P@ssw0rd;MultipleActiveResultSets=True";
+            ConnectionString = "Server=DESKTOP-M9R8T31;Initial Catalog=Test;User Id=sa;Password=P@ssw0rd;MultipleActiveResultSets=True";
         }
 
         /// <summary>

src/DotNetCore.CAP.EntityFrameworkCore/CapDbContext.cs

@@ -47,7 +47,7 @@ namespace DotNetCore.CAP.EntityFrameworkCore
         /// </param>
         protected override void OnModelCreating(ModelBuilder modelBuilder)
         {
-            //_sqlServerOptions = new SqlServerOptions();
+            _sqlServerOptions = new SqlServerOptions();
             modelBuilder.HasDefaultSchema(_sqlServerOptions.Schema);
 
             modelBuilder.Entity<CapSentMessage>(b =>
@@ -67,7 +67,8 @@ namespace DotNetCore.CAP.EntityFrameworkCore
 
         protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
         {
-           // optionsBuilder.UseSqlServer("Server=192.168.2.206;Initial Catalog=Test;User Id=cmswuliu;Password=h7xY81agBn*Veiu3;MultipleActiveResultSets=True");
+          // optionsBuilder.UseSqlServer("Server=192.168.2.206;Initial Catalog=Test;User Id=cmswuliu;Password=h7xY81agBn*Veiu3;MultipleActiveResultSets=True");
+           optionsBuilder.UseSqlServer("Server=DESKTOP-M9R8T31;Initial Catalog=Test;User Id=sa;Password=P@ssw0rd;MultipleActiveResultSets=True");
         }
     }
 }

src/DotNetCore.CAP.EntityFrameworkCore/CapPublisher.cs

@@ -0,0 +1,55 @@
+using System;
+using System.Collections.Generic;
+using System.Data;
+using System.Text;
+using System.Threading.Tasks;
+using Dapper;
+using DotNetCore.CAP.Infrastructure;
+using DotNetCore.CAP.Models;
+using DotNetCore.CAP.Processor;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Infrastructure;
+using Microsoft.EntityFrameworkCore.Storage;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace DotNetCore.CAP.EntityFrameworkCore
+{
+    public class CapPublisher : ICapPublisher
+    {
+        private readonly SqlServerOptions _options;
+        private readonly IServiceProvider _provider;
+        private readonly DbContext _dbContext;
+
+        public CapPublisher(SqlServerOptions options, IServiceProvider provider)
+        {
+            _options = options;
+            _provider = provider;
+            _dbContext = (DbContext)_provider.GetService(_options.DbContextType);
+        }
+
+        public async Task PublishAsync(string topic, string content)
+        {
+            var connection = _dbContext.Database.GetDbConnection();
+            var transaction = _dbContext.Database.CurrentTransaction;
+            transaction = transaction ?? await _dbContext.Database.BeginTransactionAsync(IsolationLevel.ReadCommitted);
+            var dbTransaction = transaction.GetDbTransaction();
+
+            var message = new CapSentMessage
+            {
+                KeyName = topic,
+                Content = content,
+                StatusName = StatusName.Scheduled
+            };
+
+            var sql = "INSERT INTO [cap].[CapSentMessages] ([Id],[Added],[Content],[KeyName],[ExpiresAt],[Retries],[StatusName])VALUES(@Id,@Added,@Content,@KeyName,@ExpiresAt,@Retries,@StatusName)";
+            await connection.ExecuteAsync(sql, message, transaction: dbTransaction);
+
+            PublishQueuer.PulseEvent.Set();
+        }
+
+        public Task PublishAsync<T>(string topic, T contentObj)
+        {
+            throw new NotImplementedException();
+        }
+    }
+}

src/DotNetCore.CAP.EntityFrameworkCore/EFStorageConnection.cs

@@ -89,7 +89,7 @@ WHERE StatusName = '{StatusName.Scheduled}'";
             var sql = $@"
 SELECT TOP (1) *
 FROM [{_options.Schema}].[{nameof(CapDbContext.CapReceivedMessages)}] WITH (readpast)
-WHERE StateName = '{StatusName.Enqueued}'";
+WHERE StatusName = '{StatusName.Enqueued}'";
 
             var connection = _context.GetDbConnection();
             var message = (await connection.QueryAsync<CapReceivedMessage>(sql)).FirstOrDefault();

src/DotNetCore.CAP/CAP.ServiceCollectionExtensions.cs

@@ -55,14 +55,14 @@ namespace Microsoft.Extensions.DependencyInjection
             //Processors
             services.AddTransient<PublishQueuer>();
             services.AddTransient<SubscribeQueuer>();
-            services.AddTransient<IMessageProcessor, DefaultMessageProcessor>();          
+            services.AddTransient<IDispatcher, DefaultDispatcher>();          
 
             //Executors
             services.AddSingleton<IQueueExecutorFactory, QueueExecutorFactory>();
             services.AddSingleton<IQueueExecutor, SubscibeQueueExecutor>();
 
 
-            services.TryAddScoped<ICapPublisher, DefaultCapPublisher>();
+           // services.TryAddScoped<ICapPublisher, DefaultCapPublisher>();
 
             return new CapBuilder(services);
         }

src/DotNetCore.CAP/DotNetCore.CAP.csproj

@@ -15,7 +15,7 @@
 
   <ItemGroup>
     <None Include="IQueueExecutor.Subscibe.cs" />
-    <None Include="Processor\IProcessor.Message.Default.cs" />
+    <None Include="Processor\IDispatcher.Default.cs" />
   </ItemGroup>
 
   <ItemGroup>

src/DotNetCore.CAP/Processor/IProcessor.Message.Default.cs → src/DotNetCore.CAP/Processor/IDispatcher.Default.cs

@@ -8,7 +8,7 @@ using Microsoft.Extensions.Options;
 
 namespace DotNetCore.CAP.Processor
 {
-    public class DefaultMessageProcessor : IMessageProcessor
+    public class DefaultDispatcher : IDispatcher
     {
         private readonly IQueueExecutorFactory _queueExecutorFactory;
         private readonly IServiceProvider _provider;
@@ -19,11 +19,11 @@ namespace DotNetCore.CAP.Processor
 
         internal static readonly AutoResetEvent PulseEvent = new AutoResetEvent(true);
 
-        public DefaultMessageProcessor(
+        public DefaultDispatcher(
                IServiceProvider provider,
                IQueueExecutorFactory queueExecutorFactory,
                IOptions<CapOptions> capOptions,
-               ILogger<DefaultMessageProcessor> logger)
+               ILogger<DefaultDispatcher> logger)
         {
             _logger = logger;
             _queueExecutorFactory = queueExecutorFactory;

src/DotNetCore.CAP/Processor/IMessageProcessor.cs → src/DotNetCore.CAP/Processor/IDispatcher.cs

@@ -4,7 +4,7 @@ using System.Text;
 
 namespace DotNetCore.CAP.Processor
 {
-    public interface IMessageProcessor : IProcessor
+    public interface IDispatcher : IProcessor
     {
         bool Waiting { get; }
     }

src/DotNetCore.CAP/Processor/IProcessingServer.Cap.cs

@@ -18,7 +18,7 @@ namespace DotNetCore.CAP.Processor
         private readonly CapOptions _options;
 
         private IProcessor[] _processors;
-        private IList<IMessageProcessor> _messageProcessors;
+        private IList<IDispatcher> _messageDispatchers;
         private ProcessingContext _context;
         private Task _compositeTask;
         private bool _disposed;
@@ -34,6 +34,7 @@ namespace DotNetCore.CAP.Processor
             _provider = provider;
             _options = options.Value;
             _cts = new CancellationTokenSource();
+            _messageDispatchers = new List<IDispatcher>();
         }
 
         public void Start()
@@ -90,7 +91,7 @@ namespace DotNetCore.CAP.Processor
 
         private bool AllProcessorsWaiting()
         {
-            foreach (var processor in _messageProcessors)
+            foreach (var processor in _messageDispatchers)
             {
                 if (!processor.Waiting)
                 {
@@ -110,10 +111,10 @@ namespace DotNetCore.CAP.Processor
             var returnedProcessors = new List<IProcessor>();
             for (int i = 0; i < processorCount; i++)
             {
-                var messageProcessors = _provider.GetService<IMessageProcessor>();
-                _messageProcessors.Add(messageProcessors);
+                var messageProcessors = _provider.GetService<IDispatcher>();
+                _messageDispatchers.Add(messageProcessors);
             }
-            returnedProcessors.AddRange(_messageProcessors);
+            returnedProcessors.AddRange(_messageDispatchers);
 
             returnedProcessors.Add(_provider.GetService<PublishQueuer>());
             returnedProcessors.Add(_provider.GetService<SubscribeQueuer>());

src/DotNetCore.CAP/Processor/IProcessor.PublishQueuer.cs

@@ -59,7 +59,7 @@ namespace DotNetCore.CAP.Processor
 
             context.ThrowIfStopping();
             
-            DefaultMessageProcessor.PulseEvent.Set();
+            DefaultDispatcher.PulseEvent.Set();
 
             await WaitHandleEx.WaitAnyAsync(PulseEvent,
                 context.CancellationToken.WaitHandle, _pollingDelay);

src/DotNetCore.CAP/Processor/IProcessor.SubscribeQueuer.cs

@@ -59,7 +59,7 @@ namespace DotNetCore.CAP.Processor
 
             context.ThrowIfStopping();
 
-            DefaultMessageProcessor.PulseEvent.Set();
+            DefaultDispatcher.PulseEvent.Set();
 
             await WaitHandleEx.WaitAnyAsync(PulseEvent,
                 context.CancellationToken.WaitHandle, _pollingDelay);