private static Logger setupLogging() {
  InternalLoggerFactory.setDefaultFactory(Log4J2LoggerFactory.INSTANCE);
  try {
    System.setProperty(
        "vertx.logger-delegate-factory-class-name",
        "io.vertx.core.logging.Log4j2LogDelegateFactory");
  } catch (SecurityException e) {
    System.out.println(
        "Could not set logging system property as the security manager prevented it:"
            + e.getMessage());
  }

  final Logger logger = getLogger();
  Thread.setDefaultUncaughtExceptionHandler(
      (thread, error) ->
          logger.error("Uncaught exception in thread \"" + thread.getName() + "\"", error));
  Thread.currentThread()
      .setUncaughtExceptionHandler(
          (thread, error) ->
              logger.error("Uncaught exception in thread \"" + thread.getName() + "\"", error));
  return logger;
}
 

@BeforeSuite
public void beforeSuite() throws InterruptedException, IOException, GeneralSecurityException {
    ConfigManager.init(new ConfigManager.Arguments());
    configManager = ConfigManager.getInstance();
    PluginsManager.getInstance().initialize();

    InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);

    server = new ModelServer(configManager);
    server.start();

    try (InputStream is = new FileInputStream("src/test/resources/inference_open_api.json")) {
        listInferenceApisResult = IOUtils.toString(is, StandardCharsets.UTF_8.name());
    }

    try (InputStream is = new FileInputStream("src/test/resources/management_open_api.json")) {
        listManagementApisResult = IOUtils.toString(is, StandardCharsets.UTF_8.name());
    }

    try (InputStream is = new FileInputStream("src/test/resources/describe_api.json")) {
        noopApiResult = IOUtils.toString(is, StandardCharsets.UTF_8.name());
    }
}
 

public void init(Container container, ServerBundleConfiguration config) {
	logger.info("Initializing the container");
	// Override the supplied one
	ServerConfiguration configuration = container.getConfiguration().getServerConfiguration();
	AbstractHttpConnector connector = null;
	
	InternalLoggerFactory.setDefaultFactory(new Slf4JLoggerFactory());
	
	logger.info("Loading the http connectors");
	for (ConnectorConfiguration connectorConfig : configuration.getConnectorConfigurations()) {
		if (connectorConfig.getScheme() == Scheme.https) {
			connector = createHttpsConnector(connectorConfig, container.getRouter());
		} else {
			connector = createHttpConnector(connectorConfig, container.getRouter());
		}
		connector.registerListener(container.getMessageObserver());
		connector.initialize();
		connectors.add(connector);
	}
}
 

ReactorNettyClient(Connection connection, MySqlSslConfiguration ssl, ConnectionContext context) {
    requireNonNull(connection, "connection must not be null");
    requireNonNull(context, "context must not be null");
    requireNonNull(ssl, "ssl must not be null");

    this.connection = connection;
    this.context = context;

    // Note: encoder/decoder should before reactor bridge.
    connection.addHandlerLast(EnvelopeSlicer.NAME, new EnvelopeSlicer())
        .addHandlerLast(MessageDuplexCodec.NAME, new MessageDuplexCodec(context, this.closing, this.requestQueue));

    if (ssl.getSslMode().startSsl()) {
        connection.addHandlerFirst(SslBridgeHandler.NAME, new SslBridgeHandler(context, ssl));
    }

    if (InternalLoggerFactory.getInstance(ReactorNettyClient.class).isTraceEnabled()) {
        // Or just use logger.isTraceEnabled()?
        logger.debug("Connection tracking logging is enabled");
        connection.addHandlerFirst(LoggingHandler.class.getSimpleName(), new LoggingHandler(ReactorNettyClient.class, LogLevel.TRACE));
    }

    Flux<ServerMessage> inbound = connection.inbound().receiveObject()
        .handle(INBOUND_HANDLE);

    if (logger.isDebugEnabled()) {
        inbound = inbound.doOnNext(DEBUG_LOGGING);
    } else if (logger.isInfoEnabled()) {
        inbound = inbound.doOnNext(INFO_LOGGING);
    }

    inbound.subscribe(this.responseProcessor::onNext, throwable -> {
        try {
            logger.error("Connection Error: {}", throwable.getMessage(), throwable);
            responseProcessor.onError(throwable);
        } finally {
            connection.dispose();
        }
    }, this.responseProcessor::onComplete);
}
 

