gaspersic/java-design-patterns
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Work on #74, added unit test cases

Browse Source
This commit is contained in:
Narendra Pathai 2015-09-02 15:08:34 +05:30
parent b94c1d37d2
commit 940a62bc01
7 changed files with 114 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
reactor/src/main/java/com/iluwatar/reactor/App.java
View File

@ -4,19 +4,32 @@ import java.io.IOException;
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);

31
reactor/src/main/java/com/iluwatar/reactor/AppClient.java
View File

@ -10,15 +10,34 @@ import java.net.InetAddress;
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 class AppClient {
}
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 class AppClient {
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));

1
reactor/src/main/java/com/iluwatar/reactor/Dispatcher.java
View File

@ -4,4 +4,5 @@ import java.nio.channels.SelectionKey;
public interface Dispatcher {
void onChannelReadEvent(AbstractNioChannel channel, Object readObject, SelectionKey key);
void stop();
}

24
reactor/src/main/java/com/iluwatar/reactor/NioReactor.java
View File

@ -9,6 +9,9 @@ import java.util.Iterator;
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 class NioReactor {
}
public void start() throws IOException {
new Thread( new Runnable() {
reactorService.execute(new Runnable() {
@Override
public void run() {
try {
@ -44,11 +48,27 @@ public class NioReactor {
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();

5
reactor/src/main/java/com/iluwatar/reactor/SameThreadDispatcher.java
View File

@ -10,4 +10,9 @@ public class SameThreadDispatcher implements Dispatcher {
channel.getHandler().handleChannelRead(channel, readObject, key);
}
}
@Override
public void stop() {
// no-op
}
}

17
reactor/src/main/java/com/iluwatar/reactor/ThreadPoolDispatcher.java
View File

@ -3,18 +3,19 @@ package com.iluwatar.reactor;
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() {
@ -22,5 +23,15 @@ public class ThreadPoolDispatcher extends SameThreadDispatcher {
}
});
}
@Override
public void stop() {
executorService.shutdownNow();
try {
executorService.awaitTermination(1000, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

27
reactor/src/test/java/com/iluwatar/reactor/AppTest.java Normal file
View File

@ -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();
}
}