聊聊reactor-netty的AccessLog

/**
 * Internal helpers for reactor-netty contracts
 *
 * @author Stephane Maldini
 */
public final class ReactorNetty {
    //......
​
    // System properties names
​
    /**
     * Default worker thread count, fallback to available processor
     * (but with a minimum value of 4)
     */
    public static final String IO_WORKER_COUNT = "reactor.netty.ioWorkerCount";
    /**
     * Default selector thread count, fallback to -1 (no selector thread)
     */
    public static final String IO_SELECT_COUNT = "reactor.netty.ioSelectCount";
    /**
     * Default worker thread count for UDP, fallback to available processor
     * (but with a minimum value of 4)
     */
    public static final String UDP_IO_THREAD_COUNT = "reactor.netty.udp.ioThreadCount";
​
​
    /**
     * Default value whether the native transport (epoll, kqueue) will be preferred,
     * fallback it will be preferred when available
     */
    public static final String NATIVE = "reactor.netty.native";
​
​
    /**
     * Default max connections, if -1 will never wait to acquire before opening a new
     * connection in an unbounded fashion. Fallback to
     * available number of processors (but with a minimum value of 16)
     */
    public static final String POOL_MAX_CONNECTIONS = "reactor.netty.pool.maxConnections";
    /**
     * Default acquisition timeout (milliseconds) before error. If -1 will never wait to
     * acquire before opening a new
     * connection in an unbounded fashion. Fallback 45 seconds
     */
    public static final String POOL_ACQUIRE_TIMEOUT = "reactor.netty.pool.acquireTimeout";
​
​
    /**
     * Default SSL handshake timeout (milliseconds), fallback to 10 seconds
     */
    public static final String SSL_HANDSHAKE_TIMEOUT = "reactor.netty.tcp.sslHandshakeTimeout";
    /**
     * Default value whether the SSL debugging on the client side will be enabled/disabled,
     * fallback to SSL debugging disabled
     */
    public static final String SSL_CLIENT_DEBUG = "reactor.netty.tcp.ssl.client.debug";
    /**
     * Default value whether the SSL debugging on the server side will be enabled/disabled,
     * fallback to SSL debugging disabled
     */
    public static final String SSL_SERVER_DEBUG = "reactor.netty.tcp.ssl.server.debug";
​
​
    /**
     * Specifies whether the Http Server access log will be enabled.
     * By default it is disabled.
     */
    public static final String ACCESS_LOG_ENABLED = "reactor.netty.http.server.accessLogEnabled";
​
    //......
}

final class HttpServerBind extends HttpServer
        implements Function<ServerBootstrap, ServerBootstrap> {
​
    static final HttpServerBind INSTANCE = new HttpServerBind();
​
    static final Function<DisposableServer, DisposableServer> CLEANUP_GLOBAL_RESOURCE = DisposableBind::new;
​
    static final boolean ACCESS_LOG =
            Boolean.parseBoolean(System.getProperty(ACCESS_LOG_ENABLED, "false"));
​
    //......
​
    static final class Http1Initializer
            implements BiConsumer<ConnectionObserver, Channel>  {
​
        final int                                                line;
        final int                                                header;
        final int                                                chunk;
        final boolean                                            validate;
        final int                                                buffer;
        final int                                                minCompressionSize;
        final BiPredicate<HttpServerRequest, HttpServerResponse> compressPredicate;
        final boolean                                            forwarded;
        final ServerCookieEncoder                                cookieEncoder;
        final ServerCookieDecoder                                cookieDecoder;
​
        Http1Initializer(int line,
                int header,
                int chunk,
                boolean validate,
                int buffer,
                int minCompressionSize,
                @Nullable BiPredicate<HttpServerRequest, HttpServerResponse> compressPredicate,
                boolean forwarded,
                ServerCookieEncoder encoder,
                ServerCookieDecoder decoder) {
            this.line = line;
            this.header = header;
            this.chunk = chunk;
            this.validate = validate;
            this.buffer = buffer;
            this.minCompressionSize = minCompressionSize;
            this.compressPredicate = compressPredicate;
            this.forwarded = forwarded;
            this.cookieEncoder = encoder;
            this.cookieDecoder = decoder;
        }
​
        @Override
        public void accept(ConnectionObserver listener, Channel channel) {
            ChannelPipeline p = channel.pipeline();
​
            p.addLast(NettyPipeline.HttpCodec, new HttpServerCodec(line, header, chunk, validate, buffer));
​
            if (ACCESS_LOG) {
                p.addLast(NettyPipeline.AccessLogHandler, new AccessLogHandler());
            }
​
            boolean alwaysCompress = compressPredicate == null && minCompressionSize == 0;
​
            if (alwaysCompress) {
                p.addLast(NettyPipeline.CompressionHandler,
                        new SimpleCompressionHandler());
            }
​
            p.addLast(NettyPipeline.HttpTrafficHandler,
                    new HttpTrafficHandler(listener, forwarded, compressPredicate, cookieEncoder, cookieDecoder));
        }
    }
​
    //......
}           