@BeforeClass
public void beforeSuite()
        throws InterruptedException, IOException, GeneralSecurityException,
                InvalidSnapshotException {
    System.setProperty("tsConfigFile", "src/test/resources/config.properties");
    FileUtils.deleteQuietly(new File(System.getProperty("LOG_LOCATION"), "config"));
    ConfigManager.init(new ConfigManager.Arguments());
    configManager = ConfigManager.getInstance();
    PluginsManager.getInstance().initialize();

    InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);
    server = new ModelServer(configManager);
    server.start();
}
 

@BeforeSuite
public void beforeSuite()
        throws InterruptedException, IOException, GeneralSecurityException,
                InvalidSnapshotException {
    ConfigManager.init(new ConfigManager.Arguments());
    configManager = ConfigManager.getInstance();
    PluginsManager.getInstance().initialize();

    InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);

    server = new ModelServer(configManager);
    server.start();
    String version = configManager.getProperty("version", null);
    try (InputStream is = new FileInputStream("src/test/resources/inference_open_api.json")) {
        listInferenceApisResult =
                String.format(IOUtils.toString(is, StandardCharsets.UTF_8.name()), version);
    }

    try (InputStream is = new FileInputStream("src/test/resources/management_open_api.json")) {
        listManagementApisResult =
                String.format(IOUtils.toString(is, StandardCharsets.UTF_8.name()), version);
    }

    try (InputStream is = new FileInputStream("src/test/resources/describe_api.json")) {
        noopApiResult = IOUtils.toString(is, StandardCharsets.UTF_8.name());
    }
}
 

public SpdyFrameLogger(InternalLogLevel level) {
    if (level == null) {
        throw new NullPointerException("level");
    }

    logger = InternalLoggerFactory.getInstance(getClass());
    this.level = level;
}
 

/**
 * Creates a new instance whose logger name is the fully qualified class
 * name of the instance.创建一个新实例，其日志记录器名称是实例的完全限定类名。
 *
 * @param level the log level
 */
public LoggingHandler(LogLevel level) {
    if (level == null) {
        throw new NullPointerException("level");
    }

    logger = InternalLoggerFactory.getInstance(getClass());
    this.level = level;
    internalLevel = level.toInternalLevel();
}
 

/**
 * Creates a new instance with the specified logger name.使用指定的日志记录器名称创建一个新实例。
 *
 * @param clazz the class type to generate the logger for
 * @param level the log level
 */
public LoggingHandler(Class<?> clazz, LogLevel level) {
    if (clazz == null) {
        throw new NullPointerException("clazz");
    }
    if (level == null) {
        throw new NullPointerException("level");
    }

    logger = InternalLoggerFactory.getInstance(clazz);
    this.level = level;
    internalLevel = level.toInternalLevel();
}
 

/**
 * Creates a new instance with the specified logger name.使用指定的日志记录器名称创建一个新实例。
 *
 * @param name the name of the class to use for the logger
 * @param level the log level
 */
public LoggingHandler(String name, LogLevel level) {
    if (name == null) {
        throw new NullPointerException("name");
    }
    if (level == null) {
        throw new NullPointerException("level");
    }

    logger = InternalLoggerFactory.getInstance(name);
    this.level = level;
    internalLevel = level.toInternalLevel();
}
 

public void initProperties() throws Exception {
	InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);
	System.setProperty("io.netty.tryReflectionSetAccessible", "true");
	// System.setProperty("io.netty.noUnsafe", "false");
	// ReferenceCountUtil.release(byteBuf);
	// ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.ADVANCED);
}
 

