diff --git a/pom.xml b/pom.xml
index 70a164c06..2da35eb46 100644
--- a/pom.xml
+++ b/pom.xml
@@ -73,12 +73,13 @@
front-controllerrepositoryasync-method-invocation
- business-delegate
- half-sync-half-async
+ business-delegate
+ half-sync-half-asyncstep-builderlayersmessage-channelfluentinterface
+ reactor
diff --git a/reactor/etc/reactor.png b/reactor/etc/reactor.png
new file mode 100644
index 000000000..0b00ec98b
Binary files /dev/null and b/reactor/etc/reactor.png differ
diff --git a/reactor/etc/reactor.ucls b/reactor/etc/reactor.ucls
new file mode 100644
index 000000000..d072e4029
--- /dev/null
+++ b/reactor/etc/reactor.ucls
@@ -0,0 +1,207 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/reactor/index.md b/reactor/index.md
new file mode 100644
index 000000000..7333c74dd
--- /dev/null
+++ b/reactor/index.md
@@ -0,0 +1,30 @@
+---
+layout: pattern
+title: Reactor
+folder: reactor
+permalink: /patterns/reactor/
+categories: Architectural
+tags:
+ - Java
+ - Difficulty-Expert
+---
+
+**Intent:** The Reactor design pattern handles service requests that are delivered concurrently to an application by one or more clients. The application can register specific handlers for processing which are called by reactor on specific events. Dispatching of event handlers is performed by an initiation dispatcher, which manages the registered event handlers. Demultiplexing of service requests is performed by a synchronous event demultiplexer.
+
+
+
+**Applicability:** Use Reactor pattern when
+
+* a server application needs to handle concurrent service requests from multiple clients.
+* a server application needs to be available for receiving requests from new clients even when handling older client requests.
+* a server must maximize throughput, minimize latency and use CPU efficiently without blocking.
+
+**Real world examples:**
+
+* [Spring Reactor](http://projectreactor.io/)
+
+**Credits**
+
+* [Douglas C. Schmidt - Reactor](https://www.dre.vanderbilt.edu/~schmidt/PDF/Reactor.pdf)
+* [Doug Lea - Scalable IO in Java](http://gee.cs.oswego.edu/dl/cpjslides/nio.pdf)
+* [Netty](http://netty.io/)
diff --git a/reactor/pom.xml b/reactor/pom.xml
new file mode 100644
index 000000000..599376e32
--- /dev/null
+++ b/reactor/pom.xml
@@ -0,0 +1,18 @@
+
+
+ 4.0.0
+
+ com.iluwatar
+ java-design-patterns
+ 1.6.0
+
+ reactor
+
+
+ junit
+ junit
+ test
+
+
+
diff --git a/reactor/src/main/java/com/iluwatar/reactor/app/App.java b/reactor/src/main/java/com/iluwatar/reactor/app/App.java
new file mode 100644
index 000000000..5c6d91ee8
--- /dev/null
+++ b/reactor/src/main/java/com/iluwatar/reactor/app/App.java
@@ -0,0 +1,131 @@
+package com.iluwatar.reactor.app;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.iluwatar.reactor.framework.AbstractNioChannel;
+import com.iluwatar.reactor.framework.ChannelHandler;
+import com.iluwatar.reactor.framework.Dispatcher;
+import com.iluwatar.reactor.framework.NioDatagramChannel;
+import com.iluwatar.reactor.framework.NioReactor;
+import com.iluwatar.reactor.framework.NioServerSocketChannel;
+import com.iluwatar.reactor.framework.ThreadPoolDispatcher;
+
+/**
+ * This application demonstrates Reactor pattern. The example demonstrated is a Distributed Logging
+ * Service where it listens on multiple TCP or UDP sockets for incoming log requests.
+ *
+ *
+ * INTENT
+ * The Reactor design pattern handles service requests that are delivered concurrently to an
+ * application by one or more clients. The application can register specific handlers for processing
+ * which are called by reactor on specific events.
+ *
+ *
+ * PROBLEM
+ * Server applications in a distributed system must handle multiple clients that send them service
+ * requests. Following forces need to be resolved:
+ *
+ *
Availability
+ *
Efficiency
+ *
Programming Simplicity
+ *
Adaptability
+ *
+ *
+ *
+ * PARTICIPANTS
+ *
+ *
Synchronous Event De-multiplexer
{@link NioReactor} plays the role of synchronous event
+ * de-multiplexer. It waits for events on multiple channels registered to it in an event loop.
+ *
+ *
+ *
Initiation Dispatcher
{@link NioReactor} plays this role as the application specific
+ * {@link ChannelHandler}s are registered to the reactor.
+ *
+ *
+ *
Handle
{@link AbstractNioChannel} acts as a handle that is registered to the reactor.
+ * When any events occur on a handle, reactor calls the appropriate handler.
+ *
+ *
+ *
Event Handler
{@link ChannelHandler} acts as an event handler, which is bound to a
+ * channel and is called back when any event occurs on any of its associated handles. Application
+ * logic resides in event handlers.
+ *
+ *
+ *
+ * The application utilizes single thread to listen for requests on all ports. It does not create a
+ * separate thread for each client, which provides better scalability under load (number of clients
+ * increase).
+ *
+ *
+ * The example uses Java NIO framework to implement the Reactor.
+ *
+ */
+public class App {
+
+ private NioReactor reactor;
+ private List channels = new ArrayList<>();
+
+ /**
+ * App entry.
+ *
+ * @throws IOException
+ */
+ public static void main(String[] args) throws IOException {
+ new App().start(new ThreadPoolDispatcher(2));
+ }
+
+ /**
+ * Starts the NIO reactor.
+ * @param threadPoolDispatcher
+ *
+ * @throws IOException if any channel fails to bind.
+ */
+ public void start(Dispatcher dispatcher) throws IOException {
+ /*
+ * The application can customize its event dispatching mechanism.
+ */
+ reactor = new NioReactor(dispatcher);
+
+ /*
+ * This represents application specific business logic that dispatcher will call on appropriate
+ * events. These events are read events in our example.
+ */
+ LoggingHandler loggingHandler = new LoggingHandler();
+
+ /*
+ * Our application binds to multiple channels and uses same logging handler to handle incoming
+ * log requests.
+ */
+ reactor.registerChannel(tcpChannel(6666, loggingHandler)).registerChannel(tcpChannel(6667, loggingHandler))
+ .registerChannel(udpChannel(6668, loggingHandler)).start();
+ }
+
+ /**
+ * Stops the NIO reactor. This is a blocking call.
+ *
+ * @throws InterruptedException if interrupted while stopping the reactor.
+ * @throws IOException if any I/O error occurs
+ */
+ public void stop() throws InterruptedException, IOException {
+ reactor.stop();
+ for (AbstractNioChannel channel : channels) {
+ channel.getChannel().close();
+ }
+ }
+
+ private AbstractNioChannel tcpChannel(int port, ChannelHandler handler) throws IOException {
+ NioServerSocketChannel channel = new NioServerSocketChannel(port, handler);
+ channel.bind();
+ channels.add(channel);
+ return channel;
+ }
+
+ private AbstractNioChannel udpChannel(int port, ChannelHandler handler) throws IOException {
+ NioDatagramChannel channel = new NioDatagramChannel(port, handler);
+ channel.bind();
+ channels.add(channel);
+ return channel;
+ }
+}
diff --git a/reactor/src/main/java/com/iluwatar/reactor/app/AppClient.java b/reactor/src/main/java/com/iluwatar/reactor/app/AppClient.java
new file mode 100644
index 000000000..659f5da21
--- /dev/null
+++ b/reactor/src/main/java/com/iluwatar/reactor/app/AppClient.java
@@ -0,0 +1,163 @@
+package com.iluwatar.reactor.app;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.PrintWriter;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.UnknownHostException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Represents the clients of Reactor pattern. Multiple clients are run concurrently and send logging
+ * requests to Reactor.
+ */
+public class AppClient {
+ private final ExecutorService service = Executors.newFixedThreadPool(4);
+
+ /**
+ * App client entry.
+ *
+ * @throws IOException if any I/O error occurs.
+ */
+ public static void main(String[] args) throws IOException {
+ AppClient appClient = new AppClient();
+ appClient.start();
+ }
+
+ /**
+ * Starts the logging clients.
+ *
+ * @throws IOException if any I/O error occurs.
+ */
+ public void start() throws IOException {
+ service.execute(new TCPLoggingClient("Client 1", 6666));
+ service.execute(new TCPLoggingClient("Client 2", 6667));
+ service.execute(new UDPLoggingClient("Client 3", 6668));
+ service.execute(new UDPLoggingClient("Client 4", 6668));
+ }
+
+ /**
+ * Stops logging clients. This is a blocking call.
+ */
+ public void stop() {
+ service.shutdown();
+ if (!service.isTerminated()) {
+ service.shutdownNow();
+ try {
+ service.awaitTermination(1000, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ private static void artificialDelayOf(long millis) {
+ try {
+ Thread.sleep(millis);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * A logging client that sends requests to Reactor on TCP socket.
+ */
+ static class TCPLoggingClient implements Runnable {
+
+ private final int serverPort;
+ private final String clientName;
+
+ /**
+ * Creates a new TCP logging client.
+ *
+ * @param clientName the name of the client to be sent in logging requests.
+ * @param port the port on which client will send logging requests.
+ */
+ public TCPLoggingClient(String clientName, int serverPort) {
+ this.clientName = clientName;
+ this.serverPort = serverPort;
+ }
+
+ public void run() {
+ try (Socket socket = new Socket(InetAddress.getLocalHost(), serverPort)) {
+ OutputStream outputStream = socket.getOutputStream();
+ PrintWriter writer = new PrintWriter(outputStream);
+ sendLogRequests(writer, socket.getInputStream());
+ } catch (IOException e) {
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void sendLogRequests(PrintWriter writer, InputStream inputStream) throws IOException {
+ for (int i = 0; i < 4; i++) {
+ writer.println(clientName + " - Log request: " + i);
+ writer.flush();
+
+ byte[] data = new byte[1024];
+ int read = inputStream.read(data, 0, data.length);
+ if (read == 0) {
+ System.out.println("Read zero bytes");
+ } else {
+ System.out.println(new String(data, 0, read));
+ }
+
+ artificialDelayOf(100);
+ }
+ }
+
+ }
+
+ /**
+ * A logging client that sends requests to Reactor on UDP socket.
+ */
+ static class UDPLoggingClient implements Runnable {
+ private final String clientName;
+ private final InetSocketAddress remoteAddress;
+
+ /**
+ * Creates a new UDP logging client.
+ *
+ * @param clientName the name of the client to be sent in logging requests.
+ * @param port the port on which client will send logging requests.
+ * @throws UnknownHostException if localhost is unknown
+ */
+ public UDPLoggingClient(String clientName, int port) throws UnknownHostException {
+ this.clientName = clientName;
+ this.remoteAddress = new InetSocketAddress(InetAddress.getLocalHost(), port);
+ }
+
+ @Override
+ public void run() {
+ try (DatagramSocket socket = new DatagramSocket()) {
+ for (int i = 0; i < 4; i++) {
+
+ String message = clientName + " - Log request: " + i;
+ DatagramPacket request = new DatagramPacket(message.getBytes(), message.getBytes().length, remoteAddress);
+
+ socket.send(request);
+
+ byte[] data = new byte[1024];
+ DatagramPacket reply = new DatagramPacket(data, data.length);
+ socket.receive(reply);
+ if (reply.getLength() == 0) {
+ System.out.println("Read zero bytes");
+ } else {
+ System.out.println(new String(reply.getData(), 0, reply.getLength()));
+ }
+
+ artificialDelayOf(100);
+ }
+ } catch (IOException e1) {
+ e1.printStackTrace();
+ }
+ }
+ }
+}
diff --git a/reactor/src/main/java/com/iluwatar/reactor/app/LoggingHandler.java b/reactor/src/main/java/com/iluwatar/reactor/app/LoggingHandler.java
new file mode 100644
index 000000000..0845303df
--- /dev/null
+++ b/reactor/src/main/java/com/iluwatar/reactor/app/LoggingHandler.java
@@ -0,0 +1,59 @@
+package com.iluwatar.reactor.app;
+
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+
+import com.iluwatar.reactor.framework.AbstractNioChannel;
+import com.iluwatar.reactor.framework.ChannelHandler;
+import com.iluwatar.reactor.framework.NioDatagramChannel.DatagramPacket;
+
+/**
+ * Logging server application logic. It logs the incoming requests on standard console and returns a
+ * canned acknowledgement back to the remote peer.
+ */
+public class LoggingHandler implements ChannelHandler {
+
+ private static final byte[] ACK = "Data logged successfully".getBytes();
+
+ /**
+ * Decodes the received data and logs it on standard console.
+ */
+ @Override
+ public void handleChannelRead(AbstractNioChannel channel, Object readObject, SelectionKey key) {
+ /*
+ * As this handler is attached with both TCP and UDP channels we need to check whether the data
+ * received is a ByteBuffer (from TCP channel) or a DatagramPacket (from UDP channel).
+ */
+ if (readObject instanceof ByteBuffer) {
+ doLogging(((ByteBuffer) readObject));
+ sendReply(channel, key);
+ } else if (readObject instanceof DatagramPacket) {
+ DatagramPacket datagram = (DatagramPacket) readObject;
+ doLogging(datagram.getData());
+ sendReply(channel, datagram, key);
+ } else {
+ throw new IllegalStateException("Unknown data received");
+ }
+ }
+
+ private void sendReply(AbstractNioChannel channel, DatagramPacket incomingPacket, SelectionKey key) {
+ /*
+ * Create a reply acknowledgement datagram packet setting the receiver to the sender of incoming
+ * message.
+ */
+ DatagramPacket replyPacket = new DatagramPacket(ByteBuffer.wrap(ACK));
+ replyPacket.setReceiver(incomingPacket.getSender());
+
+ channel.write(replyPacket, key);
+ }
+
+ private void sendReply(AbstractNioChannel channel, SelectionKey key) {
+ ByteBuffer buffer = ByteBuffer.wrap(ACK);
+ channel.write(buffer, key);
+ }
+
+ private void doLogging(ByteBuffer data) {
+ // assuming UTF-8 :(
+ System.out.println(new String(data.array(), 0, data.limit()));
+ }
+}
diff --git a/reactor/src/main/java/com/iluwatar/reactor/framework/AbstractNioChannel.java b/reactor/src/main/java/com/iluwatar/reactor/framework/AbstractNioChannel.java
new file mode 100644
index 000000000..4abacd86f
--- /dev/null
+++ b/reactor/src/main/java/com/iluwatar/reactor/framework/AbstractNioChannel.java
@@ -0,0 +1,151 @@
+package com.iluwatar.reactor.framework;
+
+import java.io.IOException;
+import java.nio.channels.SelectableChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * This represents the Handle of Reactor pattern. These are resources managed by OS which can
+ * be submitted to {@link NioReactor}.
+ *
+ *
+ * This class serves has the responsibility of reading the data when a read event occurs and writing
+ * the data back when the channel is writable. It leaves the reading and writing of data on the
+ * concrete implementation. It provides a block writing mechanism wherein when any
+ * {@link ChannelHandler} wants to write data back, it queues the data in pending write queue and
+ * clears it in block manner. This provides better throughput.
+ */
+public abstract class AbstractNioChannel {
+
+ private final SelectableChannel channel;
+ private final ChannelHandler handler;
+ private final Map> channelToPendingWrites = new ConcurrentHashMap<>();
+ private NioReactor reactor;
+
+ /**
+ * Creates a new channel.
+ *
+ * @param handler which will handle events occurring on this channel.
+ * @param channel a NIO channel to be wrapped.
+ */
+ public AbstractNioChannel(ChannelHandler handler, SelectableChannel channel) {
+ this.handler = handler;
+ this.channel = channel;
+ }
+
+ /**
+ * Injects the reactor in this channel.
+ */
+ void setReactor(NioReactor reactor) {
+ this.reactor = reactor;
+ }
+
+ /**
+ * @return the wrapped NIO channel.
+ */
+ public SelectableChannel getChannel() {
+ return channel;
+ }
+
+ /**
+ * The operation in which the channel is interested, this operation is provided to
+ * {@link Selector}.
+ *
+ * @return interested operation.
+ * @see SelectionKey
+ */
+ public abstract int getInterestedOps();
+
+ /**
+ * Binds the channel on provided port.
+ *
+ * @throws IOException if any I/O error occurs.
+ */
+ public abstract void bind() throws IOException;
+
+ /**
+ * Reads the data using the key and returns the read data. The underlying channel should be
+ * fetched using {@link SelectionKey#channel()}.
+ *
+ * @param key the key on which read event occurred.
+ * @return data read.
+ * @throws IOException if any I/O error occurs.
+ */
+ public abstract Object read(SelectionKey key) throws IOException;
+
+ /**
+ * @return the handler associated with this channel.
+ */
+ public ChannelHandler getHandler() {
+ return handler;
+ }
+
+ /*
+ * Called from the context of reactor thread when the key becomes writable. The channel writes the
+ * whole pending block of data at once.
+ */
+ void flush(SelectionKey key) throws IOException {
+ Queue