<commit_before>package com.zanox.rabbiteasy.publisher;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.zanox.rabbiteasy.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;
import java.util.Date;
import java.util.List;

/**
 * Publishes messages with certain different levels of {@link PublisherReliability}.
 *
 * @author christian.bick
 */
public class GenericPublisher implements MessagePublisher {

    private static final Logger LOGGER = LoggerFactory.getLogger(GenericPublisher.class);
    private static final int DEFAULT_RETRY_ATTEMPTS = 3;
    private static final int DEFAULT_RETRY_INTERVAL = 1000;

    private ConnectionFactory connectionFactory;
    private Channel channel;

    private PublisherReliability reliability;

    /**
     * <p>Initializes the publisher with a certain level of reliability. All messages
     * sent by the producer are sent with this level of reliability. Uses the given
     * connection factory to establish connections.</p>
     *
     * @see SimplePublisher
     * @see ConfirmedPublisher
     * @see TransactionalPublisher
     *
     * @param connectionFactory The connection factory
     * @param reliability The reliability level
     */
    public GenericPublisher(ConnectionFactory connectionFactory, PublisherReliability reliability) {
        this.connectionFactory = connectionFactory;
        this.reliability = reliability;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void send(Message message) throws IOException {
        send(message, DeliveryOptions.NONE);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void send(List<Message> messages) throws IOException {
        send(messages, DeliveryOptions.NONE);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void send(Message message, DeliveryOptions deliveryOptions)
            throws IOException {
        if (reliability == PublisherReliability.CONFIRMED) {
            publishMessageConfirmed(message, deliveryOptions);
        } else if (reliability == PublisherReliability.TRANSACTIONAL) {
            publishMessageTransactional(message, deliveryOptions);
        } else {
            publishMessage(message, deliveryOptions);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void send(List<Message> messages, DeliveryOptions deliveryOptions) throws IOException {
        if (reliability == PublisherReliability.CONFIRMED) {
            publishMessagesConfirmed(messages, deliveryOptions);
        } else if (reliability == PublisherReliability.TRANSACTIONAL) {
            publishMessagesTransactional(messages, deliveryOptions);
        } else {
            publishMessages(messages, deliveryOptions);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void close() throws IOException {
        if (channel == null) {
            LOGGER.warn("Attempt to close a publisher channel that has not been initialized");
            return;
        } else if (! channel.isOpen()) {
            LOGGER.warn("Attempt to close a publisher channel that has already been closed or is already closing");
            return;
        }
        LOGGER.debug("Closing publisher channel");
        channel.close();
        channel = null;
        LOGGER.debug("Successfully closed publisher channel");
    }

    protected void publishMessage(Message message, DeliveryOptions deliveryOptions) throws IOException {
        for (int attempt = 1; attempt <= DEFAULT_RETRY_ATTEMPTS; attempt++) {
            if (attempt > 1) {
                LOGGER.info("Attempt {} to send message", attempt);
            }

            try {
                initChannel();
                publishMessage(message, deliveryOptions, channel);
                return;
            } catch (IOException e) {
                handleIoException(attempt, e);
            }
        }
    }

    protected void publishMessages(List<Message> messages, DeliveryOptions deliveryOptions) throws IOException {
        for (Message message : messages) {
            publishMessage(message, deliveryOptions);
        }
    }

    protected void publishMessageConfirmed(Message message, DeliveryOptions deliveryOptions) throws IOException {
        for (int attempt = 1; attempt <= DEFAULT_RETRY_ATTEMPTS; attempt++) {
            if (attempt > 1) {
                LOGGER.info("Attempt {} to send message", attempt);
            }

            try {
                initConfirmedChannel();
                publishMessage(message, deliveryOptions, channel);
                LOGGER.info("Waiting for publisher ack");
                channel.waitForConfirmsOrDie();
                LOGGER.info("Received publisher ack");
                return;
            } catch (IOException e) {
                handleIoException(attempt, e);
            } catch (InterruptedException e) {
                LOGGER.warn("Publishing message interrupted while waiting for producer ack", e);
                return;
            }
        }
    }

    protected void publishMessagesConfirmed(List<Message> messages, DeliveryOptions deliveryOptions) throws IOException {
        for (Message message : messages) {
            publishMessageConfirmed(message, deliveryOptions);
        }
    }

    protected void publishMessageTransactional(Message message, DeliveryOptions deliveryOptions) throws IOException {
        publishMessagesTransactional(Collections.<Message>singletonList(message), deliveryOptions);
    }

    protected void publishMessagesTransactional(List<Message> messages, DeliveryOptions deliveryOptions) throws IOException {
        for (int attempt = 1; attempt <= DEFAULT_RETRY_ATTEMPTS; attempt++) {
            if (attempt > 1) {
                LOGGER.info("Attempt {} to send messages within transaction", attempt);
            }

            try {
                initTransactionalChannel();
                for (Message message : messages) {
                    publishMessage(message, deliveryOptions, channel);
                }
                LOGGER.info("Committing transaction");
                commitTransaction();
                LOGGER.info("Transaction committed");
                return;
            } catch (IOException e) {<commit_msg>Improved handling of failing transactional publishers.<commit_after>package com.zanox.rabbiteasy.publisher;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.zanox.rabbiteasy.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;
import java.util.Date;
import java.util.List;

/**
 * Publishes messages with certain different levels of {@link PublisherReliability}.
 *
 * @author christian.bick
 */
public class GenericPublisher implements MessagePublisher {

    private static final Logger LOGGER = LoggerFactory.getLogger(GenericPublisher.class);
    private static final int DEFAULT_RETRY_ATTEMPTS = 3;
    private static final int DEFAULT_RETRY_INTERVAL = 1000;

    private ConnectionFactory connectionFactory;
    private Channel channel;

    private PublisherReliability reliability;

    /**
     * <p>Initializes the publisher with a certain level of reliability. All messages
     * sent by the producer are sent with this level of reliability. Uses the given
     * connection factory to establish connections.</p>
     *
     * @see SimplePublisher
     * @see ConfirmedPublisher
     * @see TransactionalPublisher
     *
     * @param connectionFactory The connection factory
     * @param reliability The reliability level
     */
    public GenericPublisher(ConnectionFactory connectionFactory, PublisherReliability reliability) {
        this.connectionFactory = connectionFactory;
        this.reliability = reliability;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void send(Message message) throws IOException {
        send(message, DeliveryOptions.NONE);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void send(List<Message> messages) throws IOException {
        send(messages, DeliveryOptions.NONE);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void send(Message message, DeliveryOptions deliveryOptions)
            throws IOException {
        if (reliability == PublisherReliability.CONFIRMED) {
            publishMessageConfirmed(message, deliveryOptions);
        } else if (reliability == PublisherReliability.TRANSACTIONAL) {
            publishMessageTransactional(message, deliveryOptions);
        } else {
            publishMessage(message, deliveryOptions);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void send(List<Message> messages, DeliveryOptions deliveryOptions) throws IOException {
        if (reliability == PublisherReliability.CONFIRMED) {
            publishMessagesConfirmed(messages, deliveryOptions);
        } else if (reliability == PublisherReliability.TRANSACTIONAL) {
            publishMessagesTransactional(messages, deliveryOptions);
        } else {
            publishMessages(messages, deliveryOptions);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void close() throws IOException {
        if (channel == null) {
            LOGGER.warn("Attempt to close a publisher channel that has not been initialized");
            return;
        } else if (! channel.isOpen()) {
            LOGGER.warn("Attempt to close a publisher channel that has already been closed or is already closing");
            return;
        }
        LOGGER.debug("Closing publisher channel");
        channel.close();
        channel = null;
        LOGGER.debug("Successfully closed publisher channel");
    }

    protected void publishMessage(Message message, DeliveryOptions deliveryOptions) throws IOException {
        for (int attempt = 1; attempt <= DEFAULT_RETRY_ATTEMPTS; attempt++) {
            if (attempt > 1) {
                LOGGER.info("Attempt {} to send message", attempt);
            }

            try {
                initChannel();
                publishMessage(message, deliveryOptions, channel);
                return;
            } catch (IOException e) {
                handleIoException(attempt, e);
            }
        }
    }

    protected void publishMessages(List<Message> messages, DeliveryOptions deliveryOptions) throws IOException {
        for (Message message : messages) {
            publishMessage(message, deliveryOptions);
        }
    }

    protected void publishMessageConfirmed(Message message, DeliveryOptions deliveryOptions) throws IOException {
        for (int attempt = 1; attempt <= DEFAULT_RETRY_ATTEMPTS; attempt++) {
            if (attempt > 1) {
                LOGGER.info("Attempt {} to send message", attempt);
            }

            try {
                initConfirmedChannel();
                publishMessage(message, deliveryOptions, channel);
                LOGGER.info("Waiting for publisher ack");
                channel.waitForConfirmsOrDie();
                LOGGER.info("Received publisher ack");
                return;
            } catch (IOException e) {
                handleIoException(attempt, e);
            } catch (InterruptedException e) {
                LOGGER.warn("Publishing message interrupted while waiting for producer ack", e);
                return;
            }
        }
    }

    protected void publishMessagesConfirmed(List<Message> messages, DeliveryOptions deliveryOptions) throws IOException {
        for (Message message : messages) {
            publishMessageConfirmed(message, deliveryOptions);
        }
    }

    protected void publishMessageTransactional(Message message, DeliveryOptions deliveryOptions) throws IOException {
        publishMessagesTransactional(Collections.<Message>singletonList(message), deliveryOptions);
    }

    protected void publishMessagesTransactional(List<Message> messages, DeliveryOptions deliveryOptions) throws IOException {
        for (int attempt = 1; attempt <= DEFAULT_RETRY_ATTEMPTS; attempt++) {
            if (attempt > 1) {
                LOGGER.info("Attempt {} to send messages within transaction", attempt);
            }

            try {
                initTransactionalChannel();
                try {
                    for (Message message : messages) {
                        publishMessage(message, deliveryOptions, channel);
                    }
                    LOGGER.info("Committing transaction");
                    commitTransaction();
                    LOGGER.info("Transaction committed");
                } catch (IOException e) {
                    rollbackTransaction();
                    throw e;
                }
                return;
            } catch (IOException e) {
{'fraction_non_alphanumeric': 0.08276652655712648, 'fraction_numerical': 0.0019870080244554834, 'mean_word_length': 2.397196261682243, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 25, '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': '6443653', 'n_tokens_mistral': 3228, 'n_tokens_neox': 3026, 'n_words': 834}