<commit_before>package com.semation.jstatsd.server.messaging.disruptor;

import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Singleton;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.IgnoreExceptionHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.semation.jstatsd.StatsMessage;
import com.semation.jstatsd.server.messaging.MessageProcessor;
import com.semation.jstatsd.server.messaging.NumberOfConsumers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Singleton
class DisruptorMessageProcessor implements MessageProcessor {
    private final EventHandler<StatsMessage> eventHandler;
    private final Provider<StatsMessageParserTranslator> eventTranslatorProvider;
    private final int numberOfConsumers;
    private final Disruptor<StatsMessage> disruptor;

    private static final Logger log = LoggerFactory.getLogger(DisruptorMessageProcessor.class);

    @Inject
    public DisruptorMessageProcessor(Disruptor<StatsMessage> disruptor,
                                     EventHandler<StatsMessage> eventHandler,
                                     Provider<StatsMessageParserTranslator> eventTranslatorProvider,
                                     @NumberOfConsumers int numberOfConsumers) {
        this.disruptor = disruptor;
        this.eventHandler = eventHandler;
        this.eventTranslatorProvider = eventTranslatorProvider;
        this.numberOfConsumers = numberOfConsumers;
    }

    @Override
    public void publish(String message) {
        StatsMessageParserTranslator translator = eventTranslatorProvider.get();
        translator.setMessage(message);
        disruptor.publishEvent(translator);
    }

    @Override
    public void start() {
        log.info("Starting message processor");

        for (int i = 0; i < numberOfConsumers; i++) {
            //noinspection unchecked
            disruptor.handleEventsWith(eventHandler);
        }
        disruptor.handleExceptionsWith(new IgnoreExceptionHandler());
        disruptor.start();
        log.info("Message processor started");
    }
<commit_msg>Add injection for exception handler<commit_after>package com.semation.jstatsd.server.messaging.disruptor;

import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Singleton;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.IgnoreExceptionHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.semation.jstatsd.StatsMessage;
import com.semation.jstatsd.server.messaging.MessageProcessor;
import com.semation.jstatsd.server.messaging.NumberOfConsumers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Singleton
class DisruptorMessageProcessor implements MessageProcessor {
    private final EventHandler<StatsMessage> eventHandler;
    private final Provider<StatsMessageParserTranslator> eventTranslatorProvider;
    private final int numberOfConsumers;
    private final Disruptor<StatsMessage> disruptor;

    private ExceptionHandler exceptionHandler = new IgnoreExceptionHandler();

    private static final Logger log = LoggerFactory.getLogger(DisruptorMessageProcessor.class);

    @Inject
    public DisruptorMessageProcessor(Disruptor<StatsMessage> disruptor,
                                     EventHandler<StatsMessage> eventHandler,
                                     Provider<StatsMessageParserTranslator> eventTranslatorProvider,
                                     @NumberOfConsumers int numberOfConsumers) {
        this.disruptor = disruptor;
        this.eventHandler = eventHandler;
        this.eventTranslatorProvider = eventTranslatorProvider;
        this.numberOfConsumers = numberOfConsumers;
    }

    @Inject(optional = true)
    public void setExceptionHandler(ExceptionHandler exceptionHandler) {
        this.exceptionHandler = exceptionHandler;
    }

    @Override
    public void publish(String message) {
        StatsMessageParserTranslator translator = eventTranslatorProvider.get();
        translator.setMessage(message);
        disruptor.publishEvent(translator);
    }

    @Override
    public void start() {
        log.info("Starting message processor");

        for (int i = 0; i < numberOfConsumers; i++) {
            //noinspection unchecked
            disruptor.handleEventsWith(eventHandler);
        }
        disruptor.handleExceptionsWith(exceptionHandler);
        disruptor.start();
        log.info("Message processor started");
    }

{'fraction_non_alphanumeric': 0.07524271844660194, 'fraction_numerical': 0.001323918799646955, 'mean_word_length': 3.4792490118577075, 'pattern_counts': {'":': 0, '<': 17, '<?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': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6546488', 'n_tokens_mistral': 1210, 'n_tokens_neox': 1188, 'n_words': 224}