io.netty.handler.codec.socks.SocksInitRequestDecoder #io.netty.util.internal.logging.Slf4JLoggerFactory源码实例Demo

@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);
	}
}
 

@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 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);
}
 

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);
}
 

/**
 * 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
public InternalLogger newInstance(String name) {
    InternalLogger baseLogger = ((Slf4JLoggerFactory) Slf4JLoggerFactory.INSTANCE).newInstance(name);
    if (name.equals(ResourceLeakDetector.class.getName())) {
        return new ResourceLeakAssertionLogger(baseLogger);
    } else {
        return baseLogger;
    }
}
 

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);
  });
}
 

private void start() throws InterruptedException, IOException {
    // NioEventLoopGroup是用来处理IO操作的多线程事件循环器
    // boss用来接收进来的连接
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    // 用来处理已经被接收的连接;
    EventLoopGroup workerGroup = new NioEventLoopGroup();

    try {
        InternalLoggerFactory.setDefaultFactory(new Slf4JLoggerFactory());

        // 是一个启动NIO服务的辅助启动类
        sBootstrap = new ServerBootstrap();
        // These EventLoopGroup's are used to handle all the events and IO for ServerChannel
        // and
        // Channel's.
        // 为bootstrap设置acceptor的EventLoopGroup和client的EventLoopGroup
        // 这些EventLoopGroups用于处理所有的IO事件
        // ?这里为什么设置两个group呢?
        sBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                .childHandler(channelInboundHandler).option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);

        // 绑定端口,开始接收进来的连接
        String registHost = registration.getHost();
        future = sBootstrap.bind(registHost, messageServerConfig.getPort()).sync();

        // 获取绑定的端口号
        if (future.channel().localAddress() instanceof InetSocketAddress ) {
            InetSocketAddress socketAddress = (InetSocketAddress)future.channel().localAddress();
            this.priorIP = messageServerConfig.getIp();
            this.ipadress = socketAddress.getAddress().getHostAddress();
            this.port = socketAddress.getPort();
            this.started = true;
            logger.info("NettyChatServer 启动了，address={}:{}", socketAddress.getAddress().getHostAddress(), socketAddress.getPort());
        }
        
        // messageServerCluster
        messageServerCluster.registLocal(this);
        
        // 等待服务器socket关闭
        // 在本例子中不会发生,这时可以关闭服务器了
        future.channel().closeFuture().sync();
    } finally {
        workerGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();

        logger.info("NettyChatServer 关闭了");
    }
}
 

@Before
public void setup() throws Exception {
    InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);
    this.serviceBuilder = ServiceBuilder.newInMemoryBuilder(ServiceBuilderConfig.getDefaultConfig());
    this.serviceBuilder.initialize();
}
 

@Before
public void setup() throws Exception {
    InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);
    this.serviceBuilder = ServiceBuilder.newInMemoryBuilder(ServiceBuilderConfig.getDefaultConfig());
    this.serviceBuilder.initialize();
}
 

@BeforeClass
public static void init() {
    InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);
}
 

HttpServer(String bindAddress, boolean https, Supplier<String> contextPathSupplier,
        int numWorkerThreads, CommonHandler commonHandler, List<File> confDirs, boolean central,
        boolean offlineViewer) throws Exception {

    InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);

    ThreadFactory bossThreadFactory = new ThreadFactoryBuilder()
            .setDaemon(true)
            .setNameFormat("Glowroot-Http-Boss")
            .build();
    ThreadFactory workerThreadFactory = new ThreadFactoryBuilder()
            .setDaemon(true)
            .setNameFormat("Glowroot-Http-Worker-%d")
            .build();
    bossGroup = new NioEventLoopGroup(1, bossThreadFactory);
    workerGroup = new NioEventLoopGroup(numWorkerThreads, workerThreadFactory);

    final HttpServerHandler handler = new HttpServerHandler(contextPathSupplier, commonHandler);

    if (https) {
        // upgrade from 0.9.26 to 0.9.27
        renameHttpsConfFileIfNeeded(confDirs, "certificate.pem", "ui-cert.pem", "certificate");
        renameHttpsConfFileIfNeeded(confDirs, "private.pem", "ui-key.pem", "private key");

        File certificateFile;
        File privateKeyFile;
        if (central) {
            certificateFile = getRequiredHttpsConfFile(confDirs.get(0), "ui-cert.pem",
                    "cert.pem", "certificate");
            privateKeyFile = getRequiredHttpsConfFile(confDirs.get(0), "ui-key.pem", "key.pem",
                    "private key");
        } else {
            certificateFile = getRequiredHttpsConfFile(confDirs, "ui-cert.pem");
            privateKeyFile = getRequiredHttpsConfFile(confDirs, "ui-key.pem");
        }
        sslContext = SslContextBuilder.forServer(certificateFile, privateKeyFile)
                .build();
    }
    this.confDirs = confDirs;
    this.offlineViewer = offlineViewer;

    bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline p = ch.pipeline();
                    SslContext sslContextLocal = sslContext;
                    if (sslContextLocal != null) {
                        p.addLast(sslContextLocal.newHandler(ch.alloc()));
                    }
                    // bumping maxInitialLineLength (first arg below) from default 4096 to 65536
                    // in order to handle long urls on /jvm/gauges and /report/adhoc views
                    // bumping maxHeaderSize (second arg below) from default 8192 to 65536 for
                    // same reason due to "Referer" header once url becomes huge
                    // leaving maxChunkSize (third arg below) at default 8192
                    p.addLast(new HttpServerCodec(65536, 65536, 8192));
                    p.addLast(new HttpObjectAggregator(1048576));
                    p.addLast(new ConditionalHttpContentCompressor());
                    p.addLast(new ChunkedWriteHandler());
                    p.addLast(handler);
                }
            });
    this.handler = handler;
    this.bindAddress = bindAddress;
}