Work on iluwatar#74, added unit test cases

npathai · npathai · commit 940a62bc0183 · 2015-09-02T15:08:34.000+05:30
diff --git a/reactor/src/main/java/com/iluwatar/reactor/App.java b/reactor/src/main/java/com/iluwatar/reactor/App.java
@@ -4,19 +4,32 @@
 
 public class App {
 
+	private NioReactor reactor;
+
 	public static void main(String[] args) {
 		try {
-			NioReactor reactor = new NioReactor(new ThreadPoolDispatcher(2));
-			LoggingHandler loggingHandler = new LoggingHandler();
-			reactor
-				.registerChannel(tcpChannel(6666, loggingHandler))
-				.registerChannel(tcpChannel(6667, loggingHandler))
-				.registerChannel(udpChannel(6668, loggingHandler))
-			.start();
+			new App().start();
 		} catch (IOException e) {
 			e.printStackTrace();
 		}
 	}
+	
+	public void start() throws IOException {
+		reactor = new NioReactor(new ThreadPoolDispatcher(2));
+		
+		LoggingHandler loggingHandler = new LoggingHandler();
+		
+		reactor
+			.registerChannel(tcpChannel(6666, loggingHandler))
+			.registerChannel(tcpChannel(6667, loggingHandler))
+			.registerChannel(udpChannel(6668, loggingHandler))
+		.start();
+	}
+	
+	public void stop() {
+		reactor.stop();
+	}
+
 
 	private static AbstractNioChannel tcpChannel(int port, ChannelHandler handler) throws IOException {
 		NioServerSocketChannel channel = new NioServerSocketChannel(port, handler);
diff --git a/reactor/src/main/java/com/iluwatar/reactor/AppClient.java b/reactor/src/main/java/com/iluwatar/reactor/AppClient.java
@@ -10,15 +10,34 @@
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.net.SocketException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 public class AppClient {
-
+	private ExecutorService service = Executors.newFixedThreadPool(3);
+	
 	public static void main(String[] args) {
-		new Thread(new LoggingClient("Client 1", 6666)).start();
-		new Thread(new LoggingClient("Client 2", 6667)).start();
-		new Thread(new UDPLoggingClient(6668)).start();
+		new AppClient().start();
 	}
 
+	public void start() {
+		service.execute(new LoggingClient("Client 1", 6666));
+		service.execute(new LoggingClient("Client 2", 6667));
+		service.execute(new UDPLoggingClient(6668));
+	}
+	
+	public void stop() {
+		service.shutdown();
+		if (!service.isTerminated()) {
+			service.shutdownNow();
+			try {
+				service.awaitTermination(1000, TimeUnit.SECONDS);
+			} catch (InterruptedException e) {
+				e.printStackTrace();
+			}
+		}
+	}
 	
 	/*
 	 * A logging client that sends logging requests to logging server
@@ -55,7 +74,7 @@ public void run() {
 		}
 
 		private void writeLogs(PrintWriter writer, InputStream inputStream) throws IOException {
-			for (int i = 0; i < 1; i++) {
+			for (int i = 0; i < 4; i++) {
 				writer.println(clientName + " - Log request: " + i);
 				try {
 					Thread.sleep(100);
@@ -86,7 +105,7 @@ public void run() {
 			DatagramSocket socket = null;
 			try {
 				socket = new DatagramSocket();
-				for (int i = 0; i < 1; i++) {
+				for (int i = 0; i < 4; i++) {
 					String message = "UDP Client" + " - Log request: " + i;
 					try {
 						DatagramPacket packet = new DatagramPacket(message.getBytes(), message.getBytes().length, new InetSocketAddress(InetAddress.getLocalHost(), port));
diff --git a/reactor/src/main/java/com/iluwatar/reactor/Dispatcher.java b/reactor/src/main/java/com/iluwatar/reactor/Dispatcher.java
@@ -4,4 +4,5 @@
 
 public interface Dispatcher {
 	void onChannelReadEvent(AbstractNioChannel channel, Object readObject, SelectionKey key);
+	void stop();
 }
diff --git a/reactor/src/main/java/com/iluwatar/reactor/NioReactor.java b/reactor/src/main/java/com/iluwatar/reactor/NioReactor.java
@@ -9,6 +9,9 @@
 import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 /*
  * Abstractions
@@ -20,6 +23,7 @@ public class NioReactor {
 	private Selector selector;
 	private Dispatcher dispatcher;
 	private Queue<Command> pendingChanges = new ConcurrentLinkedQueue<>();
+	private ExecutorService reactorService = Executors.newSingleThreadExecutor();
 	
 	public NioReactor(Dispatcher dispatcher) throws IOException {
 		this.dispatcher = dispatcher;
@@ -34,7 +38,7 @@ public NioReactor registerChannel(AbstractNioChannel channel) throws IOException
 	}
 	
 	public void start() throws IOException {
-		new Thread( new Runnable() {
+		reactorService.execute(new Runnable() {
 			@Override
 			public void run() {
 				try {
@@ -44,11 +48,27 @@ public void run() {
 					e.printStackTrace();
 				}
 			}
-		}, "Reactor Main").start();
+		});
+	}
+	
+	public void stop() {
+		reactorService.shutdownNow();
+		selector.wakeup();
+		try {
+			reactorService.awaitTermination(4, TimeUnit.SECONDS);
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+		dispatcher.stop();
 	}
 
 	private void eventLoop() throws IOException {
 		while (true) {
+			
+			if (Thread.interrupted()) {
+				break;
+			}
+			
 			// honor any pending requests first
 			processPendingChanges();
 			
diff --git a/reactor/src/main/java/com/iluwatar/reactor/SameThreadDispatcher.java b/reactor/src/main/java/com/iluwatar/reactor/SameThreadDispatcher.java
@@ -10,4 +10,9 @@ public void onChannelReadEvent(AbstractNioChannel channel, Object readObject, Se
 			channel.getHandler().handleChannelRead(channel, readObject, key);
 		}
 	}
+
+	@Override
+	public void stop() {
+		// no-op
+	}
 }
diff --git a/reactor/src/main/java/com/iluwatar/reactor/ThreadPoolDispatcher.java b/reactor/src/main/java/com/iluwatar/reactor/ThreadPoolDispatcher.java
@@ -3,24 +3,35 @@
 import java.nio.channels.SelectionKey;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 public class ThreadPoolDispatcher extends SameThreadDispatcher {
 
-	private ExecutorService exectorService;
+	private ExecutorService executorService;
 
 	public ThreadPoolDispatcher(int poolSize) {
-		this.exectorService = Executors.newFixedThreadPool(poolSize);
+		this.executorService = Executors.newFixedThreadPool(poolSize);
 	}
 	
 	@Override
 	public void onChannelReadEvent(AbstractNioChannel channel, Object readObject, SelectionKey key) {
-		exectorService.execute(new Runnable() {
+		executorService.execute(new Runnable() {
 			
 			@Override
 			public void run() {
 				ThreadPoolDispatcher.super.onChannelReadEvent(channel, readObject, key);
 			}
 		});
 	}
+	
+	@Override
+	public void stop() {
+		executorService.shutdownNow();
+		try {
+			executorService.awaitTermination(1000, TimeUnit.SECONDS);
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+	}
 
 }
diff --git a/reactor/src/test/java/com/iluwatar/reactor/AppTest.java b/reactor/src/test/java/com/iluwatar/reactor/AppTest.java
@@ -0,0 +1,27 @@
+package com.iluwatar.reactor;
+
+import java.io.IOException;
+
+import org.junit.Test;
+
+public class AppTest {
+
+	@Test
+	public void testApp() throws IOException {
+		App app = new App();
+		app.start();
+		
+		AppClient client = new AppClient();
+		client.start();
+		
+		try {
+			Thread.sleep(2000);
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+		
+		client.stop();
+		
+		app.stop();
+	}
+}

Original file line number	Diff line number	Diff line change
`@@ -4,4 +4,5 @@`
`4`	`4`
`5`	`5`	`public interface Dispatcher {`
`6`	`6`	`void onChannelReadEvent(AbstractNioChannel channel, Object readObject, SelectionKey key);`
	`7`	`+ void stop();`
`7`	`8`	`}`
Original file line number	Diff line number	Diff line change
`@@ -10,4 +10,9 @@ public void onChannelReadEvent(AbstractNioChannel channel, Object readObject, Se`
`10`	`10`	`channel.getHandler().handleChannelRead(channel, readObject, key);`
`11`	`11`	`}`
`12`	`12`	`}`
	`13`	`+`
	`14`	`+ @Override`
	`15`	`+ public void stop() {`
	`16`	`+ // no-op`
	`17`	`+ }`
`13`	`18`	`}`