<commit_before>﻿using IFramework.MessageQueue.MSKafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using IFramework.MessageQueue.MSKafka.MessageFormat;
using IFramework.Config;
using Sample.Command;
using IFramework.Command;
using IFramework.IoC;
using Sample.DTO;
using IFramework.Infrastructure;
using System.Threading;

namespace MSKafka.Test
{
    class Program
    {
        static string commandQueue = "groupcommandqueue";
        static string zkConnectionString = "localhost:2181";

        static void Main(string[] args)
        {
            Configuration.Instance
                         .UseUnityContainer()
                         .UseLog4Net("log4net.config");
            GroupConsuemrTest();
        }

        static Task CreateConsumerTask(string consumerId, CancellationTokenSource cancellationTokenSource)
        {
            return Task.Run(() =>
            {
                var consumer = new KafkaConsumer(zkConnectionString, commandQueue, Environment.MachineName, consumerId);
                try
                {
                    foreach (var kafkaMessage in consumer.GetMessages(cancellationTokenSource.Token))
                    {
                        var message = Encoding.UTF8.GetString(kafkaMessage.Payload);
                        var sendTime = DateTime.Parse(message);
                        Console.WriteLine($"consumer:{consumer.ConsumerId} {DateTime.Now.ToString("HH:mm:ss.fff")} consume message: {message} cost: {(DateTime.Now - sendTime).TotalMilliseconds}");
                        consumer.CommitOffset(kafkaMessage.PartitionId.Value, kafkaMessage.Offset);
                    }
                }
                catch (OperationCanceledException)
                {
                    return;
                }
                catch (ThreadAbortException)
                {
                    return;
                }
                catch (Exception ex)
                {
                    if (!cancellationTokenSource.IsCancellationRequested)
                    {
                        Console.WriteLine(ex.GetBaseException().Message);
                    }
                }
                finally
                {
                    consumer.Stop();
                }
            });
        }

        static void GroupConsuemrTest()
        {
            var cancellationTokenSource = new CancellationTokenSource();
            var consumerTasks = new List<Task>();
            for (int i = 0; i < 3; i++)
            {
                consumerTasks.Add(CreateConsumerTask(i.ToString(), cancellationTokenSource));
            }


            var queueClient = new KafkaProducer(commandQueue, zkConnectionString);
            while (true)
            {
                var message = Console.ReadLine();
                if (message.Equals("q"))
                {
                    cancellationTokenSource.Cancel();
                    Task.WaitAll(consumerTasks.ToArray());
                    break;
                }
                message = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.ffffff");
                var kafkaMessage = new Kafka.Client.Messages.Message(Encoding.UTF8.GetBytes(message));
                var data = new Kafka.Client.Producers.ProducerData<string, Kafka.Client.Messages.Message>(commandQueue, message, kafkaMessage);
                try
                {
                    queueClient.Send(data);
                    Console.WriteLine($"send message: {message}");
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.GetBaseException().Message);
                }
            }
        }

