Merge remote-tracking branch 'origin/master'
diff --git a/apps/config/pom.xml b/apps/config/pom.xml
index 9a2d540..f2a79cb 100644
--- a/apps/config/pom.xml
+++ b/apps/config/pom.xml
@@ -18,18 +18,20 @@
<dependencies>
<dependency>
- <groupId>org.codehaus.jackson</groupId>
- <artifactId>jackson-core-asl</artifactId>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-core-asl</artifactId>
</dependency>
<dependency>
- <groupId>org.codehaus.jackson</groupId>
- <artifactId>jackson-mapper-asl</artifactId>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-mapper-asl</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-annotations</artifactId>
- <version>2.4.2</version>
- <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.onlab.onos</groupId>
+ <artifactId>onlab-misc</artifactId>
</dependency>
</dependencies>
diff --git a/apps/foo/pom.xml b/apps/foo/pom.xml
index 6109263..f0bff2d 100644
--- a/apps/foo/pom.xml
+++ b/apps/foo/pom.xml
@@ -41,5 +41,17 @@
<groupId>org.apache.karaf.shell</groupId>
<artifactId>org.apache.karaf.shell.console</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.onlab.onos</groupId>
+ <artifactId>onlab-misc</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.osgi</groupId>
+ <artifactId>org.osgi.core</artifactId>
+ </dependency>
</dependencies>
</project>
diff --git a/apps/foo/src/main/java/org/onlab/onos/foo/NettyEchoHandler.java b/apps/foo/src/main/java/org/onlab/onos/foo/NettyEchoHandler.java
index 1049a6d..5f9bfa4 100644
--- a/apps/foo/src/main/java/org/onlab/onos/foo/NettyEchoHandler.java
+++ b/apps/foo/src/main/java/org/onlab/onos/foo/NettyEchoHandler.java
@@ -4,8 +4,6 @@
import org.onlab.netty.Message;
import org.onlab.netty.MessageHandler;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
@@ -13,11 +11,8 @@
*/
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());
}
}
diff --git a/apps/foo/src/main/java/org/onlab/onos/foo/NettyLoggingHandler.java b/apps/foo/src/main/java/org/onlab/onos/foo/NettyNothingHandler.java
similarity index 70%
rename from apps/foo/src/main/java/org/onlab/onos/foo/NettyLoggingHandler.java
rename to apps/foo/src/main/java/org/onlab/onos/foo/NettyNothingHandler.java
index b35a46f..05e2cb3 100644
--- a/apps/foo/src/main/java/org/onlab/onos/foo/NettyLoggingHandler.java
+++ b/apps/foo/src/main/java/org/onlab/onos/foo/NettyNothingHandler.java
@@ -8,12 +8,12 @@
/**
* A MessageHandler that simply logs the information.
*/
-public class NettyLoggingHandler implements MessageHandler {
+public class NettyNothingHandler 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);
+ // Do nothing
}
}
diff --git a/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyClient.java b/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyClient.java
index 5e3f8f7..337d506 100644
--- a/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyClient.java
+++ b/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyClient.java
@@ -1,7 +1,10 @@
package org.onlab.onos.foo;
+import static java.lang.Thread.sleep;
+
import java.io.IOException;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.onlab.metrics.MetricsComponent;
@@ -15,14 +18,29 @@
import com.codahale.metrics.Timer;
+/**
+ * The Simple netty client test.
+ */
// FIXME: Should be move out to test or app
public final class SimpleNettyClient {
private static Logger log = LoggerFactory.getLogger(SimpleNettyClient.class);
+ static NettyMessagingService messaging;
+ static MetricsManager metrics;
+
private SimpleNettyClient() {
}
+ /**
+ * The entry point of application.
+ *
+ * @param args the input arguments
+ * @throws IOException the iO exception
+ * @throws InterruptedException the interrupted exception
+ * @throws ExecutionException the execution exception
+ * @throws TimeoutException the timeout exception
+ */
public static void main(String[] args)
throws IOException, InterruptedException, ExecutionException,
TimeoutException {
@@ -34,13 +52,20 @@
System.exit(0);
}
+
+ /**
+ * Start standalone.
+ *
+ * @param args the args
+ * @throws Exception the exception
+ */
public static void startStandalone(String[] args) throws Exception {
String host = args.length > 0 ? args[0] : "localhost";
int port = args.length > 1 ? Integer.parseInt(args[1]) : 8081;
int warmup = args.length > 2 ? Integer.parseInt(args[2]) : 1000;
int iterations = args.length > 3 ? Integer.parseInt(args[3]) : 50 * 100000;
- NettyMessagingService messaging = new TestNettyMessagingService(9081);
- MetricsManager metrics = new MetricsManager();
+ messaging = new TestNettyMessagingService(9081);
+ metrics = new MetricsManager();
Endpoint endpoint = new Endpoint(host, port);
messaging.activate();
metrics.activate();
@@ -53,30 +78,61 @@
Response response = messaging
.sendAndReceive(endpoint, "echo",
"Hello World".getBytes());
+ response.get(100000, TimeUnit.MILLISECONDS);
}
+ log.info("measuring round-trip send & receive");
+ Timer sendAndReceiveTimer = metrics.createTimer(component, feature, "SendAndReceive");
+ int timeouts = 0;
+
+ for (int i = 0; i < iterations; i++) {
+ Response response;
+ Timer.Context context = sendAndReceiveTimer.time();
+ try {
+ response = messaging
+ .sendAndReceive(endpoint, "echo",
+ "Hello World".getBytes());
+ response.get(10000, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ timeouts++;
+ log.info("timeout:" + timeouts + " at iteration:" + i);
+ } finally {
+ context.stop();
+ }
+ // System.out.println("Got back:" + new String(response.get(2, TimeUnit.SECONDS)));
+ }
+
+ sleep(1000);
log.info("measuring async sender");
Timer sendAsyncTimer = metrics.createTimer(component, feature, "AsyncSender");
for (int i = 0; i < iterations; i++) {
- Timer.Context context = sendAsyncTimer.time();
- messaging.sendAsync(endpoint, "simple", "Hello World".getBytes());
- context.stop();
+ Timer.Context context = sendAsyncTimer.time();
+ messaging.sendAsync(endpoint, "simple", "Hello World".getBytes());
+ context.stop();
}
-
- Timer sendAndReceiveTimer = metrics.createTimer(component, feature, "SendAndReceive");
- for (int i = 0; i < iterations; i++) {
- Timer.Context context = sendAndReceiveTimer.time();
- Response response = messaging
- .sendAndReceive(endpoint, "echo",
- "Hello World".getBytes());
- // System.out.println("Got back:" + new String(response.get(2, TimeUnit.SECONDS)));
- context.stop();
- }
- metrics.deactivate();
+ sleep(1000);
}
+ public static void stop() {
+ try {
+ messaging.deactivate();
+ metrics.deactivate();
+ } catch (Exception e) {
+ log.info("Unable to stop client %s", e);
+ }
+ }
+
+ /**
+ * The type Test netty messaging service.
+ */
public static class TestNettyMessagingService extends NettyMessagingService {
+ /**
+ * Instantiates a new Test netty messaging service.
+ *
+ * @param port the port
+ * @throws Exception the exception
+ */
public TestNettyMessagingService(int port) throws Exception {
super(port);
}
diff --git a/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyClientCommand.java b/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyClientCommand.java
index b63f9c1..b59ff38 100644
--- a/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyClientCommand.java
+++ b/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyClientCommand.java
@@ -1,6 +1,7 @@
package org.onlab.onos.foo;
import static org.onlab.onos.foo.SimpleNettyClient.startStandalone;
+import static org.onlab.onos.foo.SimpleNettyClient.stop;
import org.apache.karaf.shell.commands.Argument;
import org.apache.karaf.shell.commands.Command;
@@ -10,7 +11,7 @@
* Test Netty client performance.
*/
@Command(scope = "onos", name = "simple-netty-client",
- description = "Starts the simple Netty client")
+ description = "Starts simple Netty client")
public class SimpleNettyClientCommand extends AbstractShellCommand {
//FIXME: replace these arguments with proper ones needed for the test.
@@ -28,7 +29,7 @@
@Argument(index = 3, name = "messageCount", description = "Message count",
required = false, multiValued = false)
- String messageCount = "100000";
+ String messageCount = "1000000";
@Override
protected void execute() {
@@ -37,5 +38,6 @@
} catch (Exception e) {
error("Unable to start client %s", e);
}
+ stop();
}
}
diff --git a/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyServer.java b/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyServer.java
index 5578fcd..b1e14c6 100644
--- a/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyServer.java
+++ b/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyServer.java
@@ -12,16 +12,30 @@
private SimpleNettyServer() {}
- public static void main(String... args) throws Exception {
+ /**
+ * The entry point of application.
+ *
+ * @param args the input arguments
+ * @throws Exception the exception
+ */
+ public static void main(String... args) throws Exception {
startStandalone(args);
System.exit(0);
}
- public static void startStandalone(String[] args) throws Exception {
- NettyMessagingService server = new NettyMessagingService(8081);
+ /**
+ * Start standalone server.
+ *
+ * @param args the args
+ * @throws Exception the exception
+ */
+ public static void startStandalone(String[] args) throws Exception {
+ int port = args.length > 0 ? Integer.parseInt(args[0]) : 8081;
+ NettyMessagingService server = new NettyMessagingService(port);
server.activate();
- server.registerHandler("simple", new NettyLoggingHandler());
+ server.registerHandler("simple", new NettyNothingHandler());
server.registerHandler("echo", new NettyEchoHandler());
+ log.info("Netty Server server on port " + port);
}
}
diff --git a/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyServerCommand.java b/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyServerCommand.java
index 17b2586..3190d10 100644
--- a/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyServerCommand.java
+++ b/apps/foo/src/main/java/org/onlab/onos/foo/SimpleNettyServerCommand.java
@@ -10,26 +10,18 @@
* Starts the Simple Netty server.
*/
@Command(scope = "onos", name = "simple-netty-server",
- description = "Starts the simple netty server")
+ description = "Starts simple Netty server")
public class SimpleNettyServerCommand extends AbstractShellCommand {
//FIXME: Replace these with parameters for
- @Argument(index = 0, name = "serverIp", description = "Server IP address",
+ @Argument(index = 0, name = "port", description = "listen port",
required = false, multiValued = false)
- String serverIp = "127.0.0.1";
-
- @Argument(index = 1, name = "workers", description = "IO workers",
- required = false, multiValued = false)
- String workers = "6";
-
- @Argument(index = 2, name = "messageLength", description = "Message length (bytes)",
- required = false, multiValued = false)
- String messageLength = "128";
+ String port = "8081";
@Override
protected void execute() {
try {
- startStandalone(new String[]{serverIp, workers, messageLength});
+ startStandalone(new String[]{port});
} catch (Exception e) {
error("Unable to start server %s", e);
}
diff --git a/apps/mobility/pom.xml b/apps/mobility/pom.xml
index a919ff2..18b3961 100644
--- a/apps/mobility/pom.xml
+++ b/apps/mobility/pom.xml
@@ -16,4 +16,14 @@
<description>ONOS simple Mobility app</description>
+ <dependencies>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.onlab.onos</groupId>
+ <artifactId>onlab-misc</artifactId>
+ </dependency>
+ </dependencies>
</project>
diff --git a/apps/pom.xml b/apps/pom.xml
index 55a786c..eeff7b4 100644
--- a/apps/pom.xml
+++ b/apps/pom.xml
@@ -23,7 +23,8 @@
<module>foo</module>
<module>mobility</module>
<module>proxyarp</module>
- <module>config</module>
+ <module>config</module>
+ <module>sdnip</module>
</modules>
<properties>
diff --git a/apps/sdnip/pom.xml b/apps/sdnip/pom.xml
new file mode 100644
index 0000000..99960a4
--- /dev/null
+++ b/apps/sdnip/pom.xml
@@ -0,0 +1,36 @@
+<?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>onos-apps</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>onos-app-sdnip</artifactId>
+ <packaging>bundle</packaging>
+
+ <description>SDN-IP peering application</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-core-asl</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-mapper-asl</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ <version>2.4.2</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git a/apps/sdnip/src/main/java/org/onlab/onos/sdnip/SdnIp.java b/apps/sdnip/src/main/java/org/onlab/onos/sdnip/SdnIp.java
new file mode 100644
index 0000000..a4f9b0f
--- /dev/null
+++ b/apps/sdnip/src/main/java/org/onlab/onos/sdnip/SdnIp.java
@@ -0,0 +1,27 @@
+package org.onlab.onos.sdnip;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.slf4j.Logger;
+
+/**
+ * Placeholder SDN-IP component.
+ */
+@Component(immediate = true)
+public class SdnIp {
+
+ private final Logger log = getLogger(getClass());
+
+ @Activate
+ protected void activate() {
+ log.debug("SDN-IP started");
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ log.info("Stopped");
+ }
+}
diff --git a/apps/sdnip/src/main/java/org/onlab/onos/sdnip/package-info.java b/apps/sdnip/src/main/java/org/onlab/onos/sdnip/package-info.java
new file mode 100644
index 0000000..3e1bcf0
--- /dev/null
+++ b/apps/sdnip/src/main/java/org/onlab/onos/sdnip/package-info.java
@@ -0,0 +1,4 @@
+/**
+ * SDN-IP peering application.
+ */
+package org.onlab.onos.sdnip;
\ No newline at end of file
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/common/impl/ControllerNodeToNodeId.java b/core/api/src/main/java/org/onlab/onos/cluster/ControllerNodeToNodeId.java
similarity index 73%
rename from core/store/dist/src/main/java/org/onlab/onos/store/common/impl/ControllerNodeToNodeId.java
rename to core/api/src/main/java/org/onlab/onos/cluster/ControllerNodeToNodeId.java
index 73187ee..0891494 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/common/impl/ControllerNodeToNodeId.java
+++ b/core/api/src/main/java/org/onlab/onos/cluster/ControllerNodeToNodeId.java
@@ -1,7 +1,4 @@
-package org.onlab.onos.store.common.impl;
-
-import org.onlab.onos.cluster.ControllerNode;
-import org.onlab.onos.cluster.NodeId;
+package org.onlab.onos.cluster;
import com.google.common.base.Function;
@@ -18,6 +15,11 @@
return input.id();
}
+ /**
+ * Returns a Function to convert ControllerNode to NodeId.
+ *
+ * @return ControllerNodeToNodeId instance.
+ */
public static ControllerNodeToNodeId toNodeId() {
return INSTANCE;
}
diff --git a/core/api/src/test/java/org/onlab/onos/cluster/ControllerNodeToNodeIdTest.java b/core/api/src/test/java/org/onlab/onos/cluster/ControllerNodeToNodeIdTest.java
new file mode 100644
index 0000000..44261e8
--- /dev/null
+++ b/core/api/src/test/java/org/onlab/onos/cluster/ControllerNodeToNodeIdTest.java
@@ -0,0 +1,42 @@
+package org.onlab.onos.cluster;
+
+import static org.junit.Assert.*;
+import static org.onlab.onos.cluster.ControllerNodeToNodeId.toNodeId;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Test;
+import org.onlab.packet.IpPrefix;
+
+import com.google.common.collect.FluentIterable;
+
+
+public class ControllerNodeToNodeIdTest {
+
+ private static final NodeId NID1 = new NodeId("foo");
+ private static final NodeId NID2 = new NodeId("bar");
+ private static final NodeId NID3 = new NodeId("buz");
+
+ private static final IpPrefix IP1 = IpPrefix.valueOf("127.0.0.1");
+ private static final IpPrefix IP2 = IpPrefix.valueOf("127.0.0.2");
+ private static final IpPrefix IP3 = IpPrefix.valueOf("127.0.0.3");
+
+ private static final ControllerNode CN1 = new DefaultControllerNode(NID1, IP1);
+ private static final ControllerNode CN2 = new DefaultControllerNode(NID2, IP2);
+ private static final ControllerNode CN3 = new DefaultControllerNode(NID3, IP3);
+
+
+ @Test
+ public final void testToNodeId() {
+
+ final Iterable<ControllerNode> nodes = Arrays.asList(CN1, CN2, CN3);
+ final List<NodeId> nodeIds = Arrays.asList(NID1, NID2, NID3);
+
+ assertEquals(nodeIds,
+ FluentIterable.from(nodes)
+ .transform(toNodeId())
+ .toList());
+ }
+
+}
diff --git a/core/api/src/test/java/org/onlab/onos/event/AbstractEventAccumulatorTest.java b/core/api/src/test/java/org/onlab/onos/event/AbstractEventAccumulatorTest.java
index ed18195..1f57266 100644
--- a/core/api/src/test/java/org/onlab/onos/event/AbstractEventAccumulatorTest.java
+++ b/core/api/src/test/java/org/onlab/onos/event/AbstractEventAccumulatorTest.java
@@ -9,6 +9,7 @@
import java.util.List;
import java.util.Timer;
+import org.junit.Ignore;
import org.junit.Test;
/**
@@ -41,6 +42,7 @@
assertEquals("incorrect batch", "abcde", accumulator.batch);
}
+ @Ignore("FIXME: timing sensitive test failing randomly.")
@Test
public void timeTrigger() {
TestAccumulator accumulator = new TestAccumulator();
diff --git a/core/store/dist/pom.xml b/core/store/dist/pom.xml
index 1faab74..7b00833 100644
--- a/core/store/dist/pom.xml
+++ b/core/store/dist/pom.xml
@@ -54,8 +54,13 @@
<artifactId>org.apache.felix.scr.annotations</artifactId>
</dependency>
<dependency>
- <groupId>de.javakaffee</groupId>
- <artifactId>kryo-serializers</artifactId>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava-testlib</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
</dependency>
</dependencies>
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/GossipDeviceStore.java b/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/GossipDeviceStore.java
index 782bc6c..6ab529a 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/GossipDeviceStore.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/GossipDeviceStore.java
@@ -65,6 +65,7 @@
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Predicates.notNull;
+import static org.onlab.onos.cluster.ControllerNodeToNodeId.toNodeId;
import static org.onlab.onos.net.device.DeviceEvent.Type.*;
import static org.slf4j.LoggerFactory.getLogger;
import static org.apache.commons.lang3.concurrent.ConcurrentUtils.createIfAbsentUnchecked;
@@ -73,7 +74,6 @@
import static org.onlab.util.Tools.namedThreads;
import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
import static org.onlab.onos.store.device.impl.GossipDeviceStoreMessageSubjects.DEVICE_ADVERTISE;
-import static org.onlab.onos.store.common.impl.ControllerNodeToNodeId.toNodeId;
// TODO: give me a better name
/**
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/GossipLinkStore.java b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/GossipLinkStore.java
new file mode 100644
index 0000000..3c3cc5a
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/GossipLinkStore.java
@@ -0,0 +1,661 @@
+package org.onlab.onos.store.link.impl;
+
+import com.google.common.base.Function;
+import com.google.common.base.Predicate;
+import com.google.common.collect.FluentIterable;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
+import com.google.common.collect.SetMultimap;
+
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.concurrent.ConcurrentUtils;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.cluster.ClusterService;
+import org.onlab.onos.cluster.ControllerNode;
+import org.onlab.onos.cluster.NodeId;
+import org.onlab.onos.net.AnnotationsUtil;
+import org.onlab.onos.net.ConnectPoint;
+import org.onlab.onos.net.DefaultAnnotations;
+import org.onlab.onos.net.DefaultLink;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.Link;
+import org.onlab.onos.net.SparseAnnotations;
+import org.onlab.onos.net.Link.Type;
+import org.onlab.onos.net.LinkKey;
+import org.onlab.onos.net.Provided;
+import org.onlab.onos.net.link.DefaultLinkDescription;
+import org.onlab.onos.net.link.LinkDescription;
+import org.onlab.onos.net.link.LinkEvent;
+import org.onlab.onos.net.link.LinkStore;
+import org.onlab.onos.net.link.LinkStoreDelegate;
+import org.onlab.onos.net.provider.ProviderId;
+import org.onlab.onos.store.AbstractStore;
+import org.onlab.onos.store.ClockService;
+import org.onlab.onos.store.Timestamp;
+import org.onlab.onos.store.cluster.messaging.ClusterCommunicationService;
+import org.onlab.onos.store.cluster.messaging.ClusterMessage;
+import org.onlab.onos.store.cluster.messaging.ClusterMessageHandler;
+import org.onlab.onos.store.cluster.messaging.MessageSubject;
+import org.onlab.onos.store.common.impl.Timestamped;
+import org.onlab.onos.store.serializers.DistributedStoreSerializers;
+import org.onlab.onos.store.serializers.KryoSerializer;
+import org.onlab.util.KryoPool;
+import org.onlab.util.NewConcurrentHashMap;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import static org.onlab.onos.cluster.ControllerNodeToNodeId.toNodeId;
+import static org.onlab.onos.net.DefaultAnnotations.union;
+import static org.onlab.onos.net.DefaultAnnotations.merge;
+import static org.onlab.onos.net.Link.Type.DIRECT;
+import static org.onlab.onos.net.Link.Type.INDIRECT;
+import static org.onlab.onos.net.link.LinkEvent.Type.*;
+import static org.onlab.util.Tools.namedThreads;
+import static org.slf4j.LoggerFactory.getLogger;
+import static com.google.common.collect.Multimaps.synchronizedSetMultimap;
+import static com.google.common.base.Predicates.notNull;
+
+/**
+ * Manages inventory of infrastructure links in distributed data store
+ * that uses optimistic replication and gossip based techniques.
+ */
+@Component(immediate = true)
+@Service
+public class GossipLinkStore
+ extends AbstractStore<LinkEvent, LinkStoreDelegate>
+ implements LinkStore {
+
+ private final Logger log = getLogger(getClass());
+
+ // Link inventory
+ private final ConcurrentMap<LinkKey, ConcurrentMap<ProviderId, Timestamped<LinkDescription>>> linkDescs =
+ new ConcurrentHashMap<>();
+
+ // Link instance cache
+ private final ConcurrentMap<LinkKey, Link> links = new ConcurrentHashMap<>();
+
+ // Egress and ingress link sets
+ private final SetMultimap<DeviceId, LinkKey> srcLinks = createSynchronizedHashMultiMap();
+ private final SetMultimap<DeviceId, LinkKey> dstLinks = createSynchronizedHashMultiMap();
+
+ // Remove links
+ private final Map<LinkKey, Timestamp> removedLinks = Maps.newHashMap();
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClockService clockService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClusterCommunicationService clusterCommunicator;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClusterService clusterService;
+
+ private static final KryoSerializer SERIALIZER = new KryoSerializer() {
+ @Override
+ protected void setupKryoPool() {
+ serializerPool = KryoPool.newBuilder()
+ .register(DistributedStoreSerializers.COMMON)
+ .register(InternalLinkEvent.class)
+ .register(InternalLinkRemovedEvent.class)
+ .register(LinkAntiEntropyAdvertisement.class)
+ .register(LinkFragmentId.class)
+ .build()
+ .populate(1);
+ }
+ };
+
+ private ScheduledExecutorService executor;
+
+ @Activate
+ public void activate() {
+
+ clusterCommunicator.addSubscriber(
+ GossipLinkStoreMessageSubjects.LINK_UPDATE,
+ new InternalLinkEventListener());
+ clusterCommunicator.addSubscriber(
+ GossipLinkStoreMessageSubjects.LINK_REMOVED,
+ new InternalLinkRemovedEventListener());
+ clusterCommunicator.addSubscriber(
+ GossipLinkStoreMessageSubjects.LINK_ANTI_ENTROPY_ADVERTISEMENT,
+ new InternalLinkAntiEntropyAdvertisementListener());
+
+ executor =
+ newSingleThreadScheduledExecutor(namedThreads("link-anti-entropy-%d"));
+
+ // TODO: Make these configurable
+ long initialDelaySec = 5;
+ long periodSec = 5;
+ // start anti-entropy thread
+ executor.scheduleAtFixedRate(new SendAdvertisementTask(),
+ initialDelaySec, periodSec, TimeUnit.SECONDS);
+
+ log.info("Started");
+ }
+
+ @Deactivate
+ public void deactivate() {
+ linkDescs.clear();
+ links.clear();
+ srcLinks.clear();
+ dstLinks.clear();
+ log.info("Stopped");
+ }
+
+ @Override
+ public int getLinkCount() {
+ return links.size();
+ }
+
+ @Override
+ public Iterable<Link> getLinks() {
+ return Collections.unmodifiableCollection(links.values());
+ }
+
+ @Override
+ public Set<Link> getDeviceEgressLinks(DeviceId deviceId) {
+ // lock for iteration
+ synchronized (srcLinks) {
+ return FluentIterable.from(srcLinks.get(deviceId))
+ .transform(lookupLink())
+ .filter(notNull())
+ .toSet();
+ }
+ }
+
+ @Override
+ public Set<Link> getDeviceIngressLinks(DeviceId deviceId) {
+ // lock for iteration
+ synchronized (dstLinks) {
+ return FluentIterable.from(dstLinks.get(deviceId))
+ .transform(lookupLink())
+ .filter(notNull())
+ .toSet();
+ }
+ }
+
+ @Override
+ public Link getLink(ConnectPoint src, ConnectPoint dst) {
+ return links.get(new LinkKey(src, dst));
+ }
+
+ @Override
+ public Set<Link> getEgressLinks(ConnectPoint src) {
+ Set<Link> egress = new HashSet<>();
+ for (LinkKey linkKey : srcLinks.get(src.deviceId())) {
+ if (linkKey.src().equals(src)) {
+ egress.add(links.get(linkKey));
+ }
+ }
+ return egress;
+ }
+
+ @Override
+ public Set<Link> getIngressLinks(ConnectPoint dst) {
+ Set<Link> ingress = new HashSet<>();
+ for (LinkKey linkKey : dstLinks.get(dst.deviceId())) {
+ if (linkKey.dst().equals(dst)) {
+ ingress.add(links.get(linkKey));
+ }
+ }
+ return ingress;
+ }
+
+ @Override
+ public LinkEvent createOrUpdateLink(ProviderId providerId,
+ LinkDescription linkDescription) {
+
+ DeviceId dstDeviceId = linkDescription.dst().deviceId();
+ Timestamp newTimestamp = clockService.getTimestamp(dstDeviceId);
+
+ final Timestamped<LinkDescription> deltaDesc = new Timestamped<>(linkDescription, newTimestamp);
+
+ LinkEvent event = createOrUpdateLinkInternal(providerId, deltaDesc);
+
+ if (event != null) {
+ log.info("Notifying peers of a link update topology event from providerId: "
+ + "{} between src: {} and dst: {}",
+ providerId, linkDescription.src(), linkDescription.dst());
+ try {
+ notifyPeers(new InternalLinkEvent(providerId, deltaDesc));
+ } catch (IOException e) {
+ log.info("Failed to notify peers of a link update topology event from providerId: "
+ + "{} between src: {} and dst: {}",
+ providerId, linkDescription.src(), linkDescription.dst());
+ }
+ }
+ return event;
+ }
+
+ private LinkEvent createOrUpdateLinkInternal(
+ ProviderId providerId,
+ Timestamped<LinkDescription> linkDescription) {
+
+ LinkKey key = new LinkKey(linkDescription.value().src(), linkDescription.value().dst());
+ ConcurrentMap<ProviderId, Timestamped<LinkDescription>> descs = getLinkDescriptions(key);
+
+ synchronized (descs) {
+ // if the link was previously removed, we should proceed if and
+ // only if this request is more recent.
+ Timestamp linkRemovedTimestamp = removedLinks.get(key);
+ if (linkRemovedTimestamp != null) {
+ if (linkDescription.isNewer(linkRemovedTimestamp)) {
+ removedLinks.remove(key);
+ } else {
+ return null;
+ }
+ }
+
+ final Link oldLink = links.get(key);
+ // update description
+ createOrUpdateLinkDescription(descs, providerId, linkDescription);
+ final Link newLink = composeLink(descs);
+ if (oldLink == null) {
+ return createLink(key, newLink);
+ }
+ return updateLink(key, oldLink, newLink);
+ }
+ }
+
+ // Guarded by linkDescs value (=locking each Link)
+ private Timestamped<LinkDescription> createOrUpdateLinkDescription(
+ ConcurrentMap<ProviderId, Timestamped<LinkDescription>> existingLinkDescriptions,
+ ProviderId providerId,
+ Timestamped<LinkDescription> linkDescription) {
+
+ // merge existing attributes and merge
+ Timestamped<LinkDescription> existingLinkDescription = existingLinkDescriptions.get(providerId);
+ if (existingLinkDescription != null && existingLinkDescription.isNewer(linkDescription)) {
+ return null;
+ }
+ Timestamped<LinkDescription> newLinkDescription = linkDescription;
+ if (existingLinkDescription != null) {
+ SparseAnnotations merged = union(existingLinkDescription.value().annotations(),
+ linkDescription.value().annotations());
+ newLinkDescription = new Timestamped<LinkDescription>(
+ new DefaultLinkDescription(
+ linkDescription.value().src(),
+ linkDescription.value().dst(),
+ linkDescription.value().type(), merged),
+ linkDescription.timestamp());
+ }
+ return existingLinkDescriptions.put(providerId, newLinkDescription);
+ }
+
+ // Creates and stores the link and returns the appropriate event.
+ // Guarded by linkDescs value (=locking each Link)
+ private LinkEvent createLink(LinkKey key, Link newLink) {
+
+ if (newLink.providerId().isAncillary()) {
+ // TODO: revisit ancillary only Link handling
+
+ // currently treating ancillary only as down (not visible outside)
+ return null;
+ }
+
+ links.put(key, newLink);
+ srcLinks.put(newLink.src().deviceId(), key);
+ dstLinks.put(newLink.dst().deviceId(), key);
+ return new LinkEvent(LINK_ADDED, newLink);
+ }
+
+ // Updates, if necessary the specified link and returns the appropriate event.
+ // Guarded by linkDescs value (=locking each Link)
+ private LinkEvent updateLink(LinkKey key, Link oldLink, Link newLink) {
+
+ if (newLink.providerId().isAncillary()) {
+ // TODO: revisit ancillary only Link handling
+
+ // currently treating ancillary only as down (not visible outside)
+ return null;
+ }
+
+ if ((oldLink.type() == INDIRECT && newLink.type() == DIRECT) ||
+ !AnnotationsUtil.isEqual(oldLink.annotations(), newLink.annotations())) {
+
+ links.put(key, newLink);
+ // strictly speaking following can be ommitted
+ srcLinks.put(oldLink.src().deviceId(), key);
+ dstLinks.put(oldLink.dst().deviceId(), key);
+ return new LinkEvent(LINK_UPDATED, newLink);
+ }
+ return null;
+ }
+
+ @Override
+ public LinkEvent removeLink(ConnectPoint src, ConnectPoint dst) {
+ final LinkKey key = new LinkKey(src, dst);
+
+ DeviceId dstDeviceId = dst.deviceId();
+ Timestamp timestamp = clockService.getTimestamp(dstDeviceId);
+
+ LinkEvent event = removeLinkInternal(key, timestamp);
+
+ if (event != null) {
+ log.info("Notifying peers of a link removed topology event for a link "
+ + "between src: {} and dst: {}", src, dst);
+ try {
+ notifyPeers(new InternalLinkRemovedEvent(key, timestamp));
+ } catch (IOException e) {
+ log.error("Failed to notify peers of a link removed topology event for a link "
+ + "between src: {} and dst: {}", src, dst);
+ }
+ }
+ return event;
+ }
+
+ private LinkEvent removeLinkInternal(LinkKey key, Timestamp timestamp) {
+ ConcurrentMap<ProviderId, Timestamped<LinkDescription>> linkDescriptions =
+ getLinkDescriptions(key);
+ synchronized (linkDescriptions) {
+ // accept removal request if given timestamp is newer than
+ // the latest Timestamp from Primary provider
+ ProviderId primaryProviderId = pickPrimaryProviderId(linkDescriptions);
+ if (linkDescriptions.get(primaryProviderId).isNewer(timestamp)) {
+ return null;
+ }
+ removedLinks.put(key, timestamp);
+ Link link = links.remove(key);
+ linkDescriptions.clear();
+ if (link != null) {
+ srcLinks.remove(link.src().deviceId(), key);
+ dstLinks.remove(link.dst().deviceId(), key);
+ return new LinkEvent(LINK_REMOVED, link);
+ }
+ return null;
+ }
+ }
+
+ private static <K, V> SetMultimap<K, V> createSynchronizedHashMultiMap() {
+ return synchronizedSetMultimap(HashMultimap.<K, V>create());
+ }
+
+ /**
+ * @return primary ProviderID, or randomly chosen one if none exists
+ */
+ private ProviderId pickPrimaryProviderId(
+ ConcurrentMap<ProviderId, Timestamped<LinkDescription>> providerDescs) {
+
+ ProviderId fallBackPrimary = null;
+ for (Entry<ProviderId, Timestamped<LinkDescription>> e : providerDescs.entrySet()) {
+ if (!e.getKey().isAncillary()) {
+ return e.getKey();
+ } else if (fallBackPrimary == null) {
+ // pick randomly as a fallback in case there is no primary
+ fallBackPrimary = e.getKey();
+ }
+ }
+ return fallBackPrimary;
+ }
+
+ private Link composeLink(ConcurrentMap<ProviderId, Timestamped<LinkDescription>> linkDescriptions) {
+ ProviderId primaryProviderId = pickPrimaryProviderId(linkDescriptions);
+ Timestamped<LinkDescription> base = linkDescriptions.get(primaryProviderId);
+
+ ConnectPoint src = base.value().src();
+ ConnectPoint dst = base.value().dst();
+ Type type = base.value().type();
+ DefaultAnnotations annotations = DefaultAnnotations.builder().build();
+ annotations = merge(annotations, base.value().annotations());
+
+ for (Entry<ProviderId, Timestamped<LinkDescription>> e : linkDescriptions.entrySet()) {
+ if (primaryProviderId.equals(e.getKey())) {
+ continue;
+ }
+
+ // TODO: should keep track of Description timestamp
+ // and only merge conflicting keys when timestamp is newer
+ // Currently assuming there will never be a key conflict between
+ // providers
+
+ // annotation merging. not so efficient, should revisit later
+ annotations = merge(annotations, e.getValue().value().annotations());
+ }
+
+ return new DefaultLink(primaryProviderId , src, dst, type, annotations);
+ }
+
+ private ConcurrentMap<ProviderId, Timestamped<LinkDescription>> getLinkDescriptions(LinkKey key) {
+ return ConcurrentUtils.createIfAbsentUnchecked(linkDescs, key,
+ NewConcurrentHashMap.<ProviderId, Timestamped<LinkDescription>>ifNeeded());
+ }
+
+ private Timestamped<LinkDescription> getLinkDescription(LinkKey key, ProviderId providerId) {
+ return getLinkDescriptions(key).get(providerId);
+ }
+
+ private final Function<LinkKey, Link> lookupLink = new LookupLink();
+ private Function<LinkKey, Link> lookupLink() {
+ return lookupLink;
+ }
+
+ private final class LookupLink implements Function<LinkKey, Link> {
+ @Override
+ public Link apply(LinkKey input) {
+ return links.get(input);
+ }
+ }
+
+ private static final Predicate<Provided> IS_PRIMARY = new IsPrimary();
+ private static final Predicate<Provided> isPrimary() {
+ return IS_PRIMARY;
+ }
+
+ private static final class IsPrimary implements Predicate<Provided> {
+
+ @Override
+ public boolean apply(Provided input) {
+ return !input.providerId().isAncillary();
+ }
+ }
+
+ private void notifyDelegateIfNotNull(LinkEvent event) {
+ if (event != null) {
+ notifyDelegate(event);
+ }
+ }
+
+ // TODO: should we be throwing exception?
+ private void broadcastMessage(MessageSubject subject, Object event) throws IOException {
+ ClusterMessage message = new ClusterMessage(
+ clusterService.getLocalNode().id(),
+ subject,
+ SERIALIZER.encode(event));
+ clusterCommunicator.broadcast(message);
+ }
+
+ // TODO: should we be throwing exception?
+ private void unicastMessage(NodeId recipient, MessageSubject subject, Object event) {
+ try {
+ ClusterMessage message = new ClusterMessage(
+ clusterService.getLocalNode().id(),
+ subject,
+ SERIALIZER.encode(event));
+ clusterCommunicator.unicast(message, recipient);
+ } catch (IOException e) {
+ log.error("Failed to send a {} message to {}", subject.value(), recipient);
+ }
+ }
+
+ private void notifyPeers(InternalLinkEvent event) throws IOException {
+ broadcastMessage(GossipLinkStoreMessageSubjects.LINK_UPDATE, event);
+ }
+
+ private void notifyPeers(InternalLinkRemovedEvent event) throws IOException {
+ broadcastMessage(GossipLinkStoreMessageSubjects.LINK_REMOVED, event);
+ }
+
+ private void notifyPeer(NodeId peer, InternalLinkEvent event) {
+ unicastMessage(peer, GossipLinkStoreMessageSubjects.LINK_UPDATE, event);
+ }
+
+ private void notifyPeer(NodeId peer, InternalLinkRemovedEvent event) {
+ unicastMessage(peer, GossipLinkStoreMessageSubjects.LINK_REMOVED, event);
+ }
+
+ private final class SendAdvertisementTask implements Runnable {
+
+ @Override
+ public void run() {
+ if (Thread.currentThread().isInterrupted()) {
+ log.info("Interrupted, quitting");
+ return;
+ }
+
+ try {
+ final NodeId self = clusterService.getLocalNode().id();
+ Set<ControllerNode> nodes = clusterService.getNodes();
+
+ ImmutableList<NodeId> nodeIds = FluentIterable.from(nodes)
+ .transform(toNodeId())
+ .toList();
+
+ if (nodeIds.size() == 1 && nodeIds.get(0).equals(self)) {
+ log.info("No other peers in the cluster.");
+ return;
+ }
+
+ NodeId peer;
+ do {
+ int idx = RandomUtils.nextInt(0, nodeIds.size());
+ peer = nodeIds.get(idx);
+ } while (peer.equals(self));
+
+ LinkAntiEntropyAdvertisement ad = createAdvertisement();
+
+ if (Thread.currentThread().isInterrupted()) {
+ log.info("Interrupted, quitting");
+ return;
+ }
+
+ try {
+ unicastMessage(peer, GossipLinkStoreMessageSubjects.LINK_ANTI_ENTROPY_ADVERTISEMENT, ad);
+ } catch (Exception e) {
+ log.error("Failed to send anti-entropy advertisement", e);
+ return;
+ }
+ } catch (Exception e) {
+ // catch all Exception to avoid Scheduled task being suppressed.
+ log.error("Exception thrown while sending advertisement", e);
+ }
+ }
+ }
+
+ private LinkAntiEntropyAdvertisement createAdvertisement() {
+ final NodeId self = clusterService.getLocalNode().id();
+
+ Map<LinkFragmentId, Timestamp> linkTimestamps = new HashMap<>(linkDescs.size());
+ Map<LinkKey, Timestamp> linkTombstones = new HashMap<>(removedLinks.size());
+
+ for (Entry<LinkKey, ConcurrentMap<ProviderId, Timestamped<LinkDescription>>>
+ provs : linkDescs.entrySet()) {
+
+ final LinkKey linkKey = provs.getKey();
+ final ConcurrentMap<ProviderId, Timestamped<LinkDescription>> linkDesc = provs.getValue();
+ synchronized (linkDesc) {
+ for (Map.Entry<ProviderId, Timestamped<LinkDescription>> e : linkDesc.entrySet()) {
+ linkTimestamps.put(new LinkFragmentId(linkKey, e.getKey()), e.getValue().timestamp());
+ }
+ }
+ }
+
+ linkTombstones.putAll(removedLinks);
+
+ return new LinkAntiEntropyAdvertisement(self, linkTimestamps, linkTombstones);
+ }
+
+ private void handleAntiEntropyAdvertisement(LinkAntiEntropyAdvertisement advertisement) {
+
+ NodeId peer = advertisement.sender();
+
+ Map<LinkFragmentId, Timestamp> linkTimestamps = advertisement.linkTimestamps();
+ Map<LinkKey, Timestamp> linkTombstones = advertisement.linkTombstones();
+ for (Map.Entry<LinkFragmentId, Timestamp> entry : linkTimestamps.entrySet()) {
+ LinkFragmentId linkFragmentId = entry.getKey();
+ Timestamp peerTimestamp = entry.getValue();
+
+ LinkKey key = linkFragmentId.linkKey();
+ ProviderId providerId = linkFragmentId.providerId();
+
+ Timestamped<LinkDescription> linkDescription = getLinkDescription(key, providerId);
+ if (linkDescription.isNewer(peerTimestamp)) {
+ // I have more recent link description. update peer.
+ notifyPeer(peer, new InternalLinkEvent(providerId, linkDescription));
+ }
+ // else TODO: Peer has more recent link description. request it.
+
+ Timestamp linkRemovedTimestamp = removedLinks.get(key);
+ if (linkRemovedTimestamp != null && linkRemovedTimestamp.compareTo(peerTimestamp) > 0) {
+ // peer has a zombie link. update peer.
+ notifyPeer(peer, new InternalLinkRemovedEvent(key, linkRemovedTimestamp));
+ }
+ }
+
+ for (Map.Entry<LinkKey, Timestamp> entry : linkTombstones.entrySet()) {
+ LinkKey key = entry.getKey();
+ Timestamp peerTimestamp = entry.getValue();
+
+ ProviderId primaryProviderId = pickPrimaryProviderId(getLinkDescriptions(key));
+ if (primaryProviderId != null) {
+ if (!getLinkDescription(key, primaryProviderId).isNewer(peerTimestamp)) {
+ notifyDelegateIfNotNull(removeLinkInternal(key, peerTimestamp));
+ }
+ }
+ }
+ }
+
+ private class InternalLinkEventListener implements ClusterMessageHandler {
+ @Override
+ public void handle(ClusterMessage message) {
+
+ log.info("Received link event from peer: {}", message.sender());
+ InternalLinkEvent event = (InternalLinkEvent) SERIALIZER.decode(message.payload());
+
+ ProviderId providerId = event.providerId();
+ Timestamped<LinkDescription> linkDescription = event.linkDescription();
+
+ notifyDelegateIfNotNull(createOrUpdateLinkInternal(providerId, linkDescription));
+ }
+ }
+
+ private class InternalLinkRemovedEventListener implements ClusterMessageHandler {
+ @Override
+ public void handle(ClusterMessage message) {
+
+ log.info("Received link removed event from peer: {}", message.sender());
+ InternalLinkRemovedEvent event = (InternalLinkRemovedEvent) SERIALIZER.decode(message.payload());
+
+ LinkKey linkKey = event.linkKey();
+ Timestamp timestamp = event.timestamp();
+
+ notifyDelegateIfNotNull(removeLinkInternal(linkKey, timestamp));
+ }
+ }
+
+ private final class InternalLinkAntiEntropyAdvertisementListener implements ClusterMessageHandler {
+
+ @Override
+ public void handle(ClusterMessage message) {
+ log.info("Received Link Anti-Entropy advertisement from peer: {}", message.sender());
+ LinkAntiEntropyAdvertisement advertisement = SERIALIZER.decode(message.payload());
+ handleAntiEntropyAdvertisement(advertisement);
+ }
+ }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/GossipLinkStoreMessageSubjects.java b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/GossipLinkStoreMessageSubjects.java
new file mode 100644
index 0000000..817efe4
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/GossipLinkStoreMessageSubjects.java
@@ -0,0 +1,18 @@
+package org.onlab.onos.store.link.impl;
+
+import org.onlab.onos.store.cluster.messaging.MessageSubject;
+
+/**
+ * MessageSubjects used by GossipLinkStore peer-peer communication.
+ */
+public final class GossipLinkStoreMessageSubjects {
+
+ private GossipLinkStoreMessageSubjects() {}
+
+ public static final MessageSubject LINK_UPDATE =
+ new MessageSubject("peer-link-update");
+ public static final MessageSubject LINK_REMOVED =
+ new MessageSubject("peer-link-removed");
+ public static final MessageSubject LINK_ANTI_ENTROPY_ADVERTISEMENT =
+ new MessageSubject("link-enti-entropy-advertisement");
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/InternalLinkEvent.java b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/InternalLinkEvent.java
new file mode 100644
index 0000000..9bb3445
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/InternalLinkEvent.java
@@ -0,0 +1,46 @@
+package org.onlab.onos.store.link.impl;
+
+import com.google.common.base.MoreObjects;
+
+import org.onlab.onos.net.link.LinkDescription;
+import org.onlab.onos.net.provider.ProviderId;
+import org.onlab.onos.store.common.impl.Timestamped;
+
+/**
+ * Information published by GossipDeviceStore to notify peers of a device
+ * change event.
+ */
+public class InternalLinkEvent {
+
+ private final ProviderId providerId;
+ private final Timestamped<LinkDescription> linkDescription;
+
+ protected InternalLinkEvent(
+ ProviderId providerId,
+ Timestamped<LinkDescription> linkDescription) {
+ this.providerId = providerId;
+ this.linkDescription = linkDescription;
+ }
+
+ public ProviderId providerId() {
+ return providerId;
+ }
+
+ public Timestamped<LinkDescription> linkDescription() {
+ return linkDescription;
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(getClass())
+ .add("providerId", providerId)
+ .add("linkDescription", linkDescription)
+ .toString();
+ }
+
+ // for serializer
+ protected InternalLinkEvent() {
+ this.providerId = null;
+ this.linkDescription = null;
+ }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/InternalLinkRemovedEvent.java b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/InternalLinkRemovedEvent.java
new file mode 100644
index 0000000..22e65ed
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/InternalLinkRemovedEvent.java
@@ -0,0 +1,49 @@
+package org.onlab.onos.store.link.impl;
+
+import org.onlab.onos.net.LinkKey;
+import org.onlab.onos.store.Timestamp;
+
+import com.google.common.base.MoreObjects;
+
+/**
+ * Information published by GossipLinkStore to notify peers of a link
+ * being removed.
+ */
+public class InternalLinkRemovedEvent {
+
+ private final LinkKey linkKey;
+ private final Timestamp timestamp;
+
+ /**
+ * Creates a InternalLinkRemovedEvent.
+ * @param linkKey identifier of the removed link.
+ * @param timestamp timestamp of when the link was removed.
+ */
+ public InternalLinkRemovedEvent(LinkKey linkKey, Timestamp timestamp) {
+ this.linkKey = linkKey;
+ this.timestamp = timestamp;
+ }
+
+ public LinkKey linkKey() {
+ return linkKey;
+ }
+
+ public Timestamp timestamp() {
+ return timestamp;
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(getClass())
+ .add("linkKey", linkKey)
+ .add("timestamp", timestamp)
+ .toString();
+ }
+
+ // for serializer
+ @SuppressWarnings("unused")
+ private InternalLinkRemovedEvent() {
+ linkKey = null;
+ timestamp = null;
+ }
+}
\ No newline at end of file
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/LinkAntiEntropyAdvertisement.java b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/LinkAntiEntropyAdvertisement.java
new file mode 100644
index 0000000..a41f9cd
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/LinkAntiEntropyAdvertisement.java
@@ -0,0 +1,48 @@
+package org.onlab.onos.store.link.impl;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.Map;
+
+import org.onlab.onos.cluster.NodeId;
+import org.onlab.onos.net.LinkKey;
+import org.onlab.onos.store.Timestamp;
+
+/**
+ * Link AE Advertisement message.
+ */
+public class LinkAntiEntropyAdvertisement {
+
+ private final NodeId sender;
+ private final Map<LinkFragmentId, Timestamp> linkTimestamps;
+ private final Map<LinkKey, Timestamp> linkTombstones;
+
+
+ public LinkAntiEntropyAdvertisement(NodeId sender,
+ Map<LinkFragmentId, Timestamp> linkTimestamps,
+ Map<LinkKey, Timestamp> linkTombstones) {
+ this.sender = checkNotNull(sender);
+ this.linkTimestamps = checkNotNull(linkTimestamps);
+ this.linkTombstones = checkNotNull(linkTombstones);
+ }
+
+ public NodeId sender() {
+ return sender;
+ }
+
+ public Map<LinkFragmentId, Timestamp> linkTimestamps() {
+ return linkTimestamps;
+ }
+
+ public Map<LinkKey, Timestamp> linkTombstones() {
+ return linkTombstones;
+ }
+
+ // For serializer
+ @SuppressWarnings("unused")
+ private LinkAntiEntropyAdvertisement() {
+ this.sender = null;
+ this.linkTimestamps = null;
+ this.linkTombstones = null;
+ }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/LinkFragmentId.java b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/LinkFragmentId.java
new file mode 100644
index 0000000..f97bef6
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/LinkFragmentId.java
@@ -0,0 +1,62 @@
+package org.onlab.onos.store.link.impl;
+
+import java.util.Objects;
+
+import org.onlab.onos.net.LinkKey;
+import org.onlab.onos.net.provider.ProviderId;
+
+import com.google.common.base.MoreObjects;
+
+/**
+ * Identifier for LinkDescription from a Provider.
+ */
+public final class LinkFragmentId {
+ public final ProviderId providerId;
+ public final LinkKey linkKey;
+
+ public LinkFragmentId(LinkKey linkKey, ProviderId providerId) {
+ this.providerId = providerId;
+ this.linkKey = linkKey;
+ }
+
+ public LinkKey linkKey() {
+ return linkKey;
+ }
+
+ public ProviderId providerId() {
+ return providerId;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(providerId, linkKey);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof LinkFragmentId)) {
+ return false;
+ }
+ LinkFragmentId that = (LinkFragmentId) obj;
+ return Objects.equals(this.linkKey, that.linkKey) &&
+ Objects.equals(this.providerId, that.providerId);
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(getClass())
+ .add("providerId", providerId)
+ .add("linkKey", linkKey)
+ .toString();
+ }
+
+ // for serializer
+ @SuppressWarnings("unused")
+ private LinkFragmentId() {
+ this.providerId = null;
+ this.linkKey = null;
+ }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/OnosDistributedLinkStore.java b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/OnosDistributedLinkStore.java
deleted file mode 100644
index 63245a3..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/OnosDistributedLinkStore.java
+++ /dev/null
@@ -1,246 +0,0 @@
-package org.onlab.onos.store.link.impl;
-
-import static org.onlab.onos.net.Link.Type.DIRECT;
-import static org.onlab.onos.net.Link.Type.INDIRECT;
-import static org.onlab.onos.net.link.LinkEvent.Type.LINK_ADDED;
-import static org.onlab.onos.net.link.LinkEvent.Type.LINK_REMOVED;
-import static org.onlab.onos.net.link.LinkEvent.Type.LINK_UPDATED;
-import static org.slf4j.LoggerFactory.getLogger;
-
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-
-import org.apache.felix.scr.annotations.Activate;
-import org.apache.felix.scr.annotations.Component;
-import org.apache.felix.scr.annotations.Deactivate;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.ReferenceCardinality;
-import org.apache.felix.scr.annotations.Service;
-import org.onlab.onos.net.ConnectPoint;
-import org.onlab.onos.net.DefaultLink;
-import org.onlab.onos.net.DeviceId;
-import org.onlab.onos.net.Link;
-import org.onlab.onos.net.LinkKey;
-import org.onlab.onos.net.link.LinkDescription;
-import org.onlab.onos.net.link.LinkEvent;
-import org.onlab.onos.net.link.LinkStore;
-import org.onlab.onos.net.link.LinkStoreDelegate;
-import org.onlab.onos.net.provider.ProviderId;
-import org.onlab.onos.store.AbstractStore;
-import org.onlab.onos.store.ClockService;
-import org.onlab.onos.store.Timestamp;
-import org.slf4j.Logger;
-
-import com.google.common.collect.HashMultimap;
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Multimap;
-import com.google.common.collect.ImmutableSet.Builder;
-
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkState;
-
-//TODO: Add support for multiple provider and annotations
-/**
- * Manages inventory of infrastructure links using a protocol that takes into consideration
- * the order in which events occur.
- */
-// FIXME: This does not yet implement the full protocol.
-// The full protocol requires the sender of LLDP message to include the
-// version information of src device/port and the receiver to
-// take that into account when figuring out if a more recent src
-// device/port down event renders the link discovery obsolete.
-@Component(immediate = true)
-@Service
-public class OnosDistributedLinkStore
- extends AbstractStore<LinkEvent, LinkStoreDelegate>
- implements LinkStore {
-
- private final Logger log = getLogger(getClass());
-
- // Link inventory
- private ConcurrentMap<LinkKey, VersionedValue<Link>> links;
-
- public static final String LINK_NOT_FOUND = "Link between %s and %s not found";
-
- // TODO synchronize?
- // Egress and ingress link sets
- private final Multimap<DeviceId, VersionedValue<Link>> srcLinks = HashMultimap.create();
- private final Multimap<DeviceId, VersionedValue<Link>> dstLinks = HashMultimap.create();
-
- @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
- protected ClockService clockService;
-
- @Activate
- public void activate() {
-
- links = new ConcurrentHashMap<>();
-
- log.info("Started");
- }
-
- @Deactivate
- public void deactivate() {
- log.info("Stopped");
- }
-
- @Override
- public int getLinkCount() {
- return links.size();
- }
-
- @Override
- public Iterable<Link> getLinks() {
- Builder<Link> builder = ImmutableSet.builder();
- synchronized (this) {
- for (VersionedValue<Link> link : links.values()) {
- builder.add(link.entity());
- }
- return builder.build();
- }
- }
-
- @Override
- public Set<Link> getDeviceEgressLinks(DeviceId deviceId) {
- Set<VersionedValue<Link>> egressLinks = ImmutableSet.copyOf(srcLinks.get(deviceId));
- Set<Link> rawEgressLinks = new HashSet<>();
- for (VersionedValue<Link> link : egressLinks) {
- rawEgressLinks.add(link.entity());
- }
- return rawEgressLinks;
- }
-
- @Override
- public Set<Link> getDeviceIngressLinks(DeviceId deviceId) {
- Set<VersionedValue<Link>> ingressLinks = ImmutableSet.copyOf(dstLinks.get(deviceId));
- Set<Link> rawIngressLinks = new HashSet<>();
- for (VersionedValue<Link> link : ingressLinks) {
- rawIngressLinks.add(link.entity());
- }
- return rawIngressLinks;
- }
-
- @Override
- public Link getLink(ConnectPoint src, ConnectPoint dst) {
- VersionedValue<Link> link = links.get(new LinkKey(src, dst));
- checkArgument(link != null, "LINK_NOT_FOUND", src, dst);
- return link.entity();
- }
-
- @Override
- public Set<Link> getEgressLinks(ConnectPoint src) {
- Set<Link> egressLinks = new HashSet<>();
- for (VersionedValue<Link> link : srcLinks.get(src.deviceId())) {
- if (link.entity().src().equals(src)) {
- egressLinks.add(link.entity());
- }
- }
- return egressLinks;
- }
-
- @Override
- public Set<Link> getIngressLinks(ConnectPoint dst) {
- Set<Link> ingressLinks = new HashSet<>();
- for (VersionedValue<Link> link : dstLinks.get(dst.deviceId())) {
- if (link.entity().dst().equals(dst)) {
- ingressLinks.add(link.entity());
- }
- }
- return ingressLinks;
- }
-
- @Override
- public LinkEvent createOrUpdateLink(ProviderId providerId,
- LinkDescription linkDescription) {
-
- final DeviceId destinationDeviceId = linkDescription.dst().deviceId();
- final Timestamp newTimestamp = clockService.getTimestamp(destinationDeviceId);
-
- LinkKey key = new LinkKey(linkDescription.src(), linkDescription.dst());
- VersionedValue<Link> link = links.get(key);
- if (link == null) {
- return createLink(providerId, key, linkDescription, newTimestamp);
- }
-
- checkState(newTimestamp.compareTo(link.timestamp()) > 0,
- "Existing Link has a timestamp in the future!");
-
- return updateLink(providerId, link, key, linkDescription, newTimestamp);
- }
-
- // Creates and stores the link and returns the appropriate event.
- private LinkEvent createLink(ProviderId providerId, LinkKey key,
- LinkDescription linkDescription, Timestamp timestamp) {
- VersionedValue<Link> link = new VersionedValue<Link>(new DefaultLink(providerId, key.src(), key.dst(),
- linkDescription.type()), true, timestamp);
- synchronized (this) {
- links.put(key, link);
- addNewLink(link, timestamp);
- }
- // FIXME: notify peers.
- return new LinkEvent(LINK_ADDED, link.entity());
- }
-
- // update Egress and ingress link sets
- private void addNewLink(VersionedValue<Link> link, Timestamp timestamp) {
- Link rawLink = link.entity();
- synchronized (this) {
- srcLinks.put(rawLink.src().deviceId(), link);
- dstLinks.put(rawLink.dst().deviceId(), link);
- }
- }
-
- // Updates, if necessary the specified link and returns the appropriate event.
- private LinkEvent updateLink(ProviderId providerId, VersionedValue<Link> existingLink,
- LinkKey key, LinkDescription linkDescription, Timestamp timestamp) {
- // FIXME confirm Link update condition is OK
- if (existingLink.entity().type() == INDIRECT && linkDescription.type() == DIRECT) {
- synchronized (this) {
-
- VersionedValue<Link> updatedLink = new VersionedValue<Link>(
- new DefaultLink(providerId, existingLink.entity().src(), existingLink.entity().dst(),
- linkDescription.type()), true, timestamp);
- links.replace(key, existingLink, updatedLink);
-
- replaceLink(existingLink, updatedLink);
- // FIXME: notify peers.
- return new LinkEvent(LINK_UPDATED, updatedLink.entity());
- }
- }
- return null;
- }
-
- // update Egress and ingress link sets
- private void replaceLink(VersionedValue<Link> current, VersionedValue<Link> updated) {
- synchronized (this) {
- srcLinks.remove(current.entity().src().deviceId(), current);
- dstLinks.remove(current.entity().dst().deviceId(), current);
-
- srcLinks.put(current.entity().src().deviceId(), updated);
- dstLinks.put(current.entity().dst().deviceId(), updated);
- }
- }
-
- @Override
- public LinkEvent removeLink(ConnectPoint src, ConnectPoint dst) {
- synchronized (this) {
- LinkKey key = new LinkKey(src, dst);
- VersionedValue<Link> link = links.remove(key);
- if (link != null) {
- removeLink(link);
- // notify peers
- return new LinkEvent(LINK_REMOVED, link.entity());
- }
- return null;
- }
- }
-
- // update Egress and ingress link sets
- private void removeLink(VersionedValue<Link> link) {
- synchronized (this) {
- srcLinks.remove(link.entity().src().deviceId(), link);
- dstLinks.remove(link.entity().dst().deviceId(), link);
- }
- }
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/VersionedValue.java b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/VersionedValue.java
deleted file mode 100644
index 391cd1d..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/VersionedValue.java
+++ /dev/null
@@ -1,79 +0,0 @@
-package org.onlab.onos.store.link.impl;
-
-import java.util.Objects;
-
-import org.onlab.onos.store.Timestamp;
-
-// TODO: remove once we stop using this
-/**
- * Wrapper class for a entity that is versioned
- * and can either be up or down.
- *
- * @param <T> type of the value.
- */
-public class VersionedValue<T> {
- private final T entity;
- private final Timestamp timestamp;
- private final boolean isUp;
-
- public VersionedValue(T entity, boolean isUp, Timestamp timestamp) {
- this.entity = entity;
- this.isUp = isUp;
- this.timestamp = timestamp;
- }
-
- /**
- * Returns the value.
- * @return value.
- */
- public T entity() {
- return entity;
- }
-
- /**
- * Tells whether the entity is up or down.
- * @return true if up, false otherwise.
- */
- public boolean isUp() {
- return isUp;
- }
-
- /**
- * Returns the timestamp (version) associated with this entity.
- * @return timestamp.
- */
- public Timestamp timestamp() {
- return timestamp;
- }
-
-
- @Override
- public int hashCode() {
- return Objects.hash(entity, timestamp, isUp);
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null) {
- return false;
- }
- if (getClass() != obj.getClass()) {
- return false;
- }
- @SuppressWarnings("unchecked")
- VersionedValue<T> that = (VersionedValue<T>) obj;
- return Objects.equals(this.entity, that.entity) &&
- Objects.equals(this.timestamp, that.timestamp) &&
- Objects.equals(this.isUp, that.isUp);
- }
-
- // Default constructor for serializer
- protected VersionedValue() {
- this.entity = null;
- this.isUp = false;
- this.timestamp = null;
- }
-}
diff --git a/core/store/dist/src/test/java/org/onlab/onos/store/device/impl/peermsg/DeviceFragmentIdTest.java b/core/store/dist/src/test/java/org/onlab/onos/store/device/impl/peermsg/DeviceFragmentIdTest.java
new file mode 100644
index 0000000..85ba4a7
--- /dev/null
+++ b/core/store/dist/src/test/java/org/onlab/onos/store/device/impl/peermsg/DeviceFragmentIdTest.java
@@ -0,0 +1,33 @@
+package org.onlab.onos.store.device.impl.peermsg;
+
+import static org.onlab.onos.net.DeviceId.deviceId;
+
+import org.junit.Test;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.provider.ProviderId;
+
+import com.google.common.testing.EqualsTester;
+
+public class DeviceFragmentIdTest {
+
+ private static final ProviderId PID = new ProviderId("of", "foo");
+ private static final ProviderId PIDA = new ProviderId("of", "bar", true);
+ private static final DeviceId DID1 = deviceId("of:foo");
+ private static final DeviceId DID2 = deviceId("of:bar");
+
+ @Test
+ public final void testEquals() {
+
+ new EqualsTester()
+ .addEqualityGroup(new DeviceFragmentId(DID1, PID),
+ new DeviceFragmentId(DID1, PID))
+ .addEqualityGroup(new DeviceFragmentId(DID2, PID),
+ new DeviceFragmentId(DID2, PID))
+ .addEqualityGroup(new DeviceFragmentId(DID1, PIDA),
+ new DeviceFragmentId(DID1, PIDA))
+ .addEqualityGroup(new DeviceFragmentId(DID2, PIDA),
+ new DeviceFragmentId(DID2, PIDA))
+ .testEquals();
+ }
+
+}
diff --git a/core/store/dist/src/test/java/org/onlab/onos/store/device/impl/peermsg/PortFragmentIdTest.java b/core/store/dist/src/test/java/org/onlab/onos/store/device/impl/peermsg/PortFragmentIdTest.java
new file mode 100644
index 0000000..97db29d
--- /dev/null
+++ b/core/store/dist/src/test/java/org/onlab/onos/store/device/impl/peermsg/PortFragmentIdTest.java
@@ -0,0 +1,46 @@
+package org.onlab.onos.store.device.impl.peermsg;
+
+import static org.onlab.onos.net.DeviceId.deviceId;
+
+import org.junit.Test;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.PortNumber;
+import org.onlab.onos.net.provider.ProviderId;
+
+import com.google.common.testing.EqualsTester;
+
+public class PortFragmentIdTest {
+
+ private static final ProviderId PID = new ProviderId("of", "foo");
+ private static final ProviderId PIDA = new ProviderId("of", "bar", true);
+
+ private static final DeviceId DID1 = deviceId("of:foo");
+ private static final DeviceId DID2 = deviceId("of:bar");
+
+ private static final PortNumber PN1 = PortNumber.portNumber(1);
+ private static final PortNumber PN2 = PortNumber.portNumber(2);
+
+ @Test
+ public final void testEquals() {
+ new EqualsTester()
+ .addEqualityGroup(new PortFragmentId(DID1, PID, PN1),
+ new PortFragmentId(DID1, PID, PN1))
+ .addEqualityGroup(new PortFragmentId(DID2, PID, PN1),
+ new PortFragmentId(DID2, PID, PN1))
+ .addEqualityGroup(new PortFragmentId(DID1, PIDA, PN1),
+ new PortFragmentId(DID1, PIDA, PN1))
+ .addEqualityGroup(new PortFragmentId(DID2, PIDA, PN1),
+ new PortFragmentId(DID2, PIDA, PN1))
+
+ .addEqualityGroup(new PortFragmentId(DID1, PID, PN2),
+ new PortFragmentId(DID1, PID, PN2))
+ .addEqualityGroup(new PortFragmentId(DID2, PID, PN2),
+ new PortFragmentId(DID2, PID, PN2))
+ .addEqualityGroup(new PortFragmentId(DID1, PIDA, PN2),
+ new PortFragmentId(DID1, PIDA, PN2))
+ .addEqualityGroup(new PortFragmentId(DID2, PIDA, PN2),
+ new PortFragmentId(DID2, PIDA, PN2))
+ .testEquals();
+ }
+
+}
diff --git a/core/store/hz/cluster/pom.xml b/core/store/hz/cluster/pom.xml
index 95307f1..d183aed 100644
--- a/core/store/hz/cluster/pom.xml
+++ b/core/store/hz/cluster/pom.xml
@@ -46,10 +46,6 @@
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast</artifactId>
</dependency>
- <dependency>
- <groupId>de.javakaffee</groupId>
- <artifactId>kryo-serializers</artifactId>
- </dependency>
</dependencies>
<build>
diff --git a/core/store/hz/common/pom.xml b/core/store/hz/common/pom.xml
index 06aa0b7..1d79206 100644
--- a/core/store/hz/common/pom.xml
+++ b/core/store/hz/common/pom.xml
@@ -35,8 +35,8 @@
<artifactId>hazelcast</artifactId>
</dependency>
<dependency>
- <groupId>de.javakaffee</groupId>
- <artifactId>kryo-serializers</artifactId>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
</dependency>
</dependencies>
diff --git a/core/store/hz/net/pom.xml b/core/store/hz/net/pom.xml
index e3bc0e2..177e99e 100644
--- a/core/store/hz/net/pom.xml
+++ b/core/store/hz/net/pom.xml
@@ -23,11 +23,6 @@
</dependency>
<dependency>
<groupId>org.onlab.onos</groupId>
- <artifactId>onos-core-serializers</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.onlab.onos</groupId>
<artifactId>onos-core-hz-common</artifactId>
<version>${project.version}</version>
</dependency>
@@ -46,10 +41,6 @@
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast</artifactId>
</dependency>
- <dependency>
- <groupId>de.javakaffee</groupId>
- <artifactId>kryo-serializers</artifactId>
- </dependency>
</dependencies>
<build>
diff --git a/core/store/serializers/pom.xml b/core/store/serializers/pom.xml
index f222a23..fe0a501 100644
--- a/core/store/serializers/pom.xml
+++ b/core/store/serializers/pom.xml
@@ -26,8 +26,13 @@
<artifactId>org.apache.felix.scr.annotations</artifactId>
</dependency>
<dependency>
- <groupId>de.javakaffee</groupId>
- <artifactId>kryo-serializers</artifactId>
+ <groupId>com.esotericsoftware</groupId>
+ <artifactId>kryo</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava-testlib</artifactId>
+ <scope>test</scope>
</dependency>
</dependencies>
diff --git a/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/KryoPoolUtil.java b/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/KryoPoolUtil.java
index f079874..efecb6c 100644
--- a/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/KryoPoolUtil.java
+++ b/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/KryoPoolUtil.java
@@ -24,14 +24,13 @@
import org.onlab.onos.net.PortNumber;
import org.onlab.onos.net.device.DefaultDeviceDescription;
import org.onlab.onos.net.device.DefaultPortDescription;
+import org.onlab.onos.net.link.DefaultLinkDescription;
import org.onlab.onos.net.provider.ProviderId;
import org.onlab.onos.store.Timestamp;
import org.onlab.packet.IpAddress;
import org.onlab.packet.IpPrefix;
import org.onlab.util.KryoPool;
-import de.javakaffee.kryoserializers.URISerializer;
-
public final class KryoPoolUtil {
/**
@@ -60,6 +59,7 @@
DefaultControllerNode.class,
DefaultDevice.class,
DefaultDeviceDescription.class,
+ DefaultLinkDescription.class,
MastershipRole.class,
Port.class,
DefaultPortDescription.class,
diff --git a/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/URISerializer.java b/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/URISerializer.java
new file mode 100644
index 0000000..4c2d61e
--- /dev/null
+++ b/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/URISerializer.java
@@ -0,0 +1,31 @@
+package org.onlab.onos.store.serializers;
+
+import java.net.URI;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+/**
+ * Serializer for {@link URI}.
+ */
+public class URISerializer extends Serializer<URI> {
+
+ /**
+ * Creates {@link URI} serializer instance.
+ */
+ public URISerializer() {
+ super(false);
+ }
+
+ @Override
+ public void write(Kryo kryo, Output output, URI object) {
+ output.writeString(object.toString());
+ }
+
+ @Override
+ public URI read(Kryo kryo, Input input, Class<URI> type) {
+ return URI.create(input.readString());
+ }
+}
diff --git a/features/features.xml b/features/features.xml
index b240917..c9e6bdd 100644
--- a/features/features.xml
+++ b/features/features.xml
@@ -20,10 +20,11 @@
<bundle>mvn:io.dropwizard.metrics/metrics-core/3.1.0</bundle>
<bundle>mvn:com.eclipsesource.minimal-json/minimal-json/0.9.1</bundle>
- <bundle>mvn:com.esotericsoftware.kryo/kryo/2.24.0</bundle>
+ <bundle>mvn:com.esotericsoftware/kryo/3.0.0</bundle>
+ <bundle>mvn:com.esotericsoftware/reflectasm/1.10.0</bundle>
+ <bundle>mvn:org.ow2.asm/asm/4.2</bundle>
<bundle>mvn:com.esotericsoftware/minlog/1.3.0</bundle>
<bundle>mvn:org.objenesis/objenesis/2.1</bundle>
- <bundle>mvn:de.javakaffee/kryo-serializers/0.27</bundle>
<bundle>mvn:org.onlab.onos/onlab-nio/1.0.0-SNAPSHOT</bundle>
diff --git a/pom.xml b/pom.xml
index bf5e1f0..4290769 100644
--- a/pom.xml
+++ b/pom.xml
@@ -193,9 +193,20 @@
<version>0.9.1</version>
</dependency>
<dependency>
- <groupId>com.esotericsoftware.kryo</groupId>
- <artifactId>kryo</artifactId>
- <version>2.24.0</version>
+ <groupId>com.esotericsoftware</groupId>
+ <artifactId>kryo</artifactId>
+ <version>3.0.0</version>
+ </dependency>
+ <dependency>
+ <groupId>com.esotericsoftware</groupId>
+ <artifactId>reflectasm</artifactId>
+ <version>1.10.0</version>
+ <type>bundle</type>
+ </dependency>
+ <dependency>
+ <groupId>org.ow2.asm</groupId>
+ <artifactId>asm</artifactId>
+ <version>4.2</version>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
@@ -207,11 +218,6 @@
<artifactId>objenesis</artifactId>
<version>2.1</version>
</dependency>
- <dependency>
- <groupId>de.javakaffee</groupId>
- <artifactId>kryo-serializers</artifactId>
- <version>0.27</version>
- </dependency>
<!-- ONOS related -->
<dependency>
@@ -284,6 +290,10 @@
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
<artifactId>slf4j-jdk14</artifactId>
</dependency>
</dependencies>
@@ -434,9 +444,6 @@
<version>3.2</version>
<configuration>
<excludes>
- <exclude>**/datastore/serializers/**</exclude>
- <exclude>**/edu/stanford/**</exclude>
- <exclude>**/net/floodlightcontroller/**</exclude>
</excludes>
<rulesets>
<ruleset>onos/pmd.xml</ruleset>
@@ -545,9 +552,6 @@
<version>3.2</version>
<configuration>
<excludes>
- <exclude>**/datastore/serializers/**</exclude>
- <exclude>**/edu/stanford/**</exclude>
- <exclude>**/net/floodlightcontroller/**</exclude>
</excludes>
<rulesets>
<ruleset>onos/pmd.xml</ruleset>
diff --git a/tools/dev/bin/onos-build-selective-hook b/tools/dev/bin/onos-build-selective-hook
index 0b39713..eb4e482 100755
--- a/tools/dev/bin/onos-build-selective-hook
+++ b/tools/dev/bin/onos-build-selective-hook
@@ -1,3 +1,4 @@
+#!/bin/bash
#------------------------------------------------------------------------------
# Echoes project-level directory if a Java file within is newer than the
# target directory.
diff --git a/utils/misc/pom.xml b/utils/misc/pom.xml
index bd3cc08..5040fb0 100644
--- a/utils/misc/pom.xml
+++ b/utils/misc/pom.xml
@@ -44,18 +44,10 @@
<artifactId>minimal-json</artifactId>
</dependency>
<dependency>
- <groupId>com.esotericsoftware.kryo</groupId>
+ <groupId>com.esotericsoftware</groupId>
<artifactId>kryo</artifactId>
</dependency>
<dependency>
- <groupId>com.esotericsoftware</groupId>
- <artifactId>minlog</artifactId>
- </dependency>
- <dependency>
- <groupId>org.objenesis</groupId>
- <artifactId>objenesis</artifactId>
- </dependency>
- <dependency>
<groupId>io.dropwizard.metrics</groupId>
<artifactId>metrics-core</artifactId>
<version>3.1.0</version>
diff --git a/utils/netty/pom.xml b/utils/netty/pom.xml
index a980d1d..effbc3b 100644
--- a/utils/netty/pom.xml
+++ b/utils/netty/pom.xml
@@ -32,10 +32,6 @@
<scope>test</scope>
</dependency>
<dependency>
- <groupId>de.javakaffee</groupId>
- <artifactId>kryo-serializers</artifactId>
- </dependency>
- <dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
</dependency>