<commit_before>﻿using System;
using System.Collections.Generic;
using System.Globalization;
using EasyRabbitMQ.Configuration;
using EasyRabbitMQ.Infrastructure;
using EasyRabbitMQ.Serialization;
using RabbitMQ.Client;

namespace EasyRabbitMQ.Publish
{
    internal class Publisher : IPublisher
    {
        private readonly ISerializer _serializer;
        private readonly Lazy<Channel> _channel;

        private static readonly KeyValuePair<string, object> ApplicationHeader = 
            new KeyValuePair<string, object>("x-sending-application", "EasyRabbitMQ");

        internal Publisher(EasyRabbitMQConfigurer configurer)
        {
            _serializer = configurer.Serializer;
            _channel = new Lazy<Channel>(() => configurer.ChannelFactory.CreateChannel());
        }

        public void PublishQueue(string queue, dynamic message)
        {
            PublishQueue(queue, null, message);
        }

        public void PublishQueue(string queue, MessageProperties messageProperties, dynamic message)
        {
            if (string.IsNullOrWhiteSpace(queue)) throw new ArgumentNullException(nameof(queue));
            if (message == null) throw new ArgumentNullException(nameof(message));

            PublishMessageInternal("", queue, messageProperties, message);
        }

        public void PublishExchange(string exchange, dynamic message)
        {
            PublishExchange(exchange, "", null, message);
        }

        public void PublishExchange(string exchange, MessageProperties messageProperties, dynamic message)
        {
            PublishExchange(exchange, "", messageProperties, message);
        }

        public void PublishExchange(string exchange, string routingKey, dynamic message)
        {
            PublishExchange(exchange, routingKey, null, message);
        }

        public void PublishExchange(string exchange, string routingKey, MessageProperties messageProperties,
            dynamic message)
        {
            if (string.IsNullOrWhiteSpace(exchange)) throw new ArgumentNullException(nameof(exchange));
            if (routingKey == null) throw new ArgumentNullException(nameof(routingKey));
            if (message == null) throw new ArgumentNullException(nameof(message));

            PublishMessageInternal(exchange, routingKey, messageProperties, message);
        }

        private void PublishMessageInternal(string queueOrExchange, string routingKey,
            MessageProperties messageProperties, dynamic message)
        {
            var channel = _channel.Value.Instance;

            var body = _serializer.Serialize(message);

            var props = CreateBasicProperties(channel, messageProperties);

            channel.BasicPublish(queueOrExchange, routingKey, props, body);
        }

        private IBasicProperties CreateBasicProperties(IModel channel, MessageProperties messageProperties)
        {
            if (messageProperties == null) messageProperties = new MessageProperties();

            var props = channel.CreateBasicProperties();
            props.ContentType = _serializer.ContentType;
            props.CorrelationId = messageProperties.CorrelationId;
            props.Headers = messageProperties.Headers;
            props.MessageId = messageProperties.MessageId;
            props.Persistent = messageProperties.PersistentMessage;

            if (messageProperties.Expiration.HasValue)
            {
                props.Expiration = messageProperties.Expiration.Value.ToString(CultureInfo.InvariantCulture);
            }
            
            AddApplicationHeader(props);

            return props;
        }

        private static void AddApplicationHeader(IBasicProperties basicProperties)
        {
            if (basicProperties.Headers == null) basicProperties.Headers = new Dictionary<string, object>();
            basicProperties.Headers.Add(ApplicationHeader);
        }

        public void Dispose()
        {
            Dispose(true);<commit_msg>Only set values on IBasicProperties if they are not null<commit_after>﻿using System;
using System.Collections.Generic;
using System.Globalization;
using EasyRabbitMQ.Configuration;
using EasyRabbitMQ.Infrastructure;
using EasyRabbitMQ.Serialization;
using RabbitMQ.Client;

namespace EasyRabbitMQ.Publish
{
    internal class Publisher : IPublisher
    {
        private readonly ISerializer _serializer;
        private readonly Lazy<Channel> _channel;

        private static readonly KeyValuePair<string, object> ApplicationHeader = 
            new KeyValuePair<string, object>("x-sending-application", "EasyRabbitMQ");

        internal Publisher(EasyRabbitMQConfigurer configurer)
        {
            _serializer = configurer.Serializer;
            _channel = new Lazy<Channel>(() => configurer.ChannelFactory.CreateChannel());
        }

        public void PublishQueue(string queue, dynamic message)
        {
            PublishQueue(queue, null, message);
        }

        public void PublishQueue(string queue, MessageProperties messageProperties, dynamic message)
        {
            if (string.IsNullOrWhiteSpace(queue)) throw new ArgumentNullException(nameof(queue));
            if (message == null) throw new ArgumentNullException(nameof(message));

            PublishMessageInternal("", queue, messageProperties, message);
        }

        public void PublishExchange(string exchange, dynamic message)
        {
            PublishExchange(exchange, "", null, message);
        }

        public void PublishExchange(string exchange, MessageProperties messageProperties, dynamic message)
        {
            PublishExchange(exchange, "", messageProperties, message);
        }

        public void PublishExchange(string exchange, string routingKey, dynamic message)
        {
            PublishExchange(exchange, routingKey, null, message);
        }

        public void PublishExchange(string exchange, string routingKey, MessageProperties messageProperties,
            dynamic message)
        {
            if (string.IsNullOrWhiteSpace(exchange)) throw new ArgumentNullException(nameof(exchange));
            if (routingKey == null) throw new ArgumentNullException(nameof(routingKey));
            if (message == null) throw new ArgumentNullException(nameof(message));

            PublishMessageInternal(exchange, routingKey, messageProperties, message);
        }

        private void PublishMessageInternal(string queueOrExchange, string routingKey,
            MessageProperties messageProperties, dynamic message)
        {
            var channel = _channel.Value.Instance;

            var body = _serializer.Serialize(message);

            var props = CreateBasicProperties(channel, messageProperties);

            channel.BasicPublish(queueOrExchange, routingKey, props, body);
        }

        private IBasicProperties CreateBasicProperties(IModel channel, MessageProperties messageProperties)
        {
            if (messageProperties == null) messageProperties = new MessageProperties();

            var props = channel.CreateBasicProperties();
            props.ContentType = _serializer.ContentType;

            if (messageProperties.CorrelationId != null)
            {
                props.CorrelationId = messageProperties.CorrelationId;
            }

            if (messageProperties.Expiration.HasValue)
            {
                props.Expiration = messageProperties.Expiration.Value.ToString(CultureInfo.InvariantCulture);
            }

            props.Headers = messageProperties.Headers;

            if (messageProperties.MessageId != null)
            {
                props.MessageId = messageProperties.MessageId;
            }

            props.Persistent = messageProperties.PersistentMessage;
            
            AddApplicationHeader(props);

            return props;
        }

        private static void AddApplicationHeader(IBasicProperties basicProperties)
        {
            if (basicProperties.Headers == null) basicProperties.Headers = new Dictionary<string, object>();
            basicProperties.Headers.Add(ApplicationHeader);
        }

        public void Dispose()
        {
            Dispose(true);
{'fraction_non_alphanumeric': 0.07483241925655089, 'fraction_numerical': 0.0, 'mean_word_length': 2.6732318710832588, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 15, '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': '4995859', 'n_tokens_mistral': 1915, 'n_tokens_neox': 1830, 'n_words': 444}