        static void ServiceTest()
        {
            ReduceProduct reduceProduct = new ReduceProduct
            {
                ProductId = new Guid("2B6FDE83-A319-433B-9FA5-399D382D0CD3"),
                ReduceCount = 1
            };

            var _commandBus = IoCFactory.Resolve<ICommandBus>();
            _commandBus.Start();

            var t = _commandBus.SendAsync(reduceProduct, true).Result;
            Console.WriteLine(t.Reply.Result);


            var products = _commandBus.SendAsync(new GetProducts
            {
                ProductIds = new List<Guid> { reduceProduct.ProductId }
            }, true).Result.ReadAsAsync<List<Project>>().Result;

            Console.WriteLine(products.ToJson());
            Console.ReadLine();<commit_msg>add ProducerSendTPSTest<commit_after>﻿using IFramework.MessageQueue.MSKafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using IFramework.MessageQueue.MSKafka.MessageFormat;
using IFramework.Config;
using Sample.Command;
using IFramework.Command;
using IFramework.IoC;
using Sample.DTO;
using IFramework.Infrastructure;
using System.Threading;
using System.Diagnostics;

namespace MSKafka.Test
{
    class Program
    {
        static string commandQueue = "groupcommandqueue";
        static string zkConnectionString = "localhost:2181";

        static void Main(string[] args)
        {
            Configuration.Instance
                         .UseUnityContainer()
                         .UseLog4Net("log4net.config");
            ProducerSendTPSTest();
            //GroupConsuemrTest();
        }

        static Task CreateConsumerTask(string consumerId, CancellationTokenSource cancellationTokenSource)
        {
            return Task.Run(() =>
            {
                var consumer = new KafkaConsumer(zkConnectionString, commandQueue, Environment.MachineName, consumerId);
                try
                {
                    foreach (var kafkaMessage in consumer.GetMessages(cancellationTokenSource.Token))
                    {
                        var message = Encoding.UTF8.GetString(kafkaMessage.Payload);
                        var sendTime = DateTime.Parse(message);
                        Console.WriteLine($"consumer:{consumer.ConsumerId} {DateTime.Now.ToString("HH:mm:ss.fff")} consume message: {message} cost: {(DateTime.Now - sendTime).TotalMilliseconds}");
                        consumer.CommitOffset(kafkaMessage.PartitionId.Value, kafkaMessage.Offset);
                    }
                }
                catch (OperationCanceledException)
                {
                    return;
                }
                catch (ThreadAbortException)
                {
                    return;
                }
                catch (Exception ex)
                {
                    if (!cancellationTokenSource.IsCancellationRequested)
                    {
                        Console.WriteLine(ex.GetBaseException().Message);
                    }
                }
                finally
                {
                    consumer.Stop();
                }
            });
        }

        static void GroupConsuemrTest()
        {
            var cancellationTokenSource = new CancellationTokenSource();
            var consumerTasks = new List<Task>();
            for (int i = 0; i < 3; i++)
            {
                consumerTasks.Add(CreateConsumerTask(i.ToString(), cancellationTokenSource));
            }


            var queueClient = new KafkaProducer(commandQueue, zkConnectionString);
            while (true)
            {
                var message = Console.ReadLine();
                if (message.Equals("q"))
                {
                    cancellationTokenSource.Cancel();
                    Task.WaitAll(consumerTasks.ToArray());
                    break;
                }
                message = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.ffffff");
                var kafkaMessage = new Kafka.Client.Messages.Message(Encoding.UTF8.GetBytes(message));
                var data = new Kafka.Client.Producers.ProducerData<string, Kafka.Client.Messages.Message>(commandQueue, message, kafkaMessage);
                try
                {
                    queueClient.Send(data);
                    Console.WriteLine($"send message: {message}");
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.GetBaseException().Message);
                }
            }
        }
        

        static void ProducerSendTPSTest()
        {
            int batchCount = 100000;
            var queueClient = new KafkaProducer(commandQueue, zkConnectionString);
           
            var message = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.ffffff");
            var kafkaMessage = new Kafka.Client.Messages.Message(Encoding.UTF8.GetBytes(message));
            var data = new Kafka.Client.Producers.ProducerData<string, Kafka.Client.Messages.Message>(commandQueue, message, kafkaMessage);
            double totalSendRt = 0;
            Stopwatch watch = new Stopwatch();
            watch.Start();
            for (int i = 0; i < batchCount; i++)
            {
                try
                {
                    var start = DateTime.Now;
                    queueClient.Send(data);
                    totalSendRt += (DateTime.Now - start).TotalMilliseconds;
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.GetBaseException().Message);
                }
            }
            var elapsedMs = watch.ElapsedMilliseconds;

            Console.WriteLine($"cost: {elapsedMs}  tps: {batchCount * 1000 / elapsedMs} rt: {totalSendRt / (double)batchCount}");
            Console.ReadLine();
        }
        static void ServiceTest()
        {
            ReduceProduct reduceProduct = new ReduceProduct
            {
                ProductId = new Guid("2B6FDE83-A319-433B-9FA5-399D382D0CD3"),
                ReduceCount = 1
            };

            var _commandBus = IoCFactory.Resolve<ICommandBus>();
            _commandBus.Start();

            var t = _commandBus.SendAsync(reduceProduct, true).Result;
            Console.WriteLine(t.Reply.Result);


            var products = _commandBus.SendAsync(new GetProducts
            {
                ProductIds = new List<Guid> { reduceProduct.ProductId }
            }, true).Result.ReadAsAsync<List<Project>>().Result;

            Console.WriteLine(products.ToJson());
            Console.ReadLine();
{'fraction_non_alphanumeric': 0.09007618863921305, 'fraction_numerical': 0.007233098659465715, 'mean_word_length': 1.5335939408746642, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 18, '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': '5554869', 'n_tokens_mistral': 2592, 'n_tokens_neox': 2421, 'n_words': 447}