[REEF-362] Implement a Wake Transport using HTTP

lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/transport/netty/NettyDefaultLinkFactory.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.wake.remote.transport.netty;
+
+import io.netty.channel.Channel;
+import org.apache.reef.wake.remote.Encoder;
+import org.apache.reef.wake.remote.transport.Link;
+import org.apache.reef.wake.remote.transport.LinkListener;
+
+import java.net.URI;
+
+/**
+ * Factory that creates a NettyLink.
+ */
+public final class NettyDefaultLinkFactory<T> implements NettyLinkFactory {
+
+  private final URI uri;
+
+  NettyDefaultLinkFactory() {
+    this(null);
+  }
+
+  NettyDefaultLinkFactory(final URI uri){
+    this.uri = uri;
+  }
+
+  @Override
+  public Link newInstance(final Channel channel, final Encoder encoder) {
+    return uri == null ?
+        new NettyLink<T>(channel, encoder) :
+        new NettyHttpLink<T>(channel, encoder, uri);
+  }
+
+  @Override
+  public Link newInstance(final Channel channel,
+                          final Encoder encoder,
+                          final LinkListener listener) {
+    return uri == null ?
+        new NettyLink<T>(channel, encoder, listener) :
+        new NettyHttpLink<T>(channel, encoder, listener, uri);
+  }
+}

lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/transport/netty/NettyHttpLink.java

@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.wake.remote.transport.netty;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.handler.codec.http.*;
+import org.apache.reef.wake.remote.Encoder;
+import org.apache.reef.wake.remote.transport.Link;
+import org.apache.reef.wake.remote.transport.LinkListener;
+
+import java.net.SocketAddress;
+import java.net.URI;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * Link implementation with Netty.
+ *
+ * If you set a {@code LinkListener<T>}, it keeps message until writeAndFlush operation completes
+ * and notifies whether the sent message transferred successfully through the listener.
+ */
+public class NettyHttpLink<T> implements Link<T> {
+
+  public static final int INT_SIZE = Integer.SIZE / Byte.SIZE;
+
+  private static final Logger LOG = Logger.getLogger(NettyHttpLink.class.getName());
+
+  private final Channel channel;
+  private final Encoder<? super T> encoder;
+  private final LinkListener<? super T> listener;
+  private final URI uri;
+
+  /**
+   * Constructs a link.
+   *
+   * @param channel the channel
+   * @param encoder the encoder
+   * @param uri     the URI
+   */
+  public NettyHttpLink(final Channel channel,
+                       final Encoder<? super T> encoder,
+                       final URI uri) {
+    this(channel, encoder, null, uri);
+  }
+
+  /**
+   * Constructs a link.
+   *
+   * @param channel  the channel
+   * @param encoder  the encoder
+   * @param listener the link listener
+   * @param uri      the URI
+   */
+  public NettyHttpLink(
+          final Channel channel,
+          final Encoder<? super T> encoder,
+          final LinkListener<? super T> listener,
+          final URI uri) {
+    this.channel = channel;
+    this.encoder = encoder;
+    this.listener = listener;
+    this.uri = uri;
+  }
+  /**
+   * Writes the message to this link.
+   *
+   * @param message the message
+   */
+  @Override
+  public void write(final T message) {
+    LOG.log(Level.FINEST, "write {0} :: {1}", new Object[] {channel, message});
+    try {
+      final FullHttpRequest request =
+          new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, uri.getRawPath());
+      final ByteBuf buf = Unpooled.copiedBuffer(encoder.encode(message));
+      request.headers()
+          .set(HttpHeaders.Names.HOST, uri.getHost())
+          .set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE)
+          .set(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP)
+          .set(HttpHeaders.Names.CONTENT_TYPE, "application/wake-transport")
+          .set(HttpHeaders.Names.CONTENT_LENGTH, buf.readableBytes());
+      request.content().clear().writeBytes(buf);
+      final ChannelFuture future = channel.writeAndFlush(request);
+      future.sync();
+      if (listener != null) {
+        future.addListener(new NettyChannelFutureListener<>(message, listener));
+      }
+    } catch (final InterruptedException ex) {
+      LOG.log(Level.SEVERE, "Cannot send request to " + uri.getHost(), ex);
+    }
+  }
+
+  /**
+   * Gets a local address of the link.
+   *
+   * @return a local socket address
+   */
+  @Override
+  public SocketAddress getLocalAddress() {
+    return channel.localAddress();
+  }
+
+  /**
+   * Gets a remote address of the link.
+   *
+   * @return a remote socket address
+   */
+  @Override
+  public SocketAddress getRemoteAddress() {
+    return channel.remoteAddress();
+  }
+
+  @Override
+  public String toString() {
+    return "NettyLink: " + channel; // Channel has good .toString() implementation
+  }
+}

lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/transport/netty/NettyHttpServerEventListener.java

@@ -42,13 +42,15 @@ final class NettyHttpServerEventListener extends AbstractNettyEventListener {
   private HttpRequest httpRequest;
   private final StringBuilder buf = new StringBuilder();
   private final URI uri;
+  private final NettyLinkFactory linkFactory;
 
   NettyHttpServerEventListener(
       final ConcurrentMap<SocketAddress, LinkReference> addrToLinkRefMap,
       final EStage<TransportEvent> stage,
       final URI uri) {
     super(addrToLinkRefMap, stage);
     this.uri = uri;
+    this.linkFactory = new NettyDefaultLinkFactory<>(uri);
   }
 
 
@@ -61,8 +63,8 @@ public void channelActive(final ChannelHandlerContext ctx) {
     }
 
     this.addrToLinkRefMap.putIfAbsent(
-        channel.remoteAddress(), new LinkReference(new NettyLink<>(
-            channel, new ByteCodec(), new LoggingLinkListener<byte[]>(), uri)));
+        channel.remoteAddress(), new LinkReference(linkFactory.newInstance(
+            channel, new ByteCodec(), new LoggingLinkListener<byte[]>())));
 
     LOG.log(Level.FINER, "Add connected channel ref: {0}", this.addrToLinkRefMap.get(channel.remoteAddress()));
 
