Added a few command-lines for node administration.

+package org.onlab.onos.cli;
+
+import org.apache.karaf.shell.commands.Argument;
+import org.apache.karaf.shell.commands.Command;
+import org.onlab.onos.cluster.ClusterAdminService;
+import org.onlab.onos.cluster.NodeId;
+import org.onlab.packet.IpPrefix;
+
+/**
+ * Adds a new controller cluster node.
+ */
+@Command(scope = "onos", name = "add-node",
+         description = "Adds a new controller cluster node")
+public class NodeAddCommand extends AbstractShellCommand {
+
+    @Argument(index = 0, name = "nodeId", description = "Node ID",
+              required = true, multiValued = false)
+    String nodeId = null;
+
+    @Argument(index = 1, name = "ip", description = "Node IP address",
+              required = true, multiValued = false)
+    String ip = null;
+
+    @Argument(index = 2, name = "tcpPort", description = "Node TCP listen port",
+              required = false, multiValued = false)
+    int tcpPort = 9876;
+
+    @Override
+    protected void execute() {
+        ClusterAdminService service = get(ClusterAdminService.class);
+        service.addNode(new NodeId(nodeId), IpPrefix.valueOf(ip), tcpPort);
+    }
+
+}

+package org.onlab.onos.cli;
+
+import org.apache.karaf.shell.commands.Argument;
+import org.apache.karaf.shell.commands.Command;
+import org.onlab.onos.cluster.ClusterAdminService;
+import org.onlab.onos.cluster.NodeId;
+
+/**
+ * Removes a controller cluster node.
+ */
+@Command(scope = "onos", name = "remove-node",
+         description = "Removes a new controller cluster node")
+public class NodeRemoveCommand extends AbstractShellCommand {
+
+    @Argument(index = 0, name = "nodeId", description = "Node ID",
+              required = true, multiValued = false)
+    String nodeId = null;
+
+    @Override
+    protected void execute() {
+        ClusterAdminService service = get(ClusterAdminService.class);
+        service.removeNode(new NodeId(nodeId));
+    }
+
+}

@@ -5,6 +5,12 @@
             <action class="org.onlab.onos.cli.NodesListCommand"/>
         </command>
         <command>
+            <action class="org.onlab.onos.cli.NodeAddCommand"/>
+        </command>
+        <command>
+            <action class="org.onlab.onos.cli.NodeRemoveCommand"/>
+        </command>
+        <command>
             <action class="org.onlab.onos.cli.MastersListCommand"/>
             <completers>
                 <ref component-id="clusterIdCompleter"/>

@@ -54,6 +54,15 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>>
     }
 
     /**
+     * Returns the number of message stream in custody of the loop.
+     *
+     * @return number of message streams
+     */
+    public int streamCount() {
+        return streams.size();
+    }
+
+    /**
      * Creates a new message stream backed by the specified socket channel.
      *
      * @param byteChannel backing byte channel
@@ -182,9 +191,10 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>>
      * with a pending accept operation.
      *
      * @param channel backing socket channel
+     * @return newly accepted message stream
      */
-    public void acceptStream(SocketChannel channel) {
-        createAndAdmit(channel, SelectionKey.OP_READ);
+    public S acceptStream(SocketChannel channel) {
+        return createAndAdmit(channel, SelectionKey.OP_READ);
     }
 
 
@@ -193,9 +203,10 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>>
      * with a pending connect operation.
      *
      * @param channel backing socket channel
+     * @return newly connected message stream
      */
-    public void connectStream(SocketChannel channel) {
-        createAndAdmit(channel, SelectionKey.OP_CONNECT);
+    public S connectStream(SocketChannel channel) {
+        return createAndAdmit(channel, SelectionKey.OP_CONNECT);
     }
 
     /**
@@ -205,12 +216,14 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>>
      * @param channel socket channel
      * @param op      pending operations mask to be applied to the selection
      *                key as a set of initial interestedOps
+     * @return newly created message stream
      */
-    private synchronized void createAndAdmit(SocketChannel channel, int op) {
+    private synchronized S createAndAdmit(SocketChannel channel, int op) {
         S stream = createStream(channel);
         streams.add(stream);
         newStreamRequests.add(new NewStreamRequest(stream, channel, op));
         selector.wakeup();
+        return stream;
     }
 
     /**