Netty Related 26 Mar 2014

Charset utf8 = Charset.forName(“UTF-8“);
ByteBuf buf = Unpooled.copiedBuffer(“Netty in Action rocks!“, utf8);
ByteBuf sliced = buf.slice(0, 14);
System.out.println(sliced.toString(utf8);
buf.setByte(0, (byte) ’J’);
assert buf.get(0) == sliced.get(0); #1

Charset utf8 = Charset.forName(“UTF-8“);
ByteBuf buf = Unpooled.copiedBuffer(“Netty in Action rocks!“, utf8);
ByteBuf copy = buf.copy(0, 14);
System.out.println(copy.toString(utf8);
buf.setByte(0, (byte) ’J’);
assert buf.get(0) != copy.get(0); #2

public class EchoServer {
    private final int port;
    public EchoServer(int port) {
        this.port = port;
    }
    public void start() throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap(); // 1
            b.group(group)  // 2
             .channel(NioServerSocketChannel.class) // 2
             .localAddress(new InetSocketAddress(port))    // 2
             .childHandler(new ChannelInitializer<SocketChannel>() {  // 3
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new EchoServerHandler());                  // 4
                }
             });
            ChannelFuture f = b.bind().sync(); // 5
            System.out.println(EchoServer.class.getName() +
                        “ started and listen on “ + f.channel().localAddress());
            f.channel().closeFuture().sync(); // 6
        } finally {
            group.shutdownGracefully().sync(); // 7
        }
    }
}

public ChannelFuture bind(SocketAddress localAddress) {
    validate();       // 1
    if (localAddress == null) {
        throw new NullPointerException("localAddress");
    }
    return doBind(localAddress);  // 2
}

private ChannelFuture doBind(final SocketAddress localAddress) {
    final ChannelFuture regFuture = initAndRegister();  // 1
    final Channel channel = regFuture.channel();
    if (regFuture.cause() != null) {     // 2
        return regFuture;
    }
    final ChannelPromise promise;
    if (regFuture.isDone()) {      // 3
        promise = channel.newPromise();
        doBind0(regFuture, channel, localAddress, promise);
    } else {                               // 3
        // Registration future is almost always fulfilled already, but just in case it's not.
        promise = new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE);
        regFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                doBind0(regFuture, channel, localAddress, promise);
            }
        });
    }
    return promise;
}

@Override
public T newChannel(EventLoop eventLoop, EventLoopGroup childGroup) {
    try {
        Constructor<? extends T> constructor = clazz.getConstructor(EventLoop.class, EventLoopGroup.class);
        return constructor.newInstance(eventLoop, childGroup);
    } catch (Throwable t) {
        throw new ChannelException("Unable to create Channel from class " + clazz, t);
    }
}

@Override
void init(Channel channel) throws Exception {
    final Map<ChannelOption<?>, Object> options = options();
    synchronized (options) {
        channel.config().setOptions(options);  // 为channel设置options
    }
    final Map<AttributeKey<?>, Object> attrs = attrs();
    synchronized (attrs) {   // 为channel设置attributes
        for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
            @SuppressWarnings("unchecked")
            AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
            channel.attr(key).set(e.getValue());
        }
    }
    ChannelPipeline p = channel.pipeline();
    if (handler() != null) {
        p.addLast(handler());  // 为Server设置handler
    }
    final ChannelHandler currentChildHandler = childHandler;
    final Entry<ChannelOption<?>, Object>[] currentChildOptions;
    final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
    synchronized (childOptions) {
        currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
    }
    synchronized (childAttrs) {
        currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
    }
    p.addLast(new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(Channel ch) throws Exception {
            ch.pipeline().addLast(new ServerBootstrapAcceptor(currentChildHandler, currentChildOptions,
                    currentChildAttrs));  // 将这些属性设置在ServerBootstrapAcceptor，当有新的连接时，将这些属性设置到这些连接对应的channel中
        }
    });
}

@Override
public final void register(final ChannelPromise promise) {
    if (eventLoop.inEventLoop()) {
        register0(promise);   // 如果当前线程为eventLoop线程，则执行register0
    } else {
        try {
            eventLoop.execute(new OneTimeTask() {   // 如果当前线程不是eventLoop线程，则使用eventLoop执行register0
                @Override
                public void run() {
                    register0(promise);
                }
            });
        } catch (Throwable t) {
            ...
        }
    }
}

selectionKey = javaChannel().register(eventLoop().selector, 0, this);  // attachment为Channel自身

private static void doBind0(
        final ChannelFuture regFuture, final Channel channel,
        final SocketAddress localAddress, final ChannelPromise promise) {
    // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
    // the pipeline in its channelRegistered() implementation.
    // ? 未找到该处是如何做到的
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (regFuture.isSuccess()) {
                channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                promise.setFailure(regFuture.cause());
            }
        }
    });
}

unsafe.bind(localAddress, promise);

javaChannel().socket().bind(localAddress, config.getBacklog());