<commit_before>package com.github.longkerdandy.mithril.mqtt.communicator.kafka.broker;

import com.github.longkerdandy.mithril.mqtt.api.comm.BrokerListener;
import com.github.longkerdandy.mithril.mqtt.api.internal.*;
import kafka.consumer.KafkaStream;
import kafka.message.MessageAndMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Broker Communicator Worker for Kafka
 */
public class KafkaBrokerWorker implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(KafkaBrokerWorker.class);

    private final KafkaStream<String, InternalMessage> stream;
    private final BrokerListener listener;

    public KafkaBrokerWorker(KafkaStream<String, InternalMessage> stream, BrokerListener listener) {
        this.stream = stream;
        this.listener = listener;
    }

    @SuppressWarnings("unchecked")
    @Override
    public void run() {
        // reads from Kafka until stop
        for (MessageAndMetadata<String, InternalMessage> m : this.stream) {
            InternalMessage msg = m.message();
            if (msg != null) {
                switch (msg.getMessageType()) {
                    case PUBLISH:
                        this.listener.onPublish((InternalMessage<Publish>) msg);
                        break;
                    case DISCONNECT:
                        this.listener.onDisconnect(msg);
                        break;
                    default:
                        logger.warn("Broker received unexpected message type {}", msg.getMessageType());
                }
            }
        }
    }
}
<commit_msg>Reformat<commit_after>package com.github.longkerdandy.mithril.mqtt.communicator.kafka.broker;

import com.github.longkerdandy.mithril.mqtt.api.comm.BrokerListener;
import com.github.longkerdandy.mithril.mqtt.api.internal.InternalMessage;
import com.github.longkerdandy.mithril.mqtt.api.internal.Publish;
import kafka.consumer.KafkaStream;
import kafka.message.MessageAndMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Broker Communicator Worker for Kafka
 */
public class KafkaBrokerWorker implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(KafkaBrokerWorker.class);

    private final KafkaStream<String, InternalMessage> stream;
    private final BrokerListener listener;

    public KafkaBrokerWorker(KafkaStream<String, InternalMessage> stream, BrokerListener listener) {
        this.stream = stream;
        this.listener = listener;
    }

    @SuppressWarnings("unchecked")
    @Override
    public void run() {
        // reads from Kafka until stop
        for (MessageAndMetadata<String, InternalMessage> m : this.stream) {
            InternalMessage msg = m.message();
            if (msg != null) {
                switch (msg.getMessageType()) {
                    case PUBLISH:
                        this.listener.onPublish((InternalMessage<Publish>) msg);
                        break;
                    case DISCONNECT:
                        this.listener.onDisconnect(msg);
                        break;
                    default:
                        logger.warn("Broker received unexpected message type {}", msg.getMessageType());
                }
            }
        }
    }
}

{'fraction_non_alphanumeric': 0.08919330289193303, 'fraction_numerical': 0.0012176560121765602, 'mean_word_length': 2.325910931174089, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, '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': '7256211', 'n_tokens_mistral': 892, 'n_tokens_neox': 858, 'n_words': 178}