update kafka proejct

src/Cap.Consistency.Kafka/Cap.Consistency.Kafka.csproj

@@ -2,6 +2,7 @@
 
   <PropertyGroup>
     <TargetFramework>netstandard1.6</TargetFramework>
+    <NetStandardImplicitPackageVersion>1.6.1</NetStandardImplicitPackageVersion>
   </PropertyGroup>
 
   <ItemGroup>

src/Cap.Consistency.Kafka/KafkaConsumerClient.cs

@@ -38,6 +38,7 @@ namespace Cap.Consistency.Kafka
                 InitKafkaClient();
             }
             _consumerClient.Assignment.Add(new TopicPartition(topicName, partition));
+            _consumerClient.Subscribe(topicName);
         }
 
         public void Listening(TimeSpan timeout) {
@@ -65,11 +66,12 @@ namespace Cap.Consistency.Kafka
         private void ConsumerClient_OnMessage(object sender, Message<Null, string> e) {
             var message = new DeliverMessage {
                 MessageKey = e.Topic,
-                Value = e.Value
+                Value = e.Value,
+                Body = Encoding.UTF8.GetBytes(e.Value)
             };
             MessageReceieved?.Invoke(sender, message);
         }
-    
+
         #endregion
 
     }

src/Cap.Consistency.Kafka/KafkaProducerClient.cs

@@ -1,11 +1,44 @@
 using System;
 using System.Collections.Generic;
 using System.Text;
+using System.Threading.Tasks;
+using Cap.Consistency.Infrastructure;
+using Cap.Consistency.Producer;
+using Confluent.Kafka;
+using Confluent.Kafka.Serialization;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
 
 namespace Cap.Consistency.Kafka
 {
-    public class KafkaProducerClient
+    public class KafkaProducerClient : IProducerClient
     {
 
+        private readonly ConsistencyOptions _options;
+        private readonly ILogger _logger;
+
+        public KafkaProducerClient(IOptions<ConsistencyOptions> options, ILoggerFactory loggerFactory) {
+            _options = options.Value;
+            _logger = loggerFactory.CreateLogger(nameof(KafkaProducerClient));
+        }
+
+
+        public Task SendAsync(string topic, string content) {
+            var config = new Dictionary<string, object> { { "bootstrap.servers", _options.BrokerUrlList } };
+            try {
+                using (var producer = new Producer<Null, string>(config, null, new StringSerializer(Encoding.UTF8))) {
+                    var task = producer.ProduceAsync(topic, null, content);
+                    task.ContinueWith(g => {
+                        _logger.LogInformation($"publish message to topic:{topic}\r\n -------content:{content}\r\n ");
+                    });
+                    //producer.Flush(1000);
+                    return Task.CompletedTask;
+                }
+            }
+            catch (Exception e) {
+                _logger.LogError(new EventId(1), e, $"publish message to [topic:{topic}] ,content:{content}");
+                return Task.CompletedTask;
+            }
+        }
     }
 }

src/Cap.Consistency.Kafka/KafkaTopicAttribute.cs

@@ -7,7 +7,7 @@ namespace Cap.Consistency.Kafka
 {
     public class KafkaTopicAttribute : TopicAttribute
     {
-        public KafkaTopicAttribute(string topicName) 
+        public KafkaTopicAttribute(string topicName)
             : this(topicName, 0) { }
 
         public KafkaTopicAttribute(string topicName, int partition)

src/Cap.Consistency.Kafka/Microsoft.Extensions.DependencyInjection/ConsistencyBuilderExtensions.cs

@@ -0,0 +1,23 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using Cap.Consistency;
+using Cap.Consistency.Consumer;
+using Cap.Consistency.Kafka;
+using Cap.Consistency.Producer;
+
+namespace Microsoft.Extensions.DependencyInjection
+{
+    public static class ConsistencyBuilderExtensions
+    {
+        public static ConsistencyBuilder AddKafka(this ConsistencyBuilder builder) {
+
+            builder.Services.AddSingleton<IConsumerClientFactory, KafkaConsumerClientFactory>();
+
+            builder.Services.AddTransient<IProducerClient, KafkaProducerClient>();
+
+            return builder;
+        }
+
+    }
+}