Merge remote-tracking branch 'origin/master'

@@ -24,6 +24,11 @@
         </dependency>
         <dependency>
             <groupId>org.onlab.onos</groupId>
+            <artifactId>onlab-osgi</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.onlab.onos</groupId>
             <artifactId>onlab-nio</artifactId>
             <version>${project.version}</version>
         </dependency>

+package org.onlab.onos.foo;
+
+import java.io.IOException;
+
+import org.onlab.netty.Message;
+import org.onlab.netty.MessageHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Message handler that echos the message back to the sender.
+ */
+public class NettyEchoHandler implements MessageHandler {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public void handle(Message message) throws IOException {
+        //log.info("Received message. Echoing it back to the sender.");
+        message.respond(message.payload());
+    }
+}

+package org.onlab.onos.foo;
+
+import org.onlab.netty.Message;
+import org.onlab.netty.MessageHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A MessageHandler that simply logs the information.
+ */
+public class NettyLoggingHandler implements MessageHandler {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public void handle(Message message) {
+        log.info("Received message. Payload has {} bytes", message.payload().length);
+    }
+}

@@ -2,7 +2,6 @@ package org.onlab.onos.foo;
 
 import java.io.IOException;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import org.onlab.metrics.MetricsComponent;
@@ -35,28 +34,35 @@ public final class SimpleNettyClient {
         MetricsManager metrics = new MetricsManager();
         messaging.activate();
         metrics.activate();
-        MetricsFeature feature = new MetricsFeature("timers");
+        MetricsFeature feature = new MetricsFeature("latency");
         MetricsComponent component = metrics.registerComponent("NettyMessaging");
-        Timer sendAsyncTimer = metrics.createTimer(component, feature, "AsyncSender");
-        final int warmup = 100;
+
+        final int warmup = 10000;
         for (int i = 0; i < warmup; i++) {
+            messaging.sendAsync(new Endpoint("localhost", 8081), "simple", "Hello World".getBytes());
+            Response response = messaging
+                    .sendAndReceive(new Endpoint("localhost", 8081), "echo",
+                            "Hello World".getBytes());
+        }
+
+        Timer sendAsyncTimer = metrics.createTimer(component, feature, "AsyncSender");
+        Timer sendAndReceiveTimer = metrics.createTimer(component, feature, "SendAndReceive");
+
+        final int iterations = 10000000;
+        for (int i = 0; i < iterations; i++) {
             Timer.Context context = sendAsyncTimer.time();
-            messaging.sendAsync(new Endpoint("localhost", 8080), "simple", "Hello World".getBytes());
+            messaging.sendAsync(new Endpoint("localhost", 8081), "simple", "Hello World".getBytes());
             context.stop();
         }
-        metrics.registerMetric(component, feature, "AsyncTimer", sendAsyncTimer);
 
-        Timer sendAndReceiveTimer = metrics.createTimer(component, feature, "SendAndReceive");
-        final int iterations = 1000000;
         for (int i = 0; i < iterations; i++) {
             Timer.Context context = sendAndReceiveTimer.time();
             Response response = messaging
-                    .sendAndReceive(new Endpoint("localhost", 8080), "echo",
+                    .sendAndReceive(new Endpoint("localhost", 8081), "echo",
                                     "Hello World".getBytes());
-            System.out.println("Got back:" + new String(response.get(2, TimeUnit.SECONDS)));
+            // System.out.println("Got back:" + new String(response.get(2, TimeUnit.SECONDS)));
             context.stop();
         }
-        metrics.registerMetric(component, feature, "AsyncTimer", sendAndReceiveTimer);
     }
 
     public static class TestNettyMessagingService extends NettyMessagingService {

@@ -13,6 +13,7 @@ import org.onlab.onos.cli.AbstractShellCommand;
         description = "Starts the simple Netty client")
 public class SimpleNettyClientCommand extends AbstractShellCommand {
 
+    //FIXME: replace these arguments with proper ones needed for the test.
     @Argument(index = 0, name = "serverIp", description = "Server IP address",
             required = false, multiValued = false)
     String serverIp = "127.0.0.1";

 package org.onlab.onos.foo;
 
-import org.onlab.netty.EchoHandler;
 import org.onlab.netty.NettyMessagingService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -19,10 +18,10 @@ import org.slf4j.LoggerFactory;
             }
 
         public static void startStandalone(String[] args) throws Exception {
-            NettyMessagingService server = new NettyMessagingService(8080);
+            NettyMessagingService server = new NettyMessagingService(8081);
             server.activate();
-            server.registerHandler("simple", new org.onlab.netty.LoggingHandler());
-            server.registerHandler("echo", new EchoHandler());
+            server.registerHandler("simple", new NettyLoggingHandler());
+            server.registerHandler("echo", new NettyEchoHandler());
         }
     }
 

