private void start() throws IOException {
    server = NettyServerBuilder.forAddress(new InetSocketAddress(host, port))
            .addService(new GreeterImpl())
            .sslContext(getSslContextBuilder().build())
            .build()
            .start();
    logger.info("Server started, listening on " + port);
    Runtime.getRuntime().addShutdownHook(new Thread() {
        @Override
        public void run() {
            // Use stderr here since the logger may have been reset by its JVM shutdown hook.
            System.err.println("*** shutting down gRPC server since JVM is shutting down");
            HelloWorldServerTls.this.stop();
            System.err.println("*** server shut down");
        }
    });
}
 

@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  // Starts the server with HTTPS.
  try {
    SslProvider sslProvider = SslContext.defaultServerProvider();
    if (sslProvider == SslProvider.OPENSSL && !OpenSsl.isAlpnSupported()) {
      // OkHttp only supports Jetty ALPN on OpenJDK. So if OpenSSL doesn't support ALPN, then we
      // are forced to use Jetty ALPN for Netty instead of OpenSSL.
      sslProvider = SslProvider.JDK;
    }
    SslContextBuilder contextBuilder = SslContextBuilder
        .forServer(TestUtils.loadCert("server1.pem"), TestUtils.loadCert("server1.key"));
    GrpcSslContexts.configure(contextBuilder, sslProvider);
    contextBuilder.ciphers(TestUtils.preferredTestCiphers(), SupportedCipherSuiteFilter.INSTANCE);
    return NettyServerBuilder.forPort(0)
        .flowControlWindow(65 * 1024)
        .maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
        .sslContext(contextBuilder.build());
  } catch (IOException ex) {
    throw new RuntimeException(ex);
  }
}
 

@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  return NettyServerBuilder.forPort(0)
      .maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
      .compressorRegistry(compressors)
      .decompressorRegistry(decompressors)
      .intercept(new ServerInterceptor() {
          @Override
          public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
              Metadata headers, ServerCallHandler<ReqT, RespT> next) {
            Listener<ReqT> listener = next.startCall(call, headers);
            // TODO(carl-mastrangelo): check that encoding was set.
            call.setMessageCompression(true);
            return listener;
          }
        });
}
 

@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  // Starts the server with HTTPS.
  try {
    return NettyServerBuilder.forPort(0)
        .flowControlWindow(65 * 1024)
        .maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
        .sslContext(GrpcSslContexts
            .forServer(TestUtils.loadCert("server1.pem"), TestUtils.loadCert("server1.key"))
            .clientAuth(ClientAuth.REQUIRE)
            .trustManager(TestUtils.loadCert("ca.pem"))
            .ciphers(TestUtils.preferredTestCiphers(), SupportedCipherSuiteFilter.INSTANCE)
            .build());
  } catch (IOException ex) {
    throw new RuntimeException(ex);
  }
}
 

@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  // Starts the server with HTTPS.
  try {
    return NettyServerBuilder.forPort(0)
        .flowControlWindow(65 * 1024)
        .maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
        .sslContext(GrpcSslContexts
            .forServer(TestUtils.loadCert("server1.pem"), TestUtils.loadCert("server1.key"))
            .clientAuth(ClientAuth.REQUIRE)
            .trustManager(TestUtils.loadCert("ca.pem"))
            .ciphers(TestUtils.preferredTestCiphers(), SupportedCipherSuiteFilter.INSTANCE)
            .build());
  } catch (IOException ex) {
    throw new RuntimeException(ex);
  }
}
 

public void start() throws IOException {
    List<ServerInterceptor> allInterceptors = ImmutableList.<ServerInterceptor>builder()
            .add(RequestHeadersInterceptor.recordServerCallInterceptor(serverCallCapture)).build();
    //server = ServerBuilder.forPort(port)
    server = NettyServerBuilder.forPort(port)
            .addService(ServerInterceptors.intercept(new ConsenterServerImpl(), allInterceptors))
            .addService(new GossipService())
            .build()
            .start();
    log.info("consenter service start, port:" + port);

    Runtime.getRuntime().addShutdownHook(new Thread() {

        @Override
        public void run() {

            log.info("*** shutting down gRPC server since JVM is shutting down");
            ConsenterServer.this.stop();
            log.error("***consenter server shut down");
        }
    });
}
 

