[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/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/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/NettyHttpClientEventListener.java

@@ -0,0 +1,99 @@
+/*
+ * 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.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpContent;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.util.CharsetUtil;
+import org.apache.reef.wake.EStage;
+import org.apache.reef.wake.remote.impl.TransportEvent;
+
+import java.net.SocketAddress;
+import java.util.concurrent.ConcurrentMap;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * A Netty event listener for server.
+ */
+final class NettyHttpClientEventListener extends AbstractNettyEventListener {
+
+  private static final Logger LOG = Logger.getLogger(NettyHttpClientEventListener.class.getName());
+  private StringBuilder buf = new StringBuilder();
+
+  NettyHttpClientEventListener(
+      final ConcurrentMap<SocketAddress, LinkReference> addrToLinkRefMap,
+      final EStage<TransportEvent> stage) {
+    super(addrToLinkRefMap, stage);
+  }
+
+  @Override
+  public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
+    if (msg instanceof HttpResponse) {
+      System.out.println("CLIENT : IT IS HTTP RESPONSE");
+    } else if (msg instanceof HttpContent) {
+      System.out.println("CLIENT : It is HttpContent");
+      HttpContent httpContent = (HttpContent) msg;
+      ByteBuf content = httpContent.content();
+      if (content.isReadable()) {
+        buf.append("CONTENT: ");
+        buf.append(content.toString(CharsetUtil.UTF_8));
+        buf.append("\r\n");
+      }
+
+      if (msg instanceof LastHttpContent) {
+        final Channel channel = ctx.channel();
+        byte[] message = new byte[content.readableBytes()];
+        content.readBytes(message);
+        if (LOG.isLoggable(Level.FINEST)) {
+          LOG.log(Level.FINEST, "MessageEvent: local: {0} remote: {1} :: {2}", new Object[]{
+                  channel.localAddress(), channel.remoteAddress(), buf});
+        }
+
+        if (message.length > 0) {
+          // send to the dispatch stage
+          this.stage.onNext(this.getTransportEvent(message, channel));
+        }
+      }
+    } else {
+      LOG.log(Level.SEVERE, "Unknown type of message received: {0}", msg);
+    }
+
+  }
+
+  @Override
+  public void channelActive(final ChannelHandlerContext ctx) {
+    // noop
+    LOG.log(Level.FINEST, "{0}", ctx);
+  }
+
+  @Override
+  protected TransportEvent getTransportEvent(final byte[] message, final Channel channel) {
+    return new TransportEvent(message, channel.localAddress(), channel.remoteAddress());
+  }
+
+  @Override
+  protected void exceptionCleanup(final ChannelHandlerContext ctx, final Throwable cause) {
+    this.closeChannel(ctx.channel());
+  }
+}