ReactorNettyClient(Connection connection, MySqlSslConfiguration ssl, ConnectionContext context) {
    requireNonNull(connection, "connection must not be null");
    requireNonNull(context, "context must not be null");
    requireNonNull(ssl, "ssl must not be null");

    this.connection = connection;
    this.context = context;

    // Note: encoder/decoder should before reactor bridge.
    connection.addHandlerLast(EnvelopeSlicer.NAME, new EnvelopeSlicer())
        .addHandlerLast(MessageDuplexCodec.NAME, new MessageDuplexCodec(context, this.closing, this.requestQueue));

    if (ssl.getSslMode().startSsl()) {
        connection.addHandlerFirst(SslBridgeHandler.NAME, new SslBridgeHandler(context, ssl));
    }

    if (InternalLoggerFactory.getInstance(ReactorNettyClient.class).isTraceEnabled()) {
        // Or just use logger.isTraceEnabled()?
        logger.debug("Connection tracking logging is enabled");
        connection.addHandlerFirst(LoggingHandler.class.getSimpleName(), new LoggingHandler(ReactorNettyClient.class, LogLevel.TRACE));
    }

    Flux<ServerMessage> inbound = connection.inbound().receiveObject()
        .handle(INBOUND_HANDLE);

    if (logger.isDebugEnabled()) {
        inbound = inbound.doOnNext(DEBUG_LOGGING);
    } else if (logger.isInfoEnabled()) {
        inbound = inbound.doOnNext(INFO_LOGGING);
    }

    inbound.subscribe(this.responseProcessor::onNext, throwable -> {
        try {
            logger.error("Connection Error: {}", throwable.getMessage(), throwable);
            responseProcessor.onError(throwable);
        } finally {
            connection.dispose();
        }
    }, this.responseProcessor::onComplete);
}
 

public void start() {
    Configuration config = Configuration.INSTANCE;
    InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);
    bossGroup = new NioEventLoopGroup(1);
    workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                .addLast("logging", new LoggingHandler(LogLevel.DEBUG))
                                .addLast(new SocksInitRequestDecoder())
                                .addLast(new SocksMessageEncoder())
                                .addLast(new Socks5Handler())
                                .addLast(Status.TRAFFIC_HANDLER);
                    }
                });
        log.info("\tStartup {}-{}-client [{}{}]", Constants.APP_NAME, Constants.APP_VERSION, config.getMode(), config.getMode().equals("socks5") ? "" : ":" + config.getProtocol());
        new Thread(() -> new UdpServer().start()).start();
        ChannelFuture future = bootstrap.bind(config.getLocalHost(), config.getLocalPort()).sync();
        future.addListener(future1 -> log.info("\tTCP listening at {}:{}...", config.getLocalHost(), config.getLocalPort()));
        future.channel().closeFuture().sync();
    } catch (Exception e) {
        log.error("\tSocket bind failure ({})", e.getMessage());
    } finally {
        log.info("\tShutting down");
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}
 

public void start() {
    Configuration config = Configuration.INSTANCE;
    InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                .addLast("logging", new LoggingHandler(LogLevel.DEBUG))
                                .addLast(new XConnectHandler());
                        if (config.getReadLimit() != 0 || config.getWriteLimit() != 0) {
                            socketChannel.pipeline().addLast(
                                    new GlobalTrafficShapingHandler(Executors.newScheduledThreadPool(1), config.getWriteLimit(), config.getReadLimit())
                            );
                        }
                    }
                });
        log.info("\tStartup {}-{}-server [{}]", Constants.APP_NAME, Constants.APP_VERSION, config.getProtocol());
        new Thread(() -> new UdpServer().start()).start();
        ChannelFuture future = bootstrap.bind(config.getHost(), config.getPort()).sync();
        future.addListener(future1 -> log.info("\tTCP listening at {}:{}...", config.getHost(), config.getPort()));
        future.channel().closeFuture().sync();
    } catch (Exception e) {
        log.error("\tSocket bind failure ({})", e.getMessage());
    } finally {
        log.info("\tShutting down and recycling...");
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
        Configuration.shutdownRelays();
    }
    System.exit(0);
}
 

