[REEF-362] Implement a Wake Transport using HTTP

lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/RemoteConfiguration.java

@@ -47,6 +47,13 @@ public final class RemoteConfiguration {
   public static final long REMOTE_CONNECTION_RETRY_TIMEOUT =
       WakeParameters.REMOTE_EXECUTOR_SHUTDOWN_TIMEOUT / (REMOTE_CONNECTION_NUMBER_OF_RETRIES + 1);
 
+
+  /**
+   *  Unique protocol numbers for choosing protocols.
+   */
+  public static final int PROTOCOL_NETTY = 100;
+  public static final int PROTOCOL_HTTP = 101;
+
   private RemoteConfiguration() {
     // empty
   }
@@ -132,4 +139,12 @@ public static final class RemoteClientStage implements Name<EStage<TransportEven
   public static final class RemoteServerStage implements Name<EStage<TransportEvent>> {
     // Intentionally empty
   }
+
+  /**
+   * Option for use http.
+   */
+  @NamedParameter(doc = "Option for use http.", default_value = "" + PROTOCOL_NETTY)
+  public static final class Protocol implements Name<Integer> {
+    // Intentionally empty
+  }
 }

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

@@ -62,6 +62,25 @@ Transport newInstance(final String hostAddress, int port,
                         final int numberOfTries,
                         final int retryTimeout);
 
+  /**
+   * Creates a transport.
+   *
+   * @param hostAddress     a host address
+   * @param port            a listening port
+   * @param clientStage     a transport client-side stage
+   * @param serverStage     a transport server-side stage
+   * @param numberOfTries   the number of retries for connection
+   * @param retryTimeout    retry timeout
+   * @param protocol        protocol to use
+   * @return transport
+   */
+  Transport newInstance(final String hostAddress, int port,
+                        final EStage<TransportEvent> clientStage,
+                        final EStage<TransportEvent> serverStage,
+                        final int numberOfTries,
+                        final int retryTimeout,
+                        final int protocol);
+
   /**
    * Creates a transport.
    *
@@ -82,5 +101,27 @@ Transport newInstance(final String hostAddress,
                         final int retryTimeout,
                         final TcpPortProvider tcpPortProvider);
 
+  /**
+   * Creates a transport.
+   *
+   * @param hostAddress     a host address
+   * @param port            a listening port
+   * @param clientStage     a transport client-side stage
+   * @param serverStage     a transport server-side stage
+   * @param numberOfTries   the number of retries for connection
+   * @param retryTimeout    retry timeout
+   * @param tcpPortProvider tcpPortProvider
+   * @param protocol        protocol to use
+   * @return transport
+   */
+  Transport newInstance(final String hostAddress,
+                        int port,
+                        final EStage<TransportEvent> clientStage,
+                        final EStage<TransportEvent> serverStage,
+                        final int numberOfTries,
+                        final int retryTimeout,
+                        final TcpPortProvider tcpPortProvider,
+                        final int protocol);
+
 
 }

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

@@ -33,17 +33,17 @@
  * Generic functionality for the Netty event listener.
  * This is a base class for client and server versions.
  */