@@ -144,7 +146,7 @@ private static void appendDecoderResult(final StringBuilder buf, final HttpObjec
 
   @Override
   protected TransportEvent getTransportEvent(final byte[] message, final Channel channel) {
-    return new TransportEvent(message, new NettyLink<>(channel, new ByteEncoder()));
+    return new TransportEvent(message, linkFactory.newInstance(channel, new ByteEncoder()));
   }
 
   @Override

lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/transport/netty/NettyLink.java

@@ -18,17 +18,14 @@
  */
 package org.apache.reef.wake.remote.transport.netty;
 
-import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
-import io.netty.handler.codec.http.*;
 import org.apache.reef.wake.remote.Encoder;
 import org.apache.reef.wake.remote.transport.Link;
 import org.apache.reef.wake.remote.transport.LinkListener;
 
 import java.net.SocketAddress;
-import java.net.URI;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -47,7 +44,6 @@ public class NettyLink<T> implements Link<T> {
   private final Channel channel;
   private final Encoder<? super T> encoder;
   private final LinkListener<? super T> listener;
-  private final URI uri;
 
   /**
    * Constructs a link.
@@ -67,27 +63,11 @@ public NettyLink(final Channel channel, final Encoder<? super T> encoder) {
    * @param listener the link listener
    */
   public NettyLink(final Channel channel, final Encoder<? super T> encoder, final LinkListener<? super T> listener) {
-    this(channel, encoder, listener, null);
-  }
-
-  /**
-   * Constructs a link.
-   *
-   * @param channel  the channel
-   * @param encoder  the encoder
-   * @param listener the link listener
-   * @param uri the URI
-   */
-  public NettyLink(
-          final Channel channel,
-          final Encoder<? super T> encoder,
-          final LinkListener<? super T> listener,
-          final URI uri) {
     this.channel = channel;
     this.encoder = encoder;
     this.listener = listener;
-    this.uri = uri;
   }
+
   /**
    * Writes the message to this link.
    *
@@ -96,32 +76,9 @@ public NettyLink(
   @Override
   public void write(final T message) {
     LOG.log(Level.FINEST, "write {0} :: {1}", new Object[] {channel, message});
-
-    if (this.uri != null) {
-      try {
-        final FullHttpRequest request =
-            new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, uri.getRawPath());
-        final ByteBuf buf = Unpooled.copiedBuffer(encoder.encode(message));
-        request.headers()
-            .set(HttpHeaders.Names.HOST, uri.getHost())
-            .set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE)
-            .set(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP)
-            .set(HttpHeaders.Names.CONTENT_TYPE, "application/wake-transport")
-            .set(HttpHeaders.Names.CONTENT_LENGTH, buf.readableBytes());
-        request.content().clear().writeBytes(buf);
-        final ChannelFuture future = channel.writeAndFlush(request);
-        future.sync();
-        if (listener != null) {
-          future.addListener(new NettyChannelFutureListener<>(message, listener));
-        }
-      } catch (final InterruptedException ex) {
-        LOG.log(Level.SEVERE, "Cannot send request to " + uri.getHost(), ex);
-      }
-    } else {
-      final ChannelFuture future = channel.writeAndFlush(Unpooled.wrappedBuffer(encoder.encode(message)));
-      if (listener != null) {
-        future.addListener(new NettyChannelFutureListener<>(message, listener));
-      }
+    final ChannelFuture future = channel.writeAndFlush(Unpooled.wrappedBuffer(encoder.encode(message)));
+    if (listener != null) {
+      future.addListener(new NettyChannelFutureListener<>(message, listener));
     }
   }
 

lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/transport/netty/NettyLinkFactory.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.wake.remote.transport.netty;
+
+import io.netty.channel.Channel;
+import org.apache.reef.tang.annotations.DefaultImplementation;
+import org.apache.reef.wake.remote.Encoder;
+import org.apache.reef.wake.remote.transport.Link;
+import org.apache.reef.wake.remote.transport.LinkListener;
+
+/**
+ * Factory that creates a NettyLink.
+ */
+@DefaultImplementation(NettyDefaultLinkFactory.class)
+public interface NettyLinkFactory<T> {
+
+  /**
+   * Creates a NettyLink.
+   * @param channel the channel
+   * @param encoder the encoder
+   */
+  Link<T> newInstance(final Channel channel, final Encoder<? super T> encoder);
+
+  /**
+   * Creates a NettyLink.
+   * @param channel  the channel
+   * @param encoder  the encoder
+   * @param listener the listener
+   */
+  Link<T> newInstance(final Channel channel,
+                           final Encoder<? super T> encoder,
+                           final LinkListener<? super T> listener);
+}

lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/transport/netty/NettyMessagingTransport.java

@@ -362,7 +362,9 @@ public <T> Link<T> open(final SocketAddress remoteAddr, final Encoder<? super T>
         connectFuture = this.clientBootstrap.connect(remoteAddr);
         connectFuture.syncUninterruptibly();
 
-        link = new NettyLink<>(connectFuture.channel(), encoder, listener, this.uri);
+        final NettyLinkFactory linkFactory = new NettyDefaultLinkFactory<>(uri);
+
+        link = linkFactory.newInstance(connectFuture.channel(), encoder, listener);
         linkRef.setLink(link);
 
         synchronized (flag) {