@Override

    public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {

        DefaultChannelHandlerContext next;

        next = findContextOutbound(MASK_WRITE);

        ReferenceCountUtil.touch(msg, next);

        next.invoker.invokeWrite(next, msg, promise);

        next = findContextOutbound(MASK_FLUSH);

        next.invoker.invokeFlush(next);

        return promise;

}

@Override

     public void invokeWrite(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {

         if (msg == null) {

             throw new NullPointerException("msg");

         }

         if (!validatePromise(ctx, promise, true)) {

             // promise cancelled

             ReferenceCountUtil.release(msg);

             return;

         }

         if (executor.inEventLoop()) {

             invokeWriteNow(ctx, msg, promise);

         } else {

             AbstractChannel channel = (AbstractChannel) ctx.channel();

             int size = channel.estimatorHandle().size(msg);

             if (size > 0) {

                 ChannelOutboundBuffer buffer = channel.unsafe().outboundBuffer();

                 // Check for null as it may be set to null if the channel is closed already

                 if (buffer != null) {

                     buffer.incrementPendingOutboundBytes(size);

                 }

             }

             safeExecuteOutbound(WriteTask.newInstance(ctx, msg, size, promise), promise, msg);

         }

     }

private void safeExecuteOutbound(Runnable task, ChannelPromise promise, Object msg) {

         try {

             executor.execute(task);

         } catch (Throwable cause) {

             try {

                 promise.setFailure(cause);

             } finally {

                 ReferenceCountUtil.release(msg);

             }

         }

     }