public static void main(String[] args) {
	String thisVertxName = UUID.randomUUID().toString() + LocalTime.now().getNano();
	// 设置当前系统Vertx的唯一标识
	System.setProperty("thisVertxName", thisVertxName);
	InternalLoggerFactory.setDefaultFactory(Log4J2LoggerFactory.INSTANCE);
	System.setProperty("vertx.logger-delegate-factory-class-name", "io.vertx.core.logging.Log4j2LogDelegateFactory");
	System.setProperty("vertx.disableDnsResolver", "true");
	new VxApiLauncher().dispatch(args);
}
 

/**
 * Creates a new instance of the PravegaConnectionListener class.
 *
 * @param enableTls          Whether to enable SSL/TLS.
 * @param enableTlsReload    Whether to reload TLS when the X.509 certificate file is replaced.
 * @param host               The name of the host to listen to.
 * @param port               The port to listen on.
 * @param streamSegmentStore The SegmentStore to delegate all requests to.
 * @param tableStore         The TableStore to delegate all requests to.
 * @param statsRecorder      (Optional) A StatsRecorder for Metrics for Stream Segments.
 * @param tableStatsRecorder (Optional) A Table StatsRecorder for Metrics for Table Segments.
 * @param tokenVerifier      The object to verify delegation token.
 * @param certFile           Path to the certificate file to be used for TLS.
 * @param keyFile            Path to be key file to be used for TLS.
 * @param replyWithStackTraceOnError Whether to send a server-side exceptions to the client in error messages.
 * @param executor           The executor to be used for running token expiration handling tasks.
 */
public PravegaConnectionListener(boolean enableTls, boolean enableTlsReload, String host, int port, StreamSegmentStore streamSegmentStore, TableStore tableStore,
                                 SegmentStatsRecorder statsRecorder, TableSegmentStatsRecorder tableStatsRecorder,
                                 DelegationTokenVerifier tokenVerifier, String certFile, String keyFile,
                                 boolean replyWithStackTraceOnError, ScheduledExecutorService executor) {
    this.enableTls = enableTls;
    if (this.enableTls) {
        this.enableTlsReload = enableTlsReload;
    } else {
        this.enableTlsReload = false;
    }
    this.host = Exceptions.checkNotNullOrEmpty(host, "host");
    this.port = port;
    this.store = Preconditions.checkNotNull(streamSegmentStore, "streamSegmentStore");
    this.tableStore = Preconditions.checkNotNull(tableStore, "tableStore");
    this.statsRecorder = Preconditions.checkNotNull(statsRecorder, "statsRecorder");
    this.tableStatsRecorder = Preconditions.checkNotNull(tableStatsRecorder, "tableStatsRecorder");
    this.pathToTlsCertFile = certFile;
    this.pathToTlsKeyFile = keyFile;
    InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);
    if (tokenVerifier != null) {
        this.tokenVerifier = tokenVerifier;
    } else {
        this.tokenVerifier = new PassingTokenVerifier();
    }
    this.replyWithStackTraceOnError = replyWithStackTraceOnError;
    this.connectionTracker = new ConnectionTracker();
    this.tokenExpiryHandlerExecutor = executor;
}
 

@Override
@Before
public void before() {
    super.before();
    InternalLoggerFactory.setDefaultFactory(new ResourceLeakLoggerFactory());
    this.originalLevel = ResourceLeakDetector.getLevel();
    ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.PARANOID);
}
 

