<commit_before>﻿using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using RabbitMQ.Client;
using RabbitMQ.Client.Framing;
using RawRabbit.Common;
using RawRabbit.Configuration.Publish;
using RawRabbit.Context;
using RawRabbit.Context.Provider;
using RawRabbit.Logging;
using RawRabbit.Operations.Contracts;
using RawRabbit.Serialization;

namespace RawRabbit.Operations
{
	public class Publisher<TMessageContext> : OperatorBase, IPublisher where TMessageContext : IMessageContext
	{
		private readonly IMessageContextProvider<TMessageContext> _contextProvider;
		private readonly ILogger _logger = LogManager.GetLogger<Publisher<TMessageContext>>();

		public Publisher(IChannelFactory channelFactory, IMessageSerializer serializer, IMessageContextProvider<TMessageContext> contextProvider)
			: base(channelFactory, serializer)
		{
			_contextProvider = contextProvider;
		}

		public Task PublishAsync<TMessage>(TMessage message, Guid globalMessageId, PublishConfiguration config)
		{
			return _contextProvider
				.GetMessageContextAsync(globalMessageId)
				.ContinueWith(ctxTask =>
				{
					var channel = ChannelFactory.GetChannel();
					DeclareQueue(config.Queue, channel);
					DeclareExchange(config.Exchange, channel);
					var properties = new BasicProperties
					{
						MessageId = Guid.NewGuid().ToString(),
						Headers = new Dictionary<string, object>
						{
							{ _contextProvider.ContextHeaderName, ctxTask.Result },
							{PropertyHeaders.Sent, DateTime.UtcNow.ToString("u") }
						}
					};

					channel.BasicPublish(
						exchange: config.Exchange.ExchangeName,
						routingKey: config.RoutingKey,
						basicProperties: properties,
						body: Serializer.Serialize(message)
						);
					channel.Dispose();
				});
		}

		public override void Dispose()
		{
			_logger.LogDebug("Disposing Publisher");
			base.Dispose();
		}
	}
}<commit_msg>Reuse channels for burst publishes.<commit_after>﻿using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using RabbitMQ.Client;
using RabbitMQ.Client.Framing;
using RawRabbit.Common;
using RawRabbit.Configuration.Publish;
using RawRabbit.Context;
using RawRabbit.Context.Provider;
using RawRabbit.Logging;
using RawRabbit.Operations.Contracts;
using RawRabbit.Serialization;

namespace RawRabbit.Operations
{
	public class Publisher<TMessageContext> : OperatorBase, IPublisher where TMessageContext : IMessageContext
	{
		private readonly IMessageContextProvider<TMessageContext> _contextProvider;
		private ThreadLocal<IModel> _channel;
		private Timer _channelTimer;
		private readonly ILogger _logger = LogManager.GetLogger<Publisher<TMessageContext>>();

		public Publisher(IChannelFactory channelFactory, IMessageSerializer serializer, IMessageContextProvider<TMessageContext> contextProvider)
			: base(channelFactory, serializer)
		{
			_contextProvider = contextProvider;
		}

		public Task PublishAsync<TMessage>(TMessage message, Guid globalMessageId, PublishConfiguration config)
		{
			return Task.Run(() =>
			{
				var context = _contextProvider.GetMessageContext(globalMessageId);
				var channel = GetOrCreateChannel();
				DeclareQueue(config.Queue, channel);
				DeclareExchange(config.Exchange, channel);
				var properties = new BasicProperties
				{
					MessageId = Guid.NewGuid().ToString(),
					Headers = new Dictionary<string, object>
						{
							{ _contextProvider.ContextHeaderName, context },
							{PropertyHeaders.Sent, DateTime.UtcNow.ToString("u") }
						}
				};

				channel.BasicPublish(
					exchange: config.Exchange.ExchangeName,
					routingKey: config.RoutingKey,
					basicProperties: properties,
					body: Serializer.Serialize(message)
					);
			});
		}

		private IModel GetOrCreateChannel()
		{
			if (_channel == null)
			{
				_channel = new ThreadLocal<IModel>(() => ChannelFactory.CreateChannel());
				_channelTimer = new Timer(state =>
				{
					foreach (var channel in _channel.Values)
					{
						channel?.Dispose();
					}
					_channel?.Dispose();
					_channel = null;
				}, null, TimeSpan.FromSeconds(1), new TimeSpan(-1));
			}
			if (_channel.Value.IsClosed)
			{
				_channel.Value.Dispose();
				_channel.Value = ChannelFactory.CreateChannel();
			}
			_channelTimer.Change(TimeSpan.FromSeconds(1), new TimeSpan(-1));
			return _channel.Value;
		}

		public override void Dispose()
		{
			_logger.LogDebug("Disposing Publisher");
			_channelTimer?.Dispose();
			_channel?.Dispose();
			base.Dispose();
		}
	}
}
{'fraction_non_alphanumeric': 0.09837505489679403, 'fraction_numerical': 0.000878348704435661, 'mean_word_length': 4.409738717339668, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 23, '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': '114291', 'n_tokens_mistral': 1613, 'n_tokens_neox': 1288, 'n_words': 260}