<commit_before>package no.ntnu.okse.protocol.mqtt;

import io.moquette.BrokerConstants;
import io.moquette.interception.AbstractInterceptHandler;
import io.moquette.interception.InterceptHandler;
import io.moquette.interception.messages.InterceptPublishMessage;
import io.moquette.interception.messages.InterceptSubscribeMessage;
import io.moquette.proto.messages.AbstractMessage;
import io.moquette.proto.messages.PublishMessage;
import io.moquette.server.Server;
import io.moquette.server.config.IConfig;
import io.moquette.server.config.MemoryConfig;
import io.moquette.spi.ClientSession;
import no.ntnu.okse.core.messaging.Message;
import no.ntnu.okse.core.messaging.MessageService;
import no.ntnu.okse.core.subscription.Publisher;
import no.ntnu.okse.core.subscription.Subscriber;
import no.ntnu.okse.core.subscription.SubscriptionService;
import no.ntnu.okse.core.topic.TopicService;
import org.apache.log4j.Logger;

import io.moquette.spi.impl.ProtocolProcessor;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;

public class MQTTServer extends Server {

	private static Logger log = Logger.getLogger(Server.class);
	private static String protocolServerType;

	class MQTTListener extends AbstractInterceptHandler {
		@Override
		public void onPublish(InterceptPublishMessage message) {
			//TODO: We need to get the publisher that sent the message, somehow
			Publisher pub = new Publisher( message.getTopicName(), "127.0.0.1", 1883, protocolServerType);
			ByteBuffer buffer = message.getPayload();
			String payload = new String(buffer.array(), buffer.position(), buffer.limit());
			String topic = message.getTopicName();
			MessageService.getInstance().distributeMessage(
					new Message( payload, topic, pub, protocolServerType )
			);
		}

		@Override
		public void onSubscribe(InterceptSubscribeMessage message) {
			log.info("Client subscribed to: "  + message.getTopicFilter() + "   ID: " + message.getClientID());

			TopicService.getInstance().addTopic( message.getTopicFilter() );
			Subscriber sub = new Subscriber( "127.0.0.1", 1309, message.getTopicFilter(), protocolServerType );
			SubscriptionService.getInstance().addSubscriber(sub);

			super.onSubscribe(message);
		}

	}

	public void init(String host, int port) {
		List<InterceptHandler> interceptHandlers = new ArrayList<>();
		interceptHandlers.add(new MQTTListener());

		protocolServerType = "MQTT";

		Properties properties = new Properties();
		properties.setProperty(BrokerConstants.HOST_PROPERTY_NAME, host);
		properties.setProperty(BrokerConstants.PORT_PROPERTY_NAME, "" + port);
		properties.setProperty(BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, "25342");

		final IConfig config = new MemoryConfig(properties);
		try {
			startServer(config, interceptHandlers);
		} catch (IOException e) {
			e.printStackTrace();
		}

	}

	/**
	 * Sends the message to any subscriber that is subscribed to the topic that the message was sent to
	 * @param message is the message that is sent from OKSE core
	 * */
	public void sendMessage(Message message) {
		HashSet<Subscriber> subscribers = SubscriptionService.getInstance().getAllSubscribersForTopic( message.getTopic() );
		for(Subscriber subscriber : subscribers){
			PublishMessage msg = new PublishMessage();
			ByteBuffer payload = ByteBuffer.wrap( message.getMessage().getBytes() );
			String topicName = message.getTopic();

			msg.setPayload( payload );
			msg.setTopicName( topicName );

			msg.setQos(AbstractMessage.QOSType.LEAST_ONE);
			internalPublish(msg);
		}

	}
}<commit_msg>Added some TODO's<commit_after>package no.ntnu.okse.protocol.mqtt;

