package demo.benchmark;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class RabbitMqMessageBroker extends AbstractMessageBroker {

    private Connection connection;

    private Channel channel;

    public RabbitMqMessageBroker(String port) throws Exception {
        createConnection(port);
    }

    @Override public void createConnection(String port) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        connection = factory.newConnection();
        channel = connection.createChannel();
        channel.queueDeclare(QUEUE, false, false, false, null);
    }

    @Override public void enqueue() throws IOException {
        log().info("Sending....");
        for (int i = 0; i < enqueue_count; i++) {
            channel.basicPublish("", QUEUE, null, String.valueOf(i).getBytes());
        }
    }

    @Override public void dequeue() throws IOException {
        log().info("Receiving....");
        Consumer consumer = new DefaultConsumer(channel);
        for (int i = 0; i < enqueue_count; i++) {
            channel.basicConsume(QUEUE, true, consumer);
        }
    }

    @Override public void closeConnection() throws IOException, TimeoutException {
        channel.close();
        connection.close();
    }
}

@@ -122,16 +122,51 @@
 ception;
+%0Aimport java.util.stream.IntStream;
 %0A%0Apublic
@@ -794,111 +794,246 @@
 e() 
-throws IOException %7B%0A        log().info(%22Sending....%22);%0A        for (int i = 0; i %3C enqueue_count; i++)
+%7B%0A        Runnable runnable = () -%3E %7B%0A            IntStream.range(0, enqueue_count).forEach(i -%3E publish(String.valueOf(i)));%0A        %7D;%0A        submit(producerThreads, runnable);%0A    %7D%0A%0A    private void publish(String data) %7B%0A        try
  %7B%0A 
@@ -1096,17 +1096,20 @@
 valueOf(
-i
+data
 ).getByt
@@ -1111,24 +1111,124 @@
 etBytes());%0A
+        %7D catch (IOException e) %7B%0A            log().error(%22Error publishing data to the queue%22, e);%0A
         %7D%0A  
@@ -1293,45 +1293,8 @@
 n %7B%0A
-        log().info(%22Receiving....%22);%0A
     
@@ -1359,104 +1359,391 @@
     
-for (int i = 0; i %3C enqueue_count; i++) %7B%0A            channel.basicConsume(QUEUE, true, consumer
+Runnable runnable = () -%3E %7B%0A            IntStream.range(0, enqueue_count).forEach(i -%3E consume(consumer));%0A        %7D;%0A        submit(consumerThreads, runnable);%0A    %7D%0A%0A    private void consume(Consumer consumer) %7B%0A        try %7B%0A            channel.basicConsume(QUEUE, true, consumer);%0A        %7D catch (IOException e) %7B%0A            log().error(%22Error consuming data from the queue%22, e
 );%0A 
@@ -1833,32 +1833,72 @@
 eoutException %7B%0A
+        waitForThreadPoolTermination();%0A
         channel.

Refactor RabbitMqbroker