final class AccessLogHandler extends ChannelDuplexHandler {
​
    AccessLog accessLog = new AccessLog();
​
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof HttpRequest) {
            final HttpRequest request = (HttpRequest) msg;
            final SocketChannel channel = (SocketChannel) ctx.channel();
​
            accessLog = new AccessLog()
                    .address(channel.remoteAddress().getHostString())
                    .port(channel.localAddress().getPort())
                    .method(request.method().name())
                    .uri(request.uri())
                    .protocol(request.protocolVersion().text());
        }
        super.channelRead(ctx, msg);
    }
​
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        if (msg instanceof HttpResponse) {
            final HttpResponse response = (HttpResponse) msg;
            final HttpResponseStatus status = response.status();
​
            if (status.equals(HttpResponseStatus.CONTINUE)) {
                ctx.write(msg, promise);
                return;
            }
​
            final boolean chunked = HttpUtil.isTransferEncodingChunked(response);
            accessLog.status(status.codeAsText())
                     .chunked(chunked);
            if (!chunked) {
                accessLog.contentLength(HttpUtil.getContentLength(response, -1));
            }
        }
        if (msg instanceof LastHttpContent) {
            accessLog.increaseContentLength(((LastHttpContent) msg).content().readableBytes());
            ctx.write(msg, promise)
               .addListener(future -> {
                   if (future.isSuccess()) {
                       accessLog.log();
                   }
               });
            return;
        }
        if (msg instanceof ByteBuf) {
            accessLog.increaseContentLength(((ByteBuf) msg).readableBytes());
        }
        if (msg instanceof ByteBufHolder) {
            accessLog.increaseContentLength(((ByteBufHolder) msg).content().readableBytes());
        }
        ctx.write(msg, promise);
    }
}

final class AccessLog {
    static final Logger log = Loggers.getLogger("reactor.netty.http.server.AccessLog");
    static final DateTimeFormatter DATE_TIME_FORMATTER =
            DateTimeFormatter.ofPattern("dd/MMM/yyyy:HH:mm:ss Z", Locale.US);
    static final String COMMON_LOG_FORMAT =
            "{} - {} [{}] \"{} {} {}\" {} {} {} {} ms";
    static final String MISSING = "-";
​
    final String zonedDateTime;
​
    String address;
    CharSequence method;
    CharSequence uri;
    String protocol;
    String user = MISSING;
    CharSequence status;
    long contentLength;
    boolean chunked;
    long startTime = System.currentTimeMillis();
    int port;
​
    AccessLog() {
        this.zonedDateTime = ZonedDateTime.now().format(DATE_TIME_FORMATTER);
    }
​
    AccessLog address(String address) {
        this.address = Objects.requireNonNull(address, "address");
        return this;
    }
​
    AccessLog port(int port) {
        this.port = port;
        return this;
    }
​
    AccessLog method(CharSequence method) {
        this.method = Objects.requireNonNull(method, "method");
        return this;
    }
​
    AccessLog uri(CharSequence uri) {
        this.uri = Objects.requireNonNull(uri, "uri");
        return this;
    }
​
    AccessLog protocol(String protocol) {
        this.protocol = Objects.requireNonNull(protocol, "protocol");
        return this;
    }
​
    AccessLog status(CharSequence status) {
        this.status = Objects.requireNonNull(status, "status");
        return this;
    }
​
    AccessLog contentLength(long contentLength) {
        this.contentLength = contentLength;
        return this;
    }
​
    AccessLog increaseContentLength(long contentLength) {
        if (chunked) {
            this.contentLength += contentLength;
        }
        return this;
    }
​
    AccessLog chunked(boolean chunked) {
        this.chunked = chunked;
        return this;
    }
​
    long duration() {
        return System.currentTimeMillis() - startTime;
    }
​
    void log() {
        if (log.isInfoEnabled()) {
            log.info(COMMON_LOG_FORMAT, address, user, zonedDateTime,
                    method, uri, protocol, status, (contentLength > -1 ? contentLength : MISSING), port, duration());
        }
    }
}