Merge branch 'master' of ssh://gerrit.onlab.us:29418/onos-next
diff --git a/core/store/dist/pom.xml b/core/store/dist/pom.xml
index 2451955..1faab74 100644
--- a/core/store/dist/pom.xml
+++ b/core/store/dist/pom.xml
@@ -33,6 +33,12 @@
<artifactId>onlab-nio</artifactId>
<version>${project.version}</version>
</dependency>
+
+ <dependency>
+ <groupId>org.onlab.onos</groupId>
+ <artifactId>onlab-netty</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
@@ -51,15 +57,6 @@
<groupId>de.javakaffee</groupId>
<artifactId>kryo-serializers</artifactId>
</dependency>
- <dependency>
- <groupId>io.netty</groupId>
- <artifactId>netty-all</artifactId>
- </dependency>
- <dependency>
- <groupId>commons-pool</groupId>
- <artifactId>commons-pool</artifactId>
- <version>1.6</version>
- </dependency>
</dependencies>
<build>
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/OnosClusterCommunicationManager.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/OnosClusterCommunicationManager.java
index 9bd25b4..e6e4a4d 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/OnosClusterCommunicationManager.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/OnosClusterCommunicationManager.java
@@ -23,10 +23,10 @@
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.messaging.Endpoint;
-import org.onlab.onos.store.messaging.Message;
-import org.onlab.onos.store.messaging.MessageHandler;
-import org.onlab.onos.store.messaging.MessagingService;
+import org.onlab.netty.Endpoint;
+import org.onlab.netty.Message;
+import org.onlab.netty.MessageHandler;
+import org.onlab.netty.MessagingService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/Endpoint.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/Endpoint.java
deleted file mode 100644
index bd6d45f..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/Endpoint.java
+++ /dev/null
@@ -1,62 +0,0 @@
-package org.onlab.onos.store.messaging;
-
-/**
- * Representation of a TCP/UDP communication end point.
- */
-public class Endpoint {
-
- private final int port;
- private final String host;
-
- public Endpoint(String host, int port) {
- this.host = host;
- this.port = port;
- }
-
- public String host() {
- return host;
- }
-
- public int port() {
- return port;
- }
-
- @Override
- public String toString() {
- return "Endpoint [port=" + port + ", host=" + host + "]";
- }
-
- @Override
- public int hashCode() {
- final int prime = 31;
- int result = 1;
- result = prime * result + ((host == null) ? 0 : host.hashCode());
- result = prime * result + port;
- return result;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null) {
- return false;
- }
- if (getClass() != obj.getClass()) {
- return false;
- }
- Endpoint other = (Endpoint) obj;
- if (host == null) {
- if (other.host != null) {
- return false;
- }
- } else if (!host.equals(other.host)) {
- return false;
- }
- if (port != other.port) {
- return false;
- }
- return true;
- }
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/Message.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/Message.java
deleted file mode 100644
index d814927..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/Message.java
+++ /dev/null
@@ -1,23 +0,0 @@
-package org.onlab.onos.store.messaging;
-
-import java.io.IOException;
-
-/**
- * A unit of communication.
- * Has a payload. Also supports a feature to respond back to the sender.
- */
-public interface Message {
-
- /**
- * Returns the payload of this message.
- * @return message payload.
- */
- public Object payload();
-
- /**
- * Sends a reply back to the sender of this messge.
- * @param data payload of the response.
- * @throws IOException if there is a communication error.
- */
- public void respond(Object data) throws IOException;
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/MessageHandler.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/MessageHandler.java
deleted file mode 100644
index 8eaef1e..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/MessageHandler.java
+++ /dev/null
@@ -1,16 +0,0 @@
-package org.onlab.onos.store.messaging;
-
-import java.io.IOException;
-
-/**
- * Handler for a message.
- */
-public interface MessageHandler {
-
- /**
- * Handles the message.
- * @param message message.
- * @throws IOException.
- */
- public void handle(Message message) throws IOException;
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/MessagingService.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/MessagingService.java
deleted file mode 100644
index 4aa32cb..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/MessagingService.java
+++ /dev/null
@@ -1,41 +0,0 @@
-package org.onlab.onos.store.messaging;
-
-import java.io.IOException;
-
-/**
- * Interface for low level messaging primitives.
- */
-public interface MessagingService {
- /**
- * Sends a message asynchronously to the specified communication end point.
- * The message is specified using the type and payload.
- * @param ep end point to send the message to.
- * @param type type of message.
- * @param payload message payload.
- * @throws IOException
- */
- public void sendAsync(Endpoint ep, String type, Object payload) throws IOException;
-
- /**
- * Sends a message synchronously and waits for a response.
- * @param ep end point to send the message to.
- * @param type type of message.
- * @param payload message payload.
- * @return a response future
- * @throws IOException
- */
- public <T> Response<T> sendAndReceive(Endpoint ep, String type, Object payload) throws IOException;
-
- /**
- * Registers a new message handler for message type.
- * @param type message type.
- * @param handler message handler
- */
- public void registerHandler(String type, MessageHandler handler);
-
- /**
- * Unregister current handler, if one exists for message type.
- * @param type message type
- */
- public void unregisterHandler(String type);
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/Response.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/Response.java
deleted file mode 100644
index ff0d84f..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/Response.java
+++ /dev/null
@@ -1,36 +0,0 @@
-package org.onlab.onos.store.messaging;
-
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-/**
- * Response object returned when making synchronous requests.
- * Can you used to check is a response is ready and/or wait for a response
- * to become available.
- *
- * @param <T> type of response.
- */
-public interface Response<T> {
-
- /**
- * Gets the response waiting for a designated timeout period.
- * @param timeout timeout period (since request was sent out)
- * @param tu unit of time.
- * @return response
- * @throws TimeoutException if the timeout expires before the response arrives.
- */
- public T get(long timeout, TimeUnit tu) throws TimeoutException;
-
- /**
- * Gets the response waiting for indefinite timeout period.
- * @return response
- * @throws InterruptedException if the thread is interrupted before the response arrives.
- */
- public T get() throws InterruptedException;
-
- /**
- * Checks if the response is ready without blocking.
- * @return true if response is ready, false otherwise.
- */
- public boolean isReady();
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/AsyncResponse.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/AsyncResponse.java
deleted file mode 100644
index ac2337d..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/AsyncResponse.java
+++ /dev/null
@@ -1,70 +0,0 @@
-package org.onlab.onos.store.messaging.impl;
-
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import org.onlab.onos.store.messaging.Response;
-
-/**
- * An asynchronous response.
- * This class provides a base implementation of Response, with methods to retrieve the
- * result and query to see if the result is ready. The result can only be retrieved when
- * it is ready and the get methods will block if the result is not ready yet.
- * @param <T> type of response.
- */
-public class AsyncResponse<T> implements Response<T> {
-
- private T value;
- private boolean done = false;
- private final long start = System.nanoTime();
-
- @Override
- public T get(long timeout, TimeUnit tu) throws TimeoutException {
- timeout = tu.toNanos(timeout);
- boolean interrupted = false;
- try {
- synchronized (this) {
- while (!done) {
- try {
- long timeRemaining = timeout - (System.nanoTime() - start);
- if (timeRemaining <= 0) {
- throw new TimeoutException("Operation timed out.");
- }
- TimeUnit.NANOSECONDS.timedWait(this, timeRemaining);
- } catch (InterruptedException e) {
- interrupted = true;
- }
- }
- }
- } finally {
- if (interrupted) {
- Thread.currentThread().interrupt();
- }
- }
- return value;
- }
-
- @Override
- public T get() throws InterruptedException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean isReady() {
- return done;
- }
-
- /**
- * Sets response value and unblocks any thread blocking on the response to become
- * available.
- * @param data response data.
- */
- @SuppressWarnings("unchecked")
- public synchronized void setResponse(Object data) {
- if (!done) {
- done = true;
- value = (T) data;
- this.notifyAll();
- }
- }
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/EchoHandler.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/EchoHandler.java
deleted file mode 100644
index 7891c5c..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/EchoHandler.java
+++ /dev/null
@@ -1,18 +0,0 @@
-package org.onlab.onos.store.messaging.impl;
-
-import java.io.IOException;
-
-import org.onlab.onos.store.messaging.Message;
-import org.onlab.onos.store.messaging.MessageHandler;
-
-/**
- * Message handler that echos the message back to the sender.
- */
-public class EchoHandler implements MessageHandler {
-
- @Override
- public void handle(Message message) throws IOException {
- System.out.println("Received: " + message.payload() + ". Echoing it back to the sender.");
- message.respond(message.payload());
- }
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/InternalMessage.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/InternalMessage.java
deleted file mode 100644
index 8a87a3e..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/InternalMessage.java
+++ /dev/null
@@ -1,88 +0,0 @@
-package org.onlab.onos.store.messaging.impl;
-
-import java.io.IOException;
-
-import org.onlab.onos.store.messaging.Endpoint;
-import org.onlab.onos.store.messaging.Message;
-
-/**
- * Internal message representation with additional attributes
- * for supporting, synchronous request/reply behavior.
- */
-public final class InternalMessage implements Message {
-
- private long id;
- private Endpoint sender;
- private String type;
- private Object payload;
- private transient NettyMessagingService messagingService;
- public static final String REPLY_MESSAGE_TYPE = "NETTY_MESSAGIG_REQUEST_REPLY";
-
- // Must be created using the Builder.
- private InternalMessage() {}
-
- public long id() {
- return id;
- }
-
- public String type() {
- return type;
- }
-
- public Endpoint sender() {
- return sender;
- }
-
- @Override
- public Object payload() {
- return payload;
- }
-
- @Override
- public void respond(Object data) throws IOException {
- Builder builder = new Builder(messagingService);
- InternalMessage message = builder.withId(this.id)
- // FIXME: Sender should be messagingService.localEp.
- .withSender(this.sender)
- .withPayload(data)
- .withType(REPLY_MESSAGE_TYPE)
- .build();
- messagingService.sendAsync(sender, message);
- }
-
-
- /**
- * Builder for InternalMessages.
- */
- public static class Builder {
- private InternalMessage message;
-
- public Builder(NettyMessagingService messagingService) {
- message = new InternalMessage();
- message.messagingService = messagingService;
- }
-
- public Builder withId(long id) {
- message.id = id;
- return this;
- }
-
- public Builder withType(String type) {
- message.type = type;
- return this;
- }
-
- public Builder withSender(Endpoint sender) {
- message.sender = sender;
- return this;
- }
- public Builder withPayload(Object payload) {
- message.payload = payload;
- return this;
- }
-
- public InternalMessage build() {
- return message;
- }
- }
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/KryoSerializer.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/KryoSerializer.java
deleted file mode 100644
index 5d809a4..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/KryoSerializer.java
+++ /dev/null
@@ -1,47 +0,0 @@
-package org.onlab.onos.store.messaging.impl;
-
-import org.onlab.util.KryoPool;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-
-/**
- * Kryo Serializer.
- */
-public class KryoSerializer implements Serializer {
-
- private final Logger log = LoggerFactory.getLogger(getClass());
-
- private KryoPool serializerPool;
-
- public KryoSerializer() {
- setupKryoPool();
- }
-
- /**
- * Sets up the common serialzers pool.
- */
- protected void setupKryoPool() {
- // FIXME Slice out types used in common to separate pool/namespace.
- serializerPool = KryoPool.newBuilder()
- .register(ArrayList.class,
- HashMap.class,
- ArrayList.class
- )
- .build()
- .populate(1);
- }
-
-
- @Override
- public Object decode(byte[] data) {
- return serializerPool.deserialize(data);
- }
-
- @Override
- public byte[] encode(Object payload) {
- return serializerPool.serialize(payload);
- }
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/LoggingHandler.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/LoggingHandler.java
deleted file mode 100644
index bf871f8..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/LoggingHandler.java
+++ /dev/null
@@ -1,15 +0,0 @@
-package org.onlab.onos.store.messaging.impl;
-
-import org.onlab.onos.store.messaging.Message;
-import org.onlab.onos.store.messaging.MessageHandler;
-
-/**
- * A MessageHandler that simply logs the information.
- */
-public class LoggingHandler implements MessageHandler {
-
- @Override
- public void handle(Message message) {
- System.out.println("Received: " + message.payload());
- }
-}
\ No newline at end of file
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/MessageDecoder.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/MessageDecoder.java
deleted file mode 100644
index 59790f6..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/MessageDecoder.java
+++ /dev/null
@@ -1,60 +0,0 @@
-package org.onlab.onos.store.messaging.impl;
-
-import java.util.Arrays;
-import java.util.List;
-
-import static com.google.common.base.Preconditions.checkState;
-
-import org.onlab.onos.store.messaging.Endpoint;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.ByteToMessageDecoder;
-
-/**
- * Decode bytes into a InrenalMessage.
- */
-public class MessageDecoder extends ByteToMessageDecoder {
-
- private final NettyMessagingService messagingService;
- private final Serializer serializer;
-
- public MessageDecoder(NettyMessagingService messagingService, Serializer serializer) {
- this.messagingService = messagingService;
- this.serializer = serializer;
- }
-
- @Override
- protected void decode(ChannelHandlerContext context, ByteBuf in,
- List<Object> messages) throws Exception {
-
- byte[] preamble = in.readBytes(MessageEncoder.PREAMBLE.length).array();
- checkState(Arrays.equals(MessageEncoder.PREAMBLE, preamble), "Message has wrong preamble");
-
- // read message Id.
- long id = in.readLong();
-
- // read message type; first read size and then bytes.
- String type = new String(in.readBytes(in.readInt()).array());
-
- // read sender host name; first read size and then bytes.
- String host = new String(in.readBytes(in.readInt()).array());
-
- // read sender port.
- int port = in.readInt();
-
- Endpoint sender = new Endpoint(host, port);
-
- // read message payload; first read size and then bytes.
- Object payload = serializer.decode(in.readBytes(in.readInt()).array());
-
- InternalMessage message = new InternalMessage.Builder(messagingService)
- .withId(id)
- .withSender(sender)
- .withType(type)
- .withPayload(payload)
- .build();
-
- messages.add(message);
- }
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/MessageEncoder.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/MessageEncoder.java
deleted file mode 100644
index 501b70c..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/MessageEncoder.java
+++ /dev/null
@@ -1,60 +0,0 @@
-package org.onlab.onos.store.messaging.impl;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.MessageToByteEncoder;
-
-/**
- * Encode InternalMessage out into a byte buffer.
- */
-public class MessageEncoder extends MessageToByteEncoder<InternalMessage> {
-
- // onosiscool in ascii
- public static final byte[] PREAMBLE = "onosiscool".getBytes();
-
- private final Serializer serializer;
-
- public MessageEncoder(Serializer serializer) {
- this.serializer = serializer;
- }
-
- @Override
- protected void encode(ChannelHandlerContext context, InternalMessage message,
- ByteBuf out) throws Exception {
-
- // write preamble
- out.writeBytes(PREAMBLE);
-
- // write id
- out.writeLong(message.id());
-
- // write type length
- out.writeInt(message.type().length());
-
- // write type
- out.writeBytes(message.type().getBytes());
-
- // write sender host name size
- out.writeInt(message.sender().host().length());
-
- // write sender host name.
- out.writeBytes(message.sender().host().getBytes());
-
- // write port
- out.writeInt(message.sender().port());
-
- try {
- serializer.encode(message.payload());
- } catch (Exception e) {
- e.printStackTrace();
- }
-
- byte[] payload = serializer.encode(message.payload());
-
- // write payload length.
- out.writeInt(payload.length);
-
- // write payload bytes
- out.writeBytes(payload);
- }
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/NettyMessagingService.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/NettyMessagingService.java
deleted file mode 100644
index 321e0ef..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/NettyMessagingService.java
+++ /dev/null
@@ -1,259 +0,0 @@
-package org.onlab.onos.store.messaging.impl;
-
-import java.io.IOException;
-import java.net.UnknownHostException;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.TimeUnit;
-
-import io.netty.bootstrap.Bootstrap;
-import io.netty.bootstrap.ServerBootstrap;
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.EventLoopGroup;
-import io.netty.channel.SimpleChannelInboundHandler;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.channel.socket.nio.NioSocketChannel;
-
-import org.apache.commons.lang.math.RandomUtils;
-import org.apache.commons.pool.KeyedObjectPool;
-import org.apache.commons.pool.KeyedPoolableObjectFactory;
-import org.apache.commons.pool.impl.GenericKeyedObjectPool;
-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.store.messaging.Endpoint;
-import org.onlab.onos.store.messaging.MessageHandler;
-import org.onlab.onos.store.messaging.MessagingService;
-import org.onlab.onos.store.messaging.Response;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-
-/**
- * A Netty based implementation of MessagingService.
- */
-@Component(immediate = true)
-@Service
-public class NettyMessagingService implements MessagingService {
-
- private final Logger log = LoggerFactory.getLogger(getClass());
-
- private KeyedObjectPool<Endpoint, Channel> channels =
- new GenericKeyedObjectPool<Endpoint, Channel>(new OnosCommunicationChannelFactory());
- private final int port;
- private final EventLoopGroup bossGroup = new NioEventLoopGroup();
- private final EventLoopGroup workerGroup = new NioEventLoopGroup();
- private final ConcurrentMap<String, MessageHandler> handlers = new ConcurrentHashMap<>();
- private Cache<Long, AsyncResponse<?>> responseFutures;
- private final Endpoint localEp;
-
- @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
- protected Serializer serializer;
-
- public NettyMessagingService() {
- // TODO: Default port should be configurable.
- this(8080);
- }
-
- // FIXME: Constructor should not throw exceptions.
- public NettyMessagingService(int port) {
- this.port = port;
- try {
- localEp = new Endpoint(java.net.InetAddress.getLocalHost().getHostName(), port);
- } catch (UnknownHostException e) {
- // bailing out.
- throw new RuntimeException(e);
- }
- }
-
- @Activate
- public void activate() throws Exception {
- responseFutures = CacheBuilder.newBuilder()
- .maximumSize(100000)
- .weakValues()
- // TODO: Once the entry expires, notify blocking threads (if any).
- .expireAfterWrite(10, TimeUnit.MINUTES)
- .build();
- startAcceptingConnections();
- }
-
- @Deactivate
- public void deactivate() throws Exception {
- channels.close();
- bossGroup.shutdownGracefully();
- workerGroup.shutdownGracefully();
- }
-
- @Override
- public void sendAsync(Endpoint ep, String type, Object payload) throws IOException {
- InternalMessage message = new InternalMessage.Builder(this)
- .withId(RandomUtils.nextLong())
- .withSender(localEp)
- .withType(type)
- .withPayload(payload)
- .build();
- sendAsync(ep, message);
- }
-
- protected void sendAsync(Endpoint ep, InternalMessage message) throws IOException {
- Channel channel = null;
- try {
- channel = channels.borrowObject(ep);
- channel.eventLoop().execute(new WriteTask(channel, message));
- } catch (Exception e) {
- throw new IOException(e);
- } finally {
- try {
- channels.returnObject(ep, channel);
- } catch (Exception e) {
- log.warn("Error returning object back to the pool", e);
- // ignored.
- }
- }
- }
-
- @Override
- public <T> Response<T> sendAndReceive(Endpoint ep, String type, Object payload)
- throws IOException {
- AsyncResponse<T> futureResponse = new AsyncResponse<T>();
- Long messageId = RandomUtils.nextLong();
- responseFutures.put(messageId, futureResponse);
- InternalMessage message = new InternalMessage.Builder(this)
- .withId(messageId)
- .withSender(localEp)
- .withType(type)
- .withPayload(payload)
- .build();
- sendAsync(ep, message);
- return futureResponse;
- }
-
- @Override
- public void registerHandler(String type, MessageHandler handler) {
- // TODO: Is this the right semantics for handler registration?
- handlers.putIfAbsent(type, handler);
- }
-
- public void unregisterHandler(String type) {
- handlers.remove(type);
- }
-
- private MessageHandler getMessageHandler(String type) {
- return handlers.get(type);
- }
-
- private void startAcceptingConnections() throws InterruptedException {
- ServerBootstrap b = new ServerBootstrap();
- b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
- b.group(bossGroup, workerGroup)
- .channel(NioServerSocketChannel.class)
- .childHandler(new OnosCommunicationChannelInitializer())
- .option(ChannelOption.SO_BACKLOG, 128)
- .childOption(ChannelOption.SO_KEEPALIVE, true);
-
- // Bind and start to accept incoming connections.
- b.bind(port).sync();
- }
-
- private class OnosCommunicationChannelFactory
- implements KeyedPoolableObjectFactory<Endpoint, Channel> {
-
- @Override
- public void activateObject(Endpoint endpoint, Channel channel)
- throws Exception {
- }
-
- @Override
- public void destroyObject(Endpoint ep, Channel channel) throws Exception {
- channel.close();
- }
-
- @Override
- public Channel makeObject(Endpoint ep) throws Exception {
- Bootstrap b = new Bootstrap();
- b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
- b.group(workerGroup);
- // TODO: Make this faster:
- // http://normanmaurer.me/presentations/2014-facebook-eng-netty/slides.html#37.0
- b.channel(NioSocketChannel.class);
- b.option(ChannelOption.SO_KEEPALIVE, true);
- b.handler(new OnosCommunicationChannelInitializer());
-
- // Start the client.
- ChannelFuture f = b.connect(ep.host(), ep.port()).sync();
- return f.channel();
- }
-
- @Override
- public void passivateObject(Endpoint ep, Channel channel)
- throws Exception {
- }
-
- @Override
- public boolean validateObject(Endpoint ep, Channel channel) {
- return channel.isOpen();
- }
- }
-
- private class OnosCommunicationChannelInitializer extends ChannelInitializer<SocketChannel> {
-
- @Override
- protected void initChannel(SocketChannel channel) throws Exception {
- channel.pipeline()
- .addLast(new MessageEncoder(serializer))
- .addLast(new MessageDecoder(NettyMessagingService.this, serializer))
- .addLast(new NettyMessagingService.InboundMessageDispatcher());
- }
- }
-
- private class WriteTask implements Runnable {
-
- private final Object message;
- private final Channel channel;
-
- public WriteTask(Channel channel, Object message) {
- this.message = message;
- this.channel = channel;
- }
-
- @Override
- public void run() {
- channel.writeAndFlush(message);
- }
- }
-
- private class InboundMessageDispatcher extends SimpleChannelInboundHandler<InternalMessage> {
-
- @Override
- protected void channelRead0(ChannelHandlerContext ctx, InternalMessage message) throws Exception {
- String type = message.type();
- if (type.equals(InternalMessage.REPLY_MESSAGE_TYPE)) {
- try {
- AsyncResponse<?> futureResponse =
- NettyMessagingService.this.responseFutures.getIfPresent(message.id());
- if (futureResponse != null) {
- futureResponse.setResponse(message.payload());
- }
- log.warn("Received a reply. But was unable to locate the request handle");
- } finally {
- NettyMessagingService.this.responseFutures.invalidate(message.id());
- }
- return;
- }
- MessageHandler handler = NettyMessagingService.this.getMessageHandler(type);
- handler.handle(message);
- }
- }
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/Serializer.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/Serializer.java
deleted file mode 100644
index d2da7cf..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/Serializer.java
+++ /dev/null
@@ -1,24 +0,0 @@
-package org.onlab.onos.store.messaging.impl;
-
-/**
- * Interface for encoding/decoding message payloads.
- */
-public interface Serializer {
-
- /**
- * Decodes the specified byte array to a POJO.
- *
- * @param data byte array.
- * @return POJO
- */
- Object decode(byte[] data);
-
- /**
- * Encodes the specified POJO into a byte array.
- *
- * @param data POJO to be encoded
- * @return byte array.
- */
- byte[] encode(Object message);
-
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/SimpleClient.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/SimpleClient.java
deleted file mode 100644
index 746ecb2..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/SimpleClient.java
+++ /dev/null
@@ -1,27 +0,0 @@
-package org.onlab.onos.store.messaging.impl;
-
-import java.util.concurrent.TimeUnit;
-
-import org.onlab.onos.store.messaging.Endpoint;
-import org.onlab.onos.store.messaging.Response;
-
-public final class SimpleClient {
- private SimpleClient() {}
-
- public static void main(String... args) throws Exception {
- NettyMessagingService messaging = new TestNettyMessagingService(9081);
- messaging.activate();
-
- messaging.sendAsync(new Endpoint("localhost", 8080), "simple", "Hello World");
- Response<String> response = messaging.sendAndReceive(new Endpoint("localhost", 8080), "echo", "Hello World");
- System.out.println("Got back:" + response.get(2, TimeUnit.SECONDS));
- }
-
- public static class TestNettyMessagingService extends NettyMessagingService {
- public TestNettyMessagingService(int port) throws Exception {
- super(port);
- Serializer serializer = new KryoSerializer();
- this.serializer = serializer;
- }
- }
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/SimpleServer.java b/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/SimpleServer.java
deleted file mode 100644
index 96094b7..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/messaging/impl/SimpleServer.java
+++ /dev/null
@@ -1,19 +0,0 @@
-package org.onlab.onos.store.messaging.impl;
-
-public final class SimpleServer {
- private SimpleServer() {}
-
- public static void main(String... args) throws Exception {
- NettyMessagingService server = new TestNettyMessagingService();
- server.activate();
- server.registerHandler("simple", new LoggingHandler());
- server.registerHandler("echo", new EchoHandler());
- }
-
- public static class TestNettyMessagingService extends NettyMessagingService {
- protected TestNettyMessagingService() {
- Serializer serializer = new KryoSerializer();
- this.serializer = serializer;
- }
- }
-}
diff --git a/core/store/dist/src/test/java/org/onlab/onos/store/cluster/impl/ClusterCommunicationManagerTest.java b/core/store/dist/src/test/java/org/onlab/onos/store/cluster/impl/ClusterCommunicationManagerTest.java
index 3d87fb1..44e5421 100644
--- a/core/store/dist/src/test/java/org/onlab/onos/store/cluster/impl/ClusterCommunicationManagerTest.java
+++ b/core/store/dist/src/test/java/org/onlab/onos/store/cluster/impl/ClusterCommunicationManagerTest.java
@@ -7,7 +7,7 @@
import org.onlab.onos.cluster.DefaultControllerNode;
import org.onlab.onos.cluster.NodeId;
import org.onlab.onos.store.cluster.messaging.impl.OnosClusterCommunicationManager;
-import org.onlab.onos.store.messaging.impl.NettyMessagingService;
+import org.onlab.netty.NettyMessagingService;
import org.onlab.packet.IpPrefix;
import java.util.concurrent.CountDownLatch;
diff --git a/utils/netty/pom.xml b/utils/netty/pom.xml
new file mode 100644
index 0000000..d335117
--- /dev/null
+++ b/utils/netty/pom.xml
@@ -0,0 +1,49 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.onlab.onos</groupId>
+ <artifactId>onlab-utils</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>onlab-netty</artifactId>
+ <packaging>bundle</packaging>
+
+ <description>Network I/O using Netty framework</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava-testlib</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.onlab.onos</groupId>
+ <artifactId>onlab-misc</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.onlab.onos</groupId>
+ <artifactId>onlab-junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>de.javakaffee</groupId>
+ <artifactId>kryo-serializers</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-all</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>commons-pool</groupId>
+ <artifactId>commons-pool</artifactId>
+ <version>1.6</version>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git a/utils/pom.xml b/utils/pom.xml
index 2beeba8..feb60e9 100644
--- a/utils/pom.xml
+++ b/utils/pom.xml
@@ -19,6 +19,7 @@
<modules>
<module>junit</module>
<module>misc</module>
+ <module>netty</module>
<module>nio</module>
<module>osgi</module>
<module>rest</module>