@@ -9,10 +9,11 @@ import org.onlab.onos.cli.AbstractShellCommand;
 /**
  * Starts the Simple Netty server.
  */
-@Command(scope = "onos", name = "test-netty-server",
+@Command(scope = "onos", name = "simple-netty-server",
          description = "Starts the simple netty server")
 public class SimpleNettyServerCommand extends AbstractShellCommand {
 
+    //FIXME: Replace these with parameters for
     @Argument(index = 0, name = "serverIp", description = "Server IP address",
               required = false, multiValued = false)
     String serverIp = "127.0.0.1";

@@ -7,6 +7,12 @@
         <command>
             <action class="org.onlab.onos.foo.TestIOServerCommand"/>
         </command>
+        <command>
+            <action class="org.onlab.onos.foo.SimpleNettyServerCommand"/>
+        </command>
+        <command>
+            <action class="org.onlab.onos.foo.SimpleNettyClientCommand"/>
+        </command>
     </command-bundle>
 
 </blueprint>

 package org.onlab.onos.net.device.impl;
 
 import com.google.common.collect.Sets;
+
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
+import org.onlab.onos.cluster.ClusterEventListener;
+import org.onlab.onos.cluster.ClusterService;
+import org.onlab.onos.cluster.ControllerNode;
+import org.onlab.onos.cluster.DefaultControllerNode;
 import org.onlab.onos.cluster.MastershipServiceAdapter;
+import org.onlab.onos.cluster.MastershipTerm;
+import org.onlab.onos.cluster.MastershipTermService;
 import org.onlab.onos.cluster.NodeId;
+import org.onlab.onos.cluster.ControllerNode.State;
 import org.onlab.onos.event.Event;
 import org.onlab.onos.event.impl.TestEventDispatcher;
 import org.onlab.onos.net.Device;
@@ -27,7 +35,9 @@ import org.onlab.onos.net.device.DeviceService;
 import org.onlab.onos.net.device.PortDescription;
 import org.onlab.onos.net.provider.AbstractProvider;
 import org.onlab.onos.net.provider.ProviderId;
+import org.onlab.onos.store.ClockProviderService;
 import org.onlab.onos.store.trivial.impl.SimpleDeviceStore;
+import org.onlab.packet.IpPrefix;
 
 import java.util.ArrayList;
 import java.util.Iterator;
@@ -56,6 +66,8 @@ public class DeviceManagerTest {
     private static final PortNumber P1 = PortNumber.portNumber(1);
     private static final PortNumber P2 = PortNumber.portNumber(2);
     private static final PortNumber P3 = PortNumber.portNumber(3);
+    private static final NodeId NID_LOCAL = new NodeId("local");
+    private static final IpPrefix LOCALHOST = IpPrefix.valueOf("127.0.0.1");
 
     private DeviceManager mgr;
 
@@ -75,6 +87,8 @@ public class DeviceManagerTest {
         mgr.store = new SimpleDeviceStore();
         mgr.eventDispatcher = new TestEventDispatcher();
         mgr.mastershipService = new TestMastershipService();
+        mgr.clusterService = new TestClusterService();
+        mgr.clockProviderService = new TestClockProviderService();
         mgr.activate();
 
         service.addListener(listener);
@@ -273,6 +287,59 @@ public class DeviceManagerTest {
         public MastershipRole requestRoleFor(DeviceId deviceId) {
             return MastershipRole.MASTER;
         }
+
+        @Override
+        public MastershipTermService requestTermService() {
+            return new MastershipTermService() {
+                @Override
+                public MastershipTerm getMastershipTerm(DeviceId deviceId) {
+                    // FIXME: just returning something not null
+                    return MastershipTerm.of(NID_LOCAL, 1);
+                }
+            };
+        }
     }
 
+    // code clone
+    private final class TestClusterService implements ClusterService {
+
+        ControllerNode local = new DefaultControllerNode(NID_LOCAL, LOCALHOST);
+
+        @Override
+        public ControllerNode getLocalNode() {
+            return local;
+        }
+
+        @Override
+        public Set<ControllerNode> getNodes() {
+            return null;
+        }
+
+        @Override
+        public ControllerNode getNode(NodeId nodeId) {
+            return null;
+        }
+
+        @Override
+        public State getState(NodeId nodeId) {
+            return null;
+        }
+
+        @Override
+        public void addListener(ClusterEventListener listener) {
+        }
+
+        @Override
+        public void removeListener(ClusterEventListener listener) {
+        }
+    }
+
+    private final class TestClockProviderService implements
+            ClockProviderService {
+
+        @Override
+        public void setMastershipTerm(DeviceId deviceId, MastershipTerm term) {
+            // TODO Auto-generated method stub
+        }
+    }
 }

@@ -29,6 +29,7 @@ import org.onlab.onos.store.cluster.messaging.MessageSubject;
 import org.onlab.onos.store.serializers.ClusterMessageSerializer;
 import org.onlab.onos.store.serializers.KryoPoolUtil;
 import org.onlab.onos.store.serializers.KryoSerializer;
+import org.onlab.onos.store.serializers.MessageSubjectSerializer;
 import org.onlab.util.KryoPool;
 import org.onlab.netty.Endpoint;
 import org.onlab.netty.Message;
@@ -66,7 +67,7 @@ public class ClusterCommunicationManager
                     .register(ClusterMessage.class, new ClusterMessageSerializer())
                     .register(ClusterMembershipEvent.class)
                     .register(byte[].class)
-                    .register(MessageSubject.class)
+                    .register(MessageSubject.class, new MessageSubjectSerializer())
                     .build()
                     .populate(1);
         }
@@ -123,7 +124,8 @@ public class ClusterCommunicationManager
         Endpoint nodeEp = new Endpoint(node.ip().toString(), node.tcpPort());
         try {
             log.info("sending...");
-            Response resp = messagingService.sendAndReceive(nodeEp, message.subject().value(), SERIALIZER.encode(message));
+            Response resp = messagingService.sendAndReceive(nodeEp,
+                    message.subject().value(), SERIALIZER.encode(message));
             resp.get(1, TimeUnit.SECONDS);
             log.info("sent...");
             return true;

@@ -3,7 +3,6 @@ package org.onlab.onos.store.serializers;
 import org.onlab.onos.cluster.NodeId;
 import org.onlab.onos.store.cluster.messaging.ClusterMessage;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
-
 import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.Input;
@@ -11,6 +10,9 @@ import com.esotericsoftware.kryo.io.Output;
 
 public final class ClusterMessageSerializer extends Serializer<ClusterMessage> {
 
+    /**
+     * Creates a serializer for {@link ClusterMessage}.
+     */
     public ClusterMessageSerializer() {
         // does not accept null
         super(false);

@@ -14,7 +14,7 @@ import com.esotericsoftware.kryo.io.Output;
 public class MastershipBasedTimestampSerializer extends Serializer<MastershipBasedTimestamp> {
 
     /**
-     * Default constructor.
+     * Creates a serializer for {@link MastershipBasedTimestamp}.
      */
     public MastershipBasedTimestampSerializer() {
         // non-null, immutable

+package org.onlab.onos.store.serializers;
+
+import org.onlab.onos.store.cluster.messaging.MessageSubject;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+public final class MessageSubjectSerializer extends Serializer<MessageSubject> {
+
+    /**
+     * Creates a serializer for {@link MessageSubject}.
+     */
+    public MessageSubjectSerializer() {
+        // non-null, immutable
+        super(false, true);
+    }
+
+
+    @Override
+    public void write(Kryo kryo, Output output, MessageSubject object) {
+        output.writeString(object.value());
+    }
+
+    @Override
+    public MessageSubject read(Kryo kryo, Input input,
+            Class<MessageSubject> type) {
+        return new MessageSubject(input.readString());
+    }
+}

@@ -153,6 +153,7 @@
              description="ONOS sample playground application">
         <feature>onos-api</feature>
         <bundle>mvn:org.onlab.onos/onos-app-foo/1.0.0-SNAPSHOT</bundle>
+        <bundle>mvn:org.onlab.onos/onlab-netty/1.0.0-SNAPSHOT</bundle>
     </feature>
 
     <feature name="onos-app-config" version="1.0.0"

 package org.onlab.metrics;
 
-import java.io.File;
-import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -11,8 +9,8 @@ import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
 
+import com.codahale.metrics.ConsoleReporter;
 import com.codahale.metrics.Counter;
-import com.codahale.metrics.CsvReporter;
 import com.codahale.metrics.Gauge;
 import com.codahale.metrics.Histogram;
 import com.codahale.metrics.Meter;
@@ -69,15 +67,14 @@ public final class MetricsManager implements MetricsService {
     /**
      * Default Reporter for this metrics manager.
      */
-    private final CsvReporter reporter;
+    private final ConsoleReporter reporter;
 
     public MetricsManager() {
         this.metricsRegistry = new MetricRegistry();
-        this.reporter = CsvReporter.forRegistry(metricsRegistry)
-                .formatFor(Locale.US)
+        this.reporter = ConsoleReporter.forRegistry(metricsRegistry)
                 .convertRatesTo(TimeUnit.SECONDS)
                 .convertDurationsTo(TimeUnit.MICROSECONDS)
-                .build(new File("/var/onos/log/metrics/"));
+                .build();
     }
 
     @Activate

@@ -36,11 +36,6 @@ public class MessageEncoder extends MessageToByteEncoder<InternalMessage> {
         // write preamble
         out.writeBytes(PREAMBLE);
 
-        try {
-            SERIALIZER.encode(message);
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
         byte[] payload = SERIALIZER.encode(message);
 
         // write payload length