update samples.

samples/Sample.Kafka/Entity/AppKafkaLog.cs → samples/Sample.Kafka/AppDbContext.cs

@@ -2,10 +2,14 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
+using Cap.Consistency.Infrastructure;
+using JetBrains.Annotations;
+using Microsoft.EntityFrameworkCore;
 
-namespace Sample.Kafka.Entity
+namespace Sample.Kafka
 {
-    public class AppKafkaLog
+    public class AppDbContext : DbContext
     {
+        public DbSet<ConsistencyMessage> Messages { get; set; }
     }
 }

samples/Sample.Kafka/Controllers/ValuesController.cs

@@ -4,6 +4,7 @@ using System.Linq;
 using System.Threading.Tasks;
 using Cap.Consistency.Consumer;
 using Cap.Consistency.Kafka;
+using Cap.Consistency.Producer;
 using Microsoft.AspNetCore.Mvc;
 
 namespace Sample.Kafka.Controllers
@@ -11,10 +12,27 @@ namespace Sample.Kafka.Controllers
     [Route("api/[controller]")]
     public class ValuesController : Controller, IConsumerService
     {
+        private readonly IProducerClient _producer;
 
-        [KafkaTopic("zzwl.topic.finace.callBack", IsOneWay = true)]
+        public ValuesController(IProducerClient producer) {
+            _producer = producer;
+        }
+
+        [Route("/")]
+        public IActionResult Index() {
+            return Ok();
+        }
+
+        [KafkaTopic("zzwl.topic.finace.callBack", IsOneWay = true, GroupOrExchange = "test")]
+        [NonAction]
         public void KafkaTest() {
             Console.WriteLine("kafka test invoked");
         }
+
+        [Route("~/send")]
+        public async Task<IActionResult> SendTopic() {
+            await _producer.SendAsync("zzwl.topic.finace.callBack", "{\"msgBody\":\"{\\\"dealno\\\":null,\\\"businesstype\\\":\\\"1\\\",\\\"serialno\\\":\\\"435ldfhj345\\\",\\\"bankno\\\":\\\"650001\\\",\\\"amt\\\":20.0,\\\"virtualstatus\\\":1,\\\"paystatus\\\":1}\",\"callbackTopicName\":\"zzwl.topic.finace.callBack\",\"createId\":null,\"retryLimit\":0}");
+            return Ok();
+        }
     }
 }

samples/Sample.Kafka/Sample.Kafka.csproj

@@ -11,13 +11,17 @@
     <PackageReference Include="Microsoft.ApplicationInsights.AspNetCore" Version="2.0.0" />
     <PackageReference Include="Microsoft.AspNetCore" Version="1.1.2" />
     <PackageReference Include="Microsoft.AspNetCore.Mvc" Version="1.1.3" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore" Version="1.1.2" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="1.1.2" />
     <PackageReference Include="Microsoft.Extensions.Logging.Debug" Version="1.1.2" />
   </ItemGroup>
   <ItemGroup>
     <DotNetCliToolReference Include="Microsoft.VisualStudio.Web.CodeGeneration.Tools" Version="1.0.1" />
   </ItemGroup>
   <ItemGroup>
+    <ProjectReference Include="..\..\src\Cap.Consistency.EntityFrameworkCore\Cap.Consistency.EntityFrameworkCore.csproj" />
     <ProjectReference Include="..\..\src\Cap.Consistency.Kafka\Cap.Consistency.Kafka.csproj" />
+    <ProjectReference Include="..\..\src\Cap.Consistency.RabbitMQ\Cap.Consistency.RabbitMQ.csproj" />
     <ProjectReference Include="..\..\src\Cap.Consistency\Cap.Consistency.csproj" />
   </ItemGroup>
 

samples/Sample.Kafka/Startup.cs

@@ -5,10 +5,10 @@ using System.Threading.Tasks;
 using Cap.Consistency.Infrastructure;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Hosting;
+using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
-using Sample.Kafka.Entity;
 
 namespace Sample.Kafka
 {
@@ -27,12 +27,18 @@ namespace Sample.Kafka
 
         // This method gets called by the runtime. Use this method to add services to the container.
         public void ConfigureServices(IServiceCollection services) {
+
+            services.AddDbContext<AppDbContext>();
+
+            services.AddConsistency<ConsistencyMessage>()
+                    .AddEntityFrameworkStores<AppDbContext>()
+                    .AddKafka();
+
             // Add framework services.
             services.AddMvc();
-
-            services.AddConsistency<ConsistencyMessage>();
         }
 
+
         // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
         public void Configure(IApplicationBuilder app, IHostingEnvironment env, ILoggerFactory loggerFactory) {
             loggerFactory.AddConsole(Configuration.GetSection("Logging"));