Netty4.1源码 :DefaultChannelHandlerContext

 

 

 

/**
*   ChannelHandlerContext 实现了ChannelInboundInvoker和ChannelOutboundInvoker
*
* ChannelPipeline 也实现了ChannelInboundInvoker和*ChannelOutboundInvoker，
* 
*  ChannelPipeline 持有ChannelHandlerContext链表，
*  对于ChannelInboundInvoker和ChannelOutboundInvoker事件
*   ChannelPipeline 只是委派给ChannelHandlerContext链表中每一个
* ChannelHandlerContext处理，ChannelHandlerContext再调用
*  相应的ChannelInboundHandler 或ChannelOutboundHandler方法
* 
* ChannelPipeline 实现ChannelInboundInvoker和ChannelOutboundInvoker
* 目的：只是平滑地将事件委派出去。
**/

 

 

 

 

 

/**
 * Enables a {@link ChannelHandler} to interact with its {@link ChannelPipeline}
 * and other handlers. Among other things a handler can notify the next {@link ChannelHandler} in the
 * {@link ChannelPipeline} as well as modify the {@link ChannelPipeline} it belongs to dynamically.
 * 
 * ChannelHandlerContext 是ChannelPipeline与ChannelHandler之间的粘合剂，负责事件透传至ChannelHandler及调用上或下一个ChannelHandler
 * DefaultChannelPipeline 持有N个ChannelHandlerContext(链表结构)，每一个ChannelHandlerContext对应一个ChannelHandler，
 * ChannelHandlerContext创建于：channel.pipeline().addLast(ChannelHandler)方法。
 * 事件源于ChannelPipeline-》链表Head(或Tail)-》ChannelHandlerContext-》ChannelHandler
 *                           链表下一个（或上一个）->ChannelHandlerContext-》ChannelHandler
 *                           链表下一个（或上一个）->ChannelHandlerContext-》ChannelHandler
 * 调用过程：ChannelPipeline--》ChannelHandlerContext--》ChannelHandler
 * <h3>Notify</h3>
 * 你能够在ChannelHander中调用ChannelHandlerContext的各种方法来触发相邻的ChannelHanndler事件。 
 * 一般情况下：  你在ChannelHander.channelActive(ChannelHandlerContext ctx) 方法中调用ChannelHandlerContext.fireChannelActive 
 *               会触发下一个　ChannelHander.channelActive,如果没有调用，则下一个（之后的）ChannelHander不会被执行。
 *               
 *
 * <h3>Modifying a pipeline</h3>
 * 你能够通过ChannelHandlerContext.pipeline()方法获取ChannelPipeline，从而在运行时动态添加修改ChannelHandlers
 * 
 *
 * <h3>Retrieving for later use</h3>
 * 你能够保存ChannelHandlerContext对象，以使之后使用。
 * 例如在另外一个线程（非netty内部IO线程）触发事件
 * 
 * public class MyHandler extends ChannelDuplexHandler {
 *
 *     private ChannelHandlerContext ctx; 
 *
 *     public void beforeAdd(ChannelHandlerContext ctx) {
 *          this.ctx = ctx; 
 *     }
 *
 *     public void login(String username, password) {
 *         ctx.write(new LoginMessage(username, password));
 *     }
 *     ...
 * }
 * </pre>
 *
 * <h3>Storing stateful information</h3>
 * 保存状态信息（线程安全的）
 * 你可以使用 attr(AttributeKey<T> key);方法存储状态信息，以便之后需要时取出。
 * 请参考ChannelHandler了解状态信息管理
 *
 * <h3>A handler can have more than one context</h3>
 * 请注意一个ChannelHandler 实例能够添加到多个ChannelPipeline中或同一个ChannelPipeline添加多次，
 * 每一次添加，都创建一个新的ChannelHandlerContext与其对应，即同一个ChannelHandler 实例可能对应
 * 多个ChannelHandlerContext，在通过ChannelHandlerContext.attr保存或获取变量时，是保存在ChannelHandlerContext
 * 中，因而虽然只有一个ChannelHandler 实例，但添加的变量不是共享的。
 * 该变量的范围：同一个Channel.pipeline().链表中同一个ChannelHandlerContext 范围。
 * 
 */
