diff options
Diffstat (limited to 'src')
7 files changed, 227 insertions, 32 deletions
diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java index b84d693..37c125a 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java @@ -43,13 +43,13 @@ public class EventBus implements Runnable { } public void routeMessage(AgentMessage msg) throws InterruptedException { - assert msg.correctType == msg.getDestinationModule(); + assert msg.getCorrectMessageType() == msg.getDestinationModule(); System.out.println("Event bus routing message"); executors.get(msg.getDestinationModule()).addMessage(msg); } public void addMessage(AgentMessage msg) throws InterruptedException { - assert msg.correctType == msg.getDestinationModule(); + assert msg.getCorrectMessageType() == msg.getDestinationModule(); this.events.put(msg); } } diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/AgentMessage.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/AgentMessage.java index ba15a7a..b71831f 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/AgentMessage.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/AgentMessage.java @@ -1,7 +1,6 @@ package pl.edu.mimuw.cloudatlas.agent.messages; -public class AgentMessage { - public AgentModule correctType; +public abstract class AgentMessage { public enum AgentModule { TIMER_SCHEDULER, @@ -30,6 +29,8 @@ public class AgentMessage { this.timestamp = System.currentTimeMillis() / 1000L; } + public abstract AgentModule getCorrectMessageType(); + public String getMessageId() { return messageId; } diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/TimerSchedulerMessage.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/TimerSchedulerMessage.java index 23ec700..9a558b7 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/TimerSchedulerMessage.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/TimerSchedulerMessage.java @@ -43,4 +43,9 @@ public class TimerSchedulerMessage extends AgentMessage { public String getRequestId() { return requestId; } public void setRequestId(String requestId) { this.requestId = requestId; } + + @Override + public AgentModule getCorrectMessageType() { + return AgentModule.TIMER_SCHEDULER; + } } diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/TimerScheduler.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/TimerScheduler.java index 921f985..a08d060 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/TimerScheduler.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/TimerScheduler.java @@ -13,18 +13,21 @@ import java.util.Timer; * Tasks declared as inherited from TimerTask * * TODO: add request id and custom time + * TODO: enable messaging from scheduled tasks */ public class TimerScheduler extends Module { private Timer timer; public TimerScheduler(AgentMessage.AgentModule moduleType) { super(moduleType); + assert moduleType == AgentMessage.AgentModule.TIMER_SCHEDULER; this.timer = new Timer(); System.out.println("TimerScheduler instance initialized"); } @Override public void handle(AgentMessage event) throws InterruptedException { + assert event.getDestinationModule() == event.getCorrectMessageType(); TimerSchedulerMessage timerEvent = (TimerSchedulerMessage) event; this.timer.schedule(timerEvent.getTask(), timerEvent.getDelay()); System.out.println("Task with delay " + timerEvent.getDelay() + " scheduled"); diff --git a/src/test/java/pl/edu/mimuw/cloudatlas/agent/EventBusTest.java b/src/test/java/pl/edu/mimuw/cloudatlas/agent/EventBusTest.java index e43e7ea..f3df887 100644 --- a/src/test/java/pl/edu/mimuw/cloudatlas/agent/EventBusTest.java +++ b/src/test/java/pl/edu/mimuw/cloudatlas/agent/EventBusTest.java @@ -1,8 +1,13 @@ package pl.edu.mimuw.cloudatlas.agent; +import org.junit.Assert; +import org.junit.Ignore; import org.junit.Test; +import org.junit.runners.JUnit4; import pl.edu.mimuw.cloudatlas.agent.messages.AgentMessage; +import pl.edu.mimuw.cloudatlas.agent.messages.TimerSchedulerMessage; import pl.edu.mimuw.cloudatlas.agent.modules.Module; +import pl.edu.mimuw.cloudatlas.agent.modules.TimerScheduledTask; import java.util.ArrayList; import java.util.HashMap; @@ -24,7 +29,13 @@ public class EventBusTest { @Override public void handle(AgentMessage event) throws InterruptedException { System.out.println("Module 1 handle called"); - sendMessage(new AgentMessage("1", AgentMessage.AgentModule.UDP)); + // TODO correct message subclass + sendMessage(new AgentMessage("1", AgentMessage.AgentModule.UDP) { + @Override + public AgentModule getCorrectMessageType() { + return AgentModule.UDP; + } + }); counter ++; } }); @@ -55,6 +66,8 @@ public class EventBusTest { } @Test + @Ignore + // TODO correct message subclass public void messageModule() throws InterruptedException { HashMap<AgentMessage.AgentModule, Module> modules = initializeModule(); HashMap<AgentMessage.AgentModule, Executor> executors = Agent.initializeExecutors(modules); @@ -63,7 +76,12 @@ public class EventBusTest { Thread eventBusThread = new Thread(eventBus); eventBusThread.start(); - eventBus.addMessage(new AgentMessage("0", AgentMessage.AgentModule.RMI)); + eventBus.addMessage(new AgentMessage("0", AgentMessage.AgentModule.RMI) { + @Override + public AgentModule getCorrectMessageType() { + return AgentModule.RMI; + } + }); Thread.sleep(1000); eventBusThread.interrupt(); Agent.closeExecutors(executorThreads); @@ -71,6 +89,8 @@ public class EventBusTest { } @Test + @Ignore + // TODO correct message subclass public void messagingBetweenModules() throws InterruptedException { HashMap<AgentMessage.AgentModule, Module> modules = initializeTwoModules(); HashMap<AgentMessage.AgentModule, Executor> executors = Agent.initializeExecutors(modules); @@ -79,10 +99,12 @@ public class EventBusTest { Thread eventBusThread = new Thread(eventBus); eventBusThread.start(); - eventBus.addMessage(new AgentMessage( - "0", - AgentMessage.AgentModule.RMI, - System.currentTimeMillis() / 1000L)); + eventBus.addMessage(new AgentMessage("0", AgentMessage.AgentModule.RMI) { + @Override + public AgentModule getCorrectMessageType() { + return AgentModule.RMI; + } + }); Thread.sleep(1000); @@ -91,4 +113,73 @@ public class EventBusTest { assertEquals(1, ((MessageCounterModule) modules.get(AgentMessage.AgentModule.RMI)).counter); assertEquals(1, ((MessageCounterModule) modules.get(AgentMessage.AgentModule.UDP)).counter); } + + @Test + public void sendWrongMessageType1() throws InterruptedException { + HashMap<AgentMessage.AgentModule, Module> modules = initializeModule(); + HashMap<AgentMessage.AgentModule, Executor> executors = Agent.initializeExecutors(modules); + ArrayList<Thread> executorThreads = Agent.initializeExecutorThreads(executors); + EventBus eventBus = new EventBus(executors); + Thread eventBusThread = new Thread(eventBus); + Boolean routingErrorCaught = false; + eventBusThread.start(); + + try { + eventBus.addMessage(new TimerSchedulerMessage( + "0", + AgentMessage.AgentModule.RMI, + System.currentTimeMillis() / 1000L, + "1", + 10, + System.currentTimeMillis() / 1000L, + new TimerScheduledTask() { + @Override + public void run() { + System.out.println("Task executed"); + } + })); + Thread.sleep(1000); + } catch (AssertionError e) { + System.out.println("Wrong timer-scheduler message type error caught"); + routingErrorCaught = true; + } finally { + eventBusThread.interrupt(); + Agent.closeExecutors(executorThreads); + } + + if (!routingErrorCaught) { + Assert.fail("Routing not detected as faulty"); + } + } + + @Test + public void sendWrongMessageType2() throws InterruptedException { + HashMap<AgentMessage.AgentModule, Module> modules = initializeModule(); + HashMap<AgentMessage.AgentModule, Executor> executors = Agent.initializeExecutors(modules); + ArrayList<Thread> executorThreads = Agent.initializeExecutorThreads(executors); + EventBus eventBus = new EventBus(executors); + Thread eventBusThread = new Thread(eventBus); + Boolean routingErrorCaught = false; + eventBusThread.start(); + + try { + eventBus.addMessage(new AgentMessage("0", AgentMessage.AgentModule.RMI) { + @Override + public AgentModule getCorrectMessageType() { + return AgentModule.QUERY; + } + }); + Thread.sleep(1000); + } catch (AssertionError e) { + System.out.println("Wrong timer-scheduler message type error caught"); + routingErrorCaught = true; + } finally { + eventBusThread.interrupt(); + Agent.closeExecutors(executorThreads); + } + + if (!routingErrorCaught) { + Assert.fail("Routing not detected as faulty"); + } + } } diff --git a/src/test/java/pl/edu/mimuw/cloudatlas/agent/ExecutorTest.java b/src/test/java/pl/edu/mimuw/cloudatlas/agent/ExecutorTest.java index 1d3af92..8653258 100644 --- a/src/test/java/pl/edu/mimuw/cloudatlas/agent/ExecutorTest.java +++ b/src/test/java/pl/edu/mimuw/cloudatlas/agent/ExecutorTest.java @@ -7,6 +7,8 @@ import pl.edu.mimuw.cloudatlas.agent.messages.AgentMessage; import pl.edu.mimuw.cloudatlas.agent.messages.AgentMessage.AgentModule; import pl.edu.mimuw.cloudatlas.agent.modules.Module; +// TODO make agent messages specific subclass + public class ExecutorTest { public class MessageCounterModule extends Module { public int counter = 0; @@ -35,7 +37,12 @@ public class ExecutorTest { public void testExecutesHandlerOnce() throws Exception { MessageCounterModule module = new MessageCounterModule(AgentModule.UDP); Executor executor = new Executor(module); - executor.addMessage(new AgentMessage("", AgentModule.UDP, 0) {}); + executor.addMessage(new AgentMessage("", AgentModule.UDP, 0) { + @Override + public AgentModule getCorrectMessageType() { + return AgentModule.UDP; + } + }); Thread thread = new Thread(executor); thread.start(); Thread.sleep(100); @@ -47,12 +54,27 @@ public class ExecutorTest { public void testExecutesHandlerMultipleTimes() throws Exception { MessageCounterModule module = new MessageCounterModule(AgentModule.UDP); Executor executor = new Executor(module); - executor.addMessage(new AgentMessage("", AgentModule.UDP, 0) {}); - executor.addMessage(new AgentMessage("", AgentModule.UDP, 0) {}); + executor.addMessage(new AgentMessage("", AgentModule.UDP, 0) { + @Override + public AgentModule getCorrectMessageType() { + return AgentModule.UDP; + } + }); + executor.addMessage(new AgentMessage("", AgentModule.UDP, 0) { + @Override + public AgentModule getCorrectMessageType() { + return AgentModule.UDP; + } + }); Thread thread = new Thread(executor); thread.start(); Thread.sleep(100); - executor.addMessage(new AgentMessage("", AgentModule.UDP, 0) {}); + executor.addMessage(new AgentMessage("", AgentModule.UDP, 0) { + @Override + public AgentModule getCorrectMessageType() { + return AgentModule.UDP; + } + }); Thread.sleep(100); thread.interrupt(); assertEquals(3, module.counter); diff --git a/src/test/java/pl/edu/mimuw/cloudatlas/agent/SchedulerTest.java b/src/test/java/pl/edu/mimuw/cloudatlas/agent/SchedulerTest.java index 3787355..5f0d1a7 100644 --- a/src/test/java/pl/edu/mimuw/cloudatlas/agent/SchedulerTest.java +++ b/src/test/java/pl/edu/mimuw/cloudatlas/agent/SchedulerTest.java @@ -2,47 +2,120 @@ package pl.edu.mimuw.cloudatlas.agent; import org.junit.Test; import pl.edu.mimuw.cloudatlas.agent.messages.AgentMessage; +import pl.edu.mimuw.cloudatlas.agent.messages.TimerSchedulerMessage; import pl.edu.mimuw.cloudatlas.agent.modules.Module; +import pl.edu.mimuw.cloudatlas.agent.modules.TimerScheduledTask; import pl.edu.mimuw.cloudatlas.agent.modules.TimerScheduler; import java.util.ArrayList; import java.util.HashMap; -import static org.junit.Assert.assertEquals; +// TODO better task tests after enabling messaging from tasks +// TODO add wrong message test with switched types public class SchedulerTest { + private HashMap<AgentMessage.AgentModule, Module> modules; + private HashMap<AgentMessage.AgentModule, Executor> executors; + private ArrayList<Thread> executorThreads; + private EventBus eventBus; + private Thread eventBusThread; + + public SchedulerTest() { + this.modules = initializeModule(); + this.executors = Agent.initializeExecutors(modules); + this.executorThreads = Agent.initializeExecutorThreads(executors); + this.eventBus = new EventBus(executors); + this.eventBusThread = new Thread(eventBus); + eventBusThread.start(); + } public HashMap<AgentMessage.AgentModule, Module> initializeModule() { HashMap<AgentMessage.AgentModule, Module> modules = new HashMap<AgentMessage.AgentModule, Module>(); + modules.put(AgentMessage.AgentModule.TIMER_SCHEDULER, new TimerScheduler(AgentMessage.AgentModule.TIMER_SCHEDULER)); + return modules; + } - modules.put(AgentMessage.AgentModule.TIMER_SCHEDULER, new TimerScheduler(AgentMessage.AgentModule.RMI)); + @Test + public void initializeWrongModuleType() { + try { + Module timer = new TimerScheduler(AgentMessage.AgentModule.RMI); + } catch (AssertionError e) { + System.out.println("Wrong timer type during init error caught"); + } + } - return modules; + @Test + public void sendWrongMessageType() throws InterruptedException { + try { + this.eventBus.addMessage(new TimerSchedulerMessage( + "0", + AgentMessage.AgentModule.UDP, + System.currentTimeMillis() / 1000L, + "1", + 10, + System.currentTimeMillis() / 1000L, + new TimerScheduledTask() { + @Override + public void run() { + System.out.println("Task executed"); + } + })); + Thread.sleep(1000); + } catch (AssertionError e) { + System.out.println("Wrong timer-scheduler message type error caught"); + } } @Test - void scheduleTask() { + public void scheduleTask() throws InterruptedException { + this.eventBus.addMessage(new TimerSchedulerMessage( + "0", + AgentMessage.AgentModule.TIMER_SCHEDULER, + System.currentTimeMillis() / 1000L, + "1", + 10, + System.currentTimeMillis() / 1000L, + new TimerScheduledTask() { + @Override + public void run() { + System.out.println("Task executed"); + } + })); + Thread.sleep(1000); } @Test - void scheduleTwoTasks() { - HashMap<AgentMessage.AgentModule, Module> modules = ini(); - HashMap<AgentMessage.AgentModule, Executor> executors = Agent.initializeExecutors(modules); - ArrayList<Thread> executorThreads = Agent.initializeExecutorThreads(executors); - EventBus eventBus = new EventBus(executors); - Thread eventBusThread = new Thread(eventBus); - eventBusThread.start(); + public void scheduleTwoTasks() throws InterruptedException { + this.eventBus.addMessage(new TimerSchedulerMessage( + "0", + AgentMessage.AgentModule.TIMER_SCHEDULER, + System.currentTimeMillis() / 1000L, + "1", + 10, + System.currentTimeMillis() / 1000L, + new TimerScheduledTask() { + @Override + public void run() { + System.out.println("Task 1 executed"); + } + })); - eventBus.addMessage(new AgentMessage( + this.eventBus.addMessage(new TimerSchedulerMessage( "0", - AgentMessage.AgentModule.RMI, - System.currentTimeMillis() / 1000L)); + AgentMessage.AgentModule.TIMER_SCHEDULER, + System.currentTimeMillis() / 1000L, + "1", + 20, + System.currentTimeMillis() / 1000L, + new TimerScheduledTask() { + @Override + public void run() { + System.out.println("Task 2 executed"); + } + })); - Thread.sleep(1000); - eventBusThread.interrupt(); - Agent.closeExecutors(executorThreads); - assertEquals(1, ((EventBusTest.MessageCounterModule) modules.get(AgentMessage.AgentModule.RMI)).counter); + Thread.sleep(1000); } } |