public void start() throws IOException {
        server = NettyServerBuilder.forPort(port).maxMessageSize(CommConstant.MAX_GRPC_MESSAGE_SIZE)
//        server = ServerBuilder.forPort(port)
                .addService(new SmartContractSupportService())
                .build()
                .start();
        log.info("SmartContractGrpcServer start, port: " + port);

        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                log.info("Shutting down SmartContractGrpcServer since JVM is shutting down");
                SmartContractGrpcServer.this.stop();
                log.info("SmartContractGrpcServer shut down");
            }
        });
    }
 

/**
 * Create a {@link Server} if one isn't already present in the context.
 *
 * @param port               The port this server should listen on
 * @param services           The gRPC services this server should serve
 * @param serverInterceptors The {@link ServerInterceptor} implementations that should be applied to all services
 * @return A Netty server instance based on the provided information
 */
@Bean
@ConditionalOnMissingBean(Server.class)
public Server gRpcServer(
    @Value("${grpc.server.port:0}") final int port,  // TODO: finalize how to get configure this property
    final Set<BindableService> services,
    final List<ServerInterceptor> serverInterceptors
) {
    final NettyServerBuilder builder = NettyServerBuilder.forPort(port);

    // Add Service interceptors and add services to the server
    services
        .stream()
        .map(BindableService::bindService)
        .map(serviceDefinition -> ServerInterceptors.intercept(serviceDefinition, serverInterceptors))
        .forEach(builder::addService);

    return builder.build();
}
 

/**
 * Default constructor.
 * @param environment The environment
 * @param serverHost The server host
 * @param serverPort The server port
 * @param executorService The IO executor service
 */
public GrpcServerConfiguration(
        Environment environment,
        @Property(name = HOST) @Nullable String serverHost,
        @Property(name = PORT) @Nullable Integer serverPort,
        @Named(TaskExecutors.IO) ExecutorService executorService) {
    this.environment = environment;
    this.serverPort = serverPort != null ? serverPort :
            environment.getActiveNames().contains(Environment.TEST) ? SocketUtils.findAvailableTcpPort() : DEFAULT_PORT;
    this.serverHost = serverHost;
    if (serverHost != null) {
        this.serverBuilder = NettyServerBuilder.forAddress(
                new InetSocketAddress(serverHost, this.serverPort)
        );
    } else {
        this.serverBuilder = NettyServerBuilder.forPort(this.serverPort);
    }
    this.serverBuilder.executor(executorService);
}
 

public void start() throws IOException {
    server = NettyServerBuilder.forAddress(new InetSocketAddress(host, port))
    //server = ServerBuilder.forPort(port)
            .addService(new GreeterImpl())
            .build()
            .start();
    LOG.info("Server started, listening on {}:{}", host, port);
    Runtime.getRuntime().addShutdownHook(new Thread() {
        @Override
        public void run() {
            // Use stderr here since the logger may have been reset by its JVM shutdown hook.
            System.err.println("*** shutting down gRPC server since JVM is shutting down");
            this.stop();
            System.err.println("*** server shut down");
        }
    });
}
 

private void start() throws IOException {
    server = NettyServerBuilder.forPort(port)
            .addService(new GreeterImpl())
            .sslContext(getSslContextBuilder().build())
            .build()
            .start();
    logger.info("Server started, listening on " + port);
    Runtime.getRuntime().addShutdownHook(new Thread() {
        @Override
        public void run() {
            // Use stderr here since the logger may have been reset by its JVM shutdown hook.
            System.err.println("*** shutting down gRPC server since JVM is shutting down");
            HelloWorldServerTls.this.stop();
            System.err.println("*** server shut down");
        }
    });
}
 

