<commit_before>using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Reactive.Disposables;
using System.Threading;
using System.Threading.Tasks;
using kafka4net;
using Obvs.MessageProperties;
using Obvs.Serialization;

namespace Obvs.Kafka
{
    public class MessagePublisher<TMessage> : IMessagePublisher<TMessage>
        where TMessage : class
    {
        private readonly KafkaConfiguration _kafkaConfiguration;
        private readonly string _topic;
        private readonly KafkaProducerConfiguration _producerConfig;
        private readonly IMessageSerializer _serializer;
        private readonly IMessagePropertyProvider<TMessage> _propertyProvider;

        private IDisposable _disposable;
        private bool _disposed;
        private long _connected;
        private Producer _producer;

        public MessagePublisher(KafkaConfiguration kafkaConfiguration, KafkaProducerConfiguration producerConfig, string topic, IMessageSerializer serializer, IMessagePropertyProvider<TMessage> propertyProvider)
        {
            _kafkaConfiguration = kafkaConfiguration;
            _topic = topic;
            _serializer = serializer;
            _propertyProvider = propertyProvider;
            _producerConfig = producerConfig;
        }

        public Task PublishAsync(TMessage message)
        {
            if (_disposed)
            {
                throw new InvalidOperationException("Publisher has been disposed already.");
            }

            return Publish(message);
        }

        private Task Publish(TMessage message)
        {
            List<KeyValuePair<string, object>> properties = _propertyProvider.GetProperties(message).ToList();

            return Publish(message, properties);
        }

        private async Task Publish(TMessage message, List<KeyValuePair<string, object>> properties)
        {
            if (_disposed)
            {
                return;
            }

            await Connect();

            KafkaHeaderedMessage kafkaHeaderedMessage = new KafkaHeaderedMessage
            {
                PayloadType = message.GetType().Name,
            };

            using (MemoryStream stream = new MemoryStream())
            {
                _serializer.Serialize(stream, message);

                kafkaHeaderedMessage.Payload = stream.ToArray();
            }

            using (MemoryStream stream = new MemoryStream())
            {
                ProtoBuf.Serializer.Serialize(stream, kafkaHeaderedMessage);

                _producer.Send(new Message { Value = stream.ToArray() });
            }
        }

        private async Task Connect()
        {
            if (Interlocked.CompareExchange(ref _connected, 1, 0) == 0)
            {
                var producerConfiguration = new ProducerConfiguration(_topic,
                    batchFlushTime: TimeSpan.Zero,
                    batchFlushSize: _producerConfig.BatchFlushSize,
                    requiredAcks: 1,
                    autoGrowSendBuffers: true,
                    sendBuffersInitialSize: 200,
                    maxMessageSetSizeInBytes: 1073741824,
                    producerRequestTimeout: null,
                    partitioner: null);

                _producer = new Producer(_kafkaConfiguration.SeedAddresses, producerConfiguration);

                await _producer.ConnectAsync();

                _disposable = Disposable.Create(() =>
                {
                    _disposed = true;
                    _producer.CloseAsync(TimeSpan.FromSeconds(2)).Wait();
                });
            }
        }

        public void Dispose()
        {
            if (_disposable != null)
            {
                _disposable.Dispose();
            }
        }
    }
}<commit_msg>Fixed up buffer in producers.<commit_after>using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Reactive.Disposables;
using System.Threading;
using System.Threading.Tasks;
using kafka4net;
using Obvs.MessageProperties;
using Obvs.Serialization;

namespace Obvs.Kafka
{
    public class MessagePublisher<TMessage> : IMessagePublisher<TMessage>
        where TMessage : class
    {
        private readonly KafkaConfiguration _kafkaConfiguration;
        private readonly string _topic;
        private readonly KafkaProducerConfiguration _producerConfig;
        private readonly IMessageSerializer _serializer;
        private readonly IMessagePropertyProvider<TMessage> _propertyProvider;

        private IDisposable _disposable;
        private bool _disposed;
        private long _connected;
        private Producer _producer;

        public MessagePublisher(KafkaConfiguration kafkaConfiguration, KafkaProducerConfiguration producerConfig, string topic, IMessageSerializer serializer, IMessagePropertyProvider<TMessage> propertyProvider)
        {
            _kafkaConfiguration = kafkaConfiguration;
            _topic = topic;
            _serializer = serializer;
            _propertyProvider = propertyProvider;
            _producerConfig = producerConfig;
        }

        public Task PublishAsync(TMessage message)
        {
            if (_disposed)
            {
                throw new InvalidOperationException("Publisher has been disposed already.");
            }

            return Publish(message);
        }

        private Task Publish(TMessage message)
        {
            List<KeyValuePair<string, object>> properties = _propertyProvider.GetProperties(message).ToList();

            return Publish(message, properties);
        }

        private async Task Publish(TMessage message, List<KeyValuePair<string, object>> properties)
        {
            if (_disposed)
            {
                return;
            }

            await Connect();

            KafkaHeaderedMessage kafkaHeaderedMessage = new KafkaHeaderedMessage
            {
                PayloadType = message.GetType().Name,
            };

            using (MemoryStream stream = new MemoryStream())
            {
                _serializer.Serialize(stream, message);

                kafkaHeaderedMessage.Payload = stream.ToArray();
            }

            using (MemoryStream stream = new MemoryStream())
            {
                ProtoBuf.Serializer.Serialize(stream, kafkaHeaderedMessage);

                _producer.Send(new Message { Value = stream.ToArray() });
            }
        }

        private async Task Connect()
        {
            if (Interlocked.CompareExchange(ref _connected, 1, 0) == 0)
            {
                var producerConfiguration = new ProducerConfiguration(_topic,
                    batchFlushTime: TimeSpan.FromMilliseconds(50),
                    batchFlushSize: _producerConfig.BatchFlushSize,
                    requiredAcks: 1,
                    autoGrowSendBuffers: true,
                    sendBuffersInitialSize: 200,
                    maxMessageSetSizeInBytes: 1073741824,
                    producerRequestTimeout: null,
                    partitioner: null);

                _producer = new Producer(_kafkaConfiguration.SeedAddresses, producerConfiguration);

                await _producer.ConnectAsync();

                _disposable = Disposable.Create(() =>
                {
                    _disposed = true;
                    _producer.CloseAsync(TimeSpan.FromSeconds(2)).Wait();
                });
            }
        }

        public void Dispose()
        {
            if (_disposable != null)
            {
                _disposable.Dispose();
            }
        }
    }
}
{'fraction_non_alphanumeric': 0.06782087997917209, 'fraction_numerical': 0.005206977349648529, 'mean_word_length': 1.8926957831325302, 'pattern_counts': {'":': 0, '<': 19, '<?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': '5370484', 'n_tokens_mistral': 1874, 'n_tokens_neox': 1763, 'n_words': 398}