Я пытаюсь реализовать Netty 4.X с динамическим конвейером ChannelHandler.
Когда люди предложили «Использовать вызов вместо модификации конвейера во время выполнения из соображений производительности», я реализовал Сервер, RouterInboundHander и Клиент для проверки этой теории. но это не работает Вот мой код
Сервер
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import org.apache.log4j.Logger;
public class Server implements Runnable{
private static final Logger logger = Logger.getLogger(Server.class);
public static final int PORT = 9528;
public static final String TIME = "time";
public static final String REVERSE = "reverse";
public static final String ERROR = "error";
@Override
public void run() {
final EventLoopGroup boss = new NioEventLoopGroup();
final EventLoopGroup work = new NioEventLoopGroup();
try {
new ServerBootstrap()
.group(boss, work)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 128)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(final NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024 * 1024, RouterInboundHandler.DELIMINATOR));
ch.pipeline().addLast(new RouterInboundHandler());
// ch.pipeline().addLast(new RouterInboundHandler.StringWriterOutboundHandler());
}
}).bind(PORT).sync().channel().closeFuture().sync();
} catch (final Exception ex){
ex.printStackTrace();
} finally {
boss.shutdownGracefully();
work.shutdownGracefully();
}
}
public static void main(String[] args) {
new Thread(new Server()).start();
}
}
RouterInboundHandler
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import org.apache.log4j.Logger;
import java.util.Date;
public class RouterInboundHandler extends SimpleChannelInboundHandler<ByteBuf> {
private static final Logger logger = Logger.getLogger(RouterInboundHandler.class);
public static final ByteBuf DELIMINATOR = Unpooled.copiedBuffer("\r\n".getBytes());
private final ChannelInboundHandler timer = new TimePrinterInboundHandler();
private final ChannelInboundHandler string = new StringReverseHandler();
private final ChannelInboundHandler error = new ErrorInboundHander();
@Override
public void channelActive(final ChannelHandlerContext ctx) throws Exception {
System.out.println("Connection made");
}
@Override
public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
System.out.println("OOOPS");
}
@Override
protected void channelRead0(final ChannelHandlerContext ctx, final ByteBuf msg) throws Exception {
final byte[] data = new byte[msg.readableBytes()];
msg.readBytes(data);
final String command = new String(data);
if (command.equals(Server.TIME)) {
timer.channelRead(ctx, command);
} else if (command.equals(Server.REVERSE)) {
string.channelRead(ctx, command);
} else {
error.channelRead(ctx, command);
}
}
@Override
public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
public static final class StringWriterOutboundHandler extends ChannelOutboundHandlerAdapter {
@Override
public void write(final ChannelHandlerContext ctx,
final Object msg, final ChannelPromise promise) throws Exception {
if (msg.getClass().equals(String.class)) {
System.out.println("I am writing " + msg + " to the client");
ctx.writeAndFlush(Unpooled.copiedBuffer(((String) msg).getBytes()));
} else {
System.out.println("This is not supposed to be");
}
}
}
public static class TimePrinterInboundHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(final ChannelHandlerContext ctx, final String msg) throws Exception {
System.out.println("I received message " + msg);
final String time = new Date(System.currentTimeMillis()).toString();
System.out.println("TimePrinterInboundHandler invoked");
ctx.writeAndFlush(Unpooled.copiedBuffer((time + " @ " + msg).getBytes()));
}
}
public static class StringReverseHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(final ChannelHandlerContext ctx, final String msg) throws Exception {
final byte[] data = msg.getBytes();
final byte[] newData = new byte[data.length];
for (int i = 1; i <= data.length; i++) {
newData[data.length - i] = data[i - 1];
}
System.out.println("StringReverseHandler invoked");
ctx.writeAndFlush(Unpooled.copiedBuffer(new String(newData).getBytes()));
}
}
public static class ErrorInboundHander extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(final ChannelHandlerContext ctx,
final String msg) throws Exception {
System.out.println("ErrorInboundHandler invoked");
ctx.writeAndFlush(Unpooled.copiedBuffer(("Error appears, here is what you gave me [" + msg + "]").getBytes()));
}
}
}
и Клиент
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
public class Client implements Runnable{
private ChannelHandlerContext channelHandlerContext;
@Override
public void run() {
final EventLoopGroup work = new NioEventLoopGroup();
try {
new Bootstrap()
.group(work)
.channel(NioSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 128)
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(final NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
@Override
public void channelActive(final ChannelHandlerContext ctx) throws Exception {
Client.this.channelHandlerContext = ctx;
}
@Override
public void channelRead(final ChannelHandlerContext ctx, final Object msg) throws Exception {
if(msg instanceof ByteBuf){
final byte[] data = new byte[((ByteBuf) msg).readableBytes()];
((ByteBuf) msg).readBytes(data);
System.out.println(new String(data));
} else {
System.out.println("No");
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
System.out.println("ChannelReadComplete in Client");
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
System.out.println("Here");
}
});
}
}).connect("localhost", Server.PORT).sync().channel().closeFuture().sync();
} catch (final Exception ex){
ex.printStackTrace();
} finally {
work.shutdownGracefully();
}
}
public ChannelHandlerContext getChannelHandlerContext() {
return channelHandlerContext;
}
public void writeMessage(final String... message){
for(final String msg : message) {
channelHandlerContext.write(Unpooled.copiedBuffer(msg.getBytes()));
channelHandlerContext.write(RouterInboundHandler.DELIMINATOR);
}
channelHandlerContext.flush();
}
public static void main(String[] args) throws InterruptedException {
final Client client = new Client();
new Thread(client).start();
Thread.sleep(2000);
// client.writeMessage(Server.TIME, Server.REVERSE, "Hello World");
client.writeMessage("Hello World", Server.TIME, Server.REVERSE);
}
}
Как показано в коде, три подкласса ChannelInboundHandler создаются во время фазы инициализации соединения канала. Когда клиент отправляет сообщение на сервер, channelRead0 проверит команду и, соответственно, запустит другой обработчик.
Моя проблема в том, является ли это правильным способом динамического использования конвейера Netty? И почему только первый запрос от клиента когда-либо отвечает?