import io.moquette.BrokerConstants;
import io.moquette.interception.AbstractInterceptHandler;
import io.moquette.interception.InterceptHandler;
import io.moquette.interception.messages.InterceptPublishMessage;
import io.moquette.interception.messages.InterceptSubscribeMessage;
import io.moquette.proto.messages.AbstractMessage;
import io.moquette.proto.messages.PublishMessage;
import io.moquette.server.Server;
import io.moquette.server.config.IConfig;
import io.moquette.server.config.MemoryConfig;
import io.moquette.spi.ClientSession;
import no.ntnu.okse.core.messaging.Message;
import no.ntnu.okse.core.messaging.MessageService;
import no.ntnu.okse.core.subscription.Publisher;
import no.ntnu.okse.core.subscription.Subscriber;
import no.ntnu.okse.core.subscription.SubscriptionService;
import no.ntnu.okse.core.topic.TopicService;
import org.apache.log4j.Logger;

import io.moquette.spi.impl.ProtocolProcessor;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;

public class MQTTServer extends Server {

	private static Logger log = Logger.getLogger(Server.class);
	private static String protocolServerType;

	class MQTTListener extends AbstractInterceptHandler {
		@Override
		public void onPublish(InterceptPublishMessage message) {
			//TODO: We need to get the publisher that sent the message, somehow
			//TODO: So that we get the host and can send the message to the correct address, same with the port
			Publisher pub = new Publisher( message.getTopicName(), "127.0.0.1", 1883, protocolServerType);
			ByteBuffer buffer = message.getPayload();
			String payload = new String(buffer.array(), buffer.position(), buffer.limit());
			String topic = message.getTopicName();
			MessageService.getInstance().distributeMessage(
					new Message( payload, topic, pub, protocolServerType )
			);
		}

		@Override
		public void onSubscribe(InterceptSubscribeMessage message) {
			log.info("Client subscribed to: "  + message.getTopicFilter() + "   ID: " + message.getClientID());

			//TODO: We need to get the publisher that sent the message, somehow
			//TODO: So that we get the host and can send the message to the correct address, same with the port
			TopicService.getInstance().addTopic( message.getTopicFilter() );
			Subscriber sub = new Subscriber( "127.0.0.1", 1309, message.getTopicFilter(), protocolServerType );
			SubscriptionService.getInstance().addSubscriber(sub);

			super.onSubscribe(message);
		}

	}

	public void init(String host, int port) {
		List<InterceptHandler> interceptHandlers = new ArrayList<>();
		interceptHandlers.add(new MQTTListener());

		protocolServerType = "MQTT";

		Properties properties = new Properties();
		properties.setProperty(BrokerConstants.HOST_PROPERTY_NAME, host);
		properties.setProperty(BrokerConstants.PORT_PROPERTY_NAME, "" + port);
		properties.setProperty(BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, "25342");

		final IConfig config = new MemoryConfig(properties);
		try {
			startServer(config, interceptHandlers);
		} catch (IOException e) {
			e.printStackTrace();
		}

	}

	/**
	 * Sends the message to any subscriber that is subscribed to the topic that the message was sent to
	 * @param message is the message that is sent from OKSE core
	 * */
	public void sendMessage(Message message) {
		HashSet<Subscriber> subscribers = SubscriptionService.getInstance().getAllSubscribersForTopic( message.getTopic() );
		for(Subscriber subscriber : subscribers){
			if( subscriber.getOriginProtocol() == protocolServerType ){
				PublishMessage msg = new PublishMessage();
				ByteBuffer payload = ByteBuffer.wrap( message.getMessage().getBytes() );
				String topicName = message.getTopic();

				msg.setPayload( payload );
				msg.setTopicName( topicName );

				msg.setQos(AbstractMessage.QOSType.LEAST_ONE);
				internalPublish(msg);
			}
		}

	}
}
{'fraction_non_alphanumeric': 0.10465738758029978, 'fraction_numerical': 0.006959314775160599, 'mean_word_length': 5.217138103161398, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, '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': '7246491', 'n_tokens_mistral': 2405, 'n_tokens_neox': 2148, 'n_words': 505}