Merge branch 'master' of ssh://gerrit.onlab.us:29418/onos-next

@@ -33,6 +33,12 @@
             <artifactId>onlab-nio</artifactId>
             <artifactId>onlab-nio</artifactId>
             <version>${project.version}</version>
             <version>${project.version}</version>
         </dependency>
         </dependency>
+        
+        <dependency>
+            <groupId>org.onlab.onos</groupId>
+            <artifactId>onlab-netty</artifactId>
+            <version>${project.version}</version>
+        </dependency>
 
 
         <dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <groupId>com.fasterxml.jackson.core</groupId>
@@ -51,15 +57,6 @@
           <groupId>de.javakaffee</groupId>
           <groupId>de.javakaffee</groupId>
           <artifactId>kryo-serializers</artifactId>
           <artifactId>kryo-serializers</artifactId>
         </dependency>
         </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>
     </dependencies>
 
 
     <build>
     <build>

@@ -23,10 +23,10 @@ import org.onlab.onos.store.cluster.messaging.ClusterCommunicationService;
 import org.onlab.onos.store.cluster.messaging.ClusterMessage;
 import org.onlab.onos.store.cluster.messaging.ClusterMessage;
 import org.onlab.onos.store.cluster.messaging.ClusterMessageHandler;
 import org.onlab.onos.store.cluster.messaging.ClusterMessageHandler;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
 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.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.LoggerFactory;
 
 

-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;
-    }
-}

-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;
-}

-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;
-}

-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);
-}

-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();
-}

-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();
-        }
-    }
-}

-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());
-    }
-}

-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;
-        }
-    }
-}

-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);
-    }
-}

-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

-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);
-    }
-}

-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);
-    }
-}

-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);
-        }
-    }
-}

-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);
-
-}

-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;
-        }
-    }
-}

-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;
-        }
-    }
-}

@@ -7,7 +7,7 @@ import org.junit.Test;
 import org.onlab.onos.cluster.DefaultControllerNode;
 import org.onlab.onos.cluster.DefaultControllerNode;
 import org.onlab.onos.cluster.NodeId;
 import org.onlab.onos.cluster.NodeId;
 import org.onlab.onos.store.cluster.messaging.impl.OnosClusterCommunicationManager;
 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 org.onlab.packet.IpPrefix;
 
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CountDownLatch;

+<?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>

@@ -19,6 +19,7 @@
     <modules>
     <modules>
         <module>junit</module>
         <module>junit</module>
         <module>misc</module>
         <module>misc</module>
+        <module>netty</module>
         <module>nio</module>
         <module>nio</module>
         <module>osgi</module>
         <module>osgi</module>
         <module>rest</module>
         <module>rest</module>