Merge remote-tracking branch 'origin/master'

@@ -8,12 +8,16 @@ import java.io.IOException;
  */
  */
 public final class InternalMessage implements Message {
 public final class InternalMessage implements Message {
 
 
+    public static final String REPLY_MESSAGE_TYPE = "NETTY_MESSAGIG_REQUEST_REPLY";
+
     private long id;
     private long id;
     private Endpoint sender;
     private Endpoint sender;
     private String type;
     private String type;
     private Object payload;
     private Object payload;
+
     private transient NettyMessagingService messagingService;
     private transient NettyMessagingService messagingService;
-    public static final String REPLY_MESSAGE_TYPE = "NETTY_MESSAGIG_REQUEST_REPLY";
+    // TODO: add transient payload serializer or change payload type to
+    //       byte[], ByteBuffer, etc.
 
 
     // Must be created using the Builder.
     // Must be created using the Builder.
     private InternalMessage() {}
     private InternalMessage() {}

@@ -241,8 +241,9 @@ public class NettyMessagingService implements MessagingService {
                         NettyMessagingService.this.responseFutures.getIfPresent(message.id());
                         NettyMessagingService.this.responseFutures.getIfPresent(message.id());
                     if (futureResponse != null) {
                     if (futureResponse != null) {
                         futureResponse.setResponse(message.payload());
                         futureResponse.setResponse(message.payload());
-                    }
+                    } else {
                         log.warn("Received a reply. But was unable to locate the request handle");
                         log.warn("Received a reply. But was unable to locate the request handle");
+                    }
                 } finally {
                 } finally {
                     NettyMessagingService.this.responseFutures.invalidate(message.id());
                     NettyMessagingService.this.responseFutures.invalidate(message.id());
                 }
                 }

+package org.onlab.netty;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Simple ping-pong test that exercises NettyMessagingService.
+ */
+public class PingPongTest {
+
+    @Test
+    public void testPingPong() throws Exception {
+        NettyMessagingService pinger = new NettyMessagingService(8085);
+        NettyMessagingService ponger = new NettyMessagingService(9086);
+        try {
+            pinger.activate();
+            ponger.activate();
+            pinger.setPayloadSerializer(new KryoSerializer());
+            ponger.setPayloadSerializer(new KryoSerializer());
+            ponger.registerHandler("echo", new EchoHandler());
+            Response<String> response = pinger.sendAndReceive(new Endpoint("localhost", 9086), "echo", "hello");
+            Assert.assertEquals("hello", response.get(10000, TimeUnit.MILLISECONDS));
+        } finally {
+            pinger.deactivate();
+            ponger.deactivate();
+        }
+    }
+}