public interface ChannelHandlerContext extends AttributeMap, ChannelInboundInvoker, ChannelOutboundInvoker {

    /**
	 * 返回绑定到ChannelHandlerContext的Channel
     */
    Channel channel();

    /**
	 * 返回io线程对象（EventExecutor）,用于执行内部IO任务
     */
    EventExecutor executor();

    /**
     *
	 * ChannelPipeline.add("name",ChannelHandler)--这个NAME
     */
    String name();

    /**
	 * 返回ChannelHandlerContext对应的ChannelHandler（每一个ChannelHandlerContext对应一个ChannelHandler）
     */
    ChannelHandler handler();

    /**
	 * 如果对应的ChannelHandler从ChannelPipeline被迁移，则返回true
     */
    boolean isRemoved();

	/**
	* 以下方法调用 ChannelHander相应方法：如fireChannelRegistered，则调用channelRegistered
	*  DefaultChannelHandlerContext 中还实现了下一个或上一个ChannelHandler的事件触发。
	**/
    ChannelHandlerContext fireChannelRegistered();
    ChannelHandlerContext fireChannelUnregistered();
    ChannelHandlerContext fireChannelActive();
    ChannelHandlerContext fireChannelInactive();
    ChannelHandlerContext fireExceptionCaught(Throwable cause);
    ChannelHandlerContext fireUserEventTriggered(Object evt);
    ChannelHandlerContext fireChannelRead(Object msg);
    ChannelHandlerContext fireChannelReadComplete();
    ChannelHandlerContext fireChannelWritabilityChanged();
    ChannelHandlerContext read();
    ChannelHandlerContext flush();

    /**
     * 返回ChannelPipeline
     */
    ChannelPipeline pipeline();

    /**
     * Return the assigned {@link ByteBufAllocator} which will be used to allocate {@link ByteBuf}s.
     */
    ByteBufAllocator alloc();

    /**
     * @deprecated Use {@link Channel#attr(AttributeKey)}
     */
    @Deprecated
    @Override
    <T> Attribute<T> attr(AttributeKey<T> key);

    /**
     * @deprecated Use {@link Channel#hasAttr(AttributeKey)}
     */
    @Deprecated
    @Override
    <T> boolean hasAttr(AttributeKey<T> key);
}


