Starting work on I/O loop.
diff --git a/utils/nio/pom.xml b/utils/nio/pom.xml
new file mode 100644
index 0000000..561d2d4
--- /dev/null
+++ b/utils/nio/pom.xml
@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.onlab.onos</groupId>
+ <artifactId>onlab-utils</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>onlab-nio</artifactId>
+ <packaging>bundle</packaging>
+
+ <description>Fast network I/O using Java NIO</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava-testlib</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git a/utils/nio/src/main/java/org/onlab/nio/AcceptorLoop.java b/utils/nio/src/main/java/org/onlab/nio/AcceptorLoop.java
new file mode 100644
index 0000000..a5acc41
--- /dev/null
+++ b/utils/nio/src/main/java/org/onlab/nio/AcceptorLoop.java
@@ -0,0 +1,108 @@
+package org.onlab.nio;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.net.StandardSocketOptions;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.ServerSocketChannel;
+import java.util.Iterator;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Selector loop derivative tailored to acceptConnection inbound connections.
+ */
+public abstract class AcceptorLoop extends SelectorLoop {
+
+ private SocketAddress listenAddress;
+ private ServerSocketChannel socketChannel;
+
+ /**
+ * Creates an acceptor loop with the specified selection timeout and
+ * accepting connections on the the given address.
+ *
+ * @param selectTimeout selection timeout; specified in millis
+ * @param listenAddress socket address where to listen for connections
+ * @throws IOException if the backing selector cannot be opened
+ */
+ public AcceptorLoop(long selectTimeout, SocketAddress listenAddress)
+ throws IOException {
+ super(selectTimeout);
+ this.listenAddress = checkNotNull(this.listenAddress, "Address cannot be null");
+ }
+
+ /**
+ * Hook to accept an inbound connection on the specified socket channel.
+ *
+ * @param channel socketChannel where an accept operation awaits
+ * @throws IOException if the accept operation cannot be processed
+ */
+ protected abstract void acceptConnection(ServerSocketChannel channel) throws IOException;
+
+ /**
+ * Opens a new server socket channel configured in non-blocking mode and
+ * bound to the loop's listen address.
+ *
+ * @throws IOException if unable to open or configure the socket channel
+ */
+ protected synchronized void openChannel() throws IOException {
+ socketChannel = ServerSocketChannel.open();
+ socketChannel.configureBlocking(false);
+ socketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
+ socketChannel.register(selector, SelectionKey.OP_ACCEPT);
+ socketChannel.bind(listenAddress);
+ }
+
+ /**
+ * Closes the server socket channel.
+ *
+ * @throws IOException if unable to close the socketChannel
+ */
+ protected synchronized void closechannel() throws IOException {
+ if (socketChannel != null) {
+ socketChannel.close();
+ socketChannel = null;
+ }
+ }
+
+ @Override
+ public void shutdown() {
+ try {
+ closechannel();
+ } catch (IOException e) {
+ log.warn("Unable to close the socketChannel", e);
+ }
+ super.shutdown();
+ }
+
+ @Override
+ protected void loop() throws IOException {
+ openChannel();
+ notifyReady();
+
+ // Keep looping until told otherwise.
+ while (isRunning()) {
+ // Attempt a selection; if no operations selected or if signalled
+ // to shutdown, spin through.
+ int count = selector.select(selectTimeout);
+ if (count == 0 || !isRunning()) {
+ continue;
+ }
+
+ // Iterate over all keys selected for an operation and process them.
+ Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
+ while (keys.hasNext()) {
+ // Fetch the key and remove it from the pending list.
+ SelectionKey key = keys.next();
+ keys.remove();
+
+ // If the key has a pending acceptConnection operation, process it.
+ if (key.isAcceptable()) {
+ acceptConnection((ServerSocketChannel) key.channel());
+ }
+ }
+ }
+ }
+
+}
+
diff --git a/utils/nio/src/main/java/org/onlab/nio/SelectorLoop.java b/utils/nio/src/main/java/org/onlab/nio/SelectorLoop.java
new file mode 100644
index 0000000..fc88e16
--- /dev/null
+++ b/utils/nio/src/main/java/org/onlab/nio/SelectorLoop.java
@@ -0,0 +1,121 @@
+package org.onlab.nio;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.channels.Selector;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+/**
+ * Abstraction of an I/O processing loop based on an NIO selector.
+ */
+public abstract class SelectorLoop implements Runnable {
+
+ protected final Logger log = LoggerFactory.getLogger(getClass());
+
+ /**
+ * Selector used by this loop to pace the I/O operations.
+ */
+ protected final Selector selector;
+
+ /**
+ * Selection operations timeout; specified in millis.
+ */
+ protected long selectTimeout;
+
+ /**
+ * Retains the error that caused the loop to exit prematurely.
+ */
+ private Throwable error;
+
+ // State indicator
+ private enum State { STARTING, STARTED, STOPPING, STOPPED };
+ private volatile State state = State.STOPPED;
+
+ /**
+ * Creates a new selector loop with the given selection timeout.
+ *
+ * @param selectTimeout selection timeout; specified in millis
+ * @throws IOException if the backing selector cannot be opened
+ */
+ public SelectorLoop(long selectTimeout) throws IOException {
+ checkArgument(selectTimeout > 0, "Timeout must be positive");
+ this.selectTimeout = selectTimeout;
+ this.selector = openSelector();
+ }
+
+ /**
+ * Opens a new selector for the use by the loop.
+ *
+ * @return newly open selector
+ * @throws IOException if the backing selector cannot be opened
+ */
+ protected Selector openSelector() throws IOException {
+ return Selector.open();
+ }
+
+ /**
+ * Indicates that the loop is marked to run.
+ */
+ protected boolean isRunning() {
+ return state == State.STARTED || state == State.STARTING;
+ }
+
+ /**
+ * Returns the error, if there was one, that caused the loop to terminate
+ * prematurely.
+ *
+ * @return error or null if there was none
+ */
+ public Throwable getError() {
+ return error;
+ }
+
+ /**
+ * Contains the body of the I/O selector loop.
+ *
+ * @throws IOException if an error is encountered while selecting I/O
+ */
+ protected abstract void loop() throws IOException;
+
+ @Override
+ public void run() {
+ error = null;
+ state = State.STARTING;
+ try {
+ loop();
+ } catch (Throwable e) {
+ error = e;
+ log.error("Loop aborted", e);
+ }
+ notifyDone();
+ }
+
+ /**
+ * Notifies observers waiting for loop to become ready.
+ */
+ protected synchronized void notifyReady() {
+ state = State.STARTED;
+ notifyAll();
+ }
+
+ /**
+ * Triggers loop shutdown.
+ */
+ public void shutdown() {
+ // Mark the loop as no longer running and wake up the selector.
+ state = State.STOPPING;
+ selector.wakeup();
+ }
+
+ /**
+ * Notifies observers waiting for loop to fully stop.
+ */
+ private synchronized void notifyDone() {
+ state = State.STOPPED;
+ notifyAll();
+ }
+
+}
diff --git a/utils/pom.xml b/utils/pom.xml
index d9e013c..2beeba8 100644
--- a/utils/pom.xml
+++ b/utils/pom.xml
@@ -19,6 +19,7 @@
<modules>
<module>junit</module>
<module>misc</module>
+ <module>nio</module>
<module>osgi</module>
<module>rest</module>
</modules>