-abstract class AbstractNettyEventListener implements NettyEventListener {
+public abstract class AbstractNettyEventListener implements NettyEventListener {
 
   protected static final Logger LOG = Logger.getLogger(AbstractNettyEventListener.class.getName());
 
   protected final ConcurrentMap<SocketAddress, LinkReference> addrToLinkRefMap;
   protected final EStage<TransportEvent> stage;
   protected EventHandler<Exception> exceptionHandler;
 
-  AbstractNettyEventListener(
-      final ConcurrentMap<SocketAddress, LinkReference> addrToLinkRefMap,
-      final EStage<TransportEvent> stage) {
+  public AbstractNettyEventListener(
+          final ConcurrentMap<SocketAddress, LinkReference> addrToLinkRefMap,
+          final EStage<TransportEvent> stage) {
     this.addrToLinkRefMap = addrToLinkRefMap;
     this.stage = stage;
   }

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

@@ -26,27 +26,27 @@
  * A reference for a link.
  * When channel became active, LinkReference is created and mapped with remote address.
  */
-final class LinkReference {
+public final class LinkReference {
 
   private final AtomicInteger connectInProgress = new AtomicInteger(0);
   private Link<?> link;
 
-  LinkReference() {
+  public LinkReference() {
   }
 
-  LinkReference(final Link<?> link) {
+  public LinkReference(final Link<?> link) {
     this.link = link;
   }
 
-  synchronized Link<?> getLink() {
+  public synchronized Link<?> getLink() {
     return this.link;
   }
 
-  synchronized void setLink(final Link<?> link) {
+  public synchronized void setLink(final Link<?> link) {
     this.link = link;
   }
 
-  AtomicInteger getConnectInProgress() {
+  public AtomicInteger getConnectInProgress() {
     return this.connectInProgress;
   }
 }

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

@@ -101,6 +101,33 @@ public Transport newInstance(final String hostAddress,
     }
   }
 
+  /**
+   * Creates a transport.
+   *
+   * @param hostAddress   a host address
+   * @param port          a listening port
+   * @param clientStage   a client stage
+   * @param serverStage   a server stage
+   * @param numberOfTries a number of tries
+   * @param retryTimeout  a timeout for retry
+   */
+  @Override
+  public Transport newInstance(final String hostAddress,
+                               final int port,
+                               final EStage<TransportEvent> clientStage,
+                               final EStage<TransportEvent> serverStage,
+                               final int numberOfTries,
+                               final int retryTimeout,
+                               final int protocol) {
+    try {
+      TcpPortProvider tcpPortProvider = Tang.Factory.getTang().newInjector().getInstance(TcpPortProvider.class);
+      return newInstance(hostAddress, port, clientStage,
+          serverStage, numberOfTries, retryTimeout, tcpPortProvider);
+    } catch (final InjectionException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   /**
    * Creates a transport.
    *
@@ -121,6 +148,32 @@ public Transport newInstance(final String hostAddress,
                                final int retryTimeout,
                                final TcpPortProvider tcpPortProvider) {
 
+    return newInstance(hostAddress, port, clientStage,
+        serverStage, numberOfTries, retryTimeout, tcpPortProvider, RemoteConfiguration.PROTOCOL_NETTY);
+  }
+
+  /**
+   * Creates a transport.
+   *
+   * @param hostAddress     a host address
+   * @param port            a listening port
+   * @param clientStage     a client stage
+   * @param serverStage     a server stage
+   * @param numberOfTries   a number of tries
+   * @param retryTimeout    a timeout for retry
+   * @param tcpPortProvider a provider for TCP port
+   * @param protocol        a protocol to use
+   */
+  @Override
+  public Transport newInstance(final String hostAddress,
+                               final int port,
+                               final EStage<TransportEvent> clientStage,
+                               final EStage<TransportEvent> serverStage,
+                               final int numberOfTries,
+                               final int retryTimeout,
+                               final TcpPortProvider tcpPortProvider,
+                               final int protocol) {
+
     final Injector injector = Tang.Factory.getTang().newInjector();
     injector.bindVolatileParameter(RemoteConfiguration.HostAddress.class, hostAddress);
     injector.bindVolatileParameter(RemoteConfiguration.Port.class, port);
@@ -129,6 +182,7 @@ public Transport newInstance(final String hostAddress,
     injector.bindVolatileParameter(RemoteConfiguration.NumberOfTries.class, numberOfTries);
     injector.bindVolatileParameter(RemoteConfiguration.RetryTimeout.class, retryTimeout);
     injector.bindVolatileInstance(TcpPortProvider.class, tcpPortProvider);
+    injector.bindVolatileParameter(RemoteConfiguration.Protocol.class, protocol);
     try {
       return injector.getInstance(NettyMessagingTransport.class);
     } catch (final InjectionException e) {

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

@@ -0,0 +1,43 @@
+/*
+ * 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.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import org.apache.reef.wake.remote.transport.LinkListener;
+
+class NettyChannelFutureListener<T> implements ChannelFutureListener {
+
+  private final T message;
+  private LinkListener<T> listener;
+
+  NettyChannelFutureListener(final T message, final LinkListener<T> listener) {
+    this.message = message;
+    this.listener = listener;
+  }
+
+  @Override
+  public void operationComplete(final ChannelFuture channelFuture) throws Exception {
+    if (channelFuture.isSuccess()) {
+      listener.onSuccess(message);
+    } else {
+      listener.onException(channelFuture.cause(), channelFuture.channel().remoteAddress(), message);
+    }
+  }
+}

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

@@ -24,7 +24,7 @@
 /**
  * Factory that creates a Netty channel handler.
  */
-interface NettyChannelHandlerFactory {
+public interface NettyChannelHandlerFactory {
 
   /**
    * Creates a channel inbound handler.

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

@@ -24,29 +24,68 @@
 import io.netty.handler.codec.LengthFieldPrepender;
 import io.netty.handler.codec.bytes.ByteArrayDecoder;
 import io.netty.handler.codec.bytes.ByteArrayEncoder;
+import io.netty.handler.codec.http.*;
+import io.netty.handler.ssl.SslContext;
 
 /**
  * Netty channel initializer for Transport.
  */
 class NettyChannelInitializer extends ChannelInitializer<SocketChannel> {
+
+  private static final int HANDLER_NETTY = 100;
+  private static final int HANDLER_HTTP_SERVER = 101;
+  private static final int HANDLER_HTTP_CLIENT = 102;
+
   /**
    * the buffer size of the frame decoder.
    */
   public static final int MAXFRAMELENGTH = 10 * 1024 * 1024;
   private final NettyChannelHandlerFactory handlerFactory;
+  private final SslContext sslContext;
+  private final int type;
 
-  NettyChannelInitializer(final NettyChannelHandlerFactory handlerFactory) {
+  NettyChannelInitializer(
+      final NettyChannelHandlerFactory handlerFactory,
+      final SslContext sslContext,
+      final int type) {
     this.handlerFactory = handlerFactory;
+    this.sslContext = sslContext;
+    this.type = type;
   }
 
   @Override
   protected void initChannel(final SocketChannel ch) throws Exception {
-    ch.pipeline()
-        .addLast("frameDecoder", new LengthFieldBasedFrameDecoder(MAXFRAMELENGTH, 0, 4, 0, 4))
-        .addLast("bytesDecoder", new ByteArrayDecoder())
-        .addLast("frameEncoder", new LengthFieldPrepender(4))
-        .addLast("bytesEncoder", new ByteArrayEncoder())
-        .addLast("chunker", new ChunkedReadWriteHandler())
-        .addLast("handler", handlerFactory.createChannelInboundHandler());
+    switch (this.type) {
+    case HANDLER_NETTY:
+      ch.pipeline()
+          .addLast("frameDecoder", new LengthFieldBasedFrameDecoder(MAXFRAMELENGTH, 0, 4, 0, 4))
+          .addLast("bytesDecoder", new ByteArrayDecoder())
+          .addLast("frameEncoder", new LengthFieldPrepender(4))
+          .addLast("bytesEncoder", new ByteArrayEncoder())
+          .addLast("chunker", new ChunkedReadWriteHandler())
+          .addLast("handler", handlerFactory.createChannelInboundHandler());
+      break;
+    case HANDLER_HTTP_SERVER:
+      if (sslContext != null) {
+        ch.pipeline().addLast(sslContext.newHandler(ch.alloc()));
+      }
+      ch.pipeline()
+          .addLast("codec", new HttpServerCodec())
+          .addLast("requestDecoder", new HttpRequestDecoder())
+          .addLast("responseEncoder", new HttpResponseEncoder())
+          .addLast("handler", handlerFactory.createChannelInboundHandler());
+      break;
+    case HANDLER_HTTP_CLIENT:
+      if (sslContext != null) {
+        ch.pipeline().addLast(sslContext.newHandler(ch.alloc()));
+      }
+      ch.pipeline()
+          .addLast("codec", new HttpClientCodec())
+          .addLast("decompressor", new HttpContentDecompressor())
+          .addLast("handler", handlerFactory.createChannelInboundHandler());
+      break;
+    default:
+      throw new IllegalArgumentException("Invalid type of channel");
+    }
   }
 }

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

@@ -23,7 +23,7 @@
 /**
  * Netty event listener.
  */
-interface NettyEventListener {
+public interface NettyEventListener {
 
   /**
    * Handles the message.