private void startServer(ServerCalls.UnaryMethod<String, String> method) {
    try {
        server = NettyServerBuilder.forPort(0)
                .addService(ServerInterceptors.intercept(
                        ServerServiceDefinition.builder("service")
                                .addMethod(METHOD_DESCRIPTOR, ServerCalls.asyncUnaryCall(method))
                                .build(),
                        ConcurrencyLimitServerInterceptor.newBuilder(limiter)
                                .build())
                )
                .build()
                .start();

        channel = NettyChannelBuilder.forAddress("localhost", server.getPort())
                .usePlaintext(true)
                .build();
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}
 

/**
 * Before the test has started, create the server and channel.
 */
@Override
protected void before() throws Throwable {
    serviceRegistry = new MutableHandlerRegistry();

    NettyServerBuilder serverBuilder = NettyServerBuilder
            .forPort(0)
            .fallbackHandlerRegistry(serviceRegistry);

    if (useDirectExecutor) {
        serverBuilder.directExecutor();
    }

    configureServerBuilder.accept(serverBuilder);
    server = serverBuilder.build().start();
    port = server.getPort();

    NettyChannelBuilder channelBuilder = NettyChannelBuilder.forAddress("localhost", port).usePlaintext(true);
    configureChannelBuilder.accept(channelBuilder);
    channel = channelBuilder.build();
}
 

@Override
public void initialize(final Config config) throws IOException {
    this.dispatchers = loadAndInitializeDispatchers(config, Thread.currentThread().getContextClassLoader(), getName());

    final int port = config.getInt("port");
    final List<Enricher> enrichers = loadSpanEnrichers(config);

    this.server = NettyServerBuilder
            .forPort(port)
            .directExecutor()
            .permitKeepAliveWithoutCalls(true)
            .permitKeepAliveTime(KEEP_ALIVE_TIME_IN_SECONDS, TimeUnit.SECONDS)
            .addService(new SpanAgentGrpcService(dispatchers, enrichers))
            .addService(new SpanGrpcHealthService())
            .build()
            .start();

    logger.info("span agent grpc server started on port {}....", port);

    try {
        server.awaitTermination();
    } catch (InterruptedException ex) {
        logger.error("span agent server has been interrupted with exception", ex);
    }
}
 

/**
 * Creates and starts a new {@link TestServiceImpl} server.
 */
private Server newServer() throws CertificateException, IOException {
  File serverCertChainFile = TestUtils.loadCert("server1.pem");
  File serverPrivateKeyFile = TestUtils.loadCert("server1.key");
  X509Certificate[] serverTrustedCaCerts = {
    TestUtils.loadX509Cert("ca.pem")
  };

  SslContext sslContext =
      GrpcSslContexts.forServer(serverCertChainFile, serverPrivateKeyFile)
                     .trustManager(serverTrustedCaCerts)
                     .clientAuth(ClientAuth.REQUIRE)
                     .build();

  return NettyServerBuilder.forPort(0)
      .sslContext(sslContext)
      .addService(new TestServiceImpl(serverExecutor))
      .build()
      .start();
}
 

@BeforeEach
public void setUp() throws IOException, URISyntaxException {
    observerQueue = new LinkedBlockingQueue<>();
    executor = Executors.newFixedThreadPool(2);

    serviceRegistry = new MutableHandlerRegistry();
    serviceRegistry.addService(new MaintenanceImplBase() {
        @Override
        public void snapshot(SnapshotRequest request, StreamObserver<SnapshotResponse> observer) {
            try {
                observerQueue.put(observer);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    });

    fakeServer = NettyServerBuilder.forPort(TestUtil.findNextAvailablePort()).fallbackHandlerRegistry(serviceRegistry)
        .directExecutor().build().start();

    client = Client.builder().endpoints(new URI("http://127.0.0.1:" + fakeServer.getPort())).build();
    maintenance = client.getMaintenanceClient();
}
 

/**
 * Starts the RPC server.
 *
 * @throws IOException if a server cannot be successfully initialized
 */
@Override
public void start() throws IOException {
    if (useInProcessServer) {
        final ServerBuilder builder = InProcessServerBuilder.forName(address.toString());
        server = builder.addService(this)
                .executor(grpcExecutor)
                .build()
                .start();
    } else {
        server = NettyServerBuilder.forAddress(
                    new InetSocketAddress(address.getHostname().toStringUtf8(), address.getPort())
                )
                .workerEventLoopGroup(eventLoopGroup)
                .addService(this)
                .executor(grpcExecutor)
                .build()
                .start();
    }

    // Use stderr here since the logger may have been reset by its JVM shutdown hook.
    Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));
}
 

private AsyncFuture<Void> start() throws IOException {
    final Server server = NettyServerBuilder
        .forAddress(address)
        .addService(bindService())
        .intercept(new GrpcOpenCensusInterceptor(this.tracingConfig))
        .maxInboundMessageSize(maxFrameSize)
        .bossEventLoopGroup(bossGroup)
        .workerEventLoopGroup(workerGroup)
        .build();

    return async.call(() -> {
        server.start();
        this.server.set(server);
        return null;
    }).directTransform(v -> {
        final InetSocketAddress localAddress = extractInetSocketAddress(server);
        bindFuture.resolve(localAddress);
        return null;
    });
}
 

private static Server startServer(String bindAddress, int port, boolean https, File confDir,
        @Nullable ExecutorService confDirWatchExecutor, DownstreamServiceImpl downstreamService,
        CollectorServiceImpl collectorService) throws IOException {
    NettyServerBuilder builder =
            NettyServerBuilder.forAddress(new InetSocketAddress(bindAddress, port));
    if (https) {
        builder.sslContext(
                DelegatingSslContext.create(confDir, checkNotNull(confDirWatchExecutor)));
    }
    return builder.addService(collectorService.bindService())
            .addService(downstreamService.bindService())
            // need to override default max message size of 4mb until streaming is implemented
            // for DownstreamService.EntriesResponse and FullTraceResponse
            .maxInboundMessageSize(64 * 1024 * 1024)
            // aggressive keep alive is used by agent to detect silently dropped connections
            // (see org.glowroot.agent.central.CentralConnection)
            .permitKeepAliveTime(20, SECONDS)
            // aggressive max connection age forces agents to re-resolve DNS often for DNS-based
            // load balancing (e.g. to pick up and spread load across new central collectors)
            .maxConnectionAge(20, MINUTES)
            .build()
            .start();
}
 

GrpcServerWrapper(TraceCollector collector, int port) throws IOException {
    bossEventLoopGroup = EventLoopGroups.create("Glowroot-IT-Harness-GRPC-Boss-ELG");
    workerEventLoopGroup = EventLoopGroups.create("Glowroot-IT-Harness-GRPC-Worker-ELG");
    executor = Executors.newCachedThreadPool(
            new ThreadFactoryBuilder()
                    .setDaemon(true)
                    .setNameFormat("Glowroot-IT-Harness-GRPC-Executor-%d")
                    .build());
    downstreamService = new DownstreamServiceImpl();
    server = NettyServerBuilder.forPort(port)
            .bossEventLoopGroup(bossEventLoopGroup)
            .workerEventLoopGroup(workerEventLoopGroup)
            .executor(executor)
            .addService(new CollectorServiceImpl(collector).bindService())
            .addService(downstreamService.bindService())
            .maxInboundMessageSize(1024 * 1024 * 100)
            .build()
            .start();
}
 

public Server startServer() throws IOException {
  ServerInterceptor headersInterceptor = new TracingMetadataUtils.ServerHeadersInterceptor();
  NettyServerBuilder b =
      NettyServerBuilder.forPort(workerOptions.listenPort)
          .addService(ServerInterceptors.intercept(actionCacheServer, headersInterceptor))
          .addService(ServerInterceptors.intercept(bsServer, headersInterceptor))
          .addService(ServerInterceptors.intercept(casServer, headersInterceptor))
          .addService(ServerInterceptors.intercept(capabilitiesServer, headersInterceptor));

  if (workerOptions.tlsCertificate != null) {
    b.sslContext(getSslContextBuilder(workerOptions).build());
  }

  if (execServer != null) {
    b.addService(ServerInterceptors.intercept(execServer, headersInterceptor));
  } else {
    logger.atInfo().log("Execution disabled, only serving cache requests");
  }

  Server server = b.build();
  logger.atInfo().log("Starting gRPC server on port %d", workerOptions.listenPort);
  server.start();

  return server;
}
 

@Activate
    protected void activate() {
        try {
            server = NettyServerBuilder.forPort(MICRO_ONOS_PORT)
// FIXME: make this work later
//                    .useTransportSecurity(new ByteArrayInputStream(MICRO_ONOS_DEFAULT_CERT),
//                                          new ByteArrayInputStream(MICRO_ONOS_DEFAULT_KEY))
                    .addService(new MicroOnosDeviceService())
                    .build()
                    .start();
        } catch (IOException e) {
            log.error("Unable to start gRPC server", e);
            throw new IllegalStateException("Unable to start gRPC server", e);
        }
        log.info("Started");
    }
 

@Override
public Void call() throws Exception {
  OzoneConfiguration ozoneConfiguration = createOzoneConfiguration();
  CsiConfig csiConfig = ozoneConfiguration.getObject(CsiConfig.class);

  OzoneClient rpcClient = OzoneClientFactory.getRpcClient(ozoneConfiguration);

  EpollEventLoopGroup group = new EpollEventLoopGroup();

  if (csiConfig.getVolumeOwner().isEmpty()) {
    throw new IllegalArgumentException(
        "ozone.csi.owner is not set. You should set this configuration "
            + "variable to define which user should own all the created "
            + "buckets.");
  }

  Server server =
      NettyServerBuilder
          .forAddress(new DomainSocketAddress(csiConfig.getSocketPath()))
          .channelType(EpollServerDomainSocketChannel.class)
          .workerEventLoopGroup(group)
          .bossEventLoopGroup(group)
          .addService(new IdentitiyService())
          .addService(new ControllerService(rpcClient,
              csiConfig.getDefaultVolumeSize()))
          .addService(new NodeService(csiConfig))
          .build();

  server.start();
  server.awaitTermination();
  rpcClient.close();
  return null;
}
 

@Override
protected InternalServer newServer(List<ServerStreamTracer.Factory> streamTracerFactories) {
  return AccessProtectedHack.serverBuilderBuildTransportServer(
      NettyServerBuilder
        .forPort(0)
        .flowControlWindow(65 * 1024),
      streamTracerFactories,
      fakeClockTransportTracer);
}
 

@Override
protected InternalServer newServer(
    InternalServer server, List<ServerStreamTracer.Factory> streamTracerFactories) {
  int port = server.getPort();
  return AccessProtectedHack.serverBuilderBuildTransportServer(
      NettyServerBuilder
          .forPort(port)
          .flowControlWindow(65 * 1024),
      streamTracerFactories,
      fakeClockTransportTracer);
}
 

@VisibleForTesting
void start() throws Exception {
  executor = Executors.newSingleThreadScheduledExecutor();
  SslContext sslContext = null;
  if (useAlts) {
    server =
        AltsServerBuilder.forPort(port)
            .addService(
                ServerInterceptors.intercept(
                    new TestServiceImpl(executor), TestServiceImpl.interceptors()))
            .build()
            .start();
  } else {
    if (useTls) {
      sslContext =
          GrpcSslContexts.forServer(
                  TestUtils.loadCert("server1.pem"), TestUtils.loadCert("server1.key"))
              .build();
    }
    server =
        NettyServerBuilder.forPort(port)
            .sslContext(sslContext)
            .maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
            .addService(
                ServerInterceptors.intercept(
                    new TestServiceImpl(executor), TestServiceImpl.interceptors()))
            .build()
            .start();
  }
}
 

@Override
protected List<? extends InternalServer> newServer(
    int port, List<ServerStreamTracer.Factory> streamTracerFactories) {
  return AccessProtectedHack.serverBuilderBuildTransportServer(
      NettyServerBuilder
          .forAddress(new InetSocketAddress(port))
          .flowControlWindow(65 * 1024),
      streamTracerFactories,
      fakeClockTransportTracer);
}
 

private void startServer(int serverFlowControlWindow) {
  ServerBuilder<?> builder =
      NettyServerBuilder.forAddress(new InetSocketAddress("localhost", 0))
      .flowControlWindow(serverFlowControlWindow);
  builder.addService(ServerInterceptors.intercept(
      new TestServiceImpl(Executors.newScheduledThreadPool(2)),
      ImmutableList.<ServerInterceptor>of()));
  try {
    server = builder.build().start();
  } catch (IOException e) {
    throw new RuntimeException(e);
  }
}
 

@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  return NettyServerBuilder
      .forAddress(new LocalAddress("in-process-1"))
      .flowControlWindow(65 * 1024)
      .maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
      .channelType(LocalServerChannel.class);
}
 

private void start() throws IOException {
  int port = 50051;

  Executor executorPool = Executors.newFixedThreadPool(1,
          new DefaultThreadFactory("grpc-server-executor", true));

  EventLoopGroup workerEventLoopGroup = new NioEventLoopGroup(1,
          new DefaultThreadFactory("grpc-worker-group", true));

  //server = ServerBuilder.forPort(port)
  server = NettyServerBuilder.forPort(port)
          .executor(executorPool)// 自定义grpc服务端线程池
          .workerEventLoopGroup(workerEventLoopGroup)// 自定义netty的worker线程池
          .addService(new GreeterImpl())
          .build()
          .start();

  logger.info("Server started, listening on " + port);

  Runtime.getRuntime().addShutdownHook(new Thread() {
    @Override
    public void run() {
      // Use stderr here since the logger may have been reset by its JVM shutdown hook.
      System.err.println("*** shutting down gRPC server since JVM is shutting down");
      SingleThreadServer.this.stop();
      System.err.println("*** server shut down");
    }
  });
}