public SpdyFrameLogger(InternalLogLevel level) {
    if (level == null) {
        throw new NullPointerException("level");
    }

    logger = InternalLoggerFactory.getInstance(getClass());
    this.level = level;
}
 

/**
 * Creates a new instance whose logger name is the fully qualified class
 * name of the instance.
 *
 * @param level   the log level
 */
public LoggingHandler(LogLevel level) {
    if (level == null) {
        throw new NullPointerException("level");
    }

    logger = InternalLoggerFactory.getInstance(getClass());
    this.level = level;
    internalLevel = level.toInternalLevel();
}
 

/**
 * Creates a new instance with the specified logger name.
 *
 * @param level   the log level
 */
public LoggingHandler(Class<?> clazz, LogLevel level) {
    if (clazz == null) {
        throw new NullPointerException("clazz");
    }
    if (level == null) {
        throw new NullPointerException("level");
    }
    logger = InternalLoggerFactory.getInstance(clazz);
    this.level = level;
    internalLevel = level.toInternalLevel();
}
 

/**
 * Creates a new instance with the specified logger name.
 *
 * @param level   the log level
 */
public LoggingHandler(String name, LogLevel level) {
    if (name == null) {
        throw new NullPointerException("name");
    }
    if (level == null) {
        throw new NullPointerException("level");
    }
    logger = InternalLoggerFactory.getInstance(name);
    this.level = level;
    internalLevel = level.toInternalLevel();
}
 

@AfterEach
void tearDown() {
    final Logger logger = (Logger) LoggerFactory.getLogger(getClass());
    final StatusManager sm = rootLogger.getLoggerContext().getStatusManager();
    int count = 0;
    for (Status s : sm.getCopyOfStatusList()) {
        final int level = s.getEffectiveLevel();
        if (level == Status.INFO) {
            continue;
        }
        if (s.getMessage().contains(InternalLoggerFactory.class.getName())) {
            // Skip the warnings related with Netty.
            continue;
        }

        count++;
        switch (level) {
            case Status.WARN:
                if (s.getThrowable() != null) {
                    logger.warn(s.getMessage(), s.getThrowable());
                } else {
                    logger.warn(s.getMessage());
                }
                break;
            case Status.ERROR:
                if (s.getThrowable() != null) {
                    logger.warn(s.getMessage(), s.getThrowable());
                } else {
                    logger.warn(s.getMessage());
                }
                break;
        }
    }

    if (count > 0) {
        fail("Appender raised an exception.");
    }
}
 

/**
 * Creates a new instance whose logger name is the fully qualified class
 * name of the instance.
 *
 * @param level the log level
 */
public LoggingHandler(LogLevel level) {
    if (level == null) {
        throw new NullPointerException("level");
    }

    logger = InternalLoggerFactory.getInstance(getClass());
    this.level = level;
    internalLevel = InternalLogLevel.DEBUG;
}
 

/**
 * Creates a new instance with the specified logger name.
 *
 * @param clazz the class type to generate the logger for
 * @param level the log level
 */
public LoggingHandler(Class<?> clazz, LogLevel level) {
    if (clazz == null) {
        throw new NullPointerException("clazz");
    }
    if (level == null) {
        throw new NullPointerException("level");
    }

    logger = InternalLoggerFactory.getInstance(clazz);
    this.level = level;
    internalLevel = InternalLogLevel.DEBUG;
}
 

/**
 * Creates a new instance with the specified logger name.
 *
 * @param name the name of the class to use for the logger
 * @param level the log level
 */
public LoggingHandler(String name, LogLevel level) {
    if (name == null) {
        throw new NullPointerException("name");
    }
    if (level == null) {
        throw new NullPointerException("level");
    }

    logger = InternalLoggerFactory.getInstance(name);
    this.level = level;
    internalLevel = InternalLogLevel.DEBUG;
}
 

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    InternalLoggerFactory.getInstance(
            SocketConnectionAttemptTest.class).warn("Unexpected exception:", cause);
}
 