abstract class AbstractChannelHandlerContext extends DefaultAttributeMap
        implements ChannelHandlerContext, ResourceLeakHint {
	//对应的ChannelHandler 是inbound?则处理ChannelInboundHandler 事件
	private final boolean inbound
	//对应的ChannelHandler 是outbound?则处理ChannelOutboundHandler 事件
    private final boolean outbound;
	//持有pipeline，ChannelHandlerContext与pipeline 是多对一关系
    private final DefaultChannelPipeline pipeline;
	//ChannelHander 对应的名字
    private final String name;
	//ChannelHandlerContext链，指向下一个ChannelHandlerContext，
	//DefaultChannelPipeline.addLast(ChannelHandler)建立的ChannelHandlerContext链
	volatile AbstractChannelHandlerContext next;
	//ChannelHandlerContext链，指向上一个ChannelHandlerContext
	//DefaultChannelPipeline持有链头指向head
	//DefaultChannelPipeline.addLast(ChannelHandler)建立的ChannelHandlerContext链
	//DefaultChannelPipeline持有链尾指向tail
    volatile AbstractChannelHandlerContext prev;
		/**
		* 构建器 
		*
		**/
		AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor, String name,
                                  boolean inbound, boolean outbound) {
        this.name = ObjectUtil.checkNotNull(name, "name");
        this.pipeline = pipeline;
        this.executor = executor;
        this.inbound = inbound;
        this.outbound = outbound;
       
    }
	/*
	* firexxxx方法：
	* 以fireChannelActive为例 ,channelActive属于Inbound事件
	* 从当前ChannelHandlerContext向下查找每一个inbound==true的ChannelHandlerContext，找到后
	* 调用其对应handler.channelActive(ctx)，channelActive内部处理完逻辑后，如果需要调用下一个
	* ChannelHander.channelActive(ctx)，则在channelActive内部写：ctx.fireChannelActive()
	* 由此继续循环调用下一个ChannelHandler.channelActive
	* 
	*/
	
	  @Override
    public ChannelHandlerContext fireChannelActive() {
        final AbstractChannelHandlerContext next = findContextInbound();
        invokeChannelActive(next);
        return this;
    }
	private AbstractChannelHandlerContext findContextInbound() {
        AbstractChannelHandlerContext ctx = this;
        do {
            ctx = ctx.next;
        } while (!ctx.inbound);
        return ctx;
    }
    static void invokeChannelActive(final AbstractChannelHandlerContext next) {
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeChannelActive();
        } else {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    next.invokeChannelActive();
                }
            });
        }
    }

    private void invokeChannelActive() {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelActive(this);
            } catch (Throwable t) {
                notifyHandlerException(t);
            }
        } else {
            fireChannelActive();
        }
    }
	
	/**
	* wirte属于outbound事件，当通过ctx.write()方法被调用时，查找当前ChannelHandlerContext的上一个
	* ChannelHandlerContext，调用其write方法，-》nextChannelHander.write -->next ....-->链表头
	* DefaultPipeLine.HeadContext 实例.write :最终调用channel.unsafe.write 真正的向网络流中写出
	**/
	private void write(Object msg, boolean flush, ChannelPromise promise) {
        AbstractChannelHandlerContext next = findContextOutbound();
        final Object m = pipeline.touch(msg, next);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            if (flush) {
                next.invokeWriteAndFlush(m, promise);
            } else {
                next.invokeWrite(m, promise);
            }
        } else {
			//如果不是NioEventLoop内部io线程则创建任务添加到内部io线程执行队列中等待执行
			
            AbstractWriteTask task;
            if (flush) {
                task = WriteAndFlushTask.newInstance(next, m, promise);
            }  else {
                task = WriteTask.newInstance(next, m, promise);
            }
            safeExecute(executor, task, promise, m);
        }
    }
	    private void invokeWrite(Object msg, ChannelPromise promise) {
        try {
            ((ChannelOutboundHandler) handler()).write(this, msg, promise);
        } catch (Throwable t) {
            notifyOutboundHandlerException(t, promise);
        }
    }
	
	   private AbstractChannelHandlerContext findContextOutbound() {
        AbstractChannelHandlerContext ctx = this;
        do {
            ctx = ctx.prev;
        } while (!ctx.outbound);
        return ctx;
    }
	
}

/**
*DefaultChannelHandlerContext 在AbstractChannelHandlerContext基础上，添加了判断是ChannelHandler是
* ChannelInboundHandler 还是ChannelOutboundHandler的方法
* 
*/
final class DefaultChannelHandlerContext extends AbstractChannelHandlerContext {

    private final ChannelHandler handler;

    DefaultChannelHandlerContext(
            DefaultChannelPipeline pipeline, EventExecutor executor, String name, ChannelHandler handler) {
        super(pipeline, executor, name, isInbound(handler), isOutbound(handler));
        if (handler == null) {
            throw new NullPointerException("handler");
        }
        this.handler = handler;
    }

    @Override
    public ChannelHandler handler() {
        return handler;
    }

    private static boolean isInbound(ChannelHandler handler) {
        return handler instanceof ChannelInboundHandler;
    }

    private static boolean isOutbound(ChannelHandler handler) {
        return handler instanceof ChannelOutboundHandler;
    }
}