<commit_before>namespace DDDLite.CQRS.Messaging.Kafka
{
  using System;
  using System.Collections.Generic;
  using System.Text;
  using System.Threading.Tasks;

  using Confluent.Kafka;
  using Confluent.Kafka.Serialization;
  using Newtonsoft.Json;

  using DDDLite.CQRS.Events;

  using InMemory;
  using System.Threading;

  public class KafkaEventBus : InMemoryEventBus, IEventPublisher, IDisposable
  {
    private readonly string topic;
    private Producer<Null, string> producer;
    private Consumer<Null, string> consumer;
    private readonly JsonSerializerSettings settings = new JsonSerializerSettings
    {
      TypeNameHandling = TypeNameHandling.Auto
    };

    public KafkaEventBus(string host, string groupId, string topic)
    {
      this.topic = topic;
      this.producer = new Producer<Null, string>(new Dictionary<string, object>
      {
        { "bootstrap.servers", host }
      }, null, new StringSerializer(Encoding.UTF8));

      this.consumer = new Consumer<Null, string>(new Dictionary<string, object>
      {
        { "bootstrap.servers", host },
        { "group.id", groupId }
      }, null, new StringDeserializer(Encoding.UTF8));
      this.consumer.Subscribe(topic);
    }
    public void Dispose()
    {
      if (this.consumer != null)
      {
        this.consumer.Dispose();
        this.consumer = null;
      }

      if (this.producer != null)
      {
        this.producer.Flush(100);
        this.producer.Dispose();
        this.producer = null;
      }
    }

    public async override Task PublishAsync<TEvent>(TEvent @event)
    {
      var desc = new EventDescriptor(@event);
      await this.producer.ProduceAsync(topic, null, JsonConvert.SerializeObject(desc, this.settings));
    }

    public void Listening(CancellationToken cancellationToken, params string[] topics)
    {
      if (topics != null && topics.Length > 0)
      {
        this.consumer.Subscribe(topics);
      }

      while (true)
      {
        cancellationToken.ThrowIfCancellationRequested();
        Message<Null, string> msg;
        if (!consumer.Consume(out msg, TimeSpan.FromMilliseconds(100)))
        {
          continue;
        }
<commit_msg>修复KafkaEventBus监听bug<commit_after>namespace DDDLite.CQRS.Messaging.Kafka
{
  using System;
  using System.Collections.Generic;
  using System.Text;
  using System.Threading.Tasks;
  using System.Linq;

  using Confluent.Kafka;
  using Confluent.Kafka.Serialization;
  using Newtonsoft.Json;

  using DDDLite.CQRS.Events;

  using InMemory;
  using System.Threading;

  public class KafkaEventBus : InMemoryEventBus, IEventPublisher, IDisposable
  {
    private readonly string topic;
    private Producer<Null, string> producer;
    private Consumer<Null, string> consumer;
    private readonly JsonSerializerSettings settings = new JsonSerializerSettings
    {
      TypeNameHandling = TypeNameHandling.Auto
    };

    public KafkaEventBus(string host, string groupId, string topic)
    {
      this.topic = topic;
      this.producer = new Producer<Null, string>(new Dictionary<string, object>
      {
        { "bootstrap.servers", host }
      }, null, new StringSerializer(Encoding.UTF8));

      this.consumer = new Consumer<Null, string>(new Dictionary<string, object>
      {
        { "bootstrap.servers", host },
        { "group.id", groupId }
      }, null, new StringDeserializer(Encoding.UTF8));
    }
    public void Dispose()
    {
      if (this.consumer != null)
      {
        this.consumer.Dispose();
        this.consumer = null;
      }

      if (this.producer != null)
      {
        this.producer.Flush(100);
        this.producer.Dispose();
        this.producer = null;
      }
    }

    public async override Task PublishAsync<TEvent>(TEvent @event)
    {
      var desc = new EventDescriptor(@event);
      await this.producer.ProduceAsync(topic, null, JsonConvert.SerializeObject(desc, this.settings));
    }

    public void Listening(CancellationToken cancellationToken, params string[] topics)
    {
      var tmp = new List<string> { topic };
      if (topics != null)
      {
        tmp.AddRange(topics);
      }

      this.consumer.Subscribe(tmp.Distinct().ToArray());

      while (true)
      {
        cancellationToken.ThrowIfCancellationRequested();
        Message<Null, string> msg;
        if (!consumer.Consume(out msg, TimeSpan.FromMilliseconds(100)))
        {
          continue;
        }

{'fraction_non_alphanumeric': 0.10365853658536585, 'fraction_numerical': 0.0038392050587172538, 'mean_word_length': 2.7438715131022824, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3132032', 'n_tokens_mistral': 1324, 'n_tokens_neox': 1296, 'n_words': 279}