public Http2FrameLogger(LogLevel level) {
    this(level.toInternalLevel(), InternalLoggerFactory.getInstance(Http2FrameLogger.class));
}
 

public Http2FrameLogger(LogLevel level, String name) {
    this(level.toInternalLevel(), InternalLoggerFactory.getInstance(name));
}
 

public Http2FrameLogger(LogLevel level, Class<?> clazz) {
    this(level.toInternalLevel(), InternalLoggerFactory.getInstance(clazz));
}
 

public static void main(String...args) {
  InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);

  ConfigArgs configArgs = ConfigArgs.parse(args);

  // load assets or die
  try {
    weakOrBreachedPwBf = BloomFilters.deserialize(AssetRegistry.getCompromisedPasswords());
    disposableMxBf = BloomFilters.deserialize(AssetRegistry.getDisposableEmails());
    awsIpMembership = new CidrMembership<>(AssetRegistry.getAwsIps(), "aws");
    gcpIpMembership = new CidrMembership<>(AssetRegistry.getGcpIps(), "gcp");
  } catch (AssetException ex) {
    log.error("error while reading asset", ex);
    System.exit(-1);
  }

  final GeoIp4j geoIp4j = new GeoIp4jImpl();

  log.info("{} <= quarantyne => {}", configArgs.getIngress().toHuman(), configArgs.getEgress().toHuman());

  configArgs.getAdminIpPort().ifPresent(ipPort -> {
    log.info("==> admin @ http://{}:{}", ipPort.getIp(), ipPort.getPort());
  });

  log.info("see available options with --help");
  int numCpus = CpuCoreSensor.availableProcessors();

  VertxOptions vertxOptions = new VertxOptions();
  vertxOptions.setPreferNativeTransport(true);
  vertxOptions.setMetricsOptions(
      new DropwizardMetricsOptions().setEnabled(true)
  );

  log.debug("==> event loop size is {}", vertxOptions.getEventLoopPoolSize());
  log.debug("==> detected {} cpus core", numCpus);
  Vertx vertx = Vertx.vertx(vertxOptions);

  ConfigSupplier configSupplier;
  if (configArgs.getConfigFile().isPresent()) {
    configSupplier = new ConfigSupplier(vertx,
        new ConfigRetrieverOptionsSupplier(configArgs.getConfigFile().get()));
  } else {
    log.info("No configuration file was specified, using default settings");
    configSupplier = new ConfigSupplier();
  }

  // quarantyne classifiers
  List<HttpRequestClassifier> httpRequestClassifierList = Lists.newArrayList(
      new FastAgentClassifier(),
      new IpRotationClassifier(),
      new SuspiciousRequestHeadersClassifier(),
      new SuspiciousUserAgentClassifier(),
      new LargeBodySizeClassifier(),
      new CompromisedPasswordClassifier(weakOrBreachedPwBf, configSupplier),
      new DisposableEmailClassifier(disposableMxBf, configSupplier),
      new GeoDiscrepancyClassifier(geoIp4j, configSupplier),
      new PublicCloudExecutionClassifier(awsIpMembership, gcpIpMembership)
      // new SuspiciousLoginActivityClassifier(geoIp4j)
  );

  MainClassifier mainClassifier = new MainClassifier(httpRequestClassifierList);

  if (configArgs.getAdminIpPort().isPresent()) {
    vertx.deployVerticle(new AdminVerticle(configArgs.getAdminIpPort().get()));
  }

  vertx.deployVerticle(() -> new ProxyVerticle(configArgs, mainClassifier,
          configSupplier),
      new DeploymentOptions().setInstances(numCpus * 2 + 1));

  vertx.deployVerticle(() -> new WarmupVerticle(configArgs),
      new DeploymentOptions(),
      warmupVerticle -> {
        vertx.undeploy(warmupVerticle.result());
      });

  vertx.exceptionHandler(ex -> {
    log.error("uncaught exception", ex);
  });
}