[netty#844] [netty#867] Add UDT transport

example/.gitignore

@@ -0,0 +1,7 @@
+
+#
+# native udt library extract location
+#
+
+/lib
+

example/pom.xml

@@ -17,6 +17,7 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
 
     <modelVersion>4.0.0</modelVersion>
+
     <parent>
         <groupId>io.netty</groupId>
         <artifactId>netty-parent</artifactId>
@@ -54,7 +55,6 @@
             <artifactId>netty-codec-socks</artifactId>
             <version>${project.version}</version>
         </dependency>
-
         <dependency>
             <groupId>com.google.protobuf</groupId>
             <artifactId>protobuf-java</artifactId>
@@ -64,6 +64,17 @@
             <artifactId>netty-metrics-yammer</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <!-- see https://github.com/netty/netty/issues/874 -->
+        <dependency>
+            <groupId>com.yammer.metrics</groupId>
+            <artifactId>metrics-core</artifactId>
+            <version>2.2.0</version>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>netty-transport-udt</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
+
 </project>
-

example/src/main/java/io/netty/example/udt/echo/bytes/ByteEchoClient.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project 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 io.netty.example.udt.echo.bytes;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.nio.NioEventLoopGroup;
+import io.netty.example.udt.util.UtilConsoleReporter;
+import io.netty.example.udt.util.UtilThreadFactory;
+import io.netty.handler.logging.LogLevel;
+import io.netty.handler.logging.LoggingHandler;
+import io.netty.transport.udt.UdtChannel;
+import io.netty.transport.udt.nio.NioUdtProvider;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * UDT Byte Stream Client
+ * <p>
+ * Sends one message when a connection is open and echoes back any received data
+ * to the server. Simply put, the echo client initiates the ping-pong traffic
+ * between the echo client and server by sending the first message to the
+ * server.
+ */
+public class ByteEchoClient {
+
+    private static final Logger log = LoggerFactory
+            .getLogger(ByteEchoClient.class);
+
+    private final String host;
+    private final int port;
+    private final int messageSize;
+
+    public ByteEchoClient(final String host, final int port,
+            final int messageSize) {
+        this.host = host;
+        this.port = port;
+        this.messageSize = messageSize;
+    }
+
+    public void run() throws Exception {
+        // Configure the client.
+        final Bootstrap boot = new Bootstrap();
+        final ThreadFactory connectFactory = new UtilThreadFactory("connect");
+        final NioEventLoopGroup connectGroup = new NioEventLoopGroup(1,
+                connectFactory, NioUdtProvider.BYTE_PROVIDER);
+        try {
+            boot.group(connectGroup)
+                    .channelFactory(NioUdtProvider.BYTE_CONNECTOR)
+                    .localAddress("localhost", 0)
+                    .remoteAddress(new InetSocketAddress(host, port))
+                    .handler(new ChannelInitializer<UdtChannel>() {
+                        @Override
+                        public void initChannel(final UdtChannel ch)
+                                throws Exception {
+                            ch.pipeline().addLast(
+                                    new LoggingHandler(LogLevel.INFO),
+                                    new ByteEchoClientHandler(messageSize));
+                        }
+                    });
+            // Start the client.
+            final ChannelFuture f = boot.connect().sync();
+            // Wait until the connection is closed.
+            f.channel().closeFuture().sync();
+        } finally {
+            // Shut down the event loop to terminate all threads.
+            boot.shutdown();
+        }
+    }
+
+    public static void main(final String[] args) throws Exception {
+        log.info("init");
+
+        // client is reporting metrics
+        UtilConsoleReporter.enable(3, TimeUnit.SECONDS);
+
+        final String host = "localhost";
+        final int port = 1234;
+
+        final int messageSize = 64 * 1024;
+
+        new ByteEchoClient(host, port, messageSize).run();
+
+        log.info("done");
+    }
+
+}

example/src/main/java/io/netty/example/udt/echo/bytes/ByteEchoClientHandler.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project 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 io.netty.example.udt.echo.bytes;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundByteHandlerAdapter;
+import io.netty.channel.ChannelOption;
+import io.netty.transport.udt.nio.NioUdtProvider;
+
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.yammer.metrics.Metrics;
+import com.yammer.metrics.core.Meter;
+
+/**
+ * Handler implementation for the echo client. It initiates the ping-pong
+ * traffic between the echo client and server by sending the first message to
+ * the server on activation.
+ */
+public class ByteEchoClientHandler extends ChannelInboundByteHandlerAdapter {
+
+    private static final Logger log = LoggerFactory
+            .getLogger(ByteEchoClientHandler.class.getName());
+
+    private final ByteBuf message;
+
+    final Meter meter = Metrics.newMeter(ByteEchoClientHandler.class, "rate",
+            "bytes", TimeUnit.SECONDS);
+
+    public ByteEchoClientHandler(final int messageSize) {
+        message = Unpooled.buffer(messageSize);
+
+        for (int i = 0; i < message.capacity(); i++) {
+            message.writeByte((byte) i);
+        }
+    }
+
+    @Override
+    public void channelActive(final ChannelHandlerContext ctx) throws Exception {
+        log.info("ECHO active {}", NioUdtProvider.socketUDT(ctx.channel())
+                .toStringOptions());
+        ctx.write(message);
+    }
+
+    @Override
+    public void inboundBufferUpdated(final ChannelHandlerContext ctx,
+            final ByteBuf in) {
+        meter.mark(in.readableBytes());
+        final ByteBuf out = ctx.nextOutboundByteBuffer();
+        out.discardReadBytes();
+        out.writeBytes(in);
+        ctx.flush();
+    }
+
+    @Override
+    public void exceptionCaught(final ChannelHandlerContext ctx,
+            final Throwable cause) {
+        log.error("close the connection when an exception is raised", cause);
+        ctx.close();
+    }
+
+    @Override
+    public ByteBuf newInboundBuffer(final ChannelHandlerContext ctx)
+            throws Exception {
+        return ctx.alloc().directBuffer(
+                ctx.channel().config().getOption(ChannelOption.SO_RCVBUF));
+    }
+
+}

example/src/main/java/io/netty/example/udt/echo/bytes/ByteEchoServer.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project 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 io.netty.example.udt.echo.bytes;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.socket.nio.NioEventLoopGroup;
+import io.netty.example.udt.util.UtilThreadFactory;
+import io.netty.handler.logging.LogLevel;
+import io.netty.handler.logging.LoggingHandler;
+import io.netty.transport.udt.UdtChannel;
+import io.netty.transport.udt.nio.NioUdtProvider;
+
+import java.util.concurrent.ThreadFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * UDT Byte Stream Server
+ * <p>
+ * Echoes back any received data from a client.
+ */
+public class ByteEchoServer {
+
+    private static final Logger log = LoggerFactory
+            .getLogger(ByteEchoServer.class);
+
+    private final int port;
+
+    public ByteEchoServer(final int port) {
+        this.port = port;
+    }
+
+    public void run() throws Exception {
+        final ThreadFactory acceptFactory = new UtilThreadFactory("accept");
+        final ThreadFactory connectFactory = new UtilThreadFactory("connect");
+        final NioEventLoopGroup acceptGroup = new NioEventLoopGroup(1,
+                acceptFactory, NioUdtProvider.BYTE_PROVIDER);
+        final NioEventLoopGroup connectGroup = new NioEventLoopGroup(1,
+                connectFactory, NioUdtProvider.BYTE_PROVIDER);
+        // Configure the server.
+        final ServerBootstrap boot = new ServerBootstrap();
+        try {
+            boot.group(acceptGroup, connectGroup)
+                    .channelFactory(NioUdtProvider.BYTE_ACCEPTOR)
+                    .option(ChannelOption.SO_BACKLOG, 10)
+                    .localAddress("localhost", port)
+                    .handler(new LoggingHandler(LogLevel.INFO))
+                    .childHandler(new ChannelInitializer<UdtChannel>() {
+                        @Override
+                        public void initChannel(final UdtChannel ch)
+                                throws Exception {
+                            ch.pipeline().addLast(
+                                    new LoggingHandler(LogLevel.INFO),
+                                    new ByteEchoServerHandler());
+                        }
+                    });
+            // Start the server.
+            final ChannelFuture future = boot.bind().sync();
+            // Wait until the server socket is closed.
+            future.channel().closeFuture().sync();
+        } finally {
+            // Shut down all event loops to terminate all threads.
+            boot.shutdown();
+        }
+    }
+
+    public static void main(final String[] args) throws Exception {
+        log.info("init");
+
+        final int port = 1234;
+
+        new ByteEchoServer(port).run();
+
+        log.info("done");
+    }
+
+}

example/src/main/java/io/netty/example/udt/echo/bytes/ByteEchoServerHandler.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project 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 io.netty.example.udt.echo.bytes;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler.Sharable;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundByteHandlerAdapter;
+import io.netty.channel.ChannelOption;
+import io.netty.transport.udt.nio.NioUdtProvider;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Handler implementation for the echo server.
+ */
+@Sharable
+public class ByteEchoServerHandler extends ChannelInboundByteHandlerAdapter {
+
+    private static final Logger log = LoggerFactory
+            .getLogger(ByteEchoServerHandler.class.getName());
+
+    @Override
+    public void inboundBufferUpdated(final ChannelHandlerContext ctx,
+            final ByteBuf in) {
+        final ByteBuf out = ctx.nextOutboundByteBuffer();
+        out.discardReadBytes();
+        out.writeBytes(in);
+        ctx.flush();
+    }
+
+    @Override
+    public void exceptionCaught(final ChannelHandlerContext ctx,
+            final Throwable cause) {
+        log.error("close the connection when an exception is raised", cause);
+        ctx.close();
+    }
+
+    @Override
+    public void channelActive(final ChannelHandlerContext ctx) throws Exception {
+        log.info("ECHO active {}", NioUdtProvider.socketUDT(ctx.channel())
+                .toStringOptions());
+    }
+
+    @Override
+    public ByteBuf newInboundBuffer(final ChannelHandlerContext ctx)
+            throws Exception {
+        return ctx.alloc().directBuffer(
+                ctx.channel().config().getOption(ChannelOption.SO_RCVBUF));
+    }
+
+}