Refactor and add number of threads selector.

src/main/java/attack/AttackHandler.java

@@ -8,33 +8,54 @@
 import data.ConnectionMessage;
 import data.WebSocketConnectionMessage;
 import logger.Logger;
+import logger.LoggerLevel;
 import org.python.util.PythonInterpreter;
+import queue.SendMessageQueueConsumer;
+import queue.TableBlockingQueueConsumer;
 import queue.TableBlockingQueueProducer;
+import ui.attack.table.WebSocketMessageTableModel;
 
 import java.time.LocalDateTime;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 public class AttackHandler
 {
+    private final Logger logger;
+    private final BlockingQueue<WebSocketConnectionMessage> sendMessageQueue;
+    private final BlockingQueue<ConnectionMessage> tableBlockingQueue;
+    private final WebSocketMessageTableModel webSocketMessageTableModel;
+    private final WebSocketMessage baseWebSocketMessage;
+    private final AtomicBoolean isAttackRunning;
     private final PythonInterpreter interpreter;
+    private ExecutorService sendMessageExecutorService;
+    private ExecutorService tableExecutorService;
 
     public AttackHandler(
             Logger logger,
             WebSockets webSockets,
-            AtomicBoolean isProcessing,
             BlockingQueue<WebSocketConnectionMessage> sendMessageQueue,
             BlockingQueue<ConnectionMessage> tableBlockingQueue,
-            WebSocketMessage baseWebSocketMessage
+            WebSocketMessageTableModel webSocketMessageTableModel,
+            WebSocketMessage baseWebSocketMessage,
+            AtomicBoolean isAttackRunning
     )
     {
+        this.logger = logger;
+        this.sendMessageQueue = sendMessageQueue;
+        this.tableBlockingQueue = tableBlockingQueue;
+        this.webSocketMessageTableModel = webSocketMessageTableModel;
+        this.baseWebSocketMessage = baseWebSocketMessage;
+        this.isAttackRunning = isAttackRunning;
         interpreter = new PythonInterpreter();
         interpreter.setOut(logger.outputStream());
         interpreter.setErr(logger.errorStream());
 
         interpreter.set("base_websocket", baseWebSocketMessage);
 
-        interpreter.set("websocket_connection", new ConnectionFactory(logger, webSockets, isProcessing, sendMessageQueue));
+        interpreter.set("websocket_connection", new ConnectionFactory(logger, webSockets, sendMessageQueue, isAttackRunning));
 
         interpreter.set("results_table", new TableBlockingQueueProducer(logger, tableBlockingQueue));
     }
@@ -58,6 +79,57 @@ public void executeCallback(WebSocketConnectionMessage webSocketConnectionMessag
         interpreter.exec(String.format("%s(%s)", callbackMethod, messageParameterName));
     }
 
+    public BlockingQueue<WebSocketConnectionMessage> getSendMessageQueue()
+    {
+        return sendMessageQueue;
+    }
+
+    public BlockingQueue<ConnectionMessage> getTableBlockingQueue()
+    {
+        return tableBlockingQueue;
+    }
+
+    public WebSocketMessageTableModel getWebSocketMessageTableModel()
+    {
+        return webSocketMessageTableModel;
+    }
+
+    public WebSocketMessage getBaseWebSocketMessage()
+    {
+        return baseWebSocketMessage;
+    }
+
+    public AtomicBoolean getIsAttackRunning()
+    {
+        return isAttackRunning;
+    }
+
+    public void startConsumers(int numberOfSendThreads)
+    {
+
+        sendMessageExecutorService = Executors.newFixedThreadPool(numberOfSendThreads);
+        sendMessageExecutorService.execute(new SendMessageQueueConsumer(logger, this, isAttackRunning));
+
+        logger.logOutput(LoggerLevel.DEBUG, "Number of threads attack started with: " + numberOfSendThreads);
+
+        tableExecutorService = Executors.newSingleThreadExecutor();
+        tableExecutorService.execute(new TableBlockingQueueConsumer(logger, webSocketMessageTableModel, tableBlockingQueue, isAttackRunning));
+
+        logger.logOutput(LoggerLevel.DEBUG, "Table thread started.");
+    }
+
+    public void shutdownConsumers()
+    {
+        sendMessageExecutorService.shutdownNow();
+        logger.logOutput(LoggerLevel.DEBUG, "sendMessageExecutorService shutdown? " + sendMessageExecutorService.isShutdown());
+
+        tableExecutorService.shutdownNow();
+        logger.logOutput(LoggerLevel.DEBUG, "tableExecutorService shutdown? " + tableExecutorService.isShutdown());
+
+        sendMessageQueue.clear();
+        tableBlockingQueue.clear();
+    }
+
     private static class DecoratedConnectionMessage implements ConnectionMessage
     {
         private final WebSocketConnectionMessage webSocketConnectionMessage;

src/main/java/connection/ConnectionFactory.java

@@ -12,24 +12,24 @@ public class ConnectionFactory
 {
     private final Logger logger;
     private final WebSockets webSockets;
-    private final AtomicBoolean isProcessing;
     private final BlockingQueue<WebSocketConnectionMessage> sendMessageQueue;
+    private final AtomicBoolean isAttackRunning;
 
     public ConnectionFactory(
             Logger logger,
             WebSockets webSockets,
-            AtomicBoolean isProcessing,
-            BlockingQueue<WebSocketConnectionMessage> sendMessageQueue
+            BlockingQueue<WebSocketConnectionMessage> sendMessageQueue,
+            AtomicBoolean isAttackRunning
     )
     {
         this.logger = logger;
         this.webSockets = webSockets;
-        this.isProcessing = isProcessing;
         this.sendMessageQueue = sendMessageQueue;
+        this.isAttackRunning = isAttackRunning;
     }
 
     public Connection create(WebSocketMessage baseWebSocketMessage)
     {
-        return new WebSocketConnection(logger, webSockets, isProcessing, baseWebSocketMessage, sendMessageQueue);
+        return new WebSocketConnection(logger, webSockets, sendMessageQueue, baseWebSocketMessage, isAttackRunning);
     }
 }

src/main/java/connection/WebSocketConnection.java

@@ -18,30 +18,30 @@ public class WebSocketConnection implements Connection
 {
     private final Logger logger;
     private final WebSockets webSockets;
-    private final AtomicBoolean isProcessing;
     private final BlockingQueue<WebSocketConnectionMessage> sendMessageQueue;
+    private final AtomicBoolean isAttackRunning;
     private final ExtensionWebSocket extensionWebSocket;
 
     WebSocketConnection(
             Logger logger,
             WebSockets webSockets,
-            AtomicBoolean isProcessing,
+            BlockingQueue<WebSocketConnectionMessage> sendMessageQueue,
             WebSocketMessage baseWebSocketMessage,
-            BlockingQueue<WebSocketConnectionMessage> sendMessageQueue
+            AtomicBoolean isAttackRunning
     )
     {
         this.logger = logger;
         this.webSockets = webSockets;
-        this.isProcessing = isProcessing;
         this.sendMessageQueue = sendMessageQueue;
+        this.isAttackRunning = isAttackRunning;
 
         extensionWebSocket = createExtensionWebSocket(baseWebSocketMessage);
     }
 
     @Override
     public void queue(String payload)
     {
-        if (isProcessing.get())
+        if (isAttackRunning.get())
         {try
             {
                 sendMessageQueue.put(new WebSocketConnectionMessage(payload, Direction.CLIENT_TO_SERVER, LocalDateTime.now(), null, this));

src/main/java/logger/Logger.java

@@ -17,7 +17,7 @@ public Logger(Logging logging)
     {
         this.logging = logging;
 
-        debugLogLevel  = false;
+        debugLogLevel  = true;
         errorLogLevel = true;
     }
 

src/main/java/queue/SendMessageQueueConsumer.java

@@ -12,28 +12,28 @@
 public class SendMessageQueueConsumer implements Runnable
 {
     private final Logger logger;
-    private final AtomicBoolean isProcessing;
-    private final BlockingQueue<WebSocketConnectionMessage> sendMessageQueue;
     private final AttackHandler attackHandler;
+    private final AtomicBoolean isAttackRunning;
+    private final BlockingQueue<WebSocketConnectionMessage> sendMessageQueue;
 
     public SendMessageQueueConsumer(
             Logger logger,
-            AtomicBoolean isProcessing,
-            BlockingQueue<WebSocketConnectionMessage> sendMessageQueue,
-            AttackHandler attackHandler
+            AttackHandler attackHandler,
+            AtomicBoolean isAttackRunning
     )
     {
         this.logger = logger;
-        this.isProcessing = isProcessing;
-        this.sendMessageQueue = sendMessageQueue;
         this.attackHandler = attackHandler;
+        this.isAttackRunning = isAttackRunning;
+
+        sendMessageQueue = attackHandler.getSendMessageQueue();
     }
 
 
     @Override
     public void run()
     {
-        while (isProcessing.get())
+        while (isAttackRunning.get())
         {
             try
             {

src/main/java/queue/TableBlockingQueueConsumer.java

@@ -14,25 +14,25 @@ public class TableBlockingQueueConsumer implements Runnable
     private final Logger logger;
     private final BlockingQueue<ConnectionMessage> queue;
     private final WebSocketMessageTableModel tableModel;
-    private final AtomicBoolean isRunning;
+    private final AtomicBoolean isAttackRunning;
 
     public TableBlockingQueueConsumer(
             Logger logger,
-            BlockingQueue<ConnectionMessage> queue,
             WebSocketMessageTableModel tableModel,
-            AtomicBoolean isRunning
+            BlockingQueue<ConnectionMessage> queue,
+            AtomicBoolean isAttackRunning
     )
     {
         this.logger = logger;
         this.queue = queue;
         this.tableModel = tableModel;
-        this.isRunning = isRunning;
+        this.isAttackRunning = isAttackRunning;
     }
 
     @Override
     public void run()
     {
-        while (isRunning.get())
+        while (isAttackRunning.get())
         {
             try
             {

src/main/java/queue/TableBlockingQueueProducer.java

@@ -17,7 +17,6 @@ public TableBlockingQueueProducer(
             BlockingQueue<ConnectionMessage> tableBlockingQueue
     )
     {
-
         this.logger = logger;
         this.tableBlockingQueue = tableBlockingQueue;
     }

src/main/java/ui/WebSocketFrame.java

@@ -10,6 +10,7 @@
 import data.WebSocketConnectionMessage;
 import logger.Logger;
 import ui.attack.WebSocketAttackPanel;
+import ui.attack.table.WebSocketMessageTableModel;
 import ui.editor.WebSocketEditorPanel;
 
 import javax.swing.*;
@@ -27,8 +28,8 @@ public class WebSocketFrame extends JFrame
     private final Persistence persistence;
     private final WebSockets webSockets;
     private final WebSocketMessage webSocketMessage;
-    private final AtomicBoolean isProcessing;
-    private final AtomicBoolean isRunning;
+    private final AtomicBoolean isAttackRunning;
+    private AttackHandler attackHandler;
 
     public WebSocketFrame(
             Logger logger,
@@ -44,8 +45,7 @@ public WebSocketFrame(
         this.webSockets = webSockets;
         this.webSocketMessage = webSocketMessage;
 
-        isProcessing = new AtomicBoolean(true);
-        isRunning = new AtomicBoolean(true);
+        isAttackRunning = new AtomicBoolean(true);
 
         initComponents();
 
@@ -54,7 +54,8 @@ public WebSocketFrame(
             @Override
             public void windowClosed(WindowEvent e)
             {
-                isRunning.set(false);
+                isAttackRunning.set(false);
+                attackHandler.shutdownConsumers();
             }
         });
     }
@@ -72,10 +73,12 @@ private void initComponents()
         BlockingQueue<WebSocketConnectionMessage> sendMessageQueue = new LinkedBlockingQueue<>();
         BlockingQueue<ConnectionMessage> tableBlockingQueue = new LinkedBlockingQueue<>();
 
-        AttackHandler attackHandler = new AttackHandler(logger, webSockets, isProcessing, sendMessageQueue, tableBlockingQueue, webSocketMessage);
+        WebSocketMessageTableModel webSocketMessageTableModel = new WebSocketMessageTableModel();
 
-        cardDeck.add(new WebSocketEditorPanel(logger, userInterface, persistence, cardLayout, cardDeck, attackHandler, webSocketMessage), "editorPanel");
-        cardDeck.add(new WebSocketAttackPanel(logger, userInterface, cardLayout, cardDeck, attackHandler, sendMessageQueue, tableBlockingQueue, isProcessing, isRunning), "attackPanel");
+        attackHandler = new AttackHandler(logger, webSockets, sendMessageQueue, tableBlockingQueue, webSocketMessageTableModel, webSocketMessage, isAttackRunning);
+
+        cardDeck.add(new WebSocketEditorPanel(logger, userInterface, persistence, cardLayout, cardDeck, attackHandler), "editorPanel");
+        cardDeck.add(new WebSocketAttackPanel(userInterface, cardLayout, cardDeck, attackHandler), "attackPanel");
 
         this.getContentPane().add(cardDeck);
         this.pack();