diff options
4 files changed, 106 insertions, 0 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 new file mode 100644 index 0000000..47cb1ff --- /dev/null +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java @@ -0,0 +1,7 @@ +package pl.edu.mimuw.cloudatlas.agent; + +/* + * The EventBus routes messages sent between Modules. + */ +public class EventBus { +} diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/Executor.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Executor.java new file mode 100644 index 0000000..0651a9b --- /dev/null +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Executor.java @@ -0,0 +1,35 @@ +package pl.edu.mimuw.cloudatlas.agent; + +import java.util.concurrent.LinkedBlockingQueue; + +import pl.edu.mimuw.cloudatlas.agent.message.AgentMessage; + +/* + * Queues messages sent to a particular module and ensures they are eventually + * handled. + */ +public class Executor implements Runnable { + private Module module; + private LinkedBlockingQueue<AgentMessage> events; + + public Executor(Module module) { + this.module = module; + this.events = new LinkedBlockingQueue<AgentMessage>(); + } + + public void run() { + while (!Thread.currentThread().interrupted()) { + try { + AgentMessage event = events.take(); + module.handle(event); + } catch (InterruptedException e) { + System.out.println("Executor interrupted. Exiting loop."); + break; + } + } + } + + public void addMessage(AgentMessage event) throws InterruptedException { + events.put(event); + } +} diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/Module.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Module.java new file mode 100644 index 0000000..46df31b --- /dev/null +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Module.java @@ -0,0 +1,10 @@ +package pl.edu.mimuw.cloudatlas.agent; + +import pl.edu.mimuw.cloudatlas.agent.message.AgentMessage; + +/* + * A Module is a (potentially stateful) event handler. + */ +public abstract class Module { + public abstract void handle(AgentMessage event); +} diff --git a/src/test/java/pl/edu/mimuw/cloudatlas/agent/ExecutorTest.java b/src/test/java/pl/edu/mimuw/cloudatlas/agent/ExecutorTest.java new file mode 100644 index 0000000..445a5b4 --- /dev/null +++ b/src/test/java/pl/edu/mimuw/cloudatlas/agent/ExecutorTest.java @@ -0,0 +1,54 @@ +package pl.edu.mimuw.cloudatlas.agent; + +import org.junit.Test; +import static org.junit.Assert.*; + +import pl.edu.mimuw.cloudatlas.agent.message.AgentMessage; +import pl.edu.mimuw.cloudatlas.agent.message.AgentMessage.AgentModule; + +public class ExecutorTest { + public class MessageCounterModule extends Module { + public int counter = 0; + public void handle(AgentMessage m) { + counter++; + } + } + + @Test + public void testDoesntExecuteWhenNoMessages() throws Exception { + MessageCounterModule module = new MessageCounterModule(); + Executor executor = new Executor(module); + Thread thread = new Thread(executor); + thread.start(); + Thread.sleep(100); + thread.interrupt(); + assertEquals(0, module.counter); + } + + @Test + public void testExecutesHandlerOnce() throws Exception { + MessageCounterModule module = new MessageCounterModule(); + Executor executor = new Executor(module); + executor.addMessage(new AgentMessage("", AgentModule.UDP, 0) {}); + Thread thread = new Thread(executor); + thread.start(); + Thread.sleep(100); + thread.interrupt(); + assertEquals(1, module.counter); + } + + @Test + public void testExecutesHandlerMultipleTimes() throws Exception { + MessageCounterModule module = new MessageCounterModule(); + Executor executor = new Executor(module); + executor.addMessage(new AgentMessage("", AgentModule.UDP, 0) {}); + executor.addMessage(new AgentMessage("", AgentModule.UDP, 0) {}); + Thread thread = new Thread(executor); + thread.start(); + Thread.sleep(100); + executor.addMessage(new AgentMessage("", AgentModule.UDP, 0) {}); + Thread.sleep(100); + thread.interrupt(); + assertEquals(3, module.counter); + } +} |