package rabbit;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import converters.JsonConverter;
import converters.XmlToJsonConverter;
import datamodel.Job;
import datamodel.Measurement;
import mongo.MongoManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import processors.EventHubAdapter;
import processors.TlsFilter;

import java.io.*;

/**
 * Received messages that come from a queue
 */
public class Receiver {
    private static final Logger logger = LoggerFactory.getLogger(Receiver.class);
    private Channel channel;
    private QueueingConsumer consumer;
    private Connection connection;
    private final String hostName;
    private final String queueName;
    private final String userName;
    private final String password;
    private RabbitMqConfig rmqConf = new RabbitMqConfig();

    /**
     * Set parameters for RabbitMQ receiver
     */
    public Receiver() {
        hostName = rmqConf.getHost();
        queueName = rmqConf.getReceiveQueue();
        userName = rmqConf.getUsername();
        password = rmqConf.getPassword();

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(hostName);
        factory.setUsername(userName);
        factory.setPassword(password);

        try {
            connection = factory.newConnection();
            channel = connection.createChannel();

            /*
            * a client can be assigned only 1 message, the next message will be
            * assigned only after it finishes processing the first message
            */
            int prefetchCount = 1;

            channel.basicQos(prefetchCount);
            boolean durable = true;
            channel.queueDeclare(queueName, durable, false, false, null);
            consumer = new QueueingConsumer(channel);
            boolean autoAck = false;
            channel.basicConsume(queueName, autoAck, consumer);


        } catch (IOException e) {
            logger.error(e.getMessage(), e);
        }
    }

    /**
     * Listen for messages
     */
    public void startReceiving() {
        Sender sender = new Sender();
        while (true) {
            try {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                logger.info("Received message over the queue");
                Job job = JsonConverter.jsonStringToObject(message, Job.class);

                MongoManager mm = new MongoManager();
                String measurementString = mm.pullJsonById(job.getId());

                Measurement measurement = JsonConverter.jsonStringToObject(measurementString, Measurement.class);
                String jsonResult = XmlToJsonConverter.convertXmlToJson(measurement.getRawResult());
                logger.info("Converted results: " + jsonResult);

                TlsFilter tlsFilter = new TlsFilter();
                String filteredJson = tlsFilter.process(jsonResult, null);
                String eventHubMessageString =
                        new EventHubAdapter().adaptMessage(filteredJson, job, measurement);

                mm.updateJsonWithId(job.getId(), "processedResult", eventHubMessageString);
                mm.closeConnection();

                // send job over the queue
                sender.Send(job);

                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                logger.info("Sent message over queue.");
            } catch (InterruptedException e) {
                logger.error(e.getMessage());
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
            } catch (Throwable throwable) {
                throwable.printStackTrace();
            }
        }
    }

    /**
     * Close receiving connection.
     */
    public void closeConnection() {
        try {
            if (connection.isOpen()) {
                connection.close();
            }
        } catch (IOException e) {
            logger.error(e.getMessage(), e);
        }
    }

}

@@ -213,46 +213,8 @@
 er;%0A
-import converters.XmlToJsonConverter;%0A
 impo
@@ -367,51 +367,25 @@
 ors.
-EventHubAdapter;%0Aimport processors.TlsFilte
+ProcessorsExecuto
 r;%0A%0A
@@ -2720,24 +2720,213 @@
 ment.class);
+%0A%0A                ProcessorsExecutor pe = new ProcessorsExecutor();%0A                String processedString = pe.executeProcesses(measurement.getRawResult(), measurement);%0A                /*
 %0A           
@@ -3077,24 +3077,25 @@
 onResult);%0A%0A
+%0A
             
@@ -3348,24 +3348,24 @@
 surement);%0A%0A
-
             
@@ -3444,16 +3444,121 @@
 tring);%0A
+                */%0A                mm.updateJsonWithId(job.getId(), %22processedResult%22, processedString);%0A
         

Use "ProcessorExecutor".