Added EchoServer

Author: rdegnan

build.gradle

@@ -1,6 +1,6 @@
 buildscript {
     repositories { jcenter() }
-    dependencies { classpath 'io.reactivesocket:gradle-nebula-plugin-reactivesocket:1.0.5' }
+    dependencies { classpath 'io.reactivesocket:gradle-nebula-plugin-reactivesocket:1.0.6' }
 }
 
 apply plugin: 'reactivesocket-project'

reactivesocket-netty/build.gradle

@@ -1,4 +1,9 @@
 dependencies {
-    compile 'io.netty:netty-handler:4.1.0.CR6'
-    compile 'io.netty:netty-codec-http:4.1.0.CR6'
-}
+    compile 'io.netty:netty-handler:4.1.0.CR7'
+    compile 'io.netty:netty-codec-http:4.1.0.CR7'
+}
+
+task echoServer(type: JavaExec) {
+    classpath = sourceSets.examples.runtimeClasspath
+    main = 'io.reactivesocket.netty.EchoServer'
+}

reactivesocket-netty/src/examples/java/io/reactivesocket/netty/EchoServer.java

@@ -0,0 +1,38 @@
+package io.reactivesocket.netty;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.logging.LogLevel;
+import io.netty.handler.logging.LoggingHandler;
+
+public class EchoServer {
+    public static void main(String... args) throws Exception {
+        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
+        EventLoopGroup workerGroup = new NioEventLoopGroup();
+
+        try {
+            ServerBootstrap b = new ServerBootstrap();
+            b.group(bossGroup, workerGroup)
+                    .channel(NioServerSocketChannel.class)
+                    .handler(new LoggingHandler(LogLevel.INFO))
+                    .childHandler(new ChannelInitializer<Channel>() {
+                        @Override
+                        protected void initChannel(Channel ch) throws Exception {
+                            ChannelPipeline pipeline = ch.pipeline();
+                            pipeline.addLast(new EchoServerHandler());
+                        }
+                    });
+
+            Channel localhost = b.bind("localhost", 8025).sync().channel();
+            localhost.closeFuture().sync();
+        } finally {
+            bossGroup.shutdownGracefully();
+            workerGroup.shutdownGracefully();
+        }
+    }
+}

reactivesocket-netty/src/examples/java/io/reactivesocket/netty/EchoServerHandler.java

@@ -0,0 +1,67 @@
+package io.reactivesocket.netty;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.reactivesocket.RequestHandler;
+import io.reactivesocket.netty.tcp.server.ReactiveSocketServerHandler;
+
+import java.util.List;
+
+public class EchoServerHandler extends ByteToMessageDecoder {
+    private static SimpleChannelInboundHandler<FullHttpRequest> httpHandler = new HttpServerHandler();
+
+    private static ReactiveSocketServerHandler reactiveSocketHandler = ReactiveSocketServerHandler.create((setupPayload, rs) ->
+        new RequestHandler.Builder().withRequestResponse(payload -> s -> {
+            s.onNext(payload);
+            s.onComplete();
+        }).build());
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+        // Will use the first five bytes to detect a protocol.
+        if (in.readableBytes() < 5) {
+            return;
+        }
+
+        final int magic1 = in.getUnsignedByte(in.readerIndex());
+        final int magic2 = in.getUnsignedByte(in.readerIndex() + 1);
+        if (isHttp(magic1, magic2)) {
+            switchToHttp(ctx);
+        } else {
+            switchToReactiveSocket(ctx);
+        }
+    }
+
+    private static boolean isHttp(int magic1, int magic2) {
+        return
+            magic1 == 'G' && magic2 == 'E' || // GET
+            magic1 == 'P' && magic2 == 'O' || // POST
+            magic1 == 'P' && magic2 == 'U' || // PUT
+            magic1 == 'H' && magic2 == 'E' || // HEAD
+            magic1 == 'O' && magic2 == 'P' || // OPTIONS
+            magic1 == 'P' && magic2 == 'A' || // PATCH
+            magic1 == 'D' && magic2 == 'E' || // DELETE
+            magic1 == 'T' && magic2 == 'R' || // TRACE
+            magic1 == 'C' && magic2 == 'O';   // CONNECT
+    }
+
+    private void switchToHttp(ChannelHandlerContext ctx) {
+        ChannelPipeline p = ctx.pipeline();
+        p.addLast(new HttpServerCodec());
+        p.addLast(new HttpObjectAggregator(64 * 1024));
+        p.addLast(httpHandler);
+        p.remove(this);
+    }
+
+    private void switchToReactiveSocket(ChannelHandlerContext ctx) {
+        ChannelPipeline p = ctx.pipeline();
+        p.addLast(reactiveSocketHandler);
+        p.remove(this);
+    }
+}

reactivesocket-netty/src/examples/java/io/reactivesocket/netty/HttpServerHandler.java

@@ -0,0 +1,24 @@
+package io.reactivesocket.netty;
+
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+
+@ChannelHandler.Sharable
+public class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+        ctx.flush();
+        ctx.close();
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
+        ctx.write(new DefaultFullHttpResponse(HTTP_1_1, OK, msg.content